[Table Model Subquery] Support uncorrelated quantified comparison

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

@@ -254,6 +254,8 @@ public static TableAccumulator createBuiltinAccumulator(
     switch (aggregationType) {
       case COUNT:
         return new CountAccumulator();
+      case COUNT_ALL:
+        return new CountAllAccumulator();
       case COUNT_IF:
         return new CountIfAccumulator();
       case AVG:

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

@@ -0,0 +1,97 @@
+/*
+ * 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.execution.operator.source.relational.aggregation;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+public class CountAllAccumulator implements TableAccumulator {
+  private static final long INSTANCE_SIZE =
+      RamUsageEstimator.shallowSizeOfInstance(CountAllAccumulator.class);
+  private long countState = 0;
+
+  @Override
+  public long getEstimatedSize() {
+    return INSTANCE_SIZE;
+  }
+
+  @Override
+  public TableAccumulator copy() {
+    return new CountAllAccumulator();
+  }
+
+  @Override
+  public void addInput(Column[] arguments, AggregationMask mask) {
+    checkArgument(arguments.length == 1, "argument of CountAll should be one column");
+    int count = mask.getSelectedPositionCount();
+    countState += count;
+  }
+
+  @Override
+  public void removeInput(Column[] arguments) {
+    checkArgument(arguments.length == 1, "argument of Count should be one column");
+    int count = arguments[0].getPositionCount();
+    countState -= count;
+  }
+
+  @Override
+  public void addIntermediate(Column argument) {
+    for (int i = 0; i < argument.getPositionCount(); i++) {
+      if (argument.isNull(i)) {
+        continue;
+      }
+      countState += argument.getLong(i);
+    }
+  }
+
+  @Override
+  public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+    columnBuilder.writeLong(countState);
+  }
+
+  @Override
+  public void evaluateFinal(ColumnBuilder columnBuilder) {
+    columnBuilder.writeLong(countState);
+  }
+
+  @Override
+  public boolean hasFinalResult() {
+    return false;
+  }
+
+  @Override
+  public void addStatistics(Statistics[] statistics) {
+    throw new UnsupportedOperationException("CountAllAccumulator does not support statistics.");
+  }
+
+  @Override
+  public void reset() {
+    countState = 0;
+  }
+
+  @Override
+  public boolean removable() {
+    return true;
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java

@@ -493,7 +493,7 @@ protected ColumnTransformer visitGenericLiteral(GenericLiteral node, Context con
     return res;
   }
 
-  // currently, we only support Date and Timestamp
+  // currently, we only support Date/Timestamp/INT64
   // for Date, GenericLiteral.value is an int value
   // for Timestamp, GenericLiteral.value is a long value
   private static ConstantColumnTransformer getColumnTransformerForGenericLiteral(
@@ -506,6 +506,10 @@ private static ConstantColumnTransformer getColumnTransformerForGenericLiteral(
       return new ConstantColumnTransformer(
           TimestampType.TIMESTAMP,
           new LongColumn(1, Optional.empty(), new long[] {Long.parseLong(literal.getValue())}));
+    } else if (INT64.getTypeEnum().name().equals(literal.getType())) {
+      return new ConstantColumnTransformer(
+          INT64,
+          new LongColumn(1, Optional.empty(), new long[] {Long.parseLong(literal.getValue())}));
     } else {
       throw new SemanticException("Unsupported type in GenericLiteral: " + literal.getType());
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java

@@ -1533,9 +1533,11 @@ public Operator visitSemiJoin(SemiJoinNode node, LocalExecutionPlanContext conte
 
     Type sourceJoinKeyType =
         context.getTypeProvider().getTableModelType(node.getSourceJoinSymbol());
+
     checkIfJoinKeyTypeMatches(
         sourceJoinKeyType,
         context.getTypeProvider().getTableModelType(node.getFilteringSourceJoinSymbol()));
+
     OperatorContext operatorContext =
         context
             .getDriverContext()

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java

@@ -628,6 +628,7 @@ && isIntegerNumber(argumentTypes.get(2)))) {
     // get return type
     switch (functionName.toLowerCase(Locale.ENGLISH)) {
       case SqlConstant.COUNT:
+      case SqlConstant.COUNT_ALL:
       case SqlConstant.COUNT_IF:
         return INT64;
       case SqlConstant.FIRST_AGGREGATION:

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

@@ -340,7 +340,7 @@ protected Type visitLongLiteral(LongLiteral node, Context context) {
           && node.getParsedValue() <= Integer.MAX_VALUE) {
         return setExpressionType(node, INT32);
       }
-
+      // keep the original type
       return setExpressionType(node, INT64);
     }
 
@@ -361,6 +361,8 @@ protected Type visitGenericLiteral(GenericLiteral node, Context context) {
         type = DateType.DATE;
       } else if (TimestampType.TIMESTAMP.getTypeEnum().name().equals(node.getType())) {
         type = TimestampType.TIMESTAMP;
+      } else if (INT64.getTypeEnum().name().equals(node.getType())) {
+        type = INT64;
       } else {
         throw new SemanticException("Unsupported type in GenericLiteral: " + node.getType());
       }

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

@@ -0,0 +1,88 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+
+import com.google.common.collect.ImmutableList;
+
+import static com.google.common.base.Verify.verifyNotNull;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.ChildReplacer.replaceChildren;
+
+public abstract class SimplePlanRewriter<C>
+    extends PlanVisitor<PlanNode, SimplePlanRewriter.RewriteContext<C>> {
+  public static <C> PlanNode rewriteWith(SimplePlanRewriter<C> rewriter, PlanNode node) {
+    return node.accept(rewriter, new RewriteContext<>(rewriter, null));
+  }
+
+  public static <C> PlanNode rewriteWith(SimplePlanRewriter<C> rewriter, PlanNode node, C context) {
+    return node.accept(rewriter, new RewriteContext<>(rewriter, context));
+  }
+
+  @Override
+  public PlanNode visitPlan(PlanNode node, RewriteContext<C> context) {
+    return context.defaultRewrite(node, context.get());
+  }
+
+  public static class RewriteContext<C> {
+    private final C userContext;
+    private final SimplePlanRewriter<C> nodeRewriter;
+
+    private RewriteContext(SimplePlanRewriter<C> nodeRewriter, C userContext) {
+      this.nodeRewriter = nodeRewriter;
+      this.userContext = userContext;
+    }
+
+    public C get() {
+      return userContext;
+    }
+
+    /**
+     * Invoke the rewrite logic recursively on children of the given node and swap it out with an
+     * identical copy with the rewritten children
+     */
+    public PlanNode defaultRewrite(PlanNode node) {
+      return defaultRewrite(node, null);
+    }
+
+    /**
+     * Invoke the rewrite logic recursively on children of the given node and swap it out with an
+     * identical copy with the rewritten children
+     */
+    public PlanNode defaultRewrite(PlanNode node, C context) {
+      ImmutableList.Builder<PlanNode> children =
+          ImmutableList.builderWithExpectedSize(node.getChildren().size());
+      node.getChildren().forEach(source -> children.add(rewrite(source, context)));
+      return replaceChildren(node, children.build());
+    }
+
+    /** This method is meant for invoking the rewrite logic on children while processing a node */
+    public PlanNode rewrite(PlanNode node, C userContext) {
+      PlanNode result = node.accept(nodeRewriter, new RewriteContext<>(nodeRewriter, userContext));
+      return verifyNotNull(result, "nodeRewriter returned null for %s", node.getClass().getName());
+    }
+
+    /** This method is meant for invoking the rewrite logic on children while processing a node */
+    public PlanNode rewrite(PlanNode node) {
+      return rewrite(node, null);
+    }
+  }
+}

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

