Push down offset into TableScanNode when there is only one data region

Author: Wei-hao-Li

integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/old/query/IoTDBPaginationTableViewIT.java

@@ -214,5 +214,20 @@ public void limitOffsetPushDownTest() {
         expectedHeader,
         retArray,
         DATABASE_NAME);
+
+    retArray =
+        new String[] {
+          "1,",
+        };
+    tableResultSetEqualTest(
+        "select s1 from db order by time limit 1 offset 1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+    tableResultSetEqualTest(
+        "select s1 from db order by device limit 1 offset 1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PushDownOffsetIntoTableScan.java

@@ -0,0 +1,69 @@
+/*
+ * 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.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
+
+import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Capture;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Captures;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Pattern;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.offset;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.source;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.tableScan;
+import static org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Capture.newCapture;
+
+/**
+ * <b>Optimization phase:</b> Distributed plan planning.
+ *
+ * <p>The OFFSET can be eliminated when the following conditions are met:
+ * <li>Its child is DeviceTableScan, which means there OFFSET is effect on only one region
+ * <li>The query expressions are all scalar expression.
+ */
+public class PushDownOffsetIntoTableScan implements Rule<OffsetNode> {
+  private static final Capture<TableScanNode> CHILD = newCapture();
+
+  private static final Pattern<OffsetNode> PATTERN =
+      offset().with(source().matching(tableScan().capturedAs(CHILD)));
+
+  @Override
+  public Pattern<OffsetNode> getPattern() {
+    return PATTERN;
+  }
+
+  @Override
+  public Result apply(OffsetNode parent, Captures captures, Context context) {
+    TableScanNode tableScanNode = captures.get(CHILD);
+    if (tableScanNode instanceof DeviceTableScanNode
+        && !((DeviceTableScanNode) tableScanNode).isPushLimitToEachDevice()) {
+      tableScanNode.setPushDownOffset(parent.getCount());
+      // consider case that there is no limit
+      tableScanNode.setPushDownLimit(
+          tableScanNode.getPushDownLimit() == 0
+              ? 0
+              : tableScanNode.getPushDownLimit() - parent.getCount());
+      return Result.ofPlanNode(tableScanNode);
+    }
+
+    return Result.empty();
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DistributedOptimizeFactory.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.EliminateLimitWithTableScan;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.MergeLimitOverProjectWithMergeSort;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.MergeLimitWithMergeSort;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PushDownOffsetIntoTableScan;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -53,7 +54,9 @@ public DistributedOptimizeFactory(PlannerContext plannerContext) {
                 plannerContext,
                 ruleStats,
                 ImmutableSet.of(
-                    new EliminateLimitWithTableScan(), new EliminateLimitProjectWithTableScan())));
+                    new EliminateLimitWithTableScan(),
+                    new EliminateLimitProjectWithTableScan(),
+                    new PushDownOffsetIntoTableScan())));
   }
 
   public List<PlanOptimizer> getPlanOptimizers() {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/LimitOffsetPushDownTest.java

@@ -27,6 +27,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.TableLogicalPlanner;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.TableDistributedPlanner;
@@ -51,6 +52,9 @@
 import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestUtils.assertTableScan;
 import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestUtils.assertTableScanWithoutEntryOrder;
 import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestUtils.getChildrenNode;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.output;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.tableScan;
 import static org.apache.iotdb.db.queryengine.plan.statement.component.Ordering.ASC;
 import static org.apache.iotdb.db.queryengine.plan.statement.component.Ordering.DESC;
 import static org.junit.Assert.assertEquals;
@@ -303,4 +307,13 @@ public void limitDiffProjectTest() {
     assertTrue(getChildrenNode(rootNode, 2) instanceof LimitNode);
     assertTrue(getChildrenNode(rootNode, 3) instanceof DeviceTableScanNode);
   }
+
+  @Test
+  public void PushDownOffsetIntoTableScan() {
+    PlanTester planTester = new PlanTester();
+    sql = "select * from table1 where tag1='Beijing' and tag2='A1' limit 1 offset 1";
+    logicalQueryPlan = planTester.createPlan(sql);
+    // the offset node has been push down into TableScanNode
+    assertPlan(planTester.getFragmentPlan(0), output(tableScan("testdb.table1")));
+  }
 }