drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From prog...@apache.org
Subject [1/2] drill git commit: DRILL-5830: Resolve regressions to MapR DB from DRILL-5546
Date Wed, 11 Oct 2017 21:29:40 GMT
Repository: drill
Updated Branches:
  refs/heads/master fe8471316 -> fe79a633a


DRILL-5830: Resolve regressions to MapR DB from DRILL-5546

- Back out HBase changes
- Code cleanup
- Test utilities
- Fix for DRILL-5829

closes #968


Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/42f7af22
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/42f7af22
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/42f7af22

Branch: refs/heads/master
Commit: 42f7af22fc5d713aac07e057fd374ccd674e40df
Parents: fe84713
Author: Paul Rogers <progers@maprtech.com>
Authored: Thu Sep 28 09:49:38 2017 -0700
Committer: Paul Rogers <progers@maprtech.com>
Committed: Wed Oct 11 13:14:56 2017 -0700

----------------------------------------------------------------------
 .../drill/exec/store/hbase/HBaseGroupScan.java  |  95 ++---
 .../exec/store/hbase/HBaseRecordReader.java     |   6 +-
 .../drill/exec/store/hbase/HBaseScanSpec.java   |   1 -
 .../exec/store/hbase/HBaseSchemaFactory.java    |   4 +-
 .../exec/store/hbase/HBaseStoragePlugin.java    |   4 +-
 .../drill/exec/store/hbase/HBaseSubScan.java    |   9 +-
 .../org/apache/drill/hbase/BaseHBaseTest.java   |   2 +-
 .../exec/client/PrintingResultsListener.java    |  55 +--
 .../exec/physical/config/UnionExchange.java     |   7 +-
 .../UnorderedReceiverBatch.java                 |   5 +-
 .../UnorderedReceiverCreator.java               |   6 +-
 .../planner/logical/DrillPushProjIntoScan.java  |   1 -
 .../exec/planner/physical/ProjectPrel.java      |   2 +-
 .../apache/drill/exec/record/BatchSchema.java   |  15 +
 .../drill/exec/record/RecordBatchLoader.java    |  63 +++-
 .../drill/exec/record/VectorContainer.java      |   2 +-
 .../org/apache/drill/exec/util/VectorUtil.java  |  27 +-
 .../work/foreman/rm/DistributedQueryQueue.java  |   3 +
 .../drill/exec/record/vector/TestLoad.java      | 358 +++++++++++++++++--
 .../org/apache/drill/test/QueryBuilder.java     |   4 +
 .../apache/drill/test/QueryRowSetIterator.java  | 118 ++++++
 .../src/main/java/io/netty/buffer/DrillBuf.java |   6 +-
 .../drill/exec/record/MaterializedField.java    |  53 +++
 23 files changed, 691 insertions(+), 155 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
index 1ee1da8..2b8cf18 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -17,17 +17,22 @@
  */
 package org.apache.drill.exec.store.hbase;
 
-import com.fasterxml.jackson.annotation.JacksonInject;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.common.expression.SchemaPath;
@@ -50,24 +55,18 @@ import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.RegionLocator;
-import org.apache.hadoop.hbase.util.Bytes;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.NavigableMap;
-import java.util.PriorityQueue;
-import java.util.Queue;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 
 @JsonTypeName("hbase-scan")
 public class HBaseGroupScan extends AbstractGroupScan implements DrillHBaseConstants {
@@ -144,8 +143,8 @@ public class HBaseGroupScan extends AbstractGroupScan implements DrillHBaseConst
   @Override
   public GroupScan clone(List<SchemaPath> columns) {
     HBaseGroupScan newScan = new HBaseGroupScan(this);
-    newScan.columns = columns == null ? ALL_COLUMNS : columns;;
-    newScan.verifyColumnsAndConvertStar();
+    newScan.columns = columns == null ? ALL_COLUMNS : columns;
+    newScan.verifyColumns();
     return newScan;
   }
 
@@ -177,37 +176,19 @@ public class HBaseGroupScan extends AbstractGroupScan implements DrillHBaseConst
     } catch (IOException e) {
       throw new DrillRuntimeException("Error getting region info for table: " + hbaseScanSpec.getTableName(), e);
     }
-    verifyColumnsAndConvertStar();
+    verifyColumns();
   }
 
