DRILL-6381: (Part 3) Planner and Execution implementation to support Secondary Indexes

  1. Index Planning Rules and Plan generators
    - DbScanToIndexScanRule: Top level physical planning rule that drives index planning for several relational algebra patterns.
    - DbScanSortRemovalRule: Physical planning rule for index planning for Sort-based operations.
    - Plan Generators: Covering, Non-Covering and Intersect physical plan generators.
    - Support planning with functional indexes such as CAST functions.
    - Enhance PlannerSettings with several configuration options for indexes.
  2. Index Selection and Statistics
    - An IndexSelector that support cost-based index selection of covering and non-covering indexes using statistics and collation properties.
    - Costing of index intersection for comparison with single-index plans.
  3. Planning and execution operators
    - Support RangePartitioning physical operator during query planning and execution.
    - Support RowKeyJoin physical operator during query planning and execution.
    - HashTable and HashJoin changes to support RowKeyJoin and Index Intersection.
    - Enhance Materializer to keep track of subscan association with a particular rowkey join.
  4. Index Planning utilities
    - Utility classes to perform RexNode analysis, including conversion to and from SchemaPath.
    - Utility class to analyze filter condition and an input collation to determine output collation.
    - Helper classes to maintain index contexts for logical and physical planning phase.
    - IndexPlanUtils utility class for various helper methods.
  5. Miscellaneous
    - Separate physical rel for DirectScan.
    - Modify LimitExchangeTranspose rule to handle SingleMergeExchange.
    - MD-3880: Return correct status from RangePartitionRecordBatch setupNewSchema

Co-authored-by: Aman Sinha <asinha@maprtech.com>
Co-authored-by: chunhui-shi <cshi@maprtech.com>
Co-authored-by: Gautam Parai <gparai@maprtech.com>
Co-authored-by: Padma Penumarthy <ppenumar97@yahoo.com>
Co-authored-by: Hanumath Rao Maduri <hmaduri@maprtech.com>

Conflicts:
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/HashJoinPOP.java
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMergeProjectRule.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillOptiq.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjectIntoScanRule.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillScanRel.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/BroadcastExchangePrel.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/DrillDistributionTrait.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashJoinPrel.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java
	exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
	exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java
	exec/java-exec/src/main/resources/drill-module.conf
	logical/src/main/java/org/apache/drill/common/logical/StoragePluginConfig.java

Resolve merge comflicts and compilation issues.

8 files changed, 19 insertions, 14 deletions

diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/planner/index/MapRDBStatistics.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/planner/index/MapRDBStatistics.java
index dc89a4dc6..3b8de349a 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/planner/index/MapRDBStatistics.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/planner/index/MapRDBStatistics.java
@@ -185,7 +185,7 @@ public class MapRDBStatistics implements Statistics {
    *  @return approximate rows satisfying the leading filter
    */
   @Override
-  public double getLeadingRowCount(RexNode condition, String tabIdxName, RelNode scanRel) {
+  public double getLeadingRowCount(RexNode condition, String tabIdxName, DrillScanRelBase scanRel) {
     String conditionAsStr = nullConditionAsString;
     Map<String, StatisticsPayload> payloadMap;
     if ((scanRel instanceof DrillScanRel && ((DrillScanRel)scanRel).getGroupScan() instanceof DbGroupScan)
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
index cf497149e..511a111d6 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
@@ -140,7 +140,7 @@ public abstract class MapRDBPushFilterIntoScan extends StoragePluginOptimizerRul
     final JsonTableGroupScan newGroupsScan = (JsonTableGroupScan) groupScan.clone(newScanSpec);
     newGroupsScan.setFilterPushedDown(true);
 
-    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType(), scan.getTable());
 
     // Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
     final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of((RelNode)newScanPrel));
@@ -186,7 +186,7 @@ public abstract class MapRDBPushFilterIntoScan extends StoragePluginOptimizerRul
                                                                         groupScan.getTableStats());
     newGroupsScan.setFilterPushedDown(true);
 
-    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType(), scan.getTable());
 
     // Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
     final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of((RelNode)newScanPrel));;
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushLimitIntoScan.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushLimitIntoScan.java
index 79eec12dd..1f4b8c9a0 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushLimitIntoScan.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushLimitIntoScan.java
@@ -119,7 +119,7 @@ public abstract class MapRDBPushLimitIntoScan extends StoragePluginOptimizerRule
         return;
       }
       final ScanPrel newScan = new ScanPrel(scan.getCluster(), scan.getTraitSet(), newGroupScan,
-          scan.getRowType());
+          scan.getRowType(), scan.getTable());
       final RelNode newChild;
       if (project != null) {
         final ProjectPrel newProject = new ProjectPrel(project.getCluster(), project.getTraitSet(),
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushProjectIntoScan.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushProjectIntoScan.java
index 8d4f54962..2eb84e7ef 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushProjectIntoScan.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushProjectIntoScan.java
@@ -28,6 +28,7 @@ import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
+import org.apache.drill.exec.planner.common.DrillRelOptUtil;
 import org.apache.drill.exec.planner.logical.RelOptHelper;
 import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.planner.physical.PrelUtil;
@@ -36,6 +37,7 @@ import org.apache.drill.exec.planner.physical.ScanPrel;
 import org.apache.drill.exec.store.StoragePluginOptimizerRule;
 import org.apache.drill.exec.store.mapr.db.binary.BinaryTableGroupScan;
 import org.apache.drill.exec.store.mapr.db.json.JsonTableGroupScan;
+import org.apache.drill.exec.util.Utilities;
 
 import java.util.List;
 
@@ -82,9 +84,10 @@ public abstract class MapRDBPushProjectIntoScan extends StoragePluginOptimizerRu
       ProjectPrel project, ScanPrel scan, MapRDBGroupScan groupScan) {
     try {
 
-      PrelUtil.ProjectPushInfo columnInfo = PrelUtil.getColumns(scan.getRowType(), project.getProjects());
-      if (columnInfo == null || columnInfo.isStarQuery() //
-          || !groupScan.canPushdownProjects(columnInfo.columns)) {
+      DrillRelOptUtil.ProjectPushInfo columnInfo =
+          DrillRelOptUtil.getFieldsInformation(scan.getRowType(), project.getProjects());
+      if (columnInfo == null || Utilities.isStarQuery(columnInfo.getFields()) //
+          || !groupScan.canPushdownProjects(columnInfo.getFields())) {
         return;
       }
       RelTraitSet newTraits = call.getPlanner().emptyTraitSet();
@@ -95,12 +98,12 @@ public abstract class MapRDBPushProjectIntoScan extends StoragePluginOptimizerRu
         }
       }
       final ScanPrel newScan = new ScanPrel(scan.getCluster(), newTraits.plus(Prel.DRILL_PHYSICAL),
-          groupScan.clone(columnInfo.columns),
-          columnInfo.createNewRowType(project.getInput().getCluster().getTypeFactory()));
+          groupScan.clone(columnInfo.getFields()),
+          columnInfo.createNewRowType(project.getInput().getCluster().getTypeFactory()), scan.getTable());
 
       List<RexNode> newProjects = Lists.newArrayList();
       for (RexNode n : project.getChildExps()) {
-        newProjects.add(n.accept(columnInfo.getInputRewriter()));
+        newProjects.add(n.accept(columnInfo.getInputReWriter()));
       }
 
       final ProjectPrel newProj =
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
index 736b36b80..91ca7870f 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
@@ -122,7 +122,7 @@ public abstract class HBasePushFilterIntoScan extends StoragePluginOptimizerRule
         newScanSpec, groupScan.getColumns());
     newGroupsScan.setFilterPushedDown(true);
 