@@ -208,13 +208,21 @@ public LogicalOptimizeFactory(PlannerContext plannerContext) {
         new UnaliasSymbolReferences(plannerContext.getMetadata()),
         columnPruningOptimizer,
         inlineProjectionLimitFiltersOptimizer,
+        new TransformQuantifiedComparisonApplyToCorrelatedJoin(metadata),
         new IterativeOptimizer(
             plannerContext,
             ruleStats,
             ImmutableSet.of(
                 new RemoveRedundantEnforceSingleRowNode(), new RemoveUnreferencedScalarSubqueries(),
                 new TransformUncorrelatedSubqueryToJoin(),
                     new TransformUncorrelatedInPredicateSubqueryToSemiJoin())),
+        new IterativeOptimizer(
+            plannerContext,
+            ruleStats,
+            ImmutableSet.of(
+                new InlineProjections(plannerContext), new RemoveRedundantIdentityProjections()
+                /*new TransformCorrelatedSingleRowSubqueryToProject(),
+                new RemoveAggregationInSemiJoin())*/ )),
         new CheckSubqueryNodesAreRewritten(),
         new IterativeOptimizer(
             plannerContext, ruleStats, ImmutableSet.of(new PruneDistinctAggregation())),

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

@@ -212,7 +212,7 @@ public PlanNode visitFilter(FilterNode node, RewriteContext context) {
       Expression predicate = combineConjuncts(node.getPredicate(), context.inheritedPredicate);
 
       // when exist diff function, predicate can not be pushed down into DeviceTableScanNode
-      if (containsDiffFunction(predicate)) {
+      if (containsDiffFunction(predicate) || canNotPushDownBelowProjectNode(node, predicate)) {
         node.setChild(node.getChild().accept(this, new RewriteContext()));
         node.setPredicate(predicate);
         context.inheritedPredicate = TRUE_LITERAL;
@@ -234,6 +234,35 @@ public PlanNode visitFilter(FilterNode node, RewriteContext context) {
       return node;
     }
 
+    private boolean canNotPushDownBelowProjectNode(FilterNode node, Expression predicate) {
+      PlanNode child = node.getChild();
+      if (child instanceof ProjectNode) {
+        // if the inheritedPredicate is not in the output of the child of ProjectNode, we can not
+        // push it down for now.
+        // (predicate will be computed by the ProjectNode, Trino will rewrite the predicate in
+        // visitProject, but we have not implemented this for now.)
+        Set<Symbol> outputSymbolsOfProjectChild =
+            new HashSet<>(((ProjectNode) child).getChild().getOutputSymbols());
+        return missingTermsInOutputSymbols(predicate, outputSymbolsOfProjectChild);
+      }
+      return false;
+    }
+
+    private boolean missingTermsInOutputSymbols(Expression expression, Set<Symbol> outputSymbols) {
+      if (expression instanceof SymbolReference) {
+        return !outputSymbols.contains(Symbol.from(expression));
+      }
+      if (!expression.getChildren().isEmpty()) {
+        for (Node node : expression.getChildren()) {
+          if (missingTermsInOutputSymbols((Expression) node, outputSymbols)) {
+            return true;
+          }
+        }
+      }
+
+      return false;
+    }
+
     //    private boolean areExpressionsEquivalent(
     //        Expression leftExpression, Expression rightExpression) {
     //      return false;