-  private void verifyColumnsAndConvertStar() {
-    boolean hasStarCol = false;
-    LinkedHashSet<SchemaPath> requestedColumns = new LinkedHashSet<>();
-
+  private void verifyColumns() {
+    if (Utilities.isStarQuery(columns)) {
+      return;
+    }
     for (SchemaPath column : columns) {
-      // convert * into [row_key, cf1, cf2, ..., cf_n].
-      if (column.equals(Utilities.STAR_COLUMN)) {
-        hasStarCol = true;
-        Set<byte[]> families = hTableDesc.getFamiliesKeys();
-        requestedColumns.add(ROW_KEY_PATH);
-        for (byte[] family : families) {
-          SchemaPath colFamily = SchemaPath.getSimplePath(Bytes.toString(family));
-          requestedColumns.add(colFamily);
-        }
-      } else {
-        if (!(column.equals(ROW_KEY_PATH) ||
-            hTableDesc.hasFamily(HBaseUtils.getBytes(column.getRootSegment().getPath())))) {
-          DrillRuntimeException.format("The column family '%s' does not exist in HBase table: %s .",
-              column.getRootSegment().getPath(), hTableDesc.getNameAsString());
-        }
-        requestedColumns.add(column);
+      if (!(column.equals(ROW_KEY_PATH) || hTableDesc.hasFamily(HBaseUtils.getBytes(column.getRootSegment().getPath())))) {
+        DrillRuntimeException.format("The column family '%s' does not exist in HBase table: %s .",
+            column.getRootSegment().getPath(), hTableDesc.getNameAsString());
       }
     }
-
-    // since star column has been converted, reset this.cloumns.
-    if (hasStarCol) {
-      this.columns = new ArrayList<>(requestedColumns);
-    }
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java
index d6c02b5..cae7ce4 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java
@@ -126,12 +126,10 @@ public class HBaseRecordReader extends AbstractRecordReader implements DrillHBas
             HBaseUtils.andFilterAtIndex(hbaseScan.getFilter(), HBaseUtils.LAST_FILTER, new FirstKeyOnlyFilter()));
       }
     } else {
-      throw new IllegalArgumentException("HBaseRecordReader does not allow column *. Column * should have been converted to list of <row_key, column family1, column family2, ..., column family_n");
-//      rowKeyOnly = false;
-//      transformed.add(ROW_KEY_PATH);
+      rowKeyOnly = false;
+      transformed.add(ROW_KEY_PATH);
     }
 
-
     return transformed;
   }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseScanSpec.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseScanSpec.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseScanSpec.java
index f9a585e..797ec7f 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseScanSpec.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseScanSpec.java
@@ -93,5 +93,4 @@ public class HBaseScanSpec {
         + ", filter=" + (filter == null ? null : filter.toString())
         + "]";
   }
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java
index 56dfc10..548b679 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -94,7 +94,5 @@ public class HBaseSchemaFactory implements SchemaFactory {
     public String getTypeName() {
       return HBaseStoragePluginConfig.NAME;
     }
-
   }
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java
index 81899cf..62f351c 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -154,7 +154,5 @@ public class HBaseStoragePlugin extends AbstractStoragePlugin {
     private HBaseStoragePlugin getHBaseStoragePlugin() {
       return HBaseStoragePlugin.this;
     }
-
   }
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java
index 302ccca..0527391 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -41,7 +41,10 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.base.Preconditions;
 
-// Class containing information for reading a single HBase region
+/**
+ * Contains information for reading a single HBase region
+ */
+
 @JsonTypeName("hbase-region-scan")
 public class HBaseSubScan extends AbstractBase implements SubScan {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HBaseSubScan.class);
@@ -210,12 +213,10 @@ public class HBaseSubScan extends AbstractBase implements SubScan {
           + ", filter=" + (getScanFilter() == null ? null : getScanFilter().toString())
           + ", regionServer=" + regionServer + "]";
     }