-    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType(), scan.getTable());
 
     // Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
     final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of(newScanPrel));
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveMapRDBJsonScanToDrillMapRDBJsonScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveMapRDBJsonScanToDrillMapRDBJsonScan.java
index 9eb43a251..4994a7277 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveMapRDBJsonScanToDrillMapRDBJsonScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveMapRDBJsonScanToDrillMapRDBJsonScan.java
@@ -123,7 +123,7 @@ public class ConvertHiveMapRDBJsonScanToDrillMapRDBJsonScan extends StoragePlugi
     HiveScan hiveScan = (HiveScan) hiveScanRel.getGroupScan();
     Map<String, String> parameters = hiveScan.getHiveReadEntry().getHiveTableWrapper().getParameters();
 
-    JsonScanSpec scanSpec = new JsonScanSpec(parameters.get(MAPRDB_TABLE_NAME), null);
+    JsonScanSpec scanSpec = new JsonScanSpec(parameters.get(MAPRDB_TABLE_NAME), null, null);
     List<SchemaPath> hiveScanCols = hiveScanRel.getColumns().stream()
         .map(colNameSchemaPath -> replaceOverriddenSchemaPath(parameters, colNameSchemaPath))
         .collect(Collectors.toList());
diff --git a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
index 14abadff9..019a67eba 100644
--- a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
+++ b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
@@ -66,7 +66,8 @@ public class KafkaPushDownFilterIntoScan extends StoragePluginOptimizerRule {
 
     logger.info("Partitions ScanSpec after pushdown: " + newScanSpec);
     GroupScan newGroupScan = groupScan.cloneWithNewSpec(newScanSpec);
-    final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupScan, scan.getRowType());
+    final ScanPrel newScanPrel = 
+      new ScanPrel(scan, filter.getTraitSet(), newGroupScan, scan.getRowType(), scan.getTable());
     call.transformTo(filter.copy(filter.getTraitSet(), ImmutableList.of(newScanPrel)));
   }
 
diff --git a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
index afe32f225..8ad84c117 100644
--- a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
+++ b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
@@ -78,7 +78,8 @@ public class MongoPushDownFilterForScan extends StoragePluginOptimizerRule {
     newGroupsScan.setFilterPushedDown(true);
 
     final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(),
-        newGroupsScan, scan.getRowType());
+        newGroupsScan, scan.getRowType(), scan.getTable());
+
     if (mongoFilterBuilder.isAllExpressionsConverted()) {
       /*
        * Since we could convert the entire filter condition expression into an