DRILL-6130: Fix NPE during physical plan submission for various storage plugins

1. Fixed ser / de issues for Hive, Kafka, Hbase plugins.
2. Added physical plan submission unit test for all storage plugins in contrib module.
3. Refactoring.

closes #1108

3 files changed, 26 insertions, 23 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
index 69c2725dd..11782987a 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
@@ -160,7 +160,7 @@ public class HBaseGroupScan extends AbstractGroupScan implements DrillHBaseConst
       statsCalculator = new TableStatsCalculator(conn, hbaseScanSpec, storagePlugin.getContext().getConfig(), storagePluginConfig);
 
       boolean foundStartRegion = false;
-      regionsToScan = new TreeMap<HRegionInfo, ServerName>();
+      regionsToScan = new TreeMap<>();
       for (HRegionLocation regionLocation : regionLocations) {
         HRegionInfo regionInfo = regionLocation.getRegionInfo();
         if (!foundStartRegion && hbaseScanSpec.getStartRow() != null && hbaseScanSpec.getStartRow().length != 0 && !regionInfo.containsRow(hbaseScanSpec.getStartRow())) {
@@ -338,8 +338,7 @@ public class HBaseGroupScan extends AbstractGroupScan implements DrillHBaseConst
     assert minorFragmentId < endpointFragmentMapping.size() : String.format(
         "Mappings length [%d] should be greater than minor fragment id [%d] but it isn't.", endpointFragmentMapping.size(),
         minorFragmentId);
-    return new HBaseSubScan(getUserName(), storagePlugin, storagePluginConfig,
-        endpointFragmentMapping.get(minorFragmentId), columns);
+    return new HBaseSubScan(getUserName(), storagePlugin, endpointFragmentMapping.get(minorFragmentId), columns);
   }
 
   @Override
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 0527391a2..bd179fbac 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
@@ -24,7 +24,6 @@ import java.util.List;
 
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.common.logical.StoragePluginConfig;
 import org.apache.drill.exec.physical.base.AbstractBase;
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.physical.base.PhysicalVisitor;
@@ -49,9 +48,6 @@ import com.google.common.base.Preconditions;
 public class HBaseSubScan extends AbstractBase implements SubScan {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HBaseSubScan.class);
 
-  @JsonProperty
-  public final HBaseStoragePluginConfig storage;
-  @JsonIgnore
   private final HBaseStoragePlugin hbaseStoragePlugin;
   private final List<HBaseSubScanSpec> regionScanSpecList;
   private final List<SchemaPath> columns;
@@ -59,34 +55,36 @@ public class HBaseSubScan extends AbstractBase implements SubScan {
   @JsonCreator
   public HBaseSubScan(@JacksonInject StoragePluginRegistry registry,
                       @JsonProperty("userName") String userName,
-                      @JsonProperty("storage") StoragePluginConfig storage,
+                      @JsonProperty("hbaseStoragePluginConfig") HBaseStoragePluginConfig hbaseStoragePluginConfig,
                       @JsonProperty("regionScanSpecList") LinkedList<HBaseSubScanSpec> regionScanSpecList,
                       @JsonProperty("columns") List<SchemaPath> columns) throws ExecutionSetupException {
-    super(userName);
-    hbaseStoragePlugin = (HBaseStoragePlugin) registry.getPlugin(storage);
-    this.regionScanSpecList = regionScanSpecList;
-    this.storage = (HBaseStoragePluginConfig) storage;
-    this.columns = columns;
+    this(userName,
+        (HBaseStoragePlugin) registry.getPlugin(hbaseStoragePluginConfig),
+        regionScanSpecList,
+        columns);
   }
 
-  public HBaseSubScan(String userName, HBaseStoragePlugin plugin, HBaseStoragePluginConfig config,
-      List<HBaseSubScanSpec> regionInfoList, List<SchemaPath> columns) {
+  public HBaseSubScan(String userName,
+                      HBaseStoragePlugin hbaseStoragePlugin,
+                      List<HBaseSubScanSpec> regionInfoList,
+                      List<SchemaPath> columns) {
     super(userName);
-    hbaseStoragePlugin = plugin;
-    storage = config;
+    this.hbaseStoragePlugin = hbaseStoragePlugin;
     this.regionScanSpecList = regionInfoList;
     this.columns = columns;
   }
 
-  public List<HBaseSubScanSpec> getRegionScanSpecList() {
-    return regionScanSpecList;
+  @JsonProperty
+  public HBaseStoragePluginConfig getHbaseStoragePluginConfig() {
+    return hbaseStoragePlugin.getConfig();
   }
 
-  @JsonIgnore
-  public HBaseStoragePluginConfig getStorageConfig() {
-    return storage;
+  @JsonProperty
+  public List<HBaseSubScanSpec> getRegionScanSpecList() {
+    return regionScanSpecList;
   }
 
+  @JsonProperty
   public List<SchemaPath> getColumns() {
     return columns;
   }
@@ -109,7 +107,7 @@ public class HBaseSubScan extends AbstractBase implements SubScan {
   @Override
   public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
     Preconditions.checkArgument(children.isEmpty());
-    return new HBaseSubScan(getUserName(), hbaseStoragePlugin, storage, regionScanSpecList, columns);
+    return new HBaseSubScan(getUserName(), hbaseStoragePlugin, regionScanSpecList, columns);
   }
 
   @Override
diff --git a/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/TestHBaseQueries.java b/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/TestHBaseQueries.java
index ee839c5c3..0d5349943 100644
--- a/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/TestHBaseQueries.java
+++ b/contrib/storage-hbase/src/test/java/org/apache/drill/hbase/TestHBaseQueries.java
@@ -20,6 +20,7 @@ package org.apache.drill.hbase;
 import java.util.Arrays;
 import java.util.List;
 
+import org.apache.drill.PlanTestBase;
 import org.apache.drill.categories.HbaseStorageTest;
 import org.apache.drill.categories.SlowTest;
 import org.apache.drill.exec.rpc.user.QueryDataBatch;
@@ -109,4 +110,9 @@ public class TestHBaseQueries extends BaseHBaseTest {
     runHBaseSQLVerifyCount("SELECT row_key\n"
         + " FROM hbase.TestTableNullStr t WHERE row_key='a1'", 1);
   }
+
+  @Test
+  public void testPhysicalPlanSubmission() throws Exception {
+    PlanTestBase.testPhysicalPlanExecutionBasedOnQuery("select * from hbase.TestTableNullStr");
+  }
 }