-
   }
 
   @Override
   public int getOperatorType() {
     return CoreOperatorType.HBASE_SUB_SCAN_VALUE;
   }
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/BaseHBaseTest.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/BaseHBaseTest.java b/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/BaseHBaseTest.java
index e12c77c..b957347 100644
--- a/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/BaseHBaseTest.java
+++ b/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/BaseHBaseTest.java
@@ -38,7 +38,7 @@ import com.google.common.io.Files;
 
 public class BaseHBaseTest extends BaseTestQuery {
 
-  private static final String HBASE_STORAGE_PLUGIN_NAME = "hbase";
+  public static final String HBASE_STORAGE_PLUGIN_NAME = "hbase";
 
   protected static Configuration conf = HBaseConfiguration.create();
 

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
index bdd2fab..c233837 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -71,35 +71,44 @@ public class PrintingResultsListener implements UserResultsListener {
   }
 
   @Override
+  @SuppressWarnings("resource")
   public void dataArrived(QueryDataBatch result, ConnectionThrottle throttle) {
     final QueryData header = result.getHeader();
     final DrillBuf data = result.getData();
 
-    if (data != null) {
-      count.addAndGet(header.getRowCount());
-      try {
-        loader.load(header.getDef(), data);
-        // TODO:  Clean:  DRILL-2933:  That load(...) no longer throws
-        // SchemaChangeException, so check/clean catch clause below.
-      } catch (SchemaChangeException e) {
-        submissionFailed(UserException.systemError(e).build(logger));
-      }
+    try {
+      if (data != null) {
+        count.addAndGet(header.getRowCount());
+        try {
+          loader.load(header.getDef(), data);
+          // TODO:  Clean:  DRILL-2933:  That load(...) no longer throws
+          // SchemaChangeException, so check/clean catch clause below.
+        } catch (SchemaChangeException e) {
+          submissionFailed(UserException.systemError(e).build(logger));
+        }
 
-      switch(format) {
-        case TABLE:
-          VectorUtil.showVectorAccessibleContent(loader, columnWidth);
-          break;
-        case TSV:
-          VectorUtil.showVectorAccessibleContent(loader, "\t");
-          break;
-        case CSV:
-          VectorUtil.showVectorAccessibleContent(loader, ",");
-          break;
+        try {
+          switch(format) {
+            case TABLE:
+              VectorUtil.showVectorAccessibleContent(loader, columnWidth);
+              break;
+            case TSV:
+              VectorUtil.showVectorAccessibleContent(loader, "\t");
+              break;
+            case CSV:
+              VectorUtil.showVectorAccessibleContent(loader, ",");
+              break;
+            default:
+              throw new IllegalStateException(format.toString());
+          }
+        } finally {
+          loader.clear();
+        }
       }
-      loader.clear();
     }
-
-    result.release();
+    finally {
+      result.release();
+    }
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
index 318e6b1..c825fcb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -33,9 +33,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 
 @JsonTypeName("union-exchange")
-public class UnionExchange extends AbstractExchange{
-
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnionExchange.class);
+public class UnionExchange extends AbstractExchange {
 
   public UnionExchange(@JsonProperty("child") PhysicalOperator child) {
     super(child);
@@ -76,5 +74,4 @@ public class UnionExchange extends AbstractExchange{
   protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
     return new UnionExchange(child);
   }
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
index be7d4ed..cfdc06d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -181,9 +181,6 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
         return IterOutcome.OUT_OF_MEMORY;
       }
 
-
-//      logger.debug("Next received batch {}", batch);
-
       final RecordBatchDef rbd = batch.getHeader().getDef();
       final boolean schemaChanged = batchLoader.load(rbd, batch.getBody());
       // TODO:  Clean:  DRILL-2933:  That load(...) no longer throws

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
index 649ecd9..6d4f1d7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -28,8 +28,8 @@ import org.apache.drill.exec.work.batch.IncomingBuffers;
 import org.apache.drill.exec.work.batch.RawBatchBuffer;
 
 public class UnorderedReceiverCreator implements BatchCreator<UnorderedReceiver>{
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnorderedReceiverCreator.class);
 
+  @SuppressWarnings("resource")
   @Override
   public UnorderedReceiverBatch getBatch(FragmentContext context, UnorderedReceiver receiver, List<RecordBatch> children)
       throws ExecutionSetupException {
@@ -42,6 +42,4 @@ public class UnorderedReceiverCreator implements BatchCreator<UnorderedReceiver>
     RawBatchBuffer buffer = buffers[0];
     return new UnorderedReceiverBatch(context, buffer, receiver);
   }
-
-
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java
index 33c840b..b15a843 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java
@@ -22,7 +22,6 @@ import java.io.IOException;
 import java.util.List;
 
 import org.apache.calcite.adapter.enumerable.EnumerableTableScan;
-import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rel.core.TableScan;
 import org.apache.calcite.rel.logical.LogicalProject;

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
index d974bad..7f634c3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
index 63dcdb45..564aaed 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
@@ -136,6 +136,21 @@ public class BatchSchema implements Iterable<MaterializedField> {
     return true;
   }
 
+  public boolean isEquivalent(BatchSchema other) {
+    if (fields == null || other.fields == null) {
+      return fields == other.fields;
+    }
+    if (fields.size() != other.fields.size()) {
+      return false;
+    }
+    for (int i = 0; i < fields.size(); i++) {
+      if (! fields.get(i).isEquivalent(other.fields.get(i))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   /**
    * We treat fields with same set of Subtypes as equal, even if they are in a different order
    * @param t1

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
index 9a8483b..20b5cb5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
@@ -19,12 +19,15 @@ package org.apache.drill.exec.record;
 
 import io.netty.buffer.DrillBuf;
 
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.drill.common.StackTrace;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.map.CaseInsensitiveMap;
+import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.expr.TypeHelper;
 import org.apache.drill.exec.memory.BufferAllocator;
@@ -38,7 +41,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Maps;
 
 
 /**
@@ -85,7 +87,7 @@ public class RecordBatchLoader implements VectorAccessible, Iterable<VectorWrapp
     // the schema has changed since the previous call.
 
     // Set up to recognize previous fields that no longer exist.
-    final Map<String, ValueVector> oldFields = Maps.newHashMap();
+    final Map<String, ValueVector> oldFields = CaseInsensitiveMap.newHashMap();
     for(final VectorWrapper<?> wrapper : container) {
       final ValueVector vector = wrapper.getValueVector();
       oldFields.put(vector.getField().getName(), vector);
@@ -109,6 +111,17 @@ public class RecordBatchLoader implements VectorAccessible, Iterable<VectorWrapp
           vector.clear();
           schemaChanged = true;
           vector = TypeHelper.getNewVector(fieldDef, allocator);
+
+        // If the field is a map, check if the map schema changed.
+
+        } else if (vector.getField().getType().getMinorType() == MinorType.MAP  &&
+                   ! isSameSchema(vector.getField().getChildren(), field.getChildList())) {
+
+          // The map schema changed. Discard the old map and create a new one.
+
+          schemaChanged = true;
+          vector.clear();
+          vector = TypeHelper.getNewVector(fieldDef, allocator);
         }
 
         // Load the vector.
@@ -121,7 +134,6 @@ public class RecordBatchLoader implements VectorAccessible, Iterable<VectorWrapp
         newVectors.add(vector);
       }
 
-
       // rebuild the schema.
       final SchemaBuilder builder = BatchSchema.newBuilder();
       for (final VectorWrapper<?> v : newVectors) {
@@ -150,6 +162,51 @@ public class RecordBatchLoader implements VectorAccessible, Iterable<VectorWrapp
     return schemaChanged;
   }
 
+  /**
+   * Check if two schemas are the same. The schemas, given as lists, represent the
+   * children of the original and new maps (AKA structures.)
+   *
+   * @param currentChildren current children of a Drill map
+   * @param newChildren new children, in an incoming batch, of the same
+   * Drill map
+   * @return true if the schemas are identical, false if a child is missing
+   * or has changed type or cardinality (AKA "mode").
+   */
+
+  private boolean isSameSchema(Collection<MaterializedField> currentChildren,
+      List<SerializedField> newChildren) {
+    if (currentChildren.size() != newChildren.size()) {
+      return false;
+    }
+
+    // Column order can permute (see DRILL-5828). So, use a map
+    // for matching.
+
+    Map<String, MaterializedField> childMap = CaseInsensitiveMap.newHashMap();
+    for (MaterializedField currentChild : currentChildren) {
+      childMap.put(currentChild.getName(), currentChild);
+    }
+    for (SerializedField newChild : newChildren) {
+      MaterializedField currentChild = childMap.get(newChild.getNamePart().getName());
+
+      // New map member?
+
+      if (currentChild == null) {
+        return false;
+      }
+
+      // Changed data type?
+
+      if (! currentChild.getType().equals(newChild.getMajorType())) {
+        return false;
+      }
+    }
+
+    // Everything matches.
+
+    return true;
+  }
+
   @Override
   public TypedFieldId getValueVectorId(SchemaPath path) {
     return container.getValueVectorId(path);

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
index 63cac7d..abcb846 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
@@ -134,7 +134,7 @@ public class VectorContainer implements VectorAccessible {
     return addOrGet(field, null);
   }
 
-  @SuppressWarnings({ "resource", "unchecked" })
+  @SuppressWarnings("unchecked")
   public <T extends ValueVector> T addOrGet(final MaterializedField field, final SchemaChangeCallBack callBack) {
     final TypedFieldId id = getValueVectorId(SchemaPath.getSimplePath(field.getName()));
     final ValueVector vector;

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
index 4de4c2a..018653b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.util;
 import java.util.List;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.util.DrillStringUtils;
 import org.apache.drill.exec.record.MaterializedField;
 import org.apache.drill.exec.record.VectorAccessible;
@@ -42,7 +43,7 @@ public class VectorUtil {
     System.out.println(rows + " row(s):");
     List<String> columns = Lists.newArrayList();
     for (VectorWrapper<?> vw : va) {
-      columns.add(vw.getValueVector().getField().getName());
+      columns.add(formatFieldSchema(vw.getValueVector().getField()));
     }
 
     int width = columns.size();
@@ -80,6 +81,14 @@ public class VectorUtil {
     }
   }
 
+  public static String formatFieldSchema(MaterializedField field) {
+    String colName = field.getName() + "<" + field.getType().getMinorType() + "(" + field.getType().getMode() + ")" + ">";
+    if (field.getType().getMinorType() == MinorType.MAP) {
+      colName += expandMapSchema(field);
+    }
+    return colName;
+  }
+
   public static void appendVectorAccessibleContent(VectorAccessible va, StringBuilder formattedResults,
       final String delimiter, boolean includeHeader) {
     if (includeHeader) {
@@ -135,9 +144,8 @@ public class VectorUtil {
       int columnWidth = getColumnWidth(columnWidths, columnIndex);
       width += columnWidth + 2;
       formats.add("| %-" + columnWidth + "s");
-      MaterializedField field = vw.getValueVector().getField();
-      columns.add(field.getName() + "<" + field.getType().getMinorType() + "(" + field.getType().getMode() + ")" + ">");
       columnIndex++;
+      columns.add(formatFieldSchema(vw.getValueVector().getField()));
     }
 
     int rows = va.getRecordCount();
@@ -180,6 +188,19 @@ public class VectorUtil {
     }
   }
 
+  private static String expandMapSchema(MaterializedField mapField) {
+    StringBuilder buf = new StringBuilder();
+    buf.append("{");
+    String sep = "";
+    for (MaterializedField field : mapField.getChildren()) {
+      buf.append(sep);
+      sep = ",";
+      buf.append(formatFieldSchema(field));
+    }
+    buf.append("}");
+    return buf.toString();
+  }
+
   public static void allocateVectors(Iterable<ValueVector> valueVectors, int count) {
     for (final ValueVector v : valueVectors) {
       AllocationHelper.allocateNew(v, count);

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/rm/DistributedQueryQueue.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/rm/DistributedQueryQueue.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/rm/DistributedQueryQueue.java
index f4c5536..73e8de6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/rm/DistributedQueryQueue.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/rm/DistributedQueryQueue.java
@@ -183,6 +183,9 @@ public class DistributedQueryQueue implements QueryQueue {
      */
 
     public boolean isSameAs(ConfigSet otherSet) {
+      if (otherSet == null) {
+        return false;
+      }
       return queueThreshold == otherSet.queueThreshold &&
              queueTimeout == otherSet.queueTimeout &&
              largeQueueSize == otherSet.largeQueueSize &&

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
index 4febe1d..fed2914 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
@@ -18,69 +18,72 @@
 package org.apache.drill.exec.record.vector;
 
 import static org.junit.Assert.assertEquals;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.DrillBuf;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.drill.categories.VectorTest;
 import org.apache.drill.common.config.DrillConfig;
 import org.apache.drill.common.types.TypeProtos;
 import org.apache.drill.common.types.TypeProtos.MinorType;
-import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.ExecTest;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.TypeHelper;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.memory.RootAllocatorFactory;
+import org.apache.drill.exec.record.BatchSchema;
 import org.apache.drill.exec.record.MaterializedField;
 import org.apache.drill.exec.record.RecordBatchLoader;
 import org.apache.drill.exec.record.VectorWrapper;
 import org.apache.drill.exec.record.WritableBatch;
 import org.apache.drill.exec.vector.AllocationHelper;
-import org.apache.drill.exec.vector.IntVector;
-import org.apache.drill.exec.vector.NullableVarCharVector;
 import org.apache.drill.exec.vector.ValueVector;
-import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.test.rowSet.SchemaBuilder;
 import org.junit.Test;
-
-import com.google.common.collect.Lists;
 import org.junit.experimental.categories.Category;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
 @Category(VectorTest.class)
 public class TestLoad extends ExecTest {
   private final DrillConfig drillConfig = DrillConfig.create();
 
+  @SuppressWarnings("resource")
   @Test
   public void testLoadValueVector() throws Exception {
     final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
-    final ValueVector fixedV = new IntVector(MaterializedField.create("ints",
-        Types.required(MinorType.INT)), allocator);
-    final ValueVector varlenV = new VarCharVector(MaterializedField.create(
-        "chars", Types.required(MinorType.VARCHAR)), allocator);
-    final ValueVector nullableVarlenV = new NullableVarCharVector(MaterializedField.create("chars",
-        Types.optional(MinorType.VARCHAR)), allocator);
-
-    final List<ValueVector> vectors = Lists.newArrayList(fixedV, varlenV, nullableVarlenV);
-    for (final ValueVector v : vectors) {
-      AllocationHelper.allocate(v, 100, 50);
-      v.getMutator().generateTestData(100);
-    }
+    BatchSchema schema = new SchemaBuilder()
+        .add("ints", MinorType.INT)
+        .add("chars", MinorType.VARCHAR)
+        .addNullable("chars2", MinorType.VARCHAR)
+        .build();
 
-    final WritableBatch writableBatch = WritableBatch.getBatchNoHV(100, vectors, false);
-    final RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
-    final ByteBuf[] byteBufs = writableBatch.getBuffers();
-    int bytes = 0;
-    for (ByteBuf buf : byteBufs) {
-      bytes += buf.writerIndex();
-    }
-    final DrillBuf byteBuf = allocator.buffer(bytes);
-    int index = 0;
-    for (ByteBuf buf : byteBufs) {
-      buf.readBytes(byteBuf, index, buf.writerIndex());
-      index += buf.writerIndex();
-    }
-    byteBuf.writerIndex(bytes);
+    // Create vectors
+
+    final List<ValueVector> vectors = createVectors(allocator, schema, 100);
+
+    // Writeable batch now owns vector buffers
 
+     final WritableBatch writableBatch = WritableBatch.getBatchNoHV(100, vectors, false);
+
+     // Serialize the vectors
+
+    final DrillBuf byteBuf = serializeBatch(allocator, writableBatch);
+
+    // Batch loader does NOT take ownership of the serialized buffer
+
+    final RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
     batchLoader.load(writableBatch.getDef(), byteBuf);
+
+    // Release the serialized buffer.
+
+    byteBuf.release();
+
+    // TODO: Replace this with actual validation, not just dumping to the console.
+
     boolean firstColumn = true;
     int recordCount = 0;
     for (final VectorWrapper<?> v : batchLoader) {
@@ -122,7 +125,294 @@ public class TestLoad extends ExecTest {
       }
     }
     assertEquals(100, recordCount);
+
+    // Free the original vectors
+
+    writableBatch.clear();
+
+    // Free the deserialized vectors
+
     batchLoader.clear();
+
+    // The allocator will verify that the frees were done correctly.
+
+    allocator.close();
+  }
+
+  // TODO: Replace this low-level code with RowSet usage once
+  // DRILL-5657 is committed to master.
+
+  private static List<ValueVector> createVectors(BufferAllocator allocator, BatchSchema schema, int i) {
+    final List<ValueVector> vectors = new ArrayList<>();
+    for (MaterializedField field : schema) {
+      @SuppressWarnings("resource")
+      ValueVector v = TypeHelper.getNewVector(field, allocator);
+      AllocationHelper.allocate(v, 100, 50);
+      v.getMutator().generateTestData(100);
+      vectors.add(v);
+    }
+    return vectors;
+  }
+
+  private static DrillBuf serializeBatch(BufferAllocator allocator, WritableBatch writableBatch) {
+    final ByteBuf[] byteBufs = writableBatch.getBuffers();
+    int bytes = 0;
+    for (ByteBuf buf : byteBufs) {
+      bytes += buf.writerIndex();
+    }
+    final DrillBuf byteBuf = allocator.buffer(bytes);
+    int index = 0;
+    for (ByteBuf buf : byteBufs) {
+      buf.readBytes(byteBuf, index, buf.writerIndex());
+      index += buf.writerIndex();
+    }
+    byteBuf.writerIndex(bytes);
+    return byteBuf;
+  }
+
+  /**
+   * Test function to simulate loading a batch.
+   *
+   * @param allocator a memory allocator
+   * @param batchLoader the batch loader under test
+   * @param schema the schema of the new batch
+   * @return false if the same schema, true if schema changed;
+   * that is, whether the schema changed
+   * @throws SchemaChangeException should not occur
+   */
+
+  @SuppressWarnings("resource")
+  private boolean loadBatch(BufferAllocator allocator,
+      final RecordBatchLoader batchLoader,
+      BatchSchema schema) throws SchemaChangeException {
+    final List<ValueVector> vectors = createVectors(allocator, schema, 100);
+    final WritableBatch writableBatch = WritableBatch.getBatchNoHV(100, vectors, false);
+    final DrillBuf byteBuf = serializeBatch(allocator, writableBatch);
+    boolean result = batchLoader.load(writableBatch.getDef(), byteBuf);
+    byteBuf.release();
     writableBatch.clear();
+    return result;
+  }
+
+  @Test
+  public void testSchemaChange() throws SchemaChangeException {
+    final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
+    final RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
+
+    // Initial schema: a: INT, b: VARCHAR
+    // Schema change: N/A
+
+    BatchSchema schema1 = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .add("b", MinorType.VARCHAR)
+        .build();
+    {
+      assertTrue(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Same schema
+    // Schema change: No
+
+    {
+      assertFalse(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Reverse columns: b: VARCHAR, a: INT
+    // Schema change: No
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("b", MinorType.VARCHAR)
+          .add("a", MinorType.INT)
+          .build();
+      assertFalse(loadBatch(allocator, batchLoader, schema));
+
+      // Potential bug: see DRILL-5828
+
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Drop a column: a: INT
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Add a column: a: INT, b: VARCHAR, c: INT
+    // Schema change: Yes
+
+    {
+      assertTrue(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .add("b", MinorType.VARCHAR)
+          .add("c", MinorType.INT)
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Change a column type: a: INT, b: VARCHAR, c: VARCHAR
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .add("b", MinorType.VARCHAR)
+          .add("c", MinorType.VARCHAR)
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Empty schema
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    batchLoader.clear();
+    allocator.close();
+  }
+
+  @Test
+  public void testMapSchemaChange() throws SchemaChangeException {
+    final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
+    final RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
+
+    // Initial schema: a: INT, m: MAP{}
+
+    BatchSchema schema1 = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .addMap("m")
+          .buildMap()
+        .build();
+    {
+      assertTrue(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Same schema
+    // Schema change: No
+
+    {
+      assertFalse(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Add column to map: a: INT, m: MAP{b: VARCHAR}
+    // Schema change: Yes
+
+    BatchSchema schema2 = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .addMap("m")
+          .add("b", MinorType.VARCHAR)
+          .buildMap()
+        .build();
+    {
+      assertTrue(loadBatch(allocator, batchLoader, schema2));
+      assertTrue(schema2.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Same schema
+    // Schema change: No
+
+    {
+      assertFalse(loadBatch(allocator, batchLoader, schema2));
+      assertTrue(schema2.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Add column:  a: INT, m: MAP{b: VARCHAR, c: INT}
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .addMap("m")
+            .add("b", MinorType.VARCHAR)
+            .add("c", MinorType.INT)
+            .buildMap()
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Drop a column:  a: INT, m: MAP{b: VARCHAR}
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .addMap("m")
+            .add("b", MinorType.VARCHAR)
+            .buildMap()
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Change type:  a: INT, m: MAP{b: INT}
+    // Schema change: Yes
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .addMap("m")
+            .add("b", MinorType.INT)
+            .buildMap()
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Empty map: a: INT, m: MAP{}
+
+    {
+      assertTrue(loadBatch(allocator, batchLoader, schema1));
+      assertTrue(schema1.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    // Drop map: a: INT
+
+    {
+      BatchSchema schema = new SchemaBuilder()
+          .add("a", MinorType.INT)
+          .build();
+      assertTrue(loadBatch(allocator, batchLoader, schema));
+      assertTrue(schema.isEquivalent(batchLoader.getSchema()));
+      batchLoader.getContainer().zeroVectors();
+    }
+
+    batchLoader.clear();
+    allocator.close();
   }
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
index 37fcdfd..e7bf61f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
@@ -345,6 +345,10 @@ public class QueryBuilder {
     }
   }
 
+  public QueryRowSetIterator rowSetIterator( ) {
+    return new QueryRowSetIterator(client.allocator(), withEventListener());
+  }
+
   /**
    * Run the query that is expected to return (at least) one row
    * with the only (or first) column returning a long value.

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/java-exec/src/test/java/org/apache/drill/test/QueryRowSetIterator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/QueryRowSetIterator.java b/exec/java-exec/src/test/java/org/apache/drill/test/QueryRowSetIterator.java
new file mode 100644
index 0000000..c329690
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/QueryRowSetIterator.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.test;
+
+import java.util.Iterator;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+import org.apache.drill.test.BufferingQueryEventListener.QueryEvent;
+import org.apache.drill.test.rowSet.DirectRowSet;
+
+public class QueryRowSetIterator implements Iterator<DirectRowSet>, Iterable<DirectRowSet> {
+  private final BufferingQueryEventListener listener;
+  private int recordCount = 0;
+  private int batchCount = 0;
+  QueryId queryId = null;
+  private BufferAllocator allocator;
+  private QueryDataBatch batch;
+  private QueryState state;
+
+  QueryRowSetIterator(BufferAllocator allocator, BufferingQueryEventListener listener) {
+    this.allocator = allocator;
+    this.listener = listener;
+  }
+
+  public QueryId queryId() { return queryId; }
+  public String queryIdString() { return QueryIdHelper.getQueryId(queryId); }
+  public QueryState finalState() { return state; }
+  public int batchCount() { return batchCount; }
+  public int rowCount() { return recordCount; }
+
+  @Override
+  public boolean hasNext() {
+    for ( ; ; ) {
+      QueryEvent event = listener.get();
+      state = event.state;
+      batch = null;
+      switch (event.type)
+      {
+      case BATCH:
+        batchCount++;
+        recordCount += event.batch.getHeader().getRowCount();
+        batch = event.batch;
+        return true;
+      case EOF:
+        state = event.state;
+        return false;
+      case ERROR:
+        throw new RuntimeException(event.error);
+      case QUERY_ID:
+        queryId = event.queryId;
+        break;
+      default:
+        throw new IllegalStateException("Unexpected event: " + event.type);
+      }
+    }
+  }
+
+  @Override
+  public DirectRowSet next() {
+
+    if (batch == null) {
+      throw new IllegalStateException();
+    }
+
+    // Unload the batch and convert to a row set.
+
+    final RecordBatchLoader loader = new RecordBatchLoader(allocator);
+    try {
+      loader.load(batch.getHeader().getDef(), batch.getData());
+      batch.release();
+      batch = null;
+      VectorContainer container = loader.getContainer();
+      container.setRecordCount(loader.getRecordCount());
+      return new DirectRowSet(allocator, container);
+    } catch (SchemaChangeException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  public void printAll() {
+    for (DirectRowSet rowSet : this) {
+      rowSet.print();
+      rowSet.clear();
+    }
+  }
+
+  @Override
+  public Iterator<DirectRowSet> iterator() {
+    return this;
+  }
+
+  @Override
+  public void remove() {
+    throw new UnsupportedOperationException();
+  }
+}

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
----------------------------------------------------------------------
diff --git a/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java b/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
index a9feafd..5029c56 100644
--- a/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
+++ b/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -52,7 +52,7 @@ public final class DrillBuf extends AbstractByteBuf implements AutoCloseable {
   private final int offset;
   private final BufferLedger ledger;
   private final BufferManager bufManager;
-  private final ByteBufAllocator alloc;
+//  private final ByteBufAllocator alloc;
   private final boolean isEmpty;
   private volatile int length;
   private final HistoricalLog historicalLog = BaseAllocator.DEBUG ?
@@ -72,7 +72,7 @@ public final class DrillBuf extends AbstractByteBuf implements AutoCloseable {
     this.udle = byteBuf;
     this.isEmpty = isEmpty;
     this.bufManager = manager;
-    this.alloc = alloc;
+//    this.alloc = alloc;
     this.addr = byteBuf.memoryAddress() + offset;
     this.ledger = ledger;
     this.length = length;

http://git-wip-us.apache.org/repos/asf/drill/blob/42f7af22/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java b/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
index e2b44a7..4d29d55 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
@@ -25,6 +25,7 @@ import java.util.Objects;
 
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.expr.BasicTypeHelper;
 import org.apache.drill.exec.proto.UserBitShared.NamePart;
 import org.apache.drill.exec.proto.UserBitShared.SerializedField;
@@ -86,6 +87,7 @@ public class MaterializedField {
     children.add(field);
   }
 
+  @Override
   public MaterializedField clone() {
     return withPathAndType(name, getType());
   }
@@ -168,6 +170,57 @@ public class MaterializedField {
             Objects.equals(this.type, other.type);
   }
 
+  public boolean isEquivalent(MaterializedField other) {
+    if (! name.equalsIgnoreCase(other.name)) {
+      return false;
+    }
+
+    // Requires full type equality, including fields such as precision and scale.
+    // But, unset fields are equivalent to 0. Can't use the protobuf-provided
+    // isEquals(), that treats set and unset fields as different.
+
+    if (type.getMinorType() != other.type.getMinorType()) {
+      return false;
+    }
+    if (type.getMode() != other.type.getMode()) {
+      return false;
+    }
+    if (type.getScale() != other.type.getScale()) {
+      return false;
+    }
+    if (type.getPrecision() != other.type.getPrecision()) {
+      return false;
+    }
+
+    // Compare children -- but only for maps, not the internal children
+    // for Varchar, repeated or nullable types.
+
+    if (type.getMinorType() != MinorType.MAP) {
+      return true;
+    }
+
+    if (children == null  ||  other.children == null) {
+      return children == other.children;
+    }
+    if (children.size() != other.children.size()) {
+      return false;
+    }
+
+    // Maps are name-based, not position. But, for our
+    // purposes, we insist on identical ordering.
+
+    Iterator<MaterializedField> thisIter = children.iterator();
+    Iterator<MaterializedField> otherIter = other.children.iterator();
+    while (thisIter.hasNext()) {
+      MaterializedField thisChild = thisIter.next();
+      MaterializedField otherChild = otherIter.next();
+      if (! thisChild.isEquivalent(otherChild)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   /**
    * <p>Creates materialized field string representation.
    * Includes field name, its type with precision and scale if any and data mode.


Mime
View raw message