DRILL-6381: (Part 4) Enhance MapR-DB plugin to support querying secondary indexes

  1. Implementation of the index descriptor for MapR-DB.
  2. MapR-DB specific costing for covering and non-covering indexes.
  3. Discovery componenent to discover the indexes available for a MapR-DB table including CAST functional indexes.
  4. Utility functions to build a canonical index descriptor.
  5. Statistics: fetch and initialize statistcs from MapR-DB for a query condition. Maintain a query-scoped cache for the statistics. Utility functions to compute selectivity.
  6. Range Partitioning: partitioning function that takes into account the tablet map to find out where a particular rowkey belongs.
  7. Restricted Scan: support doing restricted (i.e skip) scan through lookups on the rowkey. Added a group-scan and record reader for this.
  8. MD-3726: Simple Order by queries (without limit) when an index is used are showing regression.
  9. MD-3995: Do not pushdown limit 0 past project with CONVERT_FROMJSON
  10. MD-4259 : Account for limit during hashcode computation

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:
	contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatMatcher.java
	contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushProjectIntoScan.java
	contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/JsonTableGroupScan.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/rules/DbScanSortRemovalRule.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
	exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/TopNPrel.java

Fix additional compilation issues.

14 files changed, 262 insertions, 68 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/OrderedRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/OrderedRel.java
new file mode 100644
index 000000000..5f4da7439
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/OrderedRel.java
@@ -0,0 +1,53 @@
+/*
+ * 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.exec.planner.common;
+
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rex.RexNode;
+
+/**
+ * Class implementing OrderedPrel interface guarantees to provide ordered
+ * output on certain columns. TopNPrel and SortPrel base classes which implement
+ * this interface.
+ */
+public interface OrderedRel extends DrillRelNode {
+
+  /**
+   * A method to return ordering columns of the result.
+   * @return Collation order of the output.
+   */
+  RelCollation getCollation();
+
+  /**
+   * Offset value represented in RexNode.
+   * @return offset.
+   */
+  RexNode getOffset();
+
+  /**
+   * Fetch value represented in RexNode.
+   * @return fetch
+   */
+  RexNode getFetch();
+
+  /**
+   * A method to return if this relational node can be dropped during optimization process.
+   * @return true if this node can be dropped, false otherwise.
+   */
+  boolean canBeDropped();
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexCallContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexCallContext.java
index 65788cb52..45251c641 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexCallContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexCallContext.java
@@ -20,7 +20,6 @@ package org.apache.drill.exec.planner.index;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
@@ -28,6 +27,7 @@ import org.apache.drill.exec.physical.base.DbGroupScan;
 import org.apache.drill.exec.planner.physical.DrillDistributionTrait.DistributionField;
 import org.apache.drill.exec.planner.common.DrillScanRelBase;
 import org.apache.drill.exec.planner.common.DrillProjectRelBase;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import java.util.List;
 import java.util.Set;
 
@@ -66,7 +66,7 @@ public interface IndexCallContext {
 
   RexNode getOrigCondition();
 
-  Sort getSort();
+  OrderedRel getSort();
 
   void createSortExprs();
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexLogicalPlanCallContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexLogicalPlanCallContext.java
index 27198bb02..3a6ea83d6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexLogicalPlanCallContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexLogicalPlanCallContext.java
@@ -20,7 +20,6 @@ package org.apache.drill.exec.planner.index;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.rel.RelCollation;
-import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
@@ -31,6 +30,7 @@ import org.apache.drill.exec.planner.logical.DrillProjectRel;
 import org.apache.drill.exec.planner.logical.DrillScanRel;
 import org.apache.drill.exec.planner.logical.DrillSortRel;
 import org.apache.drill.exec.planner.common.DrillScanRelBase;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.physical.DrillDistributionTrait.DistributionField;
 import org.apache.calcite.rel.RelNode;
 
@@ -164,7 +164,7 @@ public class IndexLogicalPlanCallContext implements IndexCallContext {
     return origPushedCondition;
   }
 
-  public Sort getSort() {
+  public OrderedRel getSort() {
     return sort;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPhysicalPlanCallContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPhysicalPlanCallContext.java
index 9c7b65167..91ff02c69 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPhysicalPlanCallContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPhysicalPlanCallContext.java
@@ -21,7 +21,6 @@ import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
@@ -29,10 +28,10 @@ import org.apache.drill.exec.physical.base.AbstractDbGroupScan;
 import org.apache.drill.exec.physical.base.DbGroupScan;
 import org.apache.drill.exec.planner.common.DrillProjectRelBase;
 import org.apache.drill.exec.planner.common.DrillScanRelBase;
-import org.apache.drill.exec.planner.physical.SortPrel;
 import org.apache.drill.exec.planner.physical.ProjectPrel;
 import org.apache.drill.exec.planner.physical.FilterPrel;
 import org.apache.drill.exec.planner.physical.ScanPrel;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.physical.ExchangePrel;
 import org.apache.drill.exec.planner.physical.HashToRandomExchangePrel;
 import org.apache.drill.exec.planner.physical.DrillDistributionTrait.DistributionField;
@@ -42,8 +41,9 @@ import java.util.List;
 import java.util.Set;
 
 public class IndexPhysicalPlanCallContext implements IndexCallContext {
+
   final public RelOptRuleCall call;
-  final public SortPrel sort;
+  final public OrderedRel sort;
   final public ProjectPrel upperProject;
   final public FilterPrel filter;
   final public ProjectPrel lowerProject;
@@ -67,7 +67,7 @@ public class IndexPhysicalPlanCallContext implements IndexCallContext {
   }
 
   public IndexPhysicalPlanCallContext(RelOptRuleCall call,
-                                      SortPrel sort,
+                                      OrderedRel sort,
                                       ProjectPrel capProject,
                                       FilterPrel filter,
                                       ProjectPrel project,
@@ -83,7 +83,7 @@ public class IndexPhysicalPlanCallContext implements IndexCallContext {
   }
 
   public IndexPhysicalPlanCallContext(RelOptRuleCall call,
-                                      SortPrel sort,
+                                      OrderedRel sort,
                                       ProjectPrel project,
                                       ScanPrel scan, ExchangePrel exch) {
     this.call = call;
@@ -171,7 +171,7 @@ public class IndexPhysicalPlanCallContext implements IndexCallContext {
     return origPushedCondition;
   }
 
-  public Sort getSort() {
+  public OrderedRel getSort() {
     return sort;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPlanUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPlanUtils.java
index 666e2828a..cdad63ad0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPlanUtils.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/IndexPlanUtils.java
@@ -38,6 +38,7 @@ import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.drill.common.expression.FieldReference;
 import org.apache.drill.common.expression.LogicalExpression;
@@ -56,6 +57,7 @@ import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.planner.physical.PrelUtil;
 import org.apache.drill.exec.planner.physical.ScanPrel;
 import org.apache.drill.exec.planner.physical.ProjectPrel;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexInputRef;
@@ -346,6 +348,21 @@ public class IndexPlanUtils {
     return proj.getProjects();
   }
 
+  public static boolean generateLimit(OrderedRel sort) {
+    RexNode fetchNode = sort.getFetch();
+    int fetchValue = (fetchNode == null) ? -1 : RexLiteral.intValue(fetchNode);
+    return fetchValue >=0;
+  }
+
+  public static RexNode getOffset(OrderedRel sort) {
+    return sort.getOffset();
+  }
+
+  public static RexNode getFetch(OrderedRel sort) {
+    return sort.getFetch();
+  }
+
+
   /**
    * generate logical expressions for sort rexNodes in SortRel, the result is store to IndexPlanCallContext
    * @param indexContext
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/AbstractIndexPlanGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/AbstractIndexPlanGenerator.java
index 36ff61f20..456542b31 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/AbstractIndexPlanGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/AbstractIndexPlanGenerator.java
@@ -30,7 +30,6 @@ import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.InvalidRelException;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
 import org.apache.calcite.rel.type.RelRecordType;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -43,16 +42,19 @@ import org.apache.drill.exec.planner.logical.DrillFilterRel;
 import org.apache.drill.exec.planner.logical.DrillProjectRel;
 import org.apache.drill.exec.planner.logical.DrillSortRel;
 import org.apache.drill.exec.planner.common.DrillProjectRelBase;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.common.DrillScanRelBase;
+import org.apache.drill.exec.planner.physical.SubsetTransformer;
 import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.planner.physical.Prule;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
 import org.apache.drill.exec.planner.physical.Prel;
-import org.apache.drill.exec.planner.physical.SortPrel;
 import org.apache.drill.exec.planner.physical.HashToMergeExchangePrel;
 import org.apache.drill.exec.planner.physical.SingleMergeExchangePrel;
 import org.apache.drill.exec.planner.physical.PrelUtil;
-import org.apache.drill.exec.planner.physical.Prule;
-import org.apache.drill.exec.planner.physical.SubsetTransformer;
-import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
+import org.apache.drill.exec.planner.physical.TopNPrel;
+import org.apache.drill.exec.planner.physical.SortPrel;
+import org.apache.drill.exec.planner.physical.LimitPrel;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
@@ -67,7 +69,6 @@ public abstract class AbstractIndexPlanGenerator extends SubsetTransformer<RelNo
   final protected DrillProjectRelBase origProject;
   final protected DrillScanRelBase origScan;
   final protected DrillProjectRelBase upperProject;
-  final protected RelNode origSort;
 
   final protected RexNode indexCondition;
   final protected RexNode remainderCondition;
@@ -84,7 +85,6 @@ public abstract class AbstractIndexPlanGenerator extends SubsetTransformer<RelNo
     this.origProject = indexContext.getLowerProject();
     this.origScan = indexContext.getScan();
     this.upperProject = indexContext.getUpperProject();
-    this.origSort = indexContext.getSort();
     this.indexCondition = indexCondition;
     this.remainderCondition = remainderCondition;
     this.indexContext = indexContext;
@@ -168,8 +168,8 @@ public abstract class AbstractIndexPlanGenerator extends SubsetTransformer<RelNo
     return set;
   }
 
-  protected static boolean toRemoveSort(Sort sort, RelCollation inputCollation) {
-    if ( (inputCollation != null) && inputCollation.satisfies(IndexPlanUtils.getCollation(sort))) {
+  protected static boolean toRemoveSort(RelCollation sortCollation, RelCollation inputCollation) {
+    if ( (inputCollation != null) && inputCollation.satisfies(sortCollation)) {
       return true;
     }
     return false;
@@ -194,18 +194,34 @@ public abstract class AbstractIndexPlanGenerator extends SubsetTransformer<RelNo
     }
   }
 
+  private static RelNode getSortOrTopN(IndexCallContext indexContext,
+                                       RelNode sortNode, RelNode newRel, RelNode child) {
+    if (sortNode instanceof TopNPrel) {
+      return new TopNPrel(sortNode.getCluster(),
+                    newRel.getTraitSet().replace(Prel.DRILL_PHYSICAL).plus(indexContext.getCollation()),
+                    child, ((TopNPrel)sortNode).getLimit(), indexContext.getCollation());
+    }
+    return new SortPrel(sortNode.getCluster(),
+            newRel.getTraitSet().replace(Prel.DRILL_PHYSICAL).plus(indexContext.getCollation()),
+            child, indexContext.getCollation());
+  }
+
   public static RelNode getSortNode(IndexCallContext indexContext, RelNode newRel, boolean donotGenerateSort,
                                     boolean isSingleton, boolean isExchangeRequired) {
-    Sort rel = indexContext.getSort();
+    OrderedRel rel = indexContext.getSort();
     DrillDistributionTrait hashDistribution =
         new DrillDistributionTrait(DrillDistributionTrait.DistributionType.HASH_DISTRIBUTED,
             ImmutableList.copyOf(indexContext.getDistributionFields()));
 
-    if ( toRemoveSort(indexContext.getSort(), newRel.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE))) {
+    if ( toRemoveSort(indexContext.getCollation(), newRel.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE))) {
       //we are going to remove sort
       logger.debug("Not generating SortPrel since we have the required collation");
-
-      RelTraitSet traits = newRel.getTraitSet().plus(IndexPlanUtils.getCollation(rel)).plus(Prel.DRILL_PHYSICAL);
+      if (IndexPlanUtils.generateLimit(rel)) {
+        newRel = new LimitPrel(newRel.getCluster(),
+                newRel.getTraitSet().plus(indexContext.getCollation()).plus(Prel.DRILL_PHYSICAL),
+                newRel, IndexPlanUtils.getOffset(rel), IndexPlanUtils.getFetch(rel));
+      }
+      RelTraitSet traits = newRel.getTraitSet().plus(indexContext.getCollation()).plus(Prel.DRILL_PHYSICAL);
       newRel = Prule.convert(newRel, traits);
       newRel = getExchange(newRel.getCluster(), isSingleton, isExchangeRequired,
                                  traits, hashDistribution, indexContext, newRel);
@@ -215,10 +231,9 @@ public abstract class AbstractIndexPlanGenerator extends SubsetTransformer<RelNo
         logger.debug("Not generating SortPrel and index plan, since just picking index for full index scan is not beneficial.");
         return null;
       }
-      RelTraitSet traits = newRel.getTraitSet().plus(IndexPlanUtils.getCollation(rel)).plus(Prel.DRILL_PHYSICAL);
-      newRel = new SortPrel(rel.getCluster(),
-              newRel.getTraitSet().replace(Prel.DRILL_PHYSICAL).plus(IndexPlanUtils.getCollation(rel)),
-          Prule.convert(newRel, newRel.getTraitSet().replace(Prel.DRILL_PHYSICAL)), IndexPlanUtils.getCollation(rel));
+      RelTraitSet traits = newRel.getTraitSet().plus(indexContext.getCollation()).plus(Prel.DRILL_PHYSICAL);
+      newRel = getSortOrTopN(indexContext, rel, newRel,
+                  Prule.convert(newRel, newRel.getTraitSet().replace(Prel.DRILL_PHYSICAL)));
       newRel = getExchange(newRel.getCluster(), isSingleton, isExchangeRequired,
                                  traits, hashDistribution, indexContext, newRel);
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/CoveringPlanNoFilterGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/CoveringPlanNoFilterGenerator.java
index 163aef986..e06ac8ff3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/CoveringPlanNoFilterGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/CoveringPlanNoFilterGenerator.java
@@ -18,6 +18,8 @@
 package org.apache.drill.exec.planner.index.generators;
 
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.InvalidRelException;
 import org.apache.calcite.rel.RelCollation;
@@ -30,11 +32,11 @@ import org.apache.drill.exec.planner.index.FunctionalIndexInfo;
 import org.apache.drill.exec.planner.index.IndexPlanUtils;
 import org.apache.drill.exec.planner.logical.DrillParseContext;
 import org.apache.drill.exec.planner.physical.PlannerSettings;
-import org.apache.drill.exec.planner.physical.ScanPrel;
 import org.apache.drill.exec.planner.physical.ProjectPrel;
-import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.planner.physical.PrelUtil;
+import org.apache.drill.exec.planner.physical.ScanPrel;
 import org.apache.drill.exec.planner.physical.Prule;
+import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.physical.base.DbGroupScan;
 import java.util.List;
 
@@ -62,7 +64,7 @@ public class CoveringPlanNoFilterGenerator extends AbstractIndexPlanGenerator {
   }
 
   public RelNode convertChild() throws InvalidRelException {
-
+    Preconditions.checkNotNull(indexContext.getSort());
     if (indexGroupScan == null) {
       logger.error("Null indexgroupScan in CoveringIndexPlanGenerator.convertChild");
       return null;
@@ -106,11 +108,9 @@ public class CoveringPlanNoFilterGenerator extends AbstractIndexPlanGenerator {
       }
     }
 
-    if (indexContext.getSort() != null) {
-      finalRel = getSortNode(indexContext, finalRel, true, isSingletonSortedStream, indexContext.getExchange() != null);
-      if (finalRel == null) {
-        return null;
-      }
+    finalRel = getSortNode(indexContext, finalRel, true, isSingletonSortedStream, indexContext.getExchange() != null);
+    if (finalRel == null) {
+      return null;
     }
 
     finalRel = Prule.convert(finalRel, finalRel.getTraitSet().plus(Prel.DRILL_PHYSICAL));
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/NonCoveringIndexPlanGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/NonCoveringIndexPlanGenerator.java
index db220fab4..e1337bc7e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/NonCoveringIndexPlanGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/generators/NonCoveringIndexPlanGenerator.java
@@ -319,7 +319,7 @@ public class NonCoveringIndexPlanGenerator extends AbstractIndexPlanGenerator {
     if (indexContext.getSort() != null) {
       // When ordering is required, serialize the index scan side. With parallel index scans, the rowkey join may receive
       // unsorted input because ordering is not guaranteed across different parallel inputs.
-      if (toRemoveSort(indexContext.getSort(), newRel.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE))) {
+      if (toRemoveSort(indexContext.getCollation(), newRel.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE))) {
         ((IndexGroupScan)indexScanPrel.getGroupScan()).setParallelizationWidth(1);
       }
       newRel = getSortNode(indexContext, newRel, false,true, true);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/rules/DbScanSortRemovalRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/rules/DbScanSortRemovalRule.java
index db095047d..86ed4301c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/rules/DbScanSortRemovalRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/index/rules/DbScanSortRemovalRule.java
@@ -18,6 +18,8 @@
 package org.apache.drill.exec.planner.index.rules;
 
 import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
 import org.apache.calcite.plan.RelOptRule;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.plan.RelOptRuleOperand;
@@ -39,7 +41,7 @@ import org.apache.drill.exec.planner.physical.PrelUtil;
 import org.apache.drill.exec.planner.physical.ExchangePrel;
 import org.apache.drill.exec.planner.physical.ProjectPrel;
 import org.apache.drill.exec.planner.physical.ScanPrel;
-import org.apache.drill.exec.planner.physical.SortPrel;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.planner.physical.HashToRandomExchangePrel;
 import org.apache.calcite.rel.RelNode;
@@ -52,21 +54,21 @@ public class DbScanSortRemovalRule extends Prule {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DbScanSortRemovalRule.class);
 
   public static final RelOptRule INDEX_SORT_EXCHANGE_SCAN =
-      new DbScanSortRemovalRule(RelOptHelper.some(SortPrel.class,
+      new DbScanSortRemovalRule(RelOptHelper.some(OrderedRel.class,
           RelOptHelper.some(HashToRandomExchangePrel.class,
               RelOptHelper.any(ScanPrel.class))), "DbScanSortRemovalRule:sort_exchange_Scan", new MatchSES());
 
   public static final RelOptRule INDEX_SORT_SCAN =
-          new DbScanSortRemovalRule(RelOptHelper.some(SortPrel.class,
+          new DbScanSortRemovalRule(RelOptHelper.some(OrderedRel.class,
                           RelOptHelper.any(ScanPrel.class)), "DbScanSortRemovalRule:Sort_Scan", new MatchSS());
 
   public static final RelOptRule INDEX_SORT_PROJ_SCAN =
-          new DbScanSortRemovalRule(RelOptHelper.some(SortPrel.class,
+          new DbScanSortRemovalRule(RelOptHelper.some(OrderedRel.class,
                   RelOptHelper.some(ProjectPrel.class,
                     RelOptHelper.any(ScanPrel.class))), "DbScanSortRemovalRule:Sort_Proj_Scan", new MatchSPS());
 
   public static final RelOptRule INDEX_SORT_EXCHANGE_PROJ_SCAN =
-      new DbScanSortRemovalRule(RelOptHelper.some(SortPrel.class,
+      new DbScanSortRemovalRule(RelOptHelper.some(OrderedRel.class,
           RelOptHelper.some(HashToRandomExchangePrel.class,
               RelOptHelper.some(ProjectPrel.class,
                   RelOptHelper.any(ScanPrel.class)))), "DbScanSortRemovalRule:sort_exchange_proj_Scan", new MatchSEPS());
@@ -80,16 +82,21 @@ public class DbScanSortRemovalRule extends Prule {
     this.match = match;
   }
 
+  private static boolean isRemovableRel(OrderedRel node) {
+    return node.canBeDropped();
+  }
+
   private static class MatchSES extends AbstractMatchFunction<IndexPhysicalPlanCallContext> {
 
     public boolean match(RelOptRuleCall call) {
-      final ScanPrel scan = (ScanPrel)call.rel(2);
-      return checkScan(scan.getGroupScan());
+      final OrderedRel sort = call.rel(0);
+      final ScanPrel scan = call.rel(2);
+      return sort instanceof Prel && checkScan(scan.getGroupScan()) && isRemovableRel(sort);
     }
 
     public IndexPhysicalPlanCallContext onMatch(RelOptRuleCall call) {
       final ScanPrel scan = call.rel(2);
-      final SortPrel sort = call.rel(0);
+      final OrderedRel sort = call.rel(0);
       final ExchangePrel exch = call.rel(1);
       return new IndexPhysicalPlanCallContext(call, sort, null, scan, exch);
     }
@@ -98,13 +105,14 @@ public class DbScanSortRemovalRule extends Prule {
   private static class MatchSS extends AbstractMatchFunction<IndexPhysicalPlanCallContext> {
 
     public boolean match(RelOptRuleCall call) {
-      final ScanPrel scan = (ScanPrel)call.rel(1);
-      return checkScan(scan.getGroupScan());
+      final OrderedRel sort = call.rel(0);
+      final ScanPrel scan = call.rel(1);
+      return sort instanceof Prel && checkScan(scan.getGroupScan()) && isRemovableRel(sort);
     }
 
     public IndexPhysicalPlanCallContext onMatch(RelOptRuleCall call) {
       final ScanPrel scan = call.rel(1);
-      final SortPrel sort = call.rel(0);
+      final OrderedRel sort = call.rel(0);
       return new IndexPhysicalPlanCallContext(call, sort, null, scan, null);
     }
   }
@@ -112,14 +120,15 @@ public class DbScanSortRemovalRule extends Prule {
   private static class MatchSPS extends AbstractMatchFunction<IndexPhysicalPlanCallContext> {
 
     public boolean match(RelOptRuleCall call) {
-      final ScanPrel scan = (ScanPrel)call.rel(2);
-      return checkScan(scan.getGroupScan());
+      final OrderedRel sort = call.rel(0);
+      final ScanPrel scan = call.rel(2);
+      return sort instanceof Prel && checkScan(scan.getGroupScan()) && isRemovableRel(sort);
     }
 
     public IndexPhysicalPlanCallContext onMatch(RelOptRuleCall call) {
       final ScanPrel scan = call.rel(2);
       final ProjectPrel proj = call.rel(1);
-      final SortPrel sort = call.rel(0);
+      final OrderedRel sort = call.rel(0);
       return new IndexPhysicalPlanCallContext(call, sort, proj, scan, null);
     }
   }
@@ -127,13 +136,14 @@ public class DbScanSortRemovalRule extends Prule {
   private static class MatchSEPS extends AbstractMatchFunction<IndexPhysicalPlanCallContext> {
 
     public boolean match(RelOptRuleCall call) {
-      final ScanPrel scan = (ScanPrel)call.rel(3);
-      return checkScan(scan.getGroupScan());
+      final OrderedRel sort = call.rel(0);
+      final ScanPrel scan = call.rel(3);
+      return sort instanceof Prel && checkScan(scan.getGroupScan()) && isRemovableRel(sort);
     }
 
     public IndexPhysicalPlanCallContext onMatch(RelOptRuleCall call) {
       final ScanPrel scan = call.rel(3);
-      final SortPrel sort = call.rel(0);
+      final OrderedRel sort = call.rel(0);
       final ProjectPrel proj = call.rel(2);
       final ExchangePrel exch = call.rel(1);
       return new IndexPhysicalPlanCallContext(call,  sort, proj, scan, exch);
@@ -187,12 +197,15 @@ public class DbScanSortRemovalRule extends Prule {
                           false, settings);
           if (planGen.convertChild() != null) {
             indexContext.getCall().transformTo(planGen.convertChild());
+          } else {
+            logger.debug("Not able to generate index plan in ", this.getClass().toString());
           }
         } catch (Exception e) {
           logger.warn("Exception while trying to generate indexscan to remove sort", e);
         }
       }
     } else {
+      Preconditions.checkNotNull(indexContext.getSort());
       //This case tries to use the already generated index to see if a sort can be removed.
       if (indexContext.scan.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE).getFieldCollations().size() == 0) {
         return;
@@ -204,12 +217,12 @@ public class DbScanSortRemovalRule extends Prule {
           inputs.add(finalRel);
           finalRel = indexContext.lowerProject.copy(indexContext.lowerProject.getTraitSet(), inputs);
         }
-        if (indexContext.getSort() != null) {
-          finalRel = AbstractIndexPlanGenerator.getSortNode(indexContext, finalRel, true,false,
+
+        finalRel = AbstractIndexPlanGenerator.getSortNode(indexContext, finalRel, true,false,
                   indexContext.exch != null);
-        }
 
         if (finalRel == null) {
+          logger.debug("Not able to generate index plan in ", this.getClass().toString());
           return;
         }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillSortRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillSortRel.java
index cfa0e26b4..1e380cff7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillSortRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillSortRel.java
@@ -26,6 +26,7 @@ import org.apache.drill.common.logical.data.LogicalOperator;
 import org.apache.drill.common.logical.data.Order;
 import org.apache.drill.common.logical.data.Order.Ordering;
 import org.apache.drill.exec.planner.torel.ConversionContext;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.calcite.rel.InvalidRelException;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelFieldCollation;
@@ -41,7 +42,7 @@ import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
 /**
  * Sort implemented in Drill.
  */
-public class DrillSortRel extends Sort implements DrillRel {
+public class DrillSortRel extends Sort implements DrillRel,OrderedRel {
 
   /** Creates a DrillSortRel. */
   public DrillSortRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelCollation collation) {
@@ -98,4 +99,18 @@ public class DrillSortRel extends Sort implements DrillRel {
     return new DrillSortRel(context.getCluster(), context.getLogicalTraits(), input, RelCollations.of(collations));
   }
 
+  @Override
+  public RexNode getOffset() {
+    return offset;
+  }
+
+  @Override
+  public RexNode getFetch() {
+    return fetch;
+  }
+
+  @Override
+  public boolean canBeDropped() {
+    return true;
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
index 77fb4c8bb..8064c4287 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
@@ -27,6 +27,7 @@ import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.physical.config.ExternalSort;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.cost.DrillCostBase;
 import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
 import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
@@ -40,16 +41,25 @@ import org.apache.calcite.plan.RelOptPlanner;
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rex.RexNode;
 
-public class SortPrel extends org.apache.calcite.rel.core.Sort implements Prel {
+public class SortPrel extends org.apache.calcite.rel.core.Sort implements OrderedRel,Prel {
+  private final boolean isRemovable;
 
   /** Creates a DrillSortRel. */
   public SortPrel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelCollation collation) {
     super(cluster, traits, input, collation);
+    isRemovable = true;
   }
 
   /** Creates a DrillSortRel with offset and fetch. */
   public SortPrel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelCollation collation, RexNode offset, RexNode fetch) {
     super(cluster, traits, input, collation, offset, fetch);
+    isRemovable = true;
+  }
+
+  /** Creates a DrillSortRel. */
+  public SortPrel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelCollation collation, boolean isRemovable) {
+    super(cluster, traits, input, collation);
+    this.isRemovable = isRemovable;
   }
 
   @Override
@@ -141,4 +151,20 @@ public class SortPrel extends org.apache.calcite.rel.core.Sort implements Prel {
 
     return this.copy(traits, children.get(0), collationTrait, this.offset, this.fetch);
   }
+
+  @Override
+  public RexNode getOffset() {
+    return offset;
+  }
+
+  @Override
+  public RexNode getFetch() {
+    return fetch;
+  }
+
+  @Override
+  public boolean canBeDropped() {
+    return isRemovable;
+  }
+
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrule.java
index 3fc86b3d4..bec1b6a20 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrule.java
@@ -47,26 +47,25 @@ public class SortPrule extends Prule{
 
   @Override
   public void onMatch(RelOptRuleCall call) {
-    final DrillSortRel sort = (DrillSortRel) call.rel(0);
+    final DrillSortRel sort = call.rel(0);
     final RelNode input = sort.getInput();
 
     // Keep the collation in logical sort. Convert input into a RelNode with 1) this collation, 2) Physical, 3) hash distributed on
 
     DrillDistributionTrait hashDistribution =
-        new DrillDistributionTrait(DrillDistributionTrait.DistributionType.HASH_DISTRIBUTED, ImmutableList.copyOf(getDistributionField(sort)));
+            new DrillDistributionTrait(DrillDistributionTrait.DistributionType.HASH_DISTRIBUTED, ImmutableList.copyOf(getDistributionField(sort)));
 
-    final RelTraitSet traits = sort.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(hashDistribution);
-
-    final RelNode convertedInput = convert(input, traits);
+    final RelTraitSet traits = RelTraitSet.createEmpty().plus(Prel.DRILL_PHYSICAL).plus(hashDistribution);
+    SortPrel child = new SortPrel(sort.getCluster(), traits.plus(sort.getCollation()),
+            convert(sort.getInput(), traits), sort.getCollation(), false);
 
     if(isSingleMode(call)){
-      call.transformTo(convertedInput);
+      call.transformTo(child);
     }else{
-      RelNode exch = new SingleMergeExchangePrel(sort.getCluster(), sort.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(DrillDistributionTrait.SINGLETON), convertedInput, sort.getCollation());
+      RelNode exch = new SingleMergeExchangePrel(sort.getCluster(), sort.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(DrillDistributionTrait.SINGLETON), child, sort.getCollation());
       call.transformTo(exch);  // transform logical "sort" into "SingleMergeExchange".
 
     }
-
   }
 
   private List<DistributionField> getDistributionField(DrillSortRel rel) {
@@ -76,7 +75,6 @@ public class SortPrule extends Prule{
       DistributionField field = new DistributionField(relField.getFieldIndex());
       distFields.add(field);
     }
-
     return distFields;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/TopNPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/TopNPrel.java
index e9414f174..f8f4b9d13 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/TopNPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/TopNPrel.java
@@ -26,8 +26,11 @@ import org.apache.calcite.rel.RelCollationImpl;
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.physical.config.TopN;
+import org.apache.drill.exec.planner.common.OrderedRel;
 import org.apache.drill.exec.planner.cost.DrillCostBase;
 import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
 import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
@@ -40,7 +43,7 @@ import org.apache.calcite.plan.RelOptCost;
 import org.apache.calcite.plan.RelOptPlanner;
 import org.apache.calcite.plan.RelTraitSet;
 
-public class TopNPrel extends SinglePrel {
+public class TopNPrel extends SinglePrel implements OrderedRel,Prel {
 
   protected int limit;
   protected final RelCollation collation;
@@ -66,6 +69,28 @@ public class TopNPrel extends SinglePrel {
     return creator.addMetadata(this, topN);
   }
 
+  @Override
+  public RelCollation getCollation() {
+    return collation;
+  }
+
+  @Override
+  public RexNode getOffset() {
+    return getCluster().getRexBuilder().makeExactLiteral(BigDecimal.ZERO,
+                  getCluster().getTypeFactory().createSqlType(SqlTypeName.INTEGER));
+  }
+
+  @Override
+  public RexNode getFetch() {
+    return getCluster().getRexBuilder().makeExactLiteral(BigDecimal.valueOf(limit),
+                 getCluster().getTypeFactory().createSqlType(SqlTypeName.INTEGER));
+  }
+
+  @Override
+  public boolean canBeDropped() {
+    return true;
+  }
+
   /**
    * Cost of doing Top-N is proportional to M log N where M is the total number of
    * input rows and N is the limit for Top-N.  This makes Top-N preferable to Sort
@@ -93,6 +118,10 @@ public class TopNPrel extends SinglePrel {
         .item("limit", limit);
   }
 
+  public int getLimit() {
+    return limit;
+  }
+
   @Override
   public SelectionVectorMode[] getSupportedEncodings() {
     return SelectionVectorMode.NONE_AND_TWO;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/ExcessiveExchangeIdentifier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/ExcessiveExchangeIdentifier.java
index f77a4378e..fa8e69d0b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/ExcessiveExchangeIdentifier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/ExcessiveExchangeIdentifier.java
@@ -25,6 +25,11 @@ import org.apache.drill.exec.planner.physical.ExchangePrel;
 import org.apache.drill.exec.planner.physical.Prel;
 import org.apache.drill.exec.planner.physical.ScanPrel;
 import org.apache.drill.exec.planner.physical.ScreenPrel;
+import org.apache.drill.exec.planner.physical.LimitPrel;
+import org.apache.drill.exec.planner.physical.ProjectPrel;
+import org.apache.drill.exec.planner.physical.FilterPrel;
+import org.apache.drill.exec.planner.physical.SingleMergeExchangePrel;
+import org.apache.drill.exec.planner.physical.HashToMergeExchangePrel;
 import org.apache.calcite.rel.RelNode;
 
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
@@ -48,7 +53,18 @@ public class ExcessiveExchangeIdentifier extends BasePrelVisitor<Prel, Excessive
     parent.add(prel);
     MajorFragmentStat newFrag = new MajorFragmentStat();
     newFrag.setRightSideOfLateral(parent.isRightSideOfLateral());
+
+    if (prel instanceof SingleMergeExchangePrel) {
+      newFrag.isSimpleRel = true;
+    }
+
     Prel newChild = ((Prel) prel.getInput()).accept(this, newFrag);
+
+    if (parent.isSimpleRel &&
+        prel instanceof HashToMergeExchangePrel) {
+      return newChild;
+    }
+
     if (canRemoveExchange(parent, newFrag)) {
       return newChild;
     } else {
@@ -129,6 +145,8 @@ public class ExcessiveExchangeIdentifier extends BasePrelVisitor<Prel, Excessive
       s.add(p);
     }
 
+    s.setHashDistribution(prel);
+
     for(Prel p : prel) {
       children.add(p.accept(this, s));
     }
@@ -145,6 +163,9 @@ public class ExcessiveExchangeIdentifier extends BasePrelVisitor<Prel, Excessive
     private int maxWidth = Integer.MAX_VALUE;
     private boolean isMultiSubScan = false;
     private boolean rightSideOfLateral = false;
+    //This flag if true signifies that all the Rels thus far
+    //are simple rels with no distribution requirement.
+    private boolean isSimpleRel = false;
 
     public void add(Prel prel) {
       maxRows = Math.max(prel.estimateRowCount(prel.getCluster().getMetadataQuery()), maxRows);
@@ -162,6 +183,13 @@ public class ExcessiveExchangeIdentifier extends BasePrelVisitor<Prel, Excessive
       add(prel);
     }
 
+    public void setHashDistribution(Prel prel) {
+      isSimpleRel = isSimpleRel &&
+                    (prel instanceof LimitPrel ||
+                     prel instanceof ProjectPrel ||
+                     prel instanceof FilterPrel);
+    }
+
     public boolean isSingular() {
       // do not remove exchanges when a scan has more than one subscans (e.g. SystemTableScan)
       if (isMultiSubScan) {