implement the except (distinct | all ) for table model

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

@@ -0,0 +1,159 @@
+/*
+ * 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.relational.it.query.recent;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBExceptTableIT {
+  protected static final String DATABASE_NAME = "test";
+  protected static final String[] createSqls =
+      new String[] {
+        "CREATE DATABASE " + DATABASE_NAME,
+        "USE " + DATABASE_NAME,
+        // table1:  ('d1', 1, 1) * 3, ('d1', 2, 2) * 1
+        "create table table1(device STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)",
+        "insert into table1 values (1, 'd1', 1, 1)",
+        "insert into table1 values (2, 'd1', 1, 1)",
+        "insert into table1 values (3, 'd1', 1, 1)",
+        "insert into table1 values (4, 'd1', 2, 2)",
+        // table2: ('d1', 1, 1.0) * 1, ('d1', 3, 3.0) * 1
+        "create table table2(device STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD)",
+        "insert into table2 values (1, 'd1', 1, 1.0)",
+        "insert into table2 values (2, 'd1', 3, 3.0)",
+        // table3: use for testing alias, mirrors table2
+        "create table table3(device STRING TAG, s1_testName INT64 FIELD, s2_testName DOUBLE FIELD)",
+        "insert into table3 values (1, 'd1', 1, 1.0)",
+        "insert into table3 values (2, 'd1', 3, 3.0)",
+        // table4: test type compatible
+        "create table table4(device STRING TAG, s1 TEXT FIELD, s2 DOUBLE FIELD)"
+      };
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+    prepareTableData(createSqls);
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void normalTest() {
+    String[] expectedHeader = new String[] {"device", "s1", "s2"};
+
+    // --- EXCEPT (DISTINCT) ---
+    // table1 (distinct): {('d1', 1, 1.0), ('d1', 2, 2.0)}
+    // table2 (distinct): {('d1', 1, 1.0), ('d1', 3, 3.0)}
+    // expected one tuple : ('d1', 2, 2.0)
+    String[] retArray =
+        new String[] {
+          "d1,2,2.0,",
+        };
+    tableResultSetEqualTest(
+        "select device, s1, s2 from table1 except select device, s1, s2 from table2",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+    tableResultSetEqualTest(
+        "select device, s1, s2 from table1 except distinct select device, s1, s2 from table2",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+
+    // --- EXCEPT ALL ---
+    // Row ('d1', 1, 1.0): table1 has 3, table2 has 1. max(0, 3 - 1) = 2 tuples.
+    // Row ('d1', 2, 2.0): table1 has 1, table2 has 0. max(0, 1 - 0) = 1 tuple.
+    // Row ('d1', 3, 3.0): table1 has 0, table2 has 1. max(0, 0 - 1) = 0 tuples.
+    // expected: 2 * ('d1', 1, 1.0) and 1 * ('d1', 2, 2.0)
+    retArray = new String[] {"d1,1,1.0,", "d1,1,1.0,", "d1,2,2.0,"};
+    tableResultSetEqualTest(
+        "select device, s1, s2 from table1 except all select device, s1, s2 from table2",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+    // test table3, the column name is different
+    tableResultSetEqualTest(
+        "select device, s1, s2 from table1 except all select device, s1_testName, s2_testName from table3",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void mappingTest() {
+    // table1 (aliased): (s1 as col_a) -> (1), (1), (1), (2) -> { (1.0) * 3, (2.0) * 1 }
+    // common value: (1.0)
+
+    String[] expectedHeader = new String[] {"col_a"};
+
+    // --- EXCEPT (DISTINCT) with alias ---
+    // t1_distinct = {1.0, 2.0}
+    // t2_distinct = {1.0, 3.0}
+    // Result: {2.0}
+    String[] retArray = new String[] {"2.0,"};
+    tableResultSetEqualTest(
+        "select col_a from ((select s1 as col_a, device as col_b from table1) except (select s2, device from table2)) order by col_a",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+
+    // --- EXCEPT ALL with alias ---
+    // Row (1.0): t1 has 3, t2 has 1. max(0, 3 - 1) = 2.
+    // Row (2.0): t1 has 1, t2 has 0. max(0, 1 - 0) = 1.
+    // Result: {1.0, 1.0, 2.0} (query has order by)
+    retArray = new String[] {"1.0,", "1.0,", "2.0,"};
+    tableResultSetEqualTest(
+        "select col_a from ((select s1 as col_a, device as col_b from table1) except all (select s2, device from table2)) order by col_a",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void exceptionTest() {
+    // type is incompatible (INT32 vs TEXT)
+    tableAssertTestFail(
+        "(select * from table1) except all (select * from table4)",
+        "has incompatible types: INT32, TEXT",
+        DATABASE_NAME);
+
+    tableAssertTestFail(
+        "(select * from table1) except all (select time from table4)",
+        "EXCEPT query has different number of fields: 4, 1",
+        DATABASE_NAME);
+  }
+}

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

@@ -71,6 +71,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AssignUniqueId;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExceptNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
@@ -1116,6 +1117,15 @@ public List<String> visitIntersect(IntersectNode node, GraphContext context) {
     return render(node, boxValue, context);
   }
 
+  @Override
+  public List<String> visitExcept(ExceptNode node, GraphContext context) {
+    List<String> boxValue = new ArrayList<>();
+    boxValue.add(String.format("Except-%s", node.getPlanNodeId().getId()));
+    boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
+    boxValue.add(String.format("isDistinct: %s", node.isDistinct()));
+    return render(node, boxValue, context);
+  }
+
   private List<String> render(PlanNode node, List<String> nodeBoxString, GraphContext context) {
     Box box = new Box(nodeBoxString);
     List<List<String>> children = new ArrayList<>();

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

@@ -119,6 +119,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AssignUniqueId;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExceptNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GroupNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
@@ -314,6 +315,7 @@ public enum PlanNodeType {
   TABLE_INTO_NODE((short) 1033),
   TABLE_UNION_NODE((short) 1034),
   TABLE_INTERSECT_NODE((short) 1035),
+  TABLE_EXCEPT_NODE((short) 1036),
 
   RELATIONAL_INSERT_TABLET((short) 2000),
   RELATIONAL_INSERT_ROW((short) 2001),
@@ -705,6 +707,8 @@ public static PlanNode deserialize(ByteBuffer buffer, short nodeType) {
         return UnionNode.deserialize(buffer);
       case 1035:
         return IntersectNode.deserialize(buffer);
+      case 1036:
+        return ExceptNode.deserialize(buffer);
       case 2000:
         return RelationalInsertTabletNode.deserialize(buffer);
       case 2001:

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

@@ -123,6 +123,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.GroupReference;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExceptNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GroupNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
@@ -845,4 +846,8 @@ public R visitUnion(UnionNode node, C context) {
   public R visitIntersect(IntersectNode node, C context) {
     return visitPlan(node, context);
   }
+
+  public R visitExcept(ExceptNode node, C context) {
+    return visitPlan(node, context);
+  }
 }

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

@@ -55,6 +55,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.TreeDeviceViewSchema;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExceptNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.IntersectNode;
@@ -1164,6 +1165,24 @@ protected RelationPlan visitIntersect(Intersect node, Void context) {
         intersectNode, analysis.getScope(node), intersectNode.getOutputSymbols(), outerContext);
   }
 
+  @Override
+  protected RelationPlan visitExcept(Except node, Void context) {
+    Preconditions.checkArgument(
+        !node.getRelations().isEmpty(), "No relations specified for except");
+    SetOperationPlan setOperationPlan = process(node);
+
+    PlanNode exceptNode =
+        new ExceptNode(
+            idAllocator.genPlanNodeId(),
+            setOperationPlan.getChildren(),
+            setOperationPlan.getSymbolMapping(),
+            ImmutableList.copyOf(setOperationPlan.getSymbolMapping().keySet()),
+            node.isDistinct());
+
+    return new RelationPlan(
+        exceptNode, analysis.getScope(node), exceptNode.getOutputSymbols(), outerContext);
+  }
+
   private SetOperationPlan process(SetOperation node) {
     RelationType outputFields = analysis.getOutputDescriptor(node);
     List<Symbol> outputs =
@@ -1210,11 +1229,6 @@ protected RelationPlan visitValues(Values node, Void context) {
     throw new IllegalStateException("Values is not supported in current version.");
   }
 
-  @Override
-  protected RelationPlan visitExcept(Except node, Void context) {
-    throw new IllegalStateException("Except is not supported in current version.");
-  }
-
   @Override
   protected RelationPlan visitInsertTablet(InsertTablet node, Void context) {
     final InsertTabletStatement insertTabletStatement = node.getInnerTreeStatement();

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

@@ -0,0 +1,112 @@
+/*
+ * 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.commons.udf.builtin.relational.TableBuiltinScalarFunction;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExceptNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ArithmeticBinaryExpression;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ComparisonExpression;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FunctionCall;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.GenericLiteral;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.QualifiedName;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Captures;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Pattern;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.tsfile.read.common.type.LongType;
+
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.Except.distinct;
+import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ArithmeticBinaryExpression.Operator.SUBTRACT;
+
+public class ImplementExceptAll implements Rule<ExceptNode> {
+
+  private static final Pattern<ExceptNode> PATTERN =
+      Patterns.except().with(distinct().equalTo(false));
+
+  private final Metadata metadata;
+
+  public ImplementExceptAll(Metadata metadata) {
+    this.metadata = requireNonNull(metadata, "metadata is null");
+  }
+
+  @Override
+  public Pattern<ExceptNode> getPattern() {
+    return PATTERN;
+  }
+
+  @Override
+  public Result apply(ExceptNode node, Captures captures, Context context) {
+
+    SetOperationNodeTranslator translator =
+        new SetOperationNodeTranslator(
+            metadata, context.getSymbolAllocator(), context.getIdAllocator());
+
+    // 1. translate the except(all) node to other planNodes
+    SetOperationNodeTranslator.TranslationResult translationResult =
+        translator.makeSetContainmentPlanForAll(node);
+
+    checkState(
+        !translationResult.getCountSymbols().isEmpty(),
+        "ExceptNode translation result has no count symbols");
+
+    // 2. add the filter node above the result node from translation process
+    // filter condition : row_number <= greatest(...greatest((greatest(count1 - count2, 0) - count3,
+    // 0))....)
+    Expression minusCount = translationResult.getCountSymbols().get(0).toSymbolReference();
+    QualifiedName greatest =
+        QualifiedName.of(TableBuiltinScalarFunction.GREATEST.getFunctionName());
+    for (int i = 1; i < translationResult.getCountSymbols().size(); i++) {
+      minusCount =
+          new FunctionCall(
+              greatest,
+              ImmutableList.of(
+                  new ArithmeticBinaryExpression(
+                      SUBTRACT,
+                      minusCount,
+                      translationResult.getCountSymbols().get(i).toSymbolReference()),
+                  new GenericLiteral(LongType.INT64.getDisplayName(), "0")));
+    }
+
+    FilterNode filterNode =
+        new FilterNode(
+            context.getIdAllocator().genPlanNodeId(),
+            translationResult.getPlanNode(),
+            new ComparisonExpression(
+                ComparisonExpression.Operator.LESS_THAN_OR_EQUAL,
+                translationResult.getRowNumberSymbol().toSymbolReference(),
+                minusCount));
+
+    // 3. add the project node to remove the redundant columns
+    return Result.ofPlanNode(
+        new ProjectNode(
+            context.getIdAllocator().genPlanNodeId(),
+            filterNode,
+            Assignments.identity(node.getOutputSymbols())));
+  }
+}