Support aggregation push-down optimize for table view of non aligned series

integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryIT.java

@@ -349,8 +349,8 @@ public void test() throws Exception {
           false);
       compareQueryResults(
           session,
-          "select * from (select time, battery as device1 from view1 where battery = 'b1') as t1 full outer join (select time, battery as device2 from view2 where battery = 'b') as t2 using(time)",
-          "select * from (select time, battery as device1 from table1 where battery = 'b1') as t1 full outer join (select time, battery as device2 from table1 where battery = 'b') as t2 using(time)",
+          "select * from (select time, battery as battery1 from view1 where battery = 'b1') as t1 full outer join (select time, battery as battery2 from view2 where battery = 'b') as t2 using(time)",
+          "select * from (select time, battery as battery1 from table1 where battery = 'b1') as t1 full outer join (select time, battery as battery2 from table1 where battery = 'b') as t2 using(time)",
           true);
       compareQueryResults(
           session,
@@ -380,6 +380,48 @@ public void test() throws Exception {
           "select count(distinct battery) from view4 where battery = 'b1'",
           "select count(distinct battery) from table1 where battery = 'b1'",
           true);
+
+      compareQueryResults(
+          session,
+          "select count(time) from view1 where time > 604800000",
+          "select count(time) from table1 where time > 604800000",
+          true);
+
+      compareQueryResults(
+          session,
+          "select count(battery),count(time) from view1 where time > 604800000",
+          "select count(battery),count(time) from table1 where time > 604800000",
+          true);
+
+      compareQueryResults(
+          session,
+          "select count(battery),count(time),count(current) from view1 where time > 604800000",
+          "select count(battery),count(time),count(current) from table1 where time > 604800000",
+          true);
+
+      compareQueryResults(
+          session,
+          "select distinct battery from view1 where time > 604800000",
+          "select distinct battery from table1 where time > 604800000",
+          true);
+
+      compareQueryResults(
+          session,
+          "select count(current) from view1 where time > 604800000 group by battery",
+          "select count(current) from table1 where time > 604800000 group by battery",
+          true);
+
+      compareQueryResults(
+          session,
+          "select count(current) from view1 group by date_bin(2ms,time)",
+          "select count(current) from table1 group by date_bin(2ms,time)",
+          true);
+
+      compareQueryResults(
+          session,
+          "select count(current) from view1 group by battery,date_bin(2ms,time)",
+          "select count(current) from table1 group by battery,date_bin(2ms,time)",
+          true);
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggTableScanOperator.java

@@ -34,6 +34,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
 import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
 import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
 import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
@@ -71,7 +72,7 @@
 public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOperator {
 
   private boolean finished = false;
-  private TsBlock inputTsBlock;
+  protected TsBlock inputTsBlock;
 
   protected List<TableAggregator> tableAggregators;
   protected final List<ColumnSchema> groupingKeySchemas;
@@ -104,11 +105,11 @@
   // e.g. for aggregation `last(s1), count(s2), count(s1)`, the inputChannels should be [0, 1, 0]
   protected List<Integer> aggregatorInputChannels;
 
-  private QueryDataSource queryDataSource;
+  protected QueryDataSource queryDataSource;
 
   protected ITableTimeRangeIterator timeIterator;
 
-  private boolean allAggregatorsHasFinalResult = false;
+  protected boolean allAggregatorsHasFinalResult = false;
 
   protected AbstractAggTableScanOperator(AbstractAggTableScanOperatorParameter parameter) {
 
@@ -193,7 +194,7 @@
   }
 
   /** Return true if we have the result of this timeRange. */
-  protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() {
+  protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() throws Exception {
     try {
       if (calcFromCachedData()) {
         updateResultTsBlock();
@@ -706,7 +707,7 @@
     return true;
   }
 
-  private void checkIfAllAggregatorHasFinalResult() {
+  protected void checkIfAllAggregatorHasFinalResult() throws Exception {
     if (allAggregatorsHasFinalResult
         && (timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR
             || tableAggregators.isEmpty())) {
@@ -729,7 +730,7 @@
     }
   }
 
-  private void nextDevice() {
+  protected void nextDevice() throws Exception {
     currentDeviceIndex++;
     this.operatorContext.recordSpecifiedInfo(
         CURRENT_DEVICE_INDEX_STRING, Integer.toString(currentDeviceIndex));
@@ -812,6 +813,8 @@
     protected List<DeviceEntry> deviceEntries;
     protected int deviceCount;
 
+    private List<Symbol> outputSymbols;
+
     public AbstractAggTableScanOperatorParameter(
         PlanNodeId sourceId,
         OperatorContext context,
@@ -830,7 +833,8 @@
         boolean ascending,
         boolean canUseStatistics,
         List<Integer> aggregatorInputChannels,
-        String timeColumnName) {
+        String timeColumnName,
+        List<Symbol> outputSymbols) {
       this.sourceId = sourceId;
       this.context = context;
       this.aggColumnSchemas = aggColumnSchemas;
@@ -849,6 +853,11 @@
       this.canUseStatistics = canUseStatistics;
       this.aggregatorInputChannels = aggregatorInputChannels;
       this.timeColumnName = timeColumnName;
+      this.outputSymbols = outputSymbols;
+    }
+
+    public List<Symbol> getOutputSymbols() {
+      return outputSymbols;
     }
 
     public OperatorContext getOperatorContext() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DeviceIteratorScanOperator.java

@@ -113,7 +113,7 @@ private void constructCurrentDeviceOperatorTree() {
     }
     DeviceEntry deviceEntry = this.deviceEntries.get(this.currentDeviceIndex);
 
-    deviceChildOperatorTreeGenerator.generateCurrentDeviceOperatorTree(deviceEntry);
+    deviceChildOperatorTreeGenerator.generateCurrentDeviceOperatorTree(deviceEntry, true);
     currentDeviceRootOperator = deviceChildOperatorTreeGenerator.getCurrentDeviceRootOperator();
     dataSourceOperators = deviceChildOperatorTreeGenerator.getCurrentDeviceDataSourceOperators();
     currentDeviceInit = false;
@@ -188,6 +188,10 @@ public long ramBytesUsed() {
         + RamUsageEstimator.sizeOfCollection(deviceEntries);
   }
 
+  public DeviceChildOperatorTreeGenerator getDeviceChildOperatorTreeGenerator() {
+    return deviceChildOperatorTreeGenerator;
+  }
+
   public static class TreeNonAlignedDeviceViewScanParameters {
     public final OperatorContext context;
     public final List<DeviceEntry> deviceEntries;
@@ -217,7 +221,7 @@ public interface DeviceChildOperatorTreeGenerator {
     boolean keepOffsetAndLimitOperatorAfterDeviceIterator();
 
     // Generate the following operator subtree based on the current deviceEntry
-    void generateCurrentDeviceOperatorTree(DeviceEntry deviceEntry);
+    void generateCurrentDeviceOperatorTree(DeviceEntry deviceEntry, boolean needAdaptor);
 
     // Returns the root operator of the subtree
     Operator getCurrentDeviceRootOperator();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java

@@ -156,7 +156,7 @@
   }
 
   /** Main process logic, calc the last aggregation results of current device. */
-  private void processCurrentDevice() {
+  private void processCurrentDevice() throws Exception {
     if (currentHitCacheIndex < hitCachesIndexes.size()
         && outputDeviceIndex == hitCachesIndexes.get(currentHitCacheIndex)) {
       currentDeviceEntry = cachedDeviceEntries.get(currentHitCacheIndex);
@@ -175,7 +175,7 @@
     }
   }
 
   private void buildResultUseLastRowCache() {
     appendGroupKeysToResult(cachedDeviceEntries, currentHitCacheIndex);
     Pair<OptionalLong, TsPrimitiveType[]> currentHitResult =
         lastRowCacheResults.get(currentHitCacheIndex);