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

- Back out HBase changes - Code cleanup - Test utilities - Fix for DRILL-5829 closes #968
apache · Oct 11, 2017 · 42f7af2 · 42f7af2
1 parent fe84713
commit 42f7af2
Show file tree

Hide file tree

Showing 23 changed files with 691 additions and 155 deletions.
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
@@ -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.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 @@ private HBaseGroupScan(HBaseGroupScan that) {
   @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 @@ private void init() {
     } 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

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
@@ -126,12 +126,10 @@ protected Collection<SchemaPath> transformColumns(Collection<SchemaPath> columns
             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;
   }
 

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
@@ -93,5 +93,4 @@ public String toString() {
         + ", filter=" + (filter == null ? null : filter.toString())
         + "]";
   }
-
 }
diff --git a/...rib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSchemaFactory.java b/...rib/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 Set<String> getTableNames() {
     public String getTypeName() {
       return HBaseStoragePluginConfig.NAME;
     }
-
   }
-
 }
diff --git a/...rib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java b/...rib/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 boolean equals(Object obj) {
     private HBaseStoragePlugin getHBaseStoragePlugin() {
       return HBaseStoragePlugin.this;
     }
-
   }
-
 }
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
@@ -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.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 String toString() {
           + ", filter=" + (getScanFilter() == null ? null : getScanFilter().toString())
           + ", regionServer=" + regionServer + "]";
     }
-
   }
 
   @Override
   public int getOperatorType() {
     return CoreOperatorType.HBASE_SUB_SCAN_VALUE;
   }
-
 }
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
@@ -38,7 +38,7 @@
 
 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();
 

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
@@ -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 void queryCompleted(QueryState state) {
   }
 
   @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

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
@@ -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.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 Receiver getReceiver(int minorFragmentId) {
   protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
     return new UnionExchange(child);
   }
-
 }
diff --git a/...in/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java b/...in/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 IterOutcome next() {
         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

diff --git a/.../java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java b/.../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.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 UnorderedReceiverBatch getBatch(FragmentContext context, UnorderedReceive
     RawBatchBuffer buffer = buffers[0];
     return new UnorderedReceiverBatch(context, buffer, receiver);
   }
-
-
 }
diff --git a/.../java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java b/.../java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjIntoScan.java
@@ -22,7 +22,6 @@
 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;

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
@@ -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

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
@@ -136,6 +136,21 @@ public boolean equals(Object obj) {
     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