diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBCaseWhenThenTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBCaseWhenThenTableIT.java index 50053867636bd..d570efdcb8600 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBCaseWhenThenTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBCaseWhenThenTableIT.java @@ -23,6 +23,7 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; import org.junit.AfterClass; import org.junit.BeforeClass; @@ -31,13 +32,20 @@ import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData; import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest; import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail; import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest; +import static org.junit.Assert.assertEquals; // TODO:fix the different type exception @RunWith(IoTDBTestRunner.class) @@ -695,4 +703,337 @@ public void testThenWithBinarySameConstant() { String[] retArray = new String[] {"0,", "11,", "22,", "33,"}; tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE); } + + @Test + public void testCaseWhenReusesAggregationAcrossBlocks() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_aggregation_cache (dev STRING TAG, c DOUBLE FIELD, q INT32 FIELD)", + "INSERT INTO case_aggregation_cache(time, dev, c, q) VALUES " + + "(0, 'd0', 1.0, 0), (0, 'd1', 1.0, 0), (3600000, 'd1', 1.0, 0)" + }); + + // The non-pushdown predicate and the different device/window shapes make the aggregation + // produce multiple blocks. CASE clears the shared aggregate before the next output uses it. + String suffix = " FROM case_aggregation_cache WHERE q % 2 < 1 GROUP BY dev, date_bin(1h, time)"; + for (String projection : + new String[] { + "CASE WHEN sum(c) > 0 THEN 1.0 END AS case_result, CAST(sum(c) AS DOUBLE) AS total", + "CASE WHEN sum(c) > 0 THEN 1.0 END AS case_result, sum(c) + 0 AS total", + "CASE WHEN sum(c) > 0 THEN CAST(sum(c) AS DOUBLE) / CAST(sum(c) AS DOUBLE) END " + + "AS case_result, CAST(sum(c) AS DOUBLE) AS total" + }) { + tableResultSetEqualTest( + "SELECT " + projection + suffix, + new String[] {"case_result", "total"}, + new String[] {"1.0,1.0,", "1.0,1.0,", "1.0,1.0,"}, + DATABASE); + } + } + + @Test + public void testReusingCasePreservesPendingSiblingColumn() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_sibling_cache (dev STRING TAG, x DOUBLE FIELD)", + "INSERT INTO case_sibling_cache(time, dev, x) VALUES " + + "(0, 'd0', -3.0), (1, 'd0', 0.0), (2, 'd0', 2.0)" + }); + + String sharedCase = "CASE WHEN x > 0 THEN 1.0 ELSE 0.0 END"; + tableResultSetEqualTest( + "SELECT " + + sharedCase + + " AS a, x + CASE WHEN (" + + sharedCase + + ") > 0 THEN 2.0 ELSE 3.0 END AS b FROM case_sibling_cache ORDER BY time", + new String[] {"a", "b"}, + new String[] {"0.0,0.0,", "0.0,3.0,", "1.0,4.0,"}, + DATABASE); + } + + @Test + public void testSharedDiffAcrossCaseFilterAndProjection() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_shared_diff (dev STRING TAG, x DOUBLE FIELD)", + "INSERT INTO case_shared_diff(time, dev, x) VALUES " + + "(1, 'd0', 1.0), (2, 'd0', 3.0), (3, 'd0', 6.0)", + "FLUSH", + "INSERT INTO case_shared_diff(time, dev, x) VALUES (4, 'd0', 10.0), (5, 'd0', 11.0)", + "FLUSH" + }); + tableResultSetEqualTest( + "SELECT time, diff(x) AS d FROM case_shared_diff " + + "WHERE CASE WHEN diff(x) > 1.0 THEN 1.0 ELSE 0.0 END > 0.0", + new String[] {"time", "d"}, + new String[] { + "1970-01-01T00:00:00.002Z,2.0,", + "1970-01-01T00:00:00.003Z,3.0,", + "1970-01-01T00:00:00.004Z,4.0," + }, + DATABASE); + tableResultSetEqualTest( + "SELECT diff(x) AS d FROM case_shared_diff " + + "WHERE CASE WHEN diff(x) > 1.0 THEN true ELSE true END", + new String[] {"d"}, + new String[] {"null,", "2.0,", "3.0,", "4.0,", "1.0,"}, + DATABASE); + tableResultSetEqualTest( + "SELECT CASE WHEN x > 5 THEN diff(x) ELSE 100.0 END AS a, " + + "CASE WHEN x > 5 THEN 200.0 ELSE diff(x) END AS b, diff(x) AS d " + + "FROM case_shared_diff", + new String[] {"a", "b", "d"}, + new String[] { + "100.0,null,null,", "100.0,2.0,2.0,", "3.0,200.0,3.0,", "4.0,200.0,4.0,", "1.0,200.0,1.0," + }, + DATABASE); + } + + @Test + public void testDynamicIgnoreNullDiffAcrossCaseFilterAndProjection() throws Exception { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_dynamic_diff " + + "(dev STRING TAG, x DOUBLE FIELD, ignore_null BOOLEAN FIELD)", + "INSERT INTO case_dynamic_diff(time, dev, x, ignore_null) VALUES " + + "(1, 'd0', 1.0, true), (2, 'd0', 3.0, false), (3, 'd0', null, true), " + + "(4, 'd0', 10.0, null), (5, 'd0', null, false)", + "FLUSH", + "INSERT INTO case_dynamic_diff(time, dev, x, ignore_null) VALUES " + + "(6, 'd0', 20.0, true), (7, 'd0', 25.0, false), " + + "(8, 'd0', null, true), (9, 'd0', 30.0, true)", + "FLUSH" + }); + String diff = "diff(x, ignore_null)"; + tableResultSetEqualTest( + "SELECT " + diff + " AS d FROM case_dynamic_diff ORDER BY time", + new String[] {"d"}, + new String[] {"null,", "2.0,", "null,", "7.0,", "null,", "null,", "5.0,", "null,", "5.0,"}, + DATABASE); + // The masked CASE result is unspecified where it disagrees with the unmasked DIFF. + assertDefinedDiffValues( + "SELECT time, CASE WHEN x > 5 THEN " + + diff + + " ELSE 100.0 END AS a, CASE WHEN x > 5 THEN 200.0 ELSE " + + diff + + " END AS b, " + + diff + + " AS d FROM case_dynamic_diff ORDER BY time", + new String[] {"a", "b", "d"}, + Map.of( + 1L, new String[] {"100.0", "null", "null"}, + 2L, new String[] {"100.0", "2.0", "2.0"}, + 3L, new String[] {"100.0", "null", "null"}, + 4L, new String[] {null, "200.0", "7.0"}, + 5L, new String[] {"100.0", "null", "null"}, + 6L, new String[] {"null", "200.0", "null"}, + 7L, new String[] {"5.0", "200.0", "5.0"}, + 8L, new String[] {"100.0", "null", "null"}, + 9L, new String[] {null, "200.0", "5.0"})); + // Check the rows with unambiguous DIFF values. Do not require the current omissions at + // times 4 and 9: the masked binary-DIFF predicate disagrees with a full evaluation there. + assertDefinedDiffValues( + "SELECT time, " + + diff + + " AS d FROM case_dynamic_diff WHERE CASE WHEN x IS NULL THEN false WHEN " + + diff + + " > 1.0 THEN true ELSE false END ORDER BY time", + new String[] {"d"}, + Map.of(2L, new String[] {"2.0"}, 7L, new String[] {"5.0"})); + } + + @Test + public void testLocalDiffAndCaseCastGuards() throws Exception { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_partial_diff (dev STRING TAG, x DOUBLE FIELD, ignore_null BOOLEAN FIELD)", + "INSERT INTO case_partial_diff(time, dev, x, ignore_null) VALUES " + + "(1, 'd0', 1.0, true), (2, 'd0', -2.0, false), " + + "(3, 'd0', 3.0, true), (4, 'd0', -4.0, false)", + "CREATE TABLE case_guarded_diff (dev STRING TAG, src STRING FIELD)", + "INSERT INTO case_guarded_diff(time, dev, src) VALUES " + + "(1, 'd0', 'n/a'), (2, 'd0', '5'), (3, 'd0', '8')" + }); + for (String diff : new String[] {"diff(x)", "diff(x, ignore_null)"}) { + String sql = + "SELECT time, CASE WHEN x > 0 THEN x ELSE " + + diff + + " END AS a, " + + diff + + " AS d FROM case_partial_diff ORDER BY time"; + if (diff.equals("diff(x)")) { + assertDefinedDiffValues( + sql, + new String[] {"a", "d"}, + Map.of( + 1L, + new String[] {"1.0", "null"}, + 2L, + new String[] {"-3.0", "-3.0"}, + 3L, + new String[] {"3.0", "5.0"}, + 4L, + new String[] {"-7.0", "-7.0"})); + } else { + // A null expected entry leaves the known inconsistent masked CASE value unspecified. + // The unmasked DIFF and the CASE's ordinary branches still have exact expectations. + assertDefinedDiffValues( + sql, + new String[] {"a", "d"}, + Map.of( + 1L, + new String[] {"1.0", "null"}, + 2L, + new String[] {null, "-3.0"}, + 3L, + new String[] {"3.0", "5.0"}, + 4L, + new String[] {null, "-7.0"})); + } + } + tableResultSetEqualTest( + "SELECT CASE WHEN x > 0 THEN diff(diff(x)) ELSE 0.0 END AS a, " + + "diff(diff(x)) AS d FROM case_partial_diff ORDER BY time", + new String[] {"a", "d"}, + new String[] {"null,null,", "0.0,null,", "8.0,8.0,", "0.0,-12.0,"}, + DATABASE); + String guardedDiff = "diff(CASE WHEN src <> 'n/a' THEN CAST(src AS DOUBLE) END)"; + tableResultSetEqualTest( + "SELECT CASE WHEN src = 'n/a' THEN -1.0 ELSE " + + guardedDiff + + " END AS a, " + + guardedDiff + + " AS d FROM case_guarded_diff ORDER BY time", + new String[] {"a", "d"}, + new String[] {"-1.0,null,", "null,null,", "3.0,3.0,"}, + DATABASE); + } + + @Test + public void testBinaryDiffPreservesOuterCaseCastGuard() throws Exception { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_binary_guard (dev STRING TAG, src STRING FIELD, ignore_null BOOLEAN FIELD)", + "INSERT INTO case_binary_guard(time, dev, src, ignore_null) VALUES " + + "(1, 'd0', 'n/a', true), (2, 'd0', '5', true), (3, 'd0', '8', true)", + "FLUSH", + "INSERT INTO case_binary_guard(time, dev, src, ignore_null) VALUES " + + "(4, 'd0', 'n/a', true), (5, 'd0', '10', true), (6, 'd0', '14', true)", + "FLUSH" + }); + // Keep the CAST guard and well-defined differences covered without prescribing the + // inconsistent cross-block value at time 5, where masked and full DIFF evaluation disagree. + assertDefinedDiffValues( + "SELECT time, CASE WHEN src <> 'n/a' THEN diff(CAST(src AS DOUBLE), ignore_null) END AS d " + + "FROM case_binary_guard ORDER BY time", + new String[] {"d"}, + Map.of( + 1L, + new String[] {"null"}, + 2L, + new String[] {"null"}, + 3L, + new String[] {"3.0"}, + 4L, + new String[] {"null"}, + 5L, + new String[] {null}, + 6L, + new String[] {"4.0"})); + // Literal flags use the unary transformer, which already evaluates all input rows on master. + tableAssertTestFail( + "SELECT CASE WHEN src <> 'n/a' THEN diff(CAST(src AS DOUBLE), true) END " + + "FROM case_binary_guard ORDER BY time", + "701: Cannot cast n/a to DOUBLE type", + DATABASE); + } + + @Test + public void testBetweenRetainsSharedInputsAcrossCaseEvaluation() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_between_cache (dev STRING TAG, x DOUBLE FIELD)", + "INSERT INTO case_between_cache(time,dev,x) VALUES (1,'d',1),(2,'d',2),(3,'d',-1)" + }); + for (String expression : + new String[] { + "CASE WHEN x > 0 THEN x BETWEEN x - 1 AND x + 1 ELSE false END", + "CASE WHEN x > 0 THEN x BETWEEN x - 1 AND (x - 1) + 2 ELSE false END" + }) { + tableResultSetEqualTest( + "SELECT " + expression + " AS b FROM case_between_cache ORDER BY time", + new String[] {"b"}, + new String[] {"true,", "true,", "false,"}, + DATABASE); + } + tableResultSetEqualTest( + "SELECT x BETWEEN CASE WHEN x > 0 THEN 0.0 ELSE -10.0 END AND 10.0 AS b " + + "FROM case_between_cache ORDER BY time", + new String[] {"b"}, + new String[] {"true,", "true,", "true,"}, + DATABASE); + } + + @Test + public void testProjectionRetainsInputsClearedByCaseSibling() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE case_projection_inputs (dev STRING TAG, x DOUBLE FIELD, s STRING FIELD)", + "INSERT INTO case_projection_inputs(time,dev,x,s) VALUES " + + "(1,'d',1,'match'),(2,'d',2,'other'),(3,'d',-1,'match')" + }); + tableResultSetEqualTest( + "SELECT x + CASE WHEN x > 0 THEN 2.0 ELSE 3.0 END AS a " + + "FROM case_projection_inputs ORDER BY time", + new String[] {"a"}, + new String[] {"3.0,", "4.0,", "2.0,"}, + DATABASE); + tableResultSetEqualTest( + "SELECT coalesce(x, CASE WHEN x > 0 THEN x ELSE 0.0 END) AS a " + + "FROM case_projection_inputs ORDER BY time", + new String[] {"a"}, + new String[] {"1.0,", "2.0,", "-1.0,"}, + DATABASE); + tableResultSetEqualTest( + "SELECT concat(s, CASE WHEN s = 'match' THEN '-m' ELSE '-o' END) AS a " + + "FROM case_projection_inputs ORDER BY time", + new String[] {"a"}, + new String[] {"match-m,", "other-o,", "match-m,"}, + DATABASE); + } + + private static void assertDefinedDiffValues( + String sql, String[] columns, Map expectedRows) throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + Set checked = new HashSet<>(); + try (ResultSet result = statement.executeQuery(sql)) { + while (result.next()) { + long time = result.getLong("time"); + String[] expected = expectedRows.get(time); + if (expected == null) { + continue; + } + for (int i = 0; i < columns.length; i++) { + if (expected[i] != null) { + String actual = result.getString(columns[i]); + assertEquals(expected[i], actual == null ? "null" : actual); + } + } + checked.add(time); + } + } + assertEquals(expectedRows.keySet(), checked); + } + } } diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSharedDiffStateIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSharedDiffStateIT.java new file mode 100644 index 0000000000000..e417ee66ce457 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSharedDiffStateIT.java @@ -0,0 +1,206 @@ +/* + * 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.db.it; + +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.apache.iotdb.itbase.env.BaseEnv; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.Statement; +import java.util.Arrays; +import java.util.HashSet; + +import static org.apache.iotdb.db.it.utils.TestUtils.assertResultSetEqual; +import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData; +import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBSharedDiffStateIT { + private static final String DATABASE = "shared_diff_state"; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockLineNumber(3); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setSchemaRegionGroupExtensionPolicy("CUSTOM") + .setDefaultSchemaRegionGroupNumPerDatabase(1); + EnvFactory.getEnv().initClusterEnvironment(); + prepareTableData(new String[] {"CREATE DATABASE " + DATABASE}); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testProjectionOrderDoesNotChangeCrossBlockState() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE projection_state (id STRING TAG, x DOUBLE FIELD, f BOOLEAN FIELD)", + "INSERT INTO projection_state(time,id,x,f) VALUES " + + "(1,'d',1,true),(2,'d',2,true),(3,'d',3,true)", + "FLUSH", + "INSERT INTO projection_state(time,id,x,f) VALUES " + + "(4,'d',4,true),(5,'d',5,true),(6,'d',6,true)", + "FLUSH", + "INSERT INTO projection_state(time,id,x,f) VALUES (7,'d',7,true)", + "FLUSH" + }); + for (boolean caseFirst : new boolean[] {false, true}) { + String diff = "diff(x,f) AS d"; + String caseResult = "CASE WHEN x = 7 THEN diff(x,f) ELSE 100.0 END AS a"; + String[] expected = new String[7]; + for (int i = 0; i < expected.length; i++) { + String d = i == 0 ? "null" : "1.0"; + String a = i == 6 ? "1.0" : "100.0"; + expected[i] = caseFirst ? a + "," + d + "," : d + "," + a + ","; + } + tableResultSetEqualTest( + "SELECT " + + (caseFirst ? caseResult + "," + diff : diff + "," + caseResult) + + " FROM projection_state ORDER BY time", + caseFirst ? new String[] {"a", "d"} : new String[] {"d", "a"}, + expected, + DATABASE); + } + } + + @Test + public void testAdditionalOrTermDoesNotRemoveRowsAcrossBlocks() { + prepareTableData( + new String[] { + "USE " + DATABASE, + "CREATE TABLE filter_state (id STRING TAG, x DOUBLE FIELD, f BOOLEAN FIELD)", + "INSERT INTO filter_state(time,id,x,f) VALUES " + + "(1,'d',1,true),(2,'d',3,true),(3,'d',4,true)", + "FLUSH", + "INSERT INTO filter_state(time,id,x,f) VALUES (4,'d',10,true),(5,'d',11,true)", + "FLUSH" + }); + for (String predicate : + new String[] { + "diff(x,f) IS NOT NULL", + "diff(x,f) IS NOT NULL OR CASE WHEN x > 5 THEN diff(x,f) > 0 ELSE false END" + }) { + tableResultSetEqualTest( + "SELECT x FROM filter_state WHERE " + predicate + " ORDER BY time", + new String[] {"x"}, + new String[] {"3.0,", "4.0,", "10.0,", "11.0,"}, + DATABASE); + } + } + + @Test + public void testUpdateSetOrderDoesNotWriteMaskedNulls() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE update_state (id STRING TAG, src STRING ATTRIBUTE, " + + "a STRING ATTRIBUTE, b STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO update_state(time,id,src,a,b,v) VALUES " + + "(1,'d0','1','seed','seed',1),(2,'d1','9','seed','seed',1)," + + "(3,'d2','5','seed','seed',1)"); + for (boolean diffFirst : new boolean[] {false, true}) { + statement.execute("UPDATE update_state SET a = 'seed', b = 'seed'"); + String a = "a = CAST(CAST(src AS DOUBLE) AS STRING)"; + String b = "b = CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING)"; + statement.execute( + "UPDATE update_state SET " + + (diffFirst ? b + "," + a : a + "," + b) + + " WHERE CASE WHEN src = 'n/a' THEN false ELSE " + + "diff(CAST(src AS DOUBLE), src = 'x') > 0 END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM update_state"), + "id,src,a,b,", + new HashSet<>(Arrays.asList("d0,1,seed,seed,", "d1,9,9.0,8.0,", "d2,5,seed,seed,"))); + } + } + } + + @Test + public void testGuardedUpdateWithMatchingDeviceKeepsCaseProtection() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE guarded_update (id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO guarded_update(time,id,src,a,v) VALUES (1,'d0','1','seed',1),(2,'d1','5','seed',1),(3,'d2','n/a','seed',1)"); + statement.execute( + "UPDATE guarded_update SET a = CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING) WHERE CASE WHEN src = 'n/a' THEN false ELSE diff(CAST(src AS DOUBLE), src = 'x') > 0 END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM guarded_update"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,1,seed,", "d1,5,4.0,", "d2,n/a,seed,"))); + } + } + + @Test + public void testUpdateWritesTheDiffNullUsedByWhere() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE null_update (id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO null_update(time,id,src,a,v) VALUES (1,'d0','1','seed',1),(2,'d1','3','seed',1),(3,'d2','6','seed',1)"); + statement.execute( + "UPDATE null_update SET a = coalesce(CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING), 'NULL') WHERE CASE WHEN src = '3' THEN false ELSE diff(CAST(src AS DOUBLE), src = 'x') IS NULL END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM null_update"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,1,NULL,", "d1,3,seed,", "d2,6,NULL,"))); + } + } + + @Test + public void testUpdateBatchSplitDoesNotCreateMatches() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE split_update (id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO split_update(time,id,src,a,v) VALUES (1,'d0','10','seed',1),(2,'d1','20','seed',1),(3,'d2','0','seed',1),(4,'d3','5','seed',1)"); + statement.execute( + "UPDATE split_update SET a = CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING) WHERE CASE WHEN src = '0' THEN false ELSE diff(CAST(src AS DOUBLE), src = 'x') > 0 END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM split_update"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,10,seed,", "d1,20,10.0,", "d2,0,seed,", "d3,5,seed,"))); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedScalarFunctionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedScalarFunctionIT.java index 2d1b741a7c62b..71427db1ce623 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedScalarFunctionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedScalarFunctionIT.java @@ -112,6 +112,28 @@ public void testIllegalInput() { } } + @Test + public void testCaseThenBranchSelection() { + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("USE test"); + List expectedResult = Arrays.asList("1,false", "2,false", "3,true", "5,false"); + int row = 0; + try (ResultSet resultSet = + statement.executeQuery( + "select time, CASE WHEN s1 > 1 THEN contain_null(s3, s4) ELSE false END as a from vehicle")) { + while (resultSet.next()) { + Assert.assertEquals( + expectedResult.get(row), resultSet.getLong(1) + "," + resultSet.getString(2)); + row++; + } + assertEquals(4, row); + } + } catch (Exception e) { + fail(e.getMessage()); + } + } + @Test public void testNormalQuery() { try (Connection connection = EnvFactory.getEnv().getTableConnection(); diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBCaseDeviceCacheIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBCaseDeviceCacheIT.java new file mode 100644 index 0000000000000..1552a9e226fff --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBCaseDeviceCacheIT.java @@ -0,0 +1,252 @@ +/* + * 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.schema; + +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.apache.iotdb.itbase.env.BaseEnv; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; + +import static org.apache.iotdb.db.it.utils.TestUtils.assertResultSetEqual; +import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBCaseDeviceCacheIT { + private static final String DATABASE = "case_device_cache"; + private static final String CASE_RESULT = "CASE WHEN src = 'match' THEN 'match' ELSE 'other' END"; + + @BeforeClass + public static void setUp() throws Exception { + // CUSTOM fixes the region count; PROACTIVE treats the default count as a minimum. + // Keep devices in one schema region to exercise mixed-row and repeated batches. + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setSchemaRegionGroupExtensionPolicy("CUSTOM") + .setDefaultSchemaRegionGroupNumPerDatabase(1); + EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockLineNumber(2); + EnvFactory.getEnv().initClusterEnvironment(); + prepareTableData(new String[] {"CREATE DATABASE " + DATABASE}); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testDevicePredicatesAcrossBatches() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + createTable(statement, "predicate_cache"); + for (String predicate : new String[] {CASE_RESULT + " = src", "src = " + CASE_RESULT}) { + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM predicate_cache WHERE " + predicate), + "id,src,a,b,", + new HashSet<>( + Arrays.asList( + "d0,match,null,null,", + "d1,match,null,null,", + "d2,other,null,null,", + "d3,other,null,null,"))); + assertResultSetEqual( + statement.executeQuery("COUNT DEVICES FROM predicate_cache WHERE " + predicate), + "count(devices),", + Collections.singleton("4,")); + } + } + } + + @Test + public void testAttributeProjectionAcrossBatches() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + createTable(statement, "projection_cache"); + statement.execute("UPDATE projection_cache SET a = " + CASE_RESULT + ", b = src"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM projection_cache"), + "id,src,a,b,", + new HashSet<>( + Arrays.asList( + "d0,match,match,match,", + "d1,match,match,match,", + "d2,other,other,other,", + "d3,other,other,other,"))); + } + } + + @Test + public void testAttributeConcatRetainsInputClearedByCase() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + createTable(statement, "concat_cache"); + statement.execute( + "UPDATE concat_cache SET b = concat(src, " + + "CASE WHEN src = 'match' THEN '-m' ELSE '-o' END)"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM concat_cache"), + "id,src,a,b,", + new HashSet<>( + Arrays.asList( + "d0,match,null,match-m,", + "d1,match,null,match-m,", + "d2,other,null,other-o,", + "d3,other,null,other-o,"))); + } + } + + @Test + public void testAttributeCommonExpressionClearedByCase() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + createTable(statement, "common_cache"); + statement.execute( + "UPDATE common_cache SET a = upper(src) " + + "WHERE CASE WHEN upper(src) = 'MATCH' THEN true ELSE false END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM common_cache"), + "id,src,a,b,", + new HashSet<>( + Arrays.asList( + "d0,match,MATCH,null,", + "d1,match,MATCH,null,", + "d2,other,null,null,", + "d3,other,null,null,"))); + } + } + + @Test + public void testGuardedCommonCastInAttributeUpdate() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE guarded_cast " + + "(id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO guarded_cast(time, id, src, a, v) VALUES " + + "(1, 'd0', 'n/a', 'seed', 1), (2, 'd1', '5', 'seed', 1)"); + String update = + "UPDATE guarded_cast SET a = CAST(CAST(src AS INT32) + 1 AS STRING) " + + "WHERE CASE WHEN src <> 'n/a' THEN CAST(src AS INT32) > 10 ELSE false END"; + + statement.execute(update); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM guarded_cast"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,n/a,seed,", "d1,5,seed,"))); + + statement.execute("UPDATE guarded_cast SET src = '50' WHERE id = 'd1'"); + statement.execute(update); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM guarded_cast"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,n/a,seed,", "d1,50,51,"))); + } + } + + @Test + public void testAttributeCommonDiffClearedByCase() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE diff_cache " + + "(id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO diff_cache(time, id, src, a, v) VALUES " + + "(1, 'd1', '10', 'seed', 1), (2, 'd2', '20', 'seed', 1)"); + // The shared binary diff is stateful, so the guarded recompute must re-evaluate every row + // instead of only the filter-passing rows, which would read the masked rows as nulls. + statement.execute( + "UPDATE diff_cache SET a = CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING) " + + "WHERE CASE WHEN src = 'x' THEN true ELSE " + + "diff(CAST(src AS DOUBLE), src = 'x') > 5.0 END"); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM diff_cache"), + "id,src,a,", + new HashSet<>(Arrays.asList("d1,10,seed,", "d2,20,10.0,"))); + } + } + + @Test + public void testGuardedCommonDiffWithoutMatchingDevices() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE guarded_diff_no_match " + + "(id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO guarded_diff_no_match(time, id, src, a, v) VALUES " + + "(1, 'd0', 'n/a', 'seed', 1), (2, 'd1', '5', 'seed', 1), " + + "(3, 'd2', 'n/a', 'seed', 1)"); + String update = + "UPDATE guarded_diff_no_match " + + "SET a = CAST(diff(CAST(src AS DOUBLE), src = 'x') AS STRING) " + + "WHERE CASE WHEN src = 'n/a' THEN false ELSE " + + "diff(CAST(src AS DOUBLE), src = 'x') > 0 END"; + + // Cover both a full batch and the final partial batch flushed when the updater closes. + // No device matches, so neither batch may retry the guarded CAST on 'n/a'. + statement.execute(update); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM guarded_diff_no_match"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,n/a,seed,", "d1,5,seed,", "d2,n/a,seed,"))); + + statement.execute( + "UPDATE guarded_diff_no_match SET src = CASE id WHEN 'd0' THEN '10' ELSE '20' END"); + statement.execute(update); + assertResultSetEqual( + statement.executeQuery("SHOW DEVICES FROM guarded_diff_no_match"), + "id,src,a,", + new HashSet<>(Arrays.asList("d0,10,seed,", "d1,20,10.0,", "d2,20,seed,"))); + } + } + + private static void createTable(Statement statement, String table) throws SQLException { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE " + + table + + " (id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, b STRING ATTRIBUTE, v DOUBLE FIELD)"); + statement.execute( + "INSERT INTO " + + table + + "(time, id, src, v) VALUES (1, 'd0', 'match', 1), (2, 'd1', 'match', 1), " + + "(3, 'd2', 'other', 1), (4, 'd3', 'other', 1)"); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperator.java index 67d0d69874a36..978c2a1bf9de6 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperator.java @@ -67,6 +67,10 @@ public class FilterAndProjectOperator implements ProcessOperator { private final List projectOutputTransformerList; + private final ColumnTransformer[] filterCacheTransformers; + + private final ColumnTransformer[] projectCacheTransformers; + private final TsBlockBuilder filterTsBlockBuilder; private final boolean hasNonMappableUDF; @@ -100,6 +104,13 @@ public FilterAndProjectOperator( this.hasNonMappableUDF = hasNonMappableUDF; this.filterTsBlockBuilder = new TsBlockBuilder(8, filterOutputDataTypes); this.hasFilter = hasFilter; + List filterRoots = new ArrayList<>(commonTransformerList); + if (filterOutputTransformer != null) { + filterRoots.add(filterOutputTransformer); + } + this.filterCacheTransformers = ColumnTransformer.collectTransformers(filterRoots); + this.projectCacheTransformers = + ColumnTransformer.collectTransformers(projectOutputTransformerList); } public FilterAndProjectOperator( @@ -111,6 +122,8 @@ public FilterAndProjectOperator( this.projectLeafColumnTransformerList = filterAndProjectOperator.projectLeafColumnTransformerList; this.projectOutputTransformerList = filterAndProjectOperator.projectOutputTransformerList; + this.filterCacheTransformers = filterAndProjectOperator.filterCacheTransformers; + this.projectCacheTransformers = filterAndProjectOperator.projectCacheTransformers; this.hasNonMappableUDF = filterAndProjectOperator.hasNonMappableUDF; this.hasFilter = filterAndProjectOperator.hasFilter; this.filterTsBlockBuilder = filterAndProjectOperator.filterTsBlockBuilder; @@ -152,6 +165,12 @@ public TsBlock next() throws Exception { private TsBlock getFilterTsBlock(TsBlock input) { final Column originTimeColumn = input.getTimeColumn(); final int positionCount = originTimeColumn.getPositionCount(); + // Short-circuit evaluation can reset cache reference counts when a shared expression is + // recomputed. Invalidate the whole DAG before feeding a new block, including intermediate + // expressions whose caches may still have unconsumed references from the previous block. + for (ColumnTransformer columnTransformer : filterCacheTransformers) { + columnTransformer.clearCacheForNewInput(); + } // feed Filter ColumnTransformer, including TimeStampColumnTransformer and constant for (LeafColumnTransformer leafColumnTransformer : filterLeafColumnTransformerList) { leafColumnTransformer.initFromTsBlock(input); @@ -174,8 +193,8 @@ private TsBlock getFilterTsBlock(TsBlock input) { if (!hasNonMappableUDF) { // get result of calculated common sub expressions for (ColumnTransformer columnTransformer : commonTransformerList) { - // CASE WHEN clause would clear all its cache - // evaluate again to acquire cache + // CASE evaluation can clear a common expression after the filter used it. Recompute it + // before materializing the column for projection within this same input block. columnTransformer.tryEvaluate(); resultColumns.add(columnTransformer.getColumn()); } @@ -265,6 +284,11 @@ public static boolean satisfy(Column filterColumn, int rowIndex) { private TsBlock getTransformedTsBlock(TsBlock input) { final Column originTimeColumn = input.getTimeColumn(); final int positionCount = originTimeColumn.getPositionCount(); + // The flat array was deduplicated at construction; clear each node before evaluating any + // output. + for (ColumnTransformer columnTransformer : projectCacheTransformers) { + columnTransformer.clearCacheForNewInput(); + } // feed pre calculated data for (LeafColumnTransformer leafColumnTransformer : projectLeafColumnTransformerList) { leafColumnTransformer.initFromTsBlock(input); @@ -441,6 +465,8 @@ public long ramBytesUsed() { return INSTANCE_SIZE + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(inputOperator) + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext) + + RamUsageEstimator.shallowSizeOf(filterCacheTransformers) + + RamUsageEstimator.shallowSizeOf(projectCacheTransformers) + filterTsBlockBuilder.getRetainedSizeInBytes(); } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/AbstractCaseWhenThenColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/AbstractCaseWhenThenColumnTransformer.java index 8452a590a25a1..22af23da08d2d 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/AbstractCaseWhenThenColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/AbstractCaseWhenThenColumnTransformer.java @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.function.Consumer; public abstract class AbstractCaseWhenThenColumnTransformer extends ColumnTransformer { @@ -244,4 +245,14 @@ public void close() { } elseTransformer.close(); } + + @Override + protected void forEachChild(Consumer action) { + for (Pair whenThenColumnTransformer : + whenThenTransformers) { + action.accept(whenThenColumnTransformer.left); + action.accept(whenThenColumnTransformer.right); + } + action.accept(elseTransformer); + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformer.java index d10d728cd2db5..d9926cef79fa7 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformer.java @@ -25,6 +25,13 @@ import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.read.common.type.TypeEnum; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Set; +import java.util.function.Consumer; + import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isNumericType; public abstract class ColumnTransformer { @@ -47,6 +54,39 @@ public void tryEvaluate() { } } + /** Reuses a cached result, or rebuilds a missing result after clearing descendant caches. */ + public void tryEvaluateWithFreshInputs() { + if (!columnCache.hasCached()) { + clearCache(); + evaluate(); + } + } + + /** Reuses a cached result, or recomputes only selected rows when it is missing. */ + public void tryEvaluateWithSelection(boolean[] selection) { + if (!columnCache.hasCached()) { + evaluateWithSelection(selection); + } + } + + /** + * Whether evaluating with a row-selection mask yields the same values at the selected rows + * (unselected rows are nulled by design) and the same cross-block running state as a full + * evaluation, for this node and its whole subtree. Schema UPDATE uses this property to choose + * between masked materialization and preserving a stateful result. CASE, AND and OR retain their + * own selection evaluation semantics. + */ + public boolean canEvaluateWithSelection() { + boolean[] result = {true}; + forEachChild( + child -> { + if (result[0] && !child.canEvaluateWithSelection()) { + result[0] = false; + } + }); + return result[0]; + } + public Column getColumn() { return columnCache.getColumn(); } @@ -115,4 +155,36 @@ public void close() { public void clearCache() { this.columnCache.clear(); } + + /** Clears this node only. Call on every collected node before feeding a new input block. */ + public final void clearCacheForNewInput() { + this.columnCache.clear(); + resetForNewInput(); + } + + /** Prepares per-input state while preserving state carried between input blocks. */ + protected void resetForNewInput() { + // Most transformers keep no state outside the ordinary column cache. + } + + /** Collects the DAG once, deduplicating by identity, for allocation-free block resets. */ + public static ColumnTransformer[] collectTransformers(Iterable roots) { + List transformers = new ArrayList<>(); + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + for (ColumnTransformer root : roots) { + root.collectTransformers(transformers, visited); + } + return transformers.toArray(new ColumnTransformer[0]); + } + + private void collectTransformers( + List transformers, Set visited) { + if (visited.add(this)) { + transformers.add(this); + forEachChild(child -> child.collectTransformers(transformers, visited)); + } + } + + /** Child edges used to collect block-boundary reset nodes; evaluation cleanup is unchanged. */ + protected abstract void forEachChild(Consumer action); } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/FailFunctionColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/FailFunctionColumnTransformer.java index 351c4396c0fc3..07b484deff0d4 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/FailFunctionColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/FailFunctionColumnTransformer.java @@ -24,6 +24,8 @@ import org.apache.tsfile.read.common.block.column.NullColumn; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + /** * when evaluate of FailFunctionColumnTransformer is called, throw Exception to inform the user with * specified errorMsg @@ -59,4 +61,9 @@ public void evaluateWithSelection(boolean[] selection) { protected void checkType() { // do nothing } + + @Override + protected void forEachChild(Consumer action) { + // The fail function has no children. + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/binary/BinaryColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/binary/BinaryColumnTransformer.java index 5403e9f995f13..eb80e2c6e28e4 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/binary/BinaryColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/binary/BinaryColumnTransformer.java @@ -25,6 +25,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + public abstract class BinaryColumnTransformer extends ColumnTransformer { protected final ColumnTransformer leftTransformer; @@ -42,11 +44,11 @@ protected BinaryColumnTransformer( @Override public void evaluate() { leftTransformer.tryEvaluate(); - rightTransformer.tryEvaluate(); - // attention: get positionCount before calling getColumn - int positionCount = leftTransformer.getColumnCachePositionCount(); + // A later CASE child may clear the shared input's cache. Column leftColumn = leftTransformer.getColumn(); + rightTransformer.tryEvaluate(); Column rightColumn = rightTransformer.getColumn(); + int positionCount = leftColumn.getPositionCount(); ColumnBuilder builder = returnType.createColumnBuilder(positionCount); doTransform(leftColumn, rightColumn, builder, positionCount); @@ -102,4 +104,10 @@ public void close() { this.leftTransformer.close(); this.rightTransformer.close(); } + + @Override + protected void forEachChild(Consumer action) { + action.accept(leftTransformer); + action.accept(rightTransformer); + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/leaf/LeafColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/leaf/LeafColumnTransformer.java index e9f69657a5405..ce16518ce245d 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/leaf/LeafColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/leaf/LeafColumnTransformer.java @@ -24,6 +24,8 @@ import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + public abstract class LeafColumnTransformer extends ColumnTransformer { protected TsBlock input; @@ -39,4 +41,9 @@ public void checkType() { public void initFromTsBlock(TsBlock input) { this.input = input; } + + @Override + protected void forEachChild(Consumer action) { + // Leaf transformers have no children. + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MappableUDFColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MappableUDFColumnTransformer.java index f619f5692119c..85ff134928a07 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MappableUDFColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MappableUDFColumnTransformer.java @@ -26,6 +26,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + public class MappableUDFColumnTransformer extends ColumnTransformer { private final ColumnTransformer[] inputColumnTransformers; @@ -79,6 +81,14 @@ protected void checkType() { // do nothing } + @Override + public boolean canEvaluateWithSelection() { + // User code has no purity contract: a masked evaluation feeds the executor null-padded + // columns, which a stateful UDTF would process differently from a full evaluation. This + // tree-model transformer is not used by schema UPDATE. + return false; + } + public ColumnTransformer[] getInputColumnTransformers() { return inputColumnTransformers; } @@ -100,4 +110,11 @@ public void clearCache() { columnTransformer.clearCache(); } } + + @Override + protected void forEachChild(Consumer action) { + for (ColumnTransformer columnTransformer : inputColumnTransformers) { + action.accept(columnTransformer); + } + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MultiColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MultiColumnTransformer.java index 827836a39ccb3..b26788d2de8b5 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MultiColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/multi/MultiColumnTransformer.java @@ -27,7 +27,7 @@ import java.util.ArrayList; import java.util.List; -import java.util.stream.Collectors; +import java.util.function.Consumer; public abstract class MultiColumnTransformer extends ColumnTransformer { @@ -41,21 +41,15 @@ protected MultiColumnTransformer(Type returnType, List column @Override public void evaluate() { - + List childrenColumns = new ArrayList<>(columnTransformerList.size()); for (ColumnTransformer child : columnTransformerList) { child.tryEvaluate(); + childrenColumns.add(child.getColumn()); } - - // attention: get positionCount before calling getColumn - int positionCount = columnTransformerList.get(0).getColumnCachePositionCount(); - + // Later CASE children can clear a previous child's cache. + int positionCount = childrenColumns.get(0).getPositionCount(); ColumnBuilder builder = returnType.createColumnBuilder(positionCount); - doTransform( - columnTransformerList.stream() - .map(ColumnTransformer::getColumn) - .collect(Collectors.toList()), - builder, - positionCount); + doTransform(childrenColumns, builder, positionCount); initializeColumnCache(builder.build()); } @@ -68,7 +62,8 @@ public void evaluateWithSelection(boolean[] selection) { childrenColumns.add(child.getColumn()); } - int positionCount = columnTransformerList.get(0).getColumnCachePositionCount(); + // getColumn() may have released the cache; the retained column still has its size. + int positionCount = childrenColumns.get(0).getPositionCount(); ColumnBuilder builder = returnType.createColumnBuilder(positionCount); doTransform(childrenColumns, builder, positionCount, selection); @@ -104,4 +99,9 @@ public void close() { columnTransformer.close(); } } + + @Override + protected void forEachChild(Consumer action) { + columnTransformerList.forEach(action); + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ternary/TernaryColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ternary/TernaryColumnTransformer.java index 51648f797e3f7..9aec203eaa47e 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ternary/TernaryColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/ternary/TernaryColumnTransformer.java @@ -25,6 +25,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + public abstract class TernaryColumnTransformer extends ColumnTransformer { protected ColumnTransformer firstColumnTransformer; @@ -48,12 +50,13 @@ protected TernaryColumnTransformer( @Override public void evaluate() { firstColumnTransformer.tryEvaluate(); - secondColumnTransformer.tryEvaluate(); - thirdColumnTransformer.tryEvaluate(); - int positionCount = firstColumnTransformer.getColumnCachePositionCount(); Column firstColumn = firstColumnTransformer.getColumn(); + secondColumnTransformer.tryEvaluate(); Column secondColumn = secondColumnTransformer.getColumn(); + thirdColumnTransformer.tryEvaluate(); Column thirdColumn = thirdColumnTransformer.getColumn(); + // Later children may clear shared caches; the retained columns keep their values and size. + int positionCount = firstColumn.getPositionCount(); ColumnBuilder columnBuilder = returnType.createColumnBuilder(positionCount); doTransform(firstColumn, secondColumn, thirdColumn, columnBuilder, positionCount); initializeColumnCache(columnBuilder.build()); @@ -62,12 +65,13 @@ public void evaluate() { @Override public void evaluateWithSelection(boolean[] selection) { firstColumnTransformer.evaluateWithSelection(selection); - secondColumnTransformer.evaluateWithSelection(selection); - thirdColumnTransformer.evaluateWithSelection(selection); - int positionCount = firstColumnTransformer.getColumnCachePositionCount(); Column firstColumn = firstColumnTransformer.getColumn(); + secondColumnTransformer.evaluateWithSelection(selection); Column secondColumn = secondColumnTransformer.getColumn(); + thirdColumnTransformer.evaluateWithSelection(selection); Column thirdColumn = thirdColumnTransformer.getColumn(); + // Later children may clear shared caches; the retained columns keep their values and size. + int positionCount = firstColumn.getPositionCount(); ColumnBuilder columnBuilder = returnType.createColumnBuilder(positionCount); doTransform(firstColumn, secondColumn, thirdColumn, columnBuilder, positionCount, selection); initializeColumnCache(columnBuilder.build()); @@ -118,4 +122,11 @@ public void close() { secondColumnTransformer.close(); thirdColumnTransformer.close(); } + + @Override + protected void forEachChild(Consumer action) { + action.accept(firstColumnTransformer); + action.accept(secondColumnTransformer); + action.accept(thirdColumnTransformer); + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java index 6ef156a6b707c..cc2ea120f2800 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java @@ -112,7 +112,7 @@ protected void doTransform( while (iterator.hasNext()) { try { Record input = iterator.next(); - if (selection[i++]) { + if (!selection[i++]) { builder.appendNull(); continue; } @@ -131,6 +131,14 @@ protected void doTransform( } } + @Override + public boolean canEvaluateWithSelection() { + // User code has no purity contract: it may keep state across rows, so a masked evaluation + // cannot be assumed equivalent to a full one. This is a conservative default for future + // consumers; schema UPDATE does not currently execute scalar UDFs. + return false; + } + @Override public void close() { // ensure beforeStart was called diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/UnaryColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/UnaryColumnTransformer.java index 2a88aea340a95..b1472d96dee76 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/UnaryColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/UnaryColumnTransformer.java @@ -25,6 +25,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.read.common.type.Type; +import java.util.function.Consumer; + public abstract class UnaryColumnTransformer extends ColumnTransformer { protected ColumnTransformer childColumnTransformer; @@ -78,4 +80,9 @@ public void close() { super.close(); childColumnTransformer.close(); } + + @Override + protected void forEachChild(Consumer action) { + action.accept(childColumnTransformer); + } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffColumnTransformer.java index 5aa14fb6f245d..19f9370954cda 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffColumnTransformer.java @@ -26,6 +26,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.read.common.type.Type; +import java.util.Arrays; + public class DiffColumnTransformer extends BinaryColumnTransformer { // cache the last non-null value @@ -34,11 +36,175 @@ public class DiffColumnTransformer extends BinaryColumnTransformer { // indicate whether lastValue is null private boolean lastValueIsNull = true; + // Re-evaluation in the same block must start from the state before this input. + private double lastValueBeforeCurrentInput; + private boolean lastValueIsNullBeforeCurrentInput = true; + + // A later masked pass must not replace the state carried by a complete evaluation. + private double lastValueAfterFullEvaluation; + private boolean lastValueIsNullAfterFullEvaluation; + private boolean hasFullEvaluation; + + // Schema UPDATE keeps the filter's results until its SET expressions have consumed them. + private boolean retainResults; + private boolean reuseRetainedResults; + private Column retainedResult; + private boolean[] retainedRows; + private boolean[] evaluationSelection; + public DiffColumnTransformer( Type returnType, ColumnTransformer leftTransformer, ColumnTransformer rightTransformer) { super(returnType, leftTransformer, rightTransformer); } + @Override + protected void resetForNewInput() { + lastValueBeforeCurrentInput = lastValue; + lastValueIsNullBeforeCurrentInput = lastValueIsNull; + hasFullEvaluation = false; + releaseRetainedResult(); + } + + @Override + public boolean canEvaluateWithSelection() { + // A fresh masked transform changes the running state. Schema UPDATE explicitly opts into + // reusing the filter's results instead of treating DIFF as a stateless expression. + return false; + } + + /** Enables result retention for a schema UPDATE's filter and SET phases. */ + public void retainResultForReuse() { + retainResults = true; + } + + /** Starts the SET phase without evaluating rows excluded by the filter's CASE guards. */ + public void reuseRetainedResult() { + reuseRetainedResults = true; + columnCache.clear(); + } + + /** Releases this batch's columns while keeping the state produced by its filter. */ + public void releaseRetainedResult() { + retainedResult = null; + retainedRows = null; + reuseRetainedResults = false; + evaluationSelection = null; + } + + @Override + public void initializeColumnCache(Column column) { + super.initializeColumnCache(column); + if (!retainResults || reuseRetainedResults) { + return; + } + if (retainedResult == null || evaluationSelection == null) { + retainedResult = column; + retainedRows = new boolean[column.getPositionCount()]; + if (evaluationSelection == null) { + Arrays.fill(retainedRows, true); + } else { + System.arraycopy(evaluationSelection, 0, retainedRows, 0, retainedRows.length); + } + return; + } + ColumnBuilder builder = returnType.createColumnBuilder(column.getPositionCount()); + for (int i = 0; i < column.getPositionCount(); i++) { + appendValue(builder, evaluationSelection[i] ? column : retainedResult, i); + retainedRows[i] |= evaluationSelection[i]; + } + retainedResult = builder.build(); + } + + @Override + public void evaluate() { + if (reuseRetainedResults) { + evaluateRetainedResult(null); + } else { + super.evaluate(); + } + } + + @Override + public void evaluateWithSelection(boolean[] selection) { + if (reuseRetainedResults) { + evaluateRetainedResult(selection); + return; + } + evaluationSelection = selection; + try { + super.evaluateWithSelection(selection); + } finally { + evaluationSelection = null; + } + } + + private void evaluateRetainedResult(boolean[] selection) { + boolean missing = retainedResult == null; + if (!missing) { + for (int i = 0; i < retainedRows.length; i++) { + if (!retainedRows[i] && (selection == null || selection[i])) { + missing = true; + break; + } + } + } + if (missing) { + Column previousResult = retainedResult; + boolean[] previousRows = retainedRows == null ? null : retainedRows.clone(); + boolean[] expandedSelection = selection == null ? null : selection.clone(); + if (expandedSelection != null && previousRows != null) { + for (int i = 0; i < expandedSelection.length; i++) { + expandedSelection[i] |= previousRows[i]; + } + } + double savedLastValue = lastValue; + boolean savedLastValueIsNull = lastValueIsNull; + double savedFullValue = lastValueAfterFullEvaluation; + boolean savedFullValueIsNull = lastValueIsNullAfterFullEvaluation; + boolean savedHasFullEvaluation = hasFullEvaluation; + reuseRetainedResults = false; + evaluationSelection = expandedSelection; + try { + if (expandedSelection == null) { + super.evaluate(); + } else { + super.evaluateWithSelection(expandedSelection); + } + } finally { + reuseRetainedResults = true; + evaluationSelection = null; + lastValue = savedLastValue; + lastValueIsNull = savedLastValueIsNull; + lastValueAfterFullEvaluation = savedFullValue; + lastValueIsNullAfterFullEvaluation = savedFullValueIsNull; + hasFullEvaluation = savedHasFullEvaluation; + } + // An additional SET consumer must not change values already used by WHERE. + if (previousResult != null) { + ColumnBuilder builder = returnType.createColumnBuilder(retainedRows.length); + for (int i = 0; i < retainedRows.length; i++) { + appendValue(builder, previousRows[i] ? previousResult : retainedResult, i); + } + retainedResult = builder.build(); + } + } + initializeColumnCache(retainedResult); + } + + private static void appendValue(ColumnBuilder builder, Column column, int index) { + if (column.isNull(index)) { + builder.appendNull(); + } else { + builder.write(column, index); + } + } + + @Override + public void close() { + releaseRetainedResult(); + super.close(); + } + @Override protected void checkType() { // do nothing @@ -47,6 +213,17 @@ protected void checkType() { @Override protected void doTransform( Column leftColumn, Column rightColumn, ColumnBuilder builder, int positionCount) { + transform(leftColumn, rightColumn, builder, positionCount, true); + } + + private void transform( + Column leftColumn, + Column rightColumn, + ColumnBuilder builder, + int positionCount, + boolean fullEvaluation) { + lastValue = lastValueBeforeCurrentInput; + lastValueIsNull = lastValueIsNullBeforeCurrentInput; Type leftType = leftTransformer.getType(); Type rightType = rightTransformer.getType(); for (int i = 0; i < positionCount; i++) { @@ -71,6 +248,15 @@ protected void doTransform( lastValueIsNull = false; } } + + if (fullEvaluation) { + lastValueAfterFullEvaluation = lastValue; + lastValueIsNullAfterFullEvaluation = lastValueIsNull; + hasFullEvaluation = true; + } else if (hasFullEvaluation) { + lastValue = lastValueAfterFullEvaluation; + lastValueIsNull = lastValueIsNullAfterFullEvaluation; + } } @Override @@ -80,7 +266,14 @@ protected void doTransform( ColumnBuilder builder, int positionCount, boolean[] selection) { - // Diff do not support short circuit evaluation - doTransform(leftColumn, rightColumn, builder, positionCount); + boolean fullEvaluation = true; + for (boolean selected : selection) { + if (!selected) { + fullEvaluation = false; + break; + } + } + // Preserve the selected-input results, but retain a complete pass's cross-block state. + transform(leftColumn, rightColumn, builder, positionCount, fullEvaluation); } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffFunctionColumnTransformer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffFunctionColumnTransformer.java index 04d65c594b9c5..1473f1e8f5a67 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffFunctionColumnTransformer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DiffFunctionColumnTransformer.java @@ -39,6 +39,10 @@ public class DiffFunctionColumnTransformer extends UnaryColumnTransformer { // indicate whether lastValue is null private boolean lastValueIsNull = true; + // Re-evaluation in the same block must start from the state before this input. + private double lastValueBeforeCurrentInput; + private boolean lastValueIsNullBeforeCurrentInput = true; + public DiffFunctionColumnTransformer( Type returnType, ColumnTransformer childColumnTransformer, boolean ignoreNull) { super(returnType, childColumnTransformer); @@ -58,8 +62,16 @@ public void evaluateWithSelection(boolean[] selection) { childColumnTransformer.clearCache(); } + @Override + protected void resetForNewInput() { + lastValueBeforeCurrentInput = lastValue; + lastValueIsNullBeforeCurrentInput = lastValueIsNull; + } + @Override protected void doTransform(Column column, ColumnBuilder columnBuilder) { + lastValue = lastValueBeforeCurrentInput; + lastValueIsNull = lastValueIsNullBeforeCurrentInput; for (int i = 0, n = column.getPositionCount(); i < n; i++) { if (column.isNull(i)) { columnBuilder.appendNull(); // currValue is null, append null diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperatorTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperatorTest.java new file mode 100644 index 0000000000000..cd241f02cf2d8 --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperatorTest.java @@ -0,0 +1,501 @@ +/* + * 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.calc.execution.operator.process; + +import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; +import org.apache.iotdb.calc.execution.operator.Operator; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; +import org.apache.iotdb.calc.transformation.dag.column.ColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.FailFunctionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.TableCaseWhenThenColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.ArithmeticAdditionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareGreaterThanColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.ConstantColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.IdentityColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.LeafColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.NullColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.multi.MappableUDFColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.udf.UDTFExecutor; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.column.DoubleColumn; +import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn; +import org.apache.tsfile.read.common.block.column.TimeColumn; +import org.junit.Test; + +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.function.Consumer; + +import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN; +import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class FilterAndProjectOperatorTest { + + @Test + public void testProjectionDoesNotReuseColumnsAcrossDifferentSizedBlocks() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(input, 0), graph.constant(1), graph.nullValue()); + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(-2, 3), block(4)}, caseWhen, input)) { + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {1.0}); + assertBlock(operator.next(), new Double[] {null, 1.0}, new Double[] {-2.0, 3.0}); + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {4.0}); + } + } + + @Test + public void testProjectionDoesNotReuseColumnsAcrossEqualSizedBlocks() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(input, 0), graph.constant(1), graph.nullValue()); + try (Operator operator = + project(graph, new TsBlock[] {block(1, 2), block(-3, 4)}, caseWhen, input)) { + assertBlock(operator.next(), new Double[] {1.0, 1.0}, new Double[] {1.0, 2.0}); + assertBlock(operator.next(), new Double[] {null, 1.0}, new Double[] {-3.0, 4.0}); + } + } + + @Test + public void testProjectionHandlesEmptyBlocks() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(input, 0), graph.constant(1), graph.nullValue()); + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(), block(-2)}, caseWhen, input)) { + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {1.0}); + assertBlock(operator.next(), new Double[0], new Double[0]); + assertBlock(operator.next(), new Double[] {null}, new Double[] {-2.0}); + } + } + + @Test + public void testProjectionClearsSharedIntermediateExpressions() throws Exception { + Graph graph = new Graph(); + ColumnTransformer shared = graph.add(graph.input(0), 0); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(shared, 0), graph.constant(1), graph.nullValue()); + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(-2)}, caseWhen, graph.add(shared, 1))) { + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {2.0}); + assertBlock(operator.next(), new Double[] {null}, new Double[] {-1.0}); + } + } + + @Test + public void testFilterDoesNotReuseColumnsAcrossBlocks() throws Exception { + Graph filter = new Graph(); + ColumnTransformer input = filter.input(0); + ColumnTransformer caseWhen = + filter.caseWhen(filter.greaterThan(input, 0), filter.constant(1), filter.constant(0)); + ColumnTransformer predicate = filter.greaterThan(filter.add(caseWhen, input), 0); + Graph projection = new Graph(); + try (Operator operator = + operator( + new TsBlock[] {block(1), block(-2, 3), block(-4)}, + filter, + predicate, + Collections.emptyList(), + projection, + projection.input(0))) { + assertBlock(operator.next(), new Double[] {1.0}); + assertBlock(operator.next(), new Double[] {3.0}); + assertBlock(operator.next(), new Double[0]); + } + } + + @Test + public void testFilterCommonExpressionsDoNotCrossBlocks() throws Exception { + for (boolean runLengthEncodedTime : new boolean[] {false, true}) { + Graph filter = new Graph(); + ColumnTransformer shared = filter.add(filter.input(0), 0); + ColumnTransformer caseWhen = + filter.caseWhen(filter.greaterThan(shared, 0), filter.constant(1), filter.constant(0)); + ColumnTransformer predicate = filter.greaterThan(caseWhen, 0); + Graph projection = new Graph(); + try (Operator operator = + operator( + new TsBlock[] { + block(runLengthEncodedTime, 1), + block(runLengthEncodedTime, -2, 3), + block(runLengthEncodedTime, 4, -5) + }, + filter, + predicate, + Collections.singletonList(shared), + projection, + projection.input(0), + projection.input(1))) { + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {1.0}); + assertBlock(operator.next(), new Double[] {3.0}, new Double[] {3.0}); + assertBlock(operator.next(), new Double[] {4.0}, new Double[] {4.0}); + } + } + } + + @Test + public void testProjectionClearsCachedDescendantsOfNestedCase() throws Exception { + Graph graph = new Graph(); + ColumnTransformer shared = graph.add(graph.input(0), 0); + ColumnTransformer innerCase = + graph.caseWhen(graph.greaterThan(shared, 0), graph.constant(1), graph.constant(0)); + ColumnTransformer outerCase = + graph.caseWhen(graph.greaterThan(innerCase, 0), graph.constant(2), graph.constant(0)); + // Simulate an intermediate cache retained from the previous input block. Clearing only the + // outer CASE itself must not allow this cached value to reach the new block's WHEN condition. + shared.initializeColumnCache(doubleColumn(1)); + try (Operator operator = project(graph, new TsBlock[] {block(-2)}, outerCase)) { + assertBlock(operator.next(), new Double[] {0.0}); + } + } + + @Test + public void testThenExpressionIsRecomputedWithoutSelectionForLaterOutput() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer shared = graph.add(input, 1); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(input, 0), shared, graph.constant(0)); + try (Operator operator = + project(graph, new TsBlock[] {block(-2, 3), block(-4, 5)}, caseWhen, shared)) { + assertBlock(operator.next(), new Double[] {0.0, 4.0}, new Double[] {-1.0, 4.0}); + assertBlock(operator.next(), new Double[] {0.0, 6.0}, new Double[] {-3.0, 6.0}); + } + } + + @Test + public void testCaseStillSkipsUnusedFailBranch() throws Exception { + Graph graph = new Graph(); + ColumnTransformer caseWhen = + graph.caseWhen( + graph.greaterThan(graph.input(0), 0), + graph.constant(1), + new FailFunctionColumnTransformer(DOUBLE, "The unused ELSE branch was evaluated")); + try (Operator operator = project(graph, new TsBlock[] {block(1, 2), block(3)}, caseWhen)) { + assertBlock(operator.next(), new Double[] {1.0, 1.0}); + assertBlock(operator.next(), new Double[] {1.0}); + } + } + + @Test + public void testProjectionReusesSharedExpressionsWithinEachBlock() throws Exception { + Graph graph = new Graph(); + CountingIdentityColumnTransformer input = new CountingIdentityColumnTransformer(); + graph.leaves.add(input); + ColumnTransformer shared = graph.add(input, 0); + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(2)}, shared, graph.add(shared, 1))) { + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {2.0}); + assertEquals(1, input.evaluationCount); + assertBlock(operator.next(), new Double[] {2.0}, new Double[] {3.0}); + assertEquals(2, input.evaluationCount); + } + } + + @Test + public void testCopiedOperatorReusesCollectedTransformers() throws Exception { + Graph graph = new Graph(); + CountingIdentityColumnTransformer input = new CountingIdentityColumnTransformer(); + graph.leaves.add(input); + try (FilterAndProjectOperator original = project(graph, new TsBlock[0], graph.add(input, 1)); + FilterAndProjectOperator copy = + new FilterAndProjectOperator( + original, + new ValuesOperator(new TestOperatorContext(), Arrays.asList(block(1), block(2))))) { + assertEquals(1, input.childTraversalCount); + assertBlock(copy.next(), new Double[] {2.0}); + assertBlock(copy.next(), new Double[] {3.0}); + assertEquals(1, input.childTraversalCount); + } + } + + @Test + public void testProjectionCollectsSharedDagOnlyOnce() throws Exception { + Graph graph = new Graph(); + CountingIdentityColumnTransformer input = new CountingIdentityColumnTransformer(); + graph.leaves.add(input); + ColumnTransformer shared = input; + // Each level references the same child twice. A traversal without deduplication would visit + // the input 4096 times per root, although this DAG contains only 13 distinct transformers. + for (int i = 0; i < 12; i++) { + shared = graph.add(shared, shared); + } + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(2)}, shared, graph.add(shared, 1))) { + assertEquals(1, input.childTraversalCount); + assertBlock(operator.next(), new Double[] {4096.0}, new Double[] {4097.0}); + assertEquals(1, input.childTraversalCount); + assertBlock(operator.next(), new Double[] {8192.0}, new Double[] {8193.0}); + assertEquals(1, input.childTraversalCount); + } + } + + @Test + public void testFilterAndCommonExpressionsCollectSharedDagOnlyOnce() throws Exception { + Graph filter = new Graph(); + CountingIdentityColumnTransformer input = new CountingIdentityColumnTransformer(); + filter.leaves.add(input); + ColumnTransformer shared = filter.add(input, 1); + ColumnTransformer predicate = filter.greaterThan(shared, 0); + Graph projection = new Graph(); + try (Operator operator = + operator( + new TsBlock[] {block(1), block(-2, 3)}, + filter, + predicate, + Collections.singletonList(shared), + projection, + projection.input(0), + projection.input(1))) { + assertEquals(1, input.childTraversalCount); + assertBlock(operator.next(), new Double[] {1.0}, new Double[] {2.0}); + assertEquals(1, input.childTraversalCount); + assertBlock(operator.next(), new Double[] {3.0}, new Double[] {4.0}); + assertEquals(1, input.childTraversalCount); + } + } + + @Test + public void testMappableUdfClearsCaseDescendantsAcrossBlocks() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer caseWhen = + graph.caseWhen(graph.greaterThan(input, 0), graph.constant(1), graph.constant(0)); + ColumnTransformer udf = + new MappableUDFColumnTransformer( + DOUBLE, + new ColumnTransformer[] {reference(caseWhen), reference(input)}, + new UDTFExecutor("test", ZoneId.of("UTC")) { + @Override + public void execute(Column[] columns, ColumnBuilder builder) { + for (int i = 0; i < columns[0].getPositionCount(); i++) { + builder.writeDouble(columns[0].getDouble(i) + columns[1].getDouble(i)); + } + } + + @Override + public void beforeDestroy() { + // No UDF resources are allocated by this test executor. + } + }); + try (Operator operator = + project(graph, new TsBlock[] {block(1), block(-2, 3), block(4)}, udf)) { + assertBlock(operator.next(), new Double[] {2.0}); + assertBlock(operator.next(), new Double[] {-2.0, 4.0}); + assertBlock(operator.next(), new Double[] {5.0}); + } + } + + @Test + public void testReusingCasePreservesPendingSiblingColumn() throws Exception { + Graph graph = new Graph(); + ColumnTransformer input = graph.input(0); + ColumnTransformer sharedCase = + graph.caseWhen(graph.greaterThan(input, 0), graph.constant(1), graph.constant(0)); + ColumnTransformer nestedCase = + graph.caseWhen(graph.greaterThan(sharedCase, 0), graph.constant(2), graph.constant(3)); + // The first output leaves sharedCase cached. In the second output, input is evaluated before + // nestedCase reuses sharedCase. Its cleanup must preserve input until the addition reads it. + try (Operator operator = + project( + graph, + new TsBlock[] {block(-3, 0, 2), block(4, -1)}, + sharedCase, + graph.add(input, nestedCase))) { + assertBlock(operator.next(), new Double[] {0.0, 0.0, 1.0}, new Double[] {0.0, 3.0, 4.0}); + assertBlock(operator.next(), new Double[] {1.0, 0.0}, new Double[] {6.0, 2.0}); + } + } + + private static FilterAndProjectOperator project( + Graph graph, TsBlock[] blocks, ColumnTransformer... outputs) { + return operator(blocks, null, null, Collections.emptyList(), graph, outputs); + } + + private static FilterAndProjectOperator operator( + TsBlock[] blocks, + Graph filter, + ColumnTransformer predicate, + List common, + Graph projection, + ColumnTransformer... outputs) { + List projectOutputs = new ArrayList<>(); + for (ColumnTransformer output : outputs) { + projectOutputs.add(reference(output)); + } + for (ColumnTransformer transformer : common) { + reference(transformer); + } + return new FilterAndProjectOperator( + new TestOperatorContext(), + new ValuesOperator(new TestOperatorContext(), Arrays.asList(blocks)), + Collections.nCopies(1 + common.size(), TSDataType.DOUBLE), + filter == null ? Collections.emptyList() : filter.leaves, + predicate == null ? null : reference(predicate), + common, + projection.leaves, + projectOutputs, + false, + predicate != null); + } + + private static ColumnTransformer reference(ColumnTransformer transformer) { + transformer.addReferenceCount(); + return transformer; + } + + private static DoubleColumn doubleColumn(double... values) { + return new DoubleColumn(values.length, Optional.empty(), values); + } + + private static TsBlock block(double... values) { + return block(false, values); + } + + private static TsBlock block(boolean runLengthEncodedTime, double... values) { + Column timeColumn = + runLengthEncodedTime + ? new RunLengthEncodedColumn(new TimeColumn(1, new long[] {0}), values.length) + : new TimeColumn(values.length, new long[values.length]); + return new TsBlock(timeColumn, doubleColumn(values)); + } + + private static void assertBlock(TsBlock actual, Double[]... expectedColumns) { + assertEquals(expectedColumns.length, actual.getValueColumnCount()); + assertEquals(expectedColumns[0].length, actual.getPositionCount()); + for (int columnIndex = 0; columnIndex < expectedColumns.length; columnIndex++) { + Column column = actual.getColumn(columnIndex); + Double[] expected = expectedColumns[columnIndex]; + assertEquals(expected.length, column.getPositionCount()); + for (int row = 0; row < expected.length; row++) { + if (expected[row] == null) { + assertTrue(column.isNull(row)); + } else { + assertFalse(column.isNull(row)); + assertEquals(expected[row], column.getDouble(row), 0.0); + } + } + } + } + + private static class Graph { + private final List leaves = new ArrayList<>(); + + private ColumnTransformer input(int index) { + return leaf(new IdentityColumnTransformer(DOUBLE, index)); + } + + private ColumnTransformer constant(double value) { + return leaf(new ConstantColumnTransformer(DOUBLE, doubleColumn(value))); + } + + private ColumnTransformer nullValue() { + return leaf(new NullColumnTransformer()); + } + + private ColumnTransformer leaf(LeafColumnTransformer transformer) { + leaves.add(transformer); + return transformer; + } + + private ColumnTransformer add(ColumnTransformer input, double value) { + return add(input, constant(value)); + } + + private ColumnTransformer add(ColumnTransformer left, ColumnTransformer right) { + return new ArithmeticAdditionColumnTransformer(DOUBLE, reference(left), reference(right)); + } + + private ColumnTransformer greaterThan(ColumnTransformer input, double value) { + return new CompareGreaterThanColumnTransformer( + BOOLEAN, reference(input), reference(constant(value))); + } + + private ColumnTransformer caseWhen( + ColumnTransformer when, ColumnTransformer then, ColumnTransformer otherwise) { + return new TableCaseWhenThenColumnTransformer( + DOUBLE, + Collections.singletonList(reference(when)), + Collections.singletonList(reference(then)), + reference(otherwise)); + } + } + + private static class CountingIdentityColumnTransformer extends IdentityColumnTransformer { + private int evaluationCount; + private int childTraversalCount; + + private CountingIdentityColumnTransformer() { + super(DOUBLE, 0); + } + + @Override + protected void evaluate() { + evaluationCount++; + super.evaluate(); + } + + @Override + protected void forEachChild(Consumer action) { + childTraversalCount++; + } + } + + private static class TestOperatorContext extends CommonOperatorContext { + private TestOperatorContext() { + super(0, new PlanNodeId("test"), "test"); + } + + @Override + public MemoryReservationManager getMemoryReservationContext() { + return null; + } + + @Override + public int getFragmentId() { + return 0; + } + + @Override + public int getPipelineId() { + return 0; + } + + @Override + public long ramBytesUsed() { + return 0; + } + } +} diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformerTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformerTest.java new file mode 100644 index 0000000000000..2017ff6252dbe --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformerTest.java @@ -0,0 +1,243 @@ +/* + * 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.calc.transformation.dag.column; + +import org.apache.iotdb.calc.transformation.dag.column.binary.ArithmeticAdditionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.ArithmeticSubtractionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareGreaterThanColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.ConstantColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.IdentityColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.multi.CoalesceColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.ternary.BetweenColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffFunctionColumnTransformer; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.column.BooleanColumn; +import org.apache.tsfile.read.common.block.column.DoubleColumn; +import org.apache.tsfile.read.common.block.column.TimeColumn; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Optional; + +import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN; +import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class ColumnTransformerTest { + + @Test + public void testCollectionDeduplicatesByIdentity() { + ColumnTransformer first = new EqualIdentityColumnTransformer(0); + ColumnTransformer second = new EqualIdentityColumnTransformer(1); + ColumnTransformer root = new ArithmeticAdditionColumnTransformer(DOUBLE, first, second); + + ColumnTransformer[] collected = + ColumnTransformer.collectTransformers(Arrays.asList(root, first, root)); + + assertEquals(3, collected.length); + assertSame(root, collected[0]); + assertSame(first, collected[1]); + assertSame(second, collected[2]); + } + + @Test + public void testNewInputClearingOnlyClearsOwnCache() { + ColumnTransformer left = new IdentityColumnTransformer(DOUBLE, 0); + ColumnTransformer right = new IdentityColumnTransformer(DOUBLE, 1); + ColumnTransformer root = new ArithmeticAdditionColumnTransformer(DOUBLE, left, right); + left.addReferenceCount(); + right.addReferenceCount(); + root.addReferenceCount(); + left.initializeColumnCache(column(1)); + right.initializeColumnCache(column(2)); + root.initializeColumnCache(column(3)); + + root.clearCacheForNewInput(); + + assertEquals(0, root.getColumnCachePositionCount()); + assertEquals(1.0, left.getColumn().getDouble(0), 0.0); + assertEquals(2.0, right.getColumn().getDouble(0), 0.0); + } + + @Test + public void testMaskedCommonExpressionKeepsItsInputLength() { + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + input.addReferenceCount(); + input.initFromTsBlock( + new TsBlock( + new TimeColumn(2, new long[] {0, 1}), + new DoubleColumn(2, Optional.empty(), new double[] {1, 2}))); + ColumnTransformer common = new CoalesceColumnTransformer(DOUBLE, Arrays.asList(input)); + common.addReferenceCount(); + common.tryEvaluateWithSelection(new boolean[] {false, true}); + Column result = common.getColumn(); + assertEquals(2, result.getPositionCount()); + assertEquals(2.0, result.getDouble(1), 0.0); + } + + @Test + public void testSelectionEvaluationCapabilityPropagation() { + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + ColumnTransformer flag = + new ConstantColumnTransformer( + BOOLEAN, new BooleanColumn(1, Optional.empty(), new boolean[] {true})); + ColumnTransformer diff = new DiffColumnTransformer(DOUBLE, input, flag); + ColumnTransformer root = new ArithmeticAdditionColumnTransformer(DOUBLE, diff, input); + + assertFalse(diff.canEvaluateWithSelection()); + assertFalse(root.canEvaluateWithSelection()); + assertTrue( + new ArithmeticAdditionColumnTransformer(DOUBLE, input, input).canEvaluateWithSelection()); + // The unary diff stays selection-capable: it forces an all-true child mask internally, so a + // masked evaluation is equivalent to a full one. + assertTrue(new DiffFunctionColumnTransformer(DOUBLE, input, false).canEvaluateWithSelection()); + } + + @Test + public void testMaskedBetweenRetainsSharedInput() { + Column result = sharedBetween(false, new boolean[] {true, true, false}); + assertEquals(3, result.getPositionCount()); + assertFalse(result.isNull(0)); + assertTrue(result.getBoolean(0)); + assertFalse(result.isNull(1)); + assertTrue(result.getBoolean(1)); + assertTrue(result.isNull(2)); + } + + @Test + public void testMaskedBetweenRetainsBoundSharedWithLaterChild() { + Column result = sharedBetween(true, new boolean[] {false, true, true}); + assertEquals(3, result.getPositionCount()); + assertTrue(result.isNull(0)); + assertFalse(result.isNull(1)); + assertTrue(result.getBoolean(1)); + assertFalse(result.isNull(2)); + assertTrue(result.getBoolean(2)); + } + + @Test + public void testBinaryRetainsInputClearedByCaseSibling() { + TsBlock block = inputBlock(); + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + input.initFromTsBlock(block); + ColumnTransformer sum = + reference( + new ArithmeticAdditionColumnTransformer( + DOUBLE, reference(input), reference(caseForInput(input, block)))); + sum.tryEvaluate(); + Column result = sum.getColumn(); + assertEquals(3, result.getPositionCount()); + assertEquals(3.0, result.getDouble(0), 0.0); + assertEquals(4.0, result.getDouble(1), 0.0); + assertEquals(2.0, result.getDouble(2), 0.0); + } + + @Test + public void testCoalesceRetainsInputClearedByCaseSibling() { + TsBlock block = inputBlock(); + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + input.initFromTsBlock(block); + ColumnTransformer coalesce = + reference( + new CoalesceColumnTransformer( + DOUBLE, Arrays.asList(reference(input), reference(caseForInput(input, block))))); + coalesce.tryEvaluate(); + Column result = coalesce.getColumn(); + assertEquals(3, result.getPositionCount()); + assertEquals(1.0, result.getDouble(0), 0.0); + assertEquals(2.0, result.getDouble(1), 0.0); + assertEquals(-1.0, result.getDouble(2), 0.0); + } + + private static ColumnTransformer caseForInput(ColumnTransformer input, TsBlock block) { + ColumnTransformer condition = + new CompareGreaterThanColumnTransformer( + BOOLEAN, reference(input), reference(constant(block, 0))); + return new TableCaseWhenThenColumnTransformer( + DOUBLE, + Collections.singletonList(reference(condition)), + Collections.singletonList(reference(constant(block, 2))), + reference(constant(block, 3))); + } + + private static Column sharedBetween(boolean nestedUpperBound, boolean[] selection) { + TsBlock block = inputBlock(); + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + input.initFromTsBlock(block); + ColumnTransformer lower = + new ArithmeticSubtractionColumnTransformer( + DOUBLE, reference(input), reference(constant(block, 1))); + ColumnTransformer upper = + new ArithmeticAdditionColumnTransformer( + DOUBLE, + reference(nestedUpperBound ? lower : input), + reference(constant(block, nestedUpperBound ? 2 : 1))); + ColumnTransformer between = + reference( + new BetweenColumnTransformer( + BOOLEAN, reference(input), reference(lower), reference(upper), false)); + between.evaluateWithSelection(selection); + return between.getColumn(); + } + + private static TsBlock inputBlock() { + return new TsBlock( + new TimeColumn(3, new long[] {0, 1, 2}), + new DoubleColumn(3, Optional.empty(), new double[] {1, 2, -1})); + } + + private static ConstantColumnTransformer constant(TsBlock block, double value) { + ConstantColumnTransformer constant = new ConstantColumnTransformer(DOUBLE, column(value)); + constant.initFromTsBlock(block); + return constant; + } + + private static ColumnTransformer reference(ColumnTransformer transformer) { + transformer.addReferenceCount(); + return transformer; + } + + private static Column column(double value) { + return new DoubleColumn(1, Optional.empty(), new double[] {value}); + } + + private static class EqualIdentityColumnTransformer extends IdentityColumnTransformer { + private EqualIdentityColumnTransformer(int index) { + super(DOUBLE, index); + } + + @Override + public boolean equals(Object other) { + return other instanceof EqualIdentityColumnTransformer; + } + + @Override + public int hashCode() { + return 1; + } + } +} diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/DiffCacheLifecycleTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/DiffCacheLifecycleTest.java new file mode 100644 index 0000000000000..80fa2948d99eb --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/DiffCacheLifecycleTest.java @@ -0,0 +1,768 @@ +/* + * 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.calc.transformation.dag.column; + +import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; +import org.apache.iotdb.calc.execution.operator.Operator; +import org.apache.iotdb.calc.execution.operator.process.FilterAndProjectOperator; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareGreaterThanColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.LogicOrColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.ConstantColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.IdentityColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.LeafColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.IsNullColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.CastFunctionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffFunctionColumnTransformer; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.column.TimeColumn; +import org.apache.tsfile.utils.Binary; +import org.junit.Test; + +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN; +import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE; +import static org.apache.tsfile.read.common.type.StringType.STRING; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class DiffCacheLifecycleTest { + + @Test + public void testUpdateRetainsValuesAndNullsFromDifferentFilterMasks() { + Fixture fixture = binary(); + DiffColumnTransformer diff = (DiffColumnTransformer) fixture.diff; + diff.retainResultForReuse(); + fixture.init(block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, true, true})); + assertValues(fixture.selected(new boolean[] {true, true, false}), null, 2.0, null); + assertValues(fixture.selected(new boolean[] {false, false, true}), null, null, null); + diff.reuseRetainedResult(); + assertValues(fixture.selected(new boolean[] {true, true, true}), null, 2.0, null); + fixture.init(block(new Double[] {10.0}, new Boolean[] {true})); + assertValues(fixture.full(), 4.0); + } + + @Test + public void testAdditionalSetRowsPreserveWhereValuesAndNextBatchState() { + Fixture fixture = binary(); + DiffColumnTransformer diff = (DiffColumnTransformer) fixture.diff; + diff.retainResultForReuse(); + fixture.init(block(new Double[] {1.0, 9.0, 5.0}, new Boolean[] {false, false, false})); + assertValues(fixture.selected(new boolean[] {true, false, false}), null, null, null); + diff.reuseRetainedResult(); + assertValues(fixture.selected(new boolean[] {false, true, false}), null, 8.0, null); + assertValues(fixture.selected(new boolean[] {false, false, true}), null, 8.0, -4.0); + fixture.init(block(new Double[] {10.0}, new Boolean[] {false})); + assertValues(fixture.full(), (Double) null); + } + + @Test + public void testSetCanEvaluateDiffNotVisitedByWhere() { + Fixture fixture = binary(); + DiffColumnTransformer diff = (DiffColumnTransformer) fixture.diff; + diff.retainResultForReuse(); + fixture.init(block(new Double[] {1.0, 9.0, 5.0}, new Boolean[] {false, false, false})); + diff.reuseRetainedResult(); + assertValues(fixture.selected(new boolean[] {false, true, false}), null, null, null); + diff.releaseRetainedResult(); + fixture.init(block(new Double[] {10.0}, new Boolean[] {false})); + assertValues(fixture.full(), (Double) null); + } + + @Test + public void testBinaryFullEvaluationOwnsNextInputStateRegardlessOfMaskOrder() { + for (boolean fullFirst : new boolean[] {true, false}) { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 2.0, 3.0}, new Boolean[] {true, true, true})); + if (fullFirst) { + assertValues(fixture.full(), null, 1.0, 1.0); + } + assertValues(fixture.selected(new boolean[3]), null, null, null); + if (!fullFirst) { + assertValues(fixture.full(), null, 1.0, 1.0); + } + fixture.init(block(new Double[] {4.0, 5.0, 6.0}, new Boolean[] {true, true, true})); + assertValues(fixture.full(), 1.0, 1.0, 1.0); + } + } + + @Test + public void testFreshInputsDoNotReuseMaskedChildCaches() { + Fixture fixture = binary(); + // Model inputs shared by a stateless SET item and DIFF: getColumn need not release them. + fixture.input.addReferenceCount(); + fixture.flags.addReferenceCount(); + fixture.init(block(new Double[] {1.0, 9.0, 5.0}, new Boolean[] {false, false, false})); + boolean[] selection = {false, true, false}; + fixture.input.evaluateWithSelection(selection); + fixture.flags.evaluateWithSelection(selection); + fixture.diff.tryEvaluateWithFreshInputs(); + assertValues(fixture.diff.getColumn(), null, 8.0, -4.0); + } + + @Test + public void testFullSelectionOwnsNextInputState() { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 2.0, 3.0}, new Boolean[] {true, true, true})); + assertValues(fixture.selected(new boolean[] {true, true, true}), null, 1.0, 1.0); + assertValues(fixture.selected(new boolean[3]), null, null, null); + fixture.init(block(new Double[] {4.0}, new Boolean[] {true})); + assertValues(fixture.full(), 1.0); + } + + @Test + public void testFullStateDoesNotLeakIntoLaterCaseOnlyInputs() { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 3.0}, new Boolean[] {true, true})); + assertValues(fixture.full(), null, 2.0); + assertValues(fixture.selected(new boolean[2]), null, null); + fixture.init(block(new Double[] {10.0, 20.0}, new Boolean[] {true, true})); + assertValues(fixture.selected(new boolean[] {false, true}), null, null); + fixture.init(block(new Double[] {25.0}, new Boolean[] {true})); + assertValues(fixture.full(), 5.0); + } + + @Test + public void testBinaryCaseAndFullProjectionPreserveStateInBothOrders() throws Exception { + for (boolean caseFirst : new boolean[] {true, false}) { + List leaves = new ArrayList<>(); + ColumnTransformer input = input(leaves, DOUBLE, 0); + ColumnTransformer diff = newDiff(input, input(leaves, BOOLEAN, 1)); + ColumnTransformer caseResult = + caseWhen(greater(input, constant(leaves, 6)), diff, constant(leaves, 100)); + try (Operator operator = + projection( + leaves, + caseFirst ? Arrays.asList(caseResult, diff) : Arrays.asList(diff, caseResult), + block(new Double[] {1.0, 2.0, 3.0}, new Boolean[] {true, true, true}), + block(new Double[] {4.0, 5.0, 6.0}, new Boolean[] {true, true, true}), + block(new Double[] {7.0}, new Boolean[] {true}))) { + int diffIndex = caseFirst ? 1 : 0; + int caseIndex = 1 - diffIndex; + TsBlock first = operator.next(); + assertValues(first.getColumn(diffIndex), null, 1.0, 1.0); + assertValues(first.getColumn(caseIndex), 100.0, 100.0, 100.0); + TsBlock second = operator.next(); + assertValues(second.getColumn(diffIndex), 1.0, 1.0, 1.0); + assertValues(second.getColumn(caseIndex), 100.0, 100.0, 100.0); + TsBlock third = operator.next(); + assertValues(third.getColumn(diffIndex), 1.0); + assertValues(third.getColumn(caseIndex), 1.0); + } + } + } + + @Test + public void testAddingCaseOrTermDoesNotDropNextBlockFirstRow() throws Exception { + for (boolean withCase : new boolean[] {false, true}) { + List leaves = new ArrayList<>(); + ColumnTransformer input = input(leaves, DOUBLE, 0); + ColumnTransformer diff = newDiff(input, input(leaves, BOOLEAN, 1)); + ColumnTransformer predicate = new IsNullColumnTransformer(BOOLEAN, reference(diff), true); + if (withCase) { + ColumnBuilder falseBuilder = BOOLEAN.createColumnBuilder(1); + falseBuilder.writeBoolean(false); + ConstantColumnTransformer no = new ConstantColumnTransformer(BOOLEAN, falseBuilder.build()); + leaves.add(no); + ColumnTransformer caseResult = + new TableCaseWhenThenColumnTransformer( + BOOLEAN, + Collections.singletonList(reference(greater(input, constant(leaves, 5)))), + Collections.singletonList(reference(greater(diff, constant(leaves, 0)))), + reference(no)); + predicate = + new LogicOrColumnTransformer(BOOLEAN, reference(predicate), reference(caseResult)); + } + IdentityColumnTransformer output = new IdentityColumnTransformer(DOUBLE, 0); + try (Operator operator = + new FilterAndProjectOperator( + new TestContext(), + new BlocksOperator( + block(new Double[] {1.0, 3.0, 4.0}, new Boolean[] {true, true, true}), + block(new Double[] {10.0, 11.0}, new Boolean[] {true, true})), + Arrays.asList(TSDataType.DOUBLE, TSDataType.BOOLEAN), + leaves, + reference(predicate), + Collections.emptyList(), + Collections.singletonList(output), + Collections.singletonList(reference(output)), + false, + true)) { + assertValues(operator.next().getColumn(0), 3.0, 4.0); + assertValues(operator.next().getColumn(0), 10.0, 11.0); + } + } + } + + @Test + public void testUnaryConsumedCacheDoesNotAdvanceStateAgain() { + Fixture fixture = unary(true); + fixture.init(block(1.0, 3.0, 6.0)); + Column first = fixture.full(); + assertValues(first, null, 2.0, 3.0); + assertEquals(0, fixture.diff.getColumnCachePositionCount()); + assertValues(fixture.full(), null, 2.0, 3.0); + fixture.diff.clearCache(); + assertValues(fixture.full(), null, 2.0, 3.0); + assertValues(fixture.full(), null, 2.0, 3.0); + } + + @Test + public void testUnaryChangingMasksUsesCompleteInput() { + Fixture fixture = unary(true); + fixture.init(block(1.0, 3.0, 6.0, 10.0)); + fixture.input.evaluateWithSelection(new boolean[] {true, false, true, false}); + for (boolean[] mask : masks(4)) { + fixture.diff.clearCache(); + assertValues(fixture.selected(mask), null, 2.0, 3.0, 4.0); + assertValues(fixture.full(), null, 2.0, 3.0, 4.0); + } + } + + @Test + public void testBinaryConsumedCacheDoesNotAdvanceStateAgain() { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, false, true})); + Column first = fixture.full(); + assertValues(first, null, 2.0, 3.0); + assertEquals(0, fixture.diff.getColumnCachePositionCount()); + assertValues(fixture.full(), null, 2.0, 3.0); + fixture.diff.clearCache(); + assertValues(fixture.full(), null, 2.0, 3.0); + assertValues(fixture.full(), null, 2.0, 3.0); + } + + @Test + public void testBinaryChangingMasksPreservesSelectedInputs() { + Fixture fixture = binary(); + fixture.init( + block(new Double[] {1.0, null, 3.0, 6.0}, new Boolean[] {true, true, false, true})); + boolean[] partial = {true, false, true, false}; + fixture.input.evaluateWithSelection(partial); + fixture.flags.evaluateWithSelection(partial); + Double[][] expected = { + {null, null, null, null}, {null, null, null, null}, {null, null, 2.0, 3.0} + }; + boolean[][] masks = masks(4); + for (int i = 0; i < masks.length; i++) { + // As on master, selected evaluation replaces child caches using the requested mask. + assertValues(fixture.selected(masks[i]), expected[i]); + fixture.diff.clearCache(); + assertValues(fixture.full(), null, null, 2.0, 3.0); + } + } + + @Test + public void testBinarySelectedRowsPreserveStateAtInputBoundaries() { + for (boolean selectedLastRow : new boolean[] {false, true}) { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, true, true})); + boolean[] mask = {true, true, selectedLastRow}; + assertValues(fixture.selected(mask), null, 2.0, selectedLastRow ? 3.0 : null); + fixture.diff.clearCache(); + assertValues(fixture.selected(mask), null, 2.0, selectedLastRow ? 3.0 : null); + fixture.init(block(new Double[] {10.0, 11.0}, new Boolean[] {true, true})); + assertValues(fixture.full(), selectedLastRow ? 4.0 : null, 1.0); + } + } + + @Test + public void testUnaryNullPolicyAndStateSurviveInputBoundaries() { + for (boolean ignoreNull : new boolean[] {true, false}) { + Fixture fixture = unary(ignoreNull); + fixture.init(block(1.0, 3.0, null)); + assertValues(fixture.selected(new boolean[3]), null, 2.0, null); + fixture.init(block(10.0, null, 20.0, null)); + assertValues(fixture.full(), ignoreNull ? 7.0 : null, null, ignoreNull ? 10.0 : null, null); + fixture.init(block(30.0, 32.0)); + assertValues(fixture.selected(new boolean[] {false, true}), ignoreNull ? 10.0 : null, 2.0); + fixture.init(block()); + assertValues(fixture.selected(new boolean[0])); + fixture.init(block(40.0, 41.0)); + assertValues(fixture.full(), 8.0, 1.0); + } + } + + @Test + public void testDynamicNullFlagsAndStateSurviveInputBoundaries() { + for (Boolean ignoreNull : new Boolean[] {true, false, null}) { + Fixture fixture = binary(); + fixture.init(block(new Double[] {1.0, 3.0, null}, new Boolean[] {false, null, ignoreNull})); + assertValues(fixture.selected(new boolean[] {true, true, true}), null, 2.0, null); + fixture.init( + block(new Double[] {10.0, null, 20.0, null}, new Boolean[] {false, true, false, null})); + assertValues(fixture.full(), Boolean.TRUE.equals(ignoreNull) ? 7.0 : null, null, 10.0, null); + fixture.init(block(new Double[] {30.0, 32.0}, new Boolean[] {true, false})); + assertValues(fixture.selected(new boolean[] {true, true}), null, 2.0); + fixture.init(block(new Double[0], new Boolean[0])); + assertValues(fixture.full()); + fixture.init(block(new Double[] {40.0, 41.0}, new Boolean[] {false, true})); + assertValues(fixture.full(), 8.0, 1.0); + } + } + + @Test + public void testNestedDiffResultsAreRetainedIndependently() { + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + IdentityColumnTransformer flags = new IdentityColumnTransformer(BOOLEAN, 1); + ColumnTransformer inner = new DiffColumnTransformer(DOUBLE, reference(input), reference(flags)); + ColumnTransformer outer = new DiffFunctionColumnTransformer(DOUBLE, reference(inner), true); + reference(outer); + ColumnTransformer[] nodes = + ColumnTransformer.collectTransformers(Collections.singletonList(outer)); + for (ColumnTransformer node : nodes) { + node.clearCacheForNewInput(); + } + TsBlock first = block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, false, true}); + input.initFromTsBlock(first); + flags.initFromTsBlock(first); + outer.evaluateWithSelection(new boolean[3]); + assertValues(outer.getColumn(), null, null, 1.0); + outer.clearCache(); + outer.tryEvaluate(); + assertValues(outer.getColumn(), null, null, 1.0); + for (ColumnTransformer node : nodes) { + node.clearCacheForNewInput(); + } + TsBlock second = block(new Double[] {10.0, 11.0}, new Boolean[] {false, true}); + input.initFromTsBlock(second); + flags.initFromTsBlock(second); + outer.evaluateWithSelection(new boolean[] {false, true}); + assertValues(outer.getColumn(), 1.0, -3.0); + outer.tryEvaluate(); + assertValues(outer.getColumn(), 1.0, -3.0); + } + + @Test + public void testAlreadyReturnedColumnsSurviveReevaluationAndNewInputs() { + for (Fixture fixture : Arrays.asList(unary(true), binary())) { + fixture.init(block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, false, true})); + Column first = fixture.full(); + fixture.selected(new boolean[] {false, true, false}); + fixture.diff.clearCache(); + assertValues(fixture.full(), null, 2.0, 3.0); + fixture.init(block(new Double[] {10.0, 11.0}, new Boolean[] {true, true})); + assertValues(fixture.full(), 4.0, 1.0); + assertValues(first, null, 2.0, 3.0); + } + } + + @Test + public void testBinarySelectionPreservesOuterCaseCastGuard() throws Exception { + List leaves = new ArrayList<>(); + ColumnTransformer text = input(leaves, STRING, 0); + ColumnTransformer flags = input(leaves, BOOLEAN, 1); + ColumnTransformer valid = input(leaves, BOOLEAN, 2); + ColumnTransformer cast = + new CastFunctionColumnTransformer(DOUBLE, reference(text), ZoneId.of("UTC")); + ColumnTransformer diff = newDiff(cast, flags); + ColumnTransformer result = caseWhen(valid, diff, constant(leaves, 0)); + ColumnBuilder strings = STRING.createColumnBuilder(3); + for (String value : new String[] {"n/a", "5", "8"}) { + strings.writeBinary(new Binary(value, java.nio.charset.StandardCharsets.UTF_8)); + } + ColumnBuilder flagsBuilder = BOOLEAN.createColumnBuilder(3); + ColumnBuilder validBuilder = BOOLEAN.createColumnBuilder(3); + for (int i = 0; i < 3; i++) { + flagsBuilder.writeBoolean(true); + validBuilder.writeBoolean(i > 0); + } + TsBlock block = + new TsBlock( + new TimeColumn(3, new long[3]), + strings.build(), + flagsBuilder.build(), + validBuilder.build()); + try (Operator operator = projection(leaves, Collections.singletonList(result), block, block)) { + assertValues(operator.next().getColumn(0), 0.0, null, 3.0); + assertValues(operator.next().getColumn(0), 0.0, null, 3.0); + } + } + + @Test + public void testUnaryCaseFilterAndCommonProjectionAdvanceStateOnce() throws Exception { + assertCaseFilterAndCommonProjection(false); + } + + @Test + public void testDynamicFlagCaseFilterAndCommonProjectionAdvanceStateOnce() throws Exception { + assertCaseFilterAndCommonProjection(true); + } + + @Test + public void testTwoCasesAndFullProjectionPreserveEachSelection() throws Exception { + for (boolean dynamic : new boolean[] {false, true}) { + List leaves = new ArrayList<>(); + ColumnTransformer input = input(leaves, DOUBLE, 0); + ColumnTransformer flags = dynamic ? input(leaves, BOOLEAN, 1) : null; + ColumnTransformer diff = newDiff(input, flags); + ColumnTransformer first = + caseWhen(greater(input, constant(leaves, 2)), diff, constant(leaves, 0)); + ColumnTransformer second = + caseWhen(greater(input, constant(leaves, 2)), constant(leaves, 0), diff); + try (Operator operator = + projection( + leaves, + Arrays.asList(first, second, diff), + block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, false, true}), + block(new Double[] {10.0, 11.0}, new Boolean[] {false, true}))) { + TsBlock firstBlock = operator.next(); + assertValues(firstBlock.getColumn(0), 0.0, dynamic ? null : 2.0, 3.0); + assertValues(firstBlock.getColumn(1), null, 0.0, 0.0); + assertValues(firstBlock.getColumn(2), null, 2.0, 3.0); + TsBlock secondBlock = operator.next(); + assertValues(secondBlock.getColumn(0), 4.0, 1.0); + assertValues(secondBlock.getColumn(1), 0.0, 0.0); + assertValues(secondBlock.getColumn(2), 4.0, 1.0); + } + } + } + + @Test + public void testDiffPreservesMasterInputSelection() throws Exception { + for (boolean dynamic : new boolean[] {false, true}) { + List leaves = new ArrayList<>(); + ColumnTransformer input = input(leaves, DOUBLE, 0); + ColumnTransformer flags = dynamic ? input(leaves, BOOLEAN, 1) : null; + ColumnTransformer diff = newDiff(input, flags); + ColumnTransformer result = caseWhen(greater(input, constant(leaves, 0)), input, diff); + try (Operator operator = + projection( + leaves, + Collections.singletonList(result), + block( + new Double[] {1.0, -2.0, 3.0, -4.0}, new Boolean[] {true, false, true, false}))) { + assertValues( + operator.next().getColumn(0), 1.0, dynamic ? null : -3.0, 3.0, dynamic ? null : -7.0); + } + } + } + + @Test + public void testBinaryCaseFlagsKeepMasterSelectedEvaluation() { + List leaves = new ArrayList<>(); + IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + leaves.add(input); + ColumnBuilder trueBuilder = BOOLEAN.createColumnBuilder(1); + trueBuilder.writeBoolean(true); + ConstantColumnTransformer yes = new ConstantColumnTransformer(BOOLEAN, trueBuilder.build()); + ColumnBuilder falseBuilder = BOOLEAN.createColumnBuilder(1); + falseBuilder.writeBoolean(false); + ConstantColumnTransformer no = new ConstantColumnTransformer(BOOLEAN, falseBuilder.build()); + leaves.add(yes); + leaves.add(no); + ColumnTransformer flags = + new TableCaseWhenThenColumnTransformer( + BOOLEAN, + Collections.singletonList(reference(greater(input, constant(leaves, 0)))), + Collections.singletonList(reference(yes)), + reference(no)); + ColumnTransformer diff = new DiffColumnTransformer(DOUBLE, reference(input), reference(flags)); + reference(diff); + ColumnTransformer[] nodes = + ColumnTransformer.collectTransformers(Collections.singletonList(diff)); + TsBlock[] blocks = {block(1.0, 3.0, null, 6.0), block(10.0, null, 14.0), block(20.0, 21.0)}; + Double[][] expected = {{null, 2.0, null, null}, {4.0, null, null}, {6.0, 1.0}}; + for (int i = 0; i < blocks.length; i++) { + for (ColumnTransformer node : nodes) { + node.clearCacheForNewInput(); + } + for (LeafColumnTransformer leaf : leaves) { + leaf.initFromTsBlock(blocks[i]); + } + diff.evaluateWithSelection(masks(blocks[i].getPositionCount())[2]); + assertValues(diff.getColumn(), expected[i]); + diff.clearCache(); + diff.evaluateWithSelection(masks(blocks[i].getPositionCount())[2]); + assertValues(diff.getColumn(), expected[i]); + } + diff.close(); + } + + private static void assertCaseFilterAndCommonProjection(boolean dynamic) throws Exception { + List filterLeaves = new ArrayList<>(); + ColumnTransformer input = input(filterLeaves, DOUBLE, 0); + ColumnTransformer flags = dynamic ? input(filterLeaves, BOOLEAN, 1) : null; + ColumnTransformer diff = newDiff(input, flags); + ColumnTransformer predicate = + greater( + caseWhen( + greater(diff, constant(filterLeaves, 1)), + constant(filterLeaves, 1), + constant(filterLeaves, 0)), + constant(filterLeaves, 0)); + IdentityColumnTransformer output = new IdentityColumnTransformer(DOUBLE, 2); + try (Operator operator = + new FilterAndProjectOperator( + new TestContext(), + new BlocksOperator( + block(new Double[] {1.0, 3.0, 6.0}, new Boolean[] {true, false, true}), + block(new Double[] {10.0, 11.0}, new Boolean[] {false, true})), + Arrays.asList(TSDataType.DOUBLE, TSDataType.BOOLEAN, TSDataType.DOUBLE), + filterLeaves, + reference(predicate), + Collections.singletonList(reference(diff)), + Collections.singletonList(output), + Collections.singletonList(reference(output)), + false, + true)) { + assertValues(operator.next().getColumn(0), 2.0, 3.0); + assertValues(operator.next().getColumn(0), 4.0); + } + } + + private static Operator projection( + List leaves, List outputs, TsBlock... blocks) { + List roots = new ArrayList<>(); + for (ColumnTransformer output : outputs) { + roots.add(reference(output)); + } + return new FilterAndProjectOperator( + new TestContext(), + new BlocksOperator(blocks), + Collections.emptyList(), + Collections.emptyList(), + null, + Collections.emptyList(), + leaves, + roots, + false, + false); + } + + private static Fixture unary(boolean ignoreNull) { + return new Fixture(false, ignoreNull); + } + + private static Fixture binary() { + return new Fixture(true, true); + } + + private static ColumnTransformer newDiff(ColumnTransformer input, ColumnTransformer flags) { + return flags == null + ? new DiffFunctionColumnTransformer(DOUBLE, reference(input), true) + : new DiffColumnTransformer(DOUBLE, reference(input), reference(flags)); + } + + private static ColumnTransformer reference(ColumnTransformer transformer) { + transformer.addReferenceCount(); + return transformer; + } + + private static ColumnTransformer input( + List leaves, org.apache.tsfile.read.common.type.Type type, int index) { + IdentityColumnTransformer input = new IdentityColumnTransformer(type, index); + leaves.add(input); + return input; + } + + private static ColumnTransformer constant(List leaves, double value) { + ConstantColumnTransformer constant = + new ConstantColumnTransformer(DOUBLE, block(value).getColumn(0)); + leaves.add(constant); + return constant; + } + + private static ColumnTransformer greater(ColumnTransformer left, ColumnTransformer right) { + return new CompareGreaterThanColumnTransformer(BOOLEAN, reference(left), reference(right)); + } + + private static ColumnTransformer caseWhen( + ColumnTransformer when, ColumnTransformer then, ColumnTransformer otherwise) { + return new TableCaseWhenThenColumnTransformer( + DOUBLE, + Collections.singletonList(reference(when)), + Collections.singletonList(reference(then)), + reference(otherwise)); + } + + private static boolean[][] masks(int size) { + boolean[] partial = new boolean[size]; + boolean[] all = new boolean[size]; + for (int i = 0; i < size; i++) { + partial[i] = (i & 1) == 1; + all[i] = true; + } + return new boolean[][] {new boolean[size], partial, all}; + } + + private static TsBlock block(Double... values) { + return block(values, null); + } + + private static TsBlock block(Double[] values, Boolean[] flags) { + ColumnBuilder numbers = DOUBLE.createColumnBuilder(values.length); + ColumnBuilder booleans = BOOLEAN.createColumnBuilder(values.length); + for (int i = 0; i < values.length; i++) { + if (values[i] == null) { + numbers.appendNull(); + } else { + numbers.writeDouble(values[i]); + } + if (flags == null || flags[i] == null) { + booleans.appendNull(); + } else { + booleans.writeBoolean(flags[i]); + } + } + return new TsBlock( + new TimeColumn(values.length, new long[values.length]), numbers.build(), booleans.build()); + } + + private static void assertValues(Column column, Double... expected) { + assertEquals(expected.length, column.getPositionCount()); + for (int i = 0; i < expected.length; i++) { + if (expected[i] == null) { + assertTrue(column.isNull(i)); + } else { + assertFalse(column.isNull(i)); + assertEquals(expected[i], column.getDouble(i), 0.0); + } + } + } + + private static class Fixture { + final IdentityColumnTransformer input = new IdentityColumnTransformer(DOUBLE, 0); + final IdentityColumnTransformer flags; + final ColumnTransformer diff; + final ColumnTransformer[] nodes; + + Fixture(boolean dynamic, boolean ignoreNull) { + flags = dynamic ? new IdentityColumnTransformer(BOOLEAN, 1) : null; + diff = + dynamic + ? new DiffColumnTransformer(DOUBLE, reference(input), reference(flags)) + : new DiffFunctionColumnTransformer(DOUBLE, reference(input), ignoreNull); + reference(diff); + nodes = ColumnTransformer.collectTransformers(Collections.singletonList(diff)); + } + + void init(TsBlock block) { + for (ColumnTransformer node : nodes) { + node.clearCacheForNewInput(); + } + input.initFromTsBlock(block); + if (flags != null) { + flags.initFromTsBlock(block); + } + } + + Column full() { + diff.tryEvaluate(); + return diff.getColumn(); + } + + Column selected(boolean[] mask) { + diff.evaluateWithSelection(mask); + return diff.getColumn(); + } + } + + private static class TestContext extends CommonOperatorContext { + TestContext() { + super(0, new PlanNodeId("diff-cache-lifecycle"), "diff-cache-lifecycle"); + } + + @Override + public MemoryReservationManager getMemoryReservationContext() { + return null; + } + + @Override + public int getFragmentId() { + return 0; + } + + @Override + public int getPipelineId() { + return 0; + } + + @Override + public long ramBytesUsed() { + return 0; + } + } + + private static class BlocksOperator implements Operator { + private final TsBlock[] blocks; + private final CommonOperatorContext context = new TestContext(); + private int position; + + BlocksOperator(TsBlock... blocks) { + this.blocks = blocks; + } + + @Override + public CommonOperatorContext getOperatorContext() { + return context; + } + + @Override + public TsBlock next() { + return blocks[position++]; + } + + @Override + public boolean hasNext() { + return position < blocks.length; + } + + @Override + public boolean isFinished() { + return position >= blocks.length; + } + + @Override + public void close() {} + + @Override + public long calculateMaxPeekMemory() { + return 0; + } + + @Override + public long calculateMaxReturnSize() { + return 0; + } + + @Override + public long calculateRetainedSizeAfterCallingNext() { + return 0; + } + + @Override + public long ramBytesUsed() { + return 0; + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java index 9120444ba0d23..b2ad08edfe391 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java @@ -21,6 +21,7 @@ import org.apache.iotdb.calc.transformation.dag.column.ColumnTransformer; import org.apache.iotdb.calc.transformation.dag.column.leaf.LeafColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffColumnTransformer; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.schema.node.role.IDeviceMNode; import org.apache.iotdb.commons.schema.table.TsTable; @@ -40,6 +41,7 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.function.BiFunction; @@ -49,8 +51,12 @@ public class DeviceAttributeUpdater extends DeviceUpdater { private final List commonTransformerList; + private final boolean[] commonSupportsSelection; + private final boolean[] commonUsesRetainedDiff; + private final List commonDiffTransformers = new ArrayList<>(); private final List projectLeafColumnTransformerList; private final List projectOutputTransformerList; + private final ColumnTransformer[] projectCacheTransformers; private final TriConsumer attributeUpdater; private final TsBlockBuilder filterTsBlockBuilder; private final List attributePointers = new ArrayList<>(); @@ -75,13 +81,39 @@ public DeviceAttributeUpdater( super( filterLeafColumnTransformerList, filterOutputTransformer, + commonTransformerList, columnSchemaList, attributeProvider, database, table); this.commonTransformerList = commonTransformerList; + this.commonSupportsSelection = new boolean[commonTransformerList.size()]; + this.commonUsesRetainedDiff = new boolean[commonTransformerList.size()]; + for (int i = 0; i < commonTransformerList.size(); i++) { + ColumnTransformer common = commonTransformerList.get(i); + commonSupportsSelection[i] = common.canEvaluateWithSelection(); + if (!commonSupportsSelection[i]) { + for (ColumnTransformer transformer : + ColumnTransformer.collectTransformers(Collections.singletonList(common))) { + if (transformer instanceof DiffColumnTransformer) { + commonUsesRetainedDiff[i] = true; + break; + } + } + } + } + for (ColumnTransformer transformer : + ColumnTransformer.collectTransformers(commonTransformerList)) { + if (transformer instanceof DiffColumnTransformer) { + DiffColumnTransformer diff = (DiffColumnTransformer) transformer; + diff.retainResultForReuse(); + commonDiffTransformers.add(diff); + } + } this.projectLeafColumnTransformerList = projectLeafColumnTransformerList; this.projectOutputTransformerList = projectOutputTransformerList; + this.projectCacheTransformers = + ColumnTransformer.collectTransformers(projectOutputTransformerList); this.attributeUpdater = attributeUpdater; this.filterTsBlockBuilder = new TsBlockBuilder(filterOutputDataTypes); this.attributeNames = attributeNames; @@ -95,8 +127,26 @@ public void handleDeviceNode(final IDeviceMNode node) throws Metadata } protected void update() throws MetadataException { + try { + updateAttributes(); + } finally { + commonDiffTransformers.forEach(DiffColumnTransformer::releaseRetainedResult); + } + } + + private void updateAttributes() throws MetadataException { + if (!withoutFilter() && indexes.isEmpty()) { + // No device needs an update. Do not retry expressions excluded by the filter's CASE guard. + attributePointers.clear(); + super.clear(); + return; + } + final TsBlock block = getFilterTsBlock(); + for (ColumnTransformer transformer : projectCacheTransformers) { + transformer.clearCacheForNewInput(); + } projectLeafColumnTransformerList.forEach( leafColumnTransformer -> leafColumnTransformer.initFromTsBlock(block)); @@ -145,9 +195,40 @@ private TsBlock getFilterTsBlock() { final List filterResultColumns = new ArrayList<>(Arrays.asList(curBlock.getValueColumns())); - // Get result of calculated common sub expressions - commonTransformerList.forEach( - columnTransformer -> filterResultColumns.add(columnTransformer.getColumn())); + if (!commonTransformerList.isEmpty()) { + boolean[] selection = null; + // CASE may clear a common expression. Recompute only rows whose filter guard passed. + commonDiffTransformers.forEach(DiffColumnTransformer::reuseRetainedResult); + for (int i = 0; i < commonTransformerList.size(); i++) { + ColumnTransformer transformer = commonTransformerList.get(i); + if (commonSupportsSelection[i] && transformer.getColumnCachePositionCount() > 0) { + filterResultColumns.add(transformer.getColumn()); + continue; + } + if (commonSupportsSelection[i] || commonUsesRetainedDiff[i]) { + if (selection == null) { + selection = new boolean[deviceSchemaBatch.size()]; + for (int index : indexes) { + selection[index] = true; + } + } + if (commonSupportsSelection[i]) { + transformer.tryEvaluateWithSelection(selection); + } else { + // Rebuild wrappers from the DIFF results actually used by WHERE, including nulls. + transformer.evaluateWithSelection(selection); + } + } else { + // Stateful transformers must be re-evaluated over every row: a masked evaluation + // would read the masked rows as nulls and corrupt their running state. The snapshot + // taken at the input boundary makes the full re-evaluation idempotent. + // A preceding common may have left a masked child cached. Rebuild the subtree as well, + // otherwise this supposedly full pass can still read masked nulls as actual input. + transformer.tryEvaluateWithFreshInputs(); + } + filterResultColumns.add(transformer.getColumn()); + } + } final ColumnBuilder[] columnBuilders = filterTsBlockBuilder.getValueColumnBuilders(); final int rowCount = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java index db7d0a2b374bf..37f2efcd98342 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java @@ -35,6 +35,7 @@ import org.apache.tsfile.read.common.block.TsBlockBuilder; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.stream.Collectors; @@ -45,6 +46,7 @@ public abstract class DevicePredicateHandler implements AutoCloseable { private final List filterLeafColumnTransformerList; protected final ColumnTransformer filterOutputTransformer; + private final ColumnTransformer[] filterCacheTransformers; private final List inputDataTypes; private final List columnSchemaList; private final int tagIndex; @@ -59,14 +61,36 @@ public abstract class DevicePredicateHandler implements AutoCloseable { protected TsBlock curBlock; protected Column curFilterColumn; + // Convenience constructor for callers that do not share filter and projection expressions. protected DevicePredicateHandler( final List filterLeafColumnTransformerList, final ColumnTransformer filterOutputTransformer, final List columnSchemaList, final String database, final TsTable table) { + this( + filterLeafColumnTransformerList, + filterOutputTransformer, + Collections.emptyList(), + columnSchemaList, + database, + table); + } + + protected DevicePredicateHandler( + final List filterLeafColumnTransformerList, + final ColumnTransformer filterOutputTransformer, + final List commonTransformerList, + final List columnSchemaList, + final String database, + final TsTable table) { this.filterLeafColumnTransformerList = filterLeafColumnTransformerList; this.filterOutputTransformer = filterOutputTransformer; + List filterRoots = new ArrayList<>(commonTransformerList); + if (filterOutputTransformer != null) { + filterRoots.add(filterOutputTransformer); + } + this.filterCacheTransformers = ColumnTransformer.collectTransformers(filterRoots); this.columnSchemaList = columnSchemaList; this.inputDataTypes = columnSchemaList.stream() @@ -106,6 +130,10 @@ public void prepareBatchResult() { return; } + // A CASE can leave shared expressions cached after the previous batch was consumed. + for (ColumnTransformer transformer : filterCacheTransformers) { + transformer.clearCacheForNewInput(); + } // feed Filter ColumnTransformer, including TimeStampColumnTransformer and constant filterLeafColumnTransformerList.forEach( leafColumnTransformer -> leafColumnTransformer.initFromTsBlock(curBlock)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java index 2679f501dd060..68bff8ea81398 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java @@ -31,6 +31,7 @@ import org.apache.tsfile.utils.Binary; +import java.util.Collections; import java.util.List; import java.util.function.BiFunction; @@ -38,6 +39,7 @@ public abstract class DeviceUpdater extends DevicePredicateHandler { private final BiFunction attributeProvider; + // Convenience constructor for callers that do not share filter and projection expressions. protected DeviceUpdater( final List filterLeafColumnTransformerList, final ColumnTransformer filterOutputTransformer, @@ -45,9 +47,28 @@ protected DeviceUpdater( final BiFunction attributeProvider, final String database, final TsTable table) { + this( + filterLeafColumnTransformerList, + filterOutputTransformer, + Collections.emptyList(), + columnSchemaList, + attributeProvider, + database, + table); + } + + protected DeviceUpdater( + final List filterLeafColumnTransformerList, + final ColumnTransformer filterOutputTransformer, + final List commonTransformerList, + final List columnSchemaList, + final BiFunction attributeProvider, + final String database, + final TsTable table) { super( filterLeafColumnTransformerList, filterOutputTransformer, + commonTransformerList, columnSchemaList, database, table); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandlerTest.java new file mode 100644 index 0000000000000..8d2e6749a70d8 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandlerTest.java @@ -0,0 +1,568 @@ +/* + * 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.schema.source; + +import org.apache.iotdb.calc.transformation.dag.column.ColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.FailFunctionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.TableCaseWhenThenColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareEqualToColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareGreaterThanColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.ConstantColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.IdentityColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.leaf.LeafColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.IsNullColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.CastFunctionColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DiffColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.unary.scalar.UpperColumnTransformer; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.schema.node.role.IDeviceMNode; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema; +import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TableDeviceInfo; +import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.impl.ShowDevicesResult; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.column.BinaryColumn; +import org.apache.tsfile.read.common.block.column.BooleanColumn; +import org.apache.tsfile.read.common.block.column.DoubleColumn; +import org.apache.tsfile.utils.Binary; +import org.junit.Test; + +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN; +import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE; +import static org.apache.tsfile.read.common.type.StringType.STRING; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class DevicePredicateHandlerTest { + + private static final List COLUMNS = + Collections.singletonList(new TagColumnSchema("tag", TSDataType.STRING)); + + @Test + public void testPredicateFilterDoesNotReuseFullBatches() throws Exception { + Graph graph = new Graph(); + ColumnTransformer predicate = graph.matchesCase(graph.input(0)); + try (DevicePredicateFilter filter = predicateFilter(graph, predicate)) { + int batchSize = DevicePredicateHandler.DEFAULT_MAX_TS_BLOCK_LINE_NUMBER; + for (String value : new String[] {"match", "other"}) { + for (int i = 0; i < batchSize; i++) { + filter.addBatch(device(value)); + } + // addBatch evaluates full batches automatically; exhausting the iterator resets the batch. + List result = consume(filter); + assertEquals(batchSize, result.size()); + for (String actual : result) { + assertEquals(value, actual); + } + } + } + } + + @Test + public void testPredicateFilterDoesNotReuseDifferentSizedBatches() throws Exception { + Graph graph = new Graph(); + ColumnTransformer predicate = graph.matchesCase(graph.input(0)); + try (DevicePredicateFilter filter = predicateFilter(graph, predicate)) { + assertEquals(Collections.singletonList("match"), filterBatch(filter, "match")); + assertEquals(Arrays.asList("other", "match"), filterBatch(filter, "other", "match")); + assertEquals(Collections.emptyList(), filterBatch(filter, "skip")); + } + } + + @Test + public void testBlackListConstructorDoesNotReusePredicateAcrossBatches() throws Exception { + Graph graph = new Graph(); + ColumnTransformer predicate = graph.matchesCase(graph.input(0)); + List> nodes = + Arrays.asList(node("match", 0), node("other", 1), node("match", 2)); + try (DeviceBlackListConstructor constructor = + new DeviceBlackListConstructor( + graph.leaves, + reference(predicate), + "table", + COLUMNS, + (pointer, name) -> null, + mock(MemSchemaRegionStatistics.class), + "test", + new TsTable("table"))) { + constructor.handleDeviceNode(nodes.get(0)); + constructor.prepareBatchResult(); + constructor.update(); + constructor.handleDeviceNode(nodes.get(1)); + constructor.handleDeviceNode(nodes.get(2)); + constructor.prepareBatchResult(); + constructor.update(); + assertEquals(3, constructor.getPreDeletedNum()); + for (IDeviceMNode node : nodes) { + verify(node).preDeactivateSelfOrTemplate(); + } + } + } + + @Test + public void testAttributeUpdaterDoesNotReuseFilterAcrossBatches() throws Exception { + Graph filter = new Graph(); + ColumnTransformer predicate = filter.matchesCase(filter.input(0)); + Graph projection = new Graph(); + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + attributeUpdater( + filter, predicate, Collections.emptyList(), projection, updates, projection.input(0))) { + updateBatch(updater, "match"); + updateBatch(updater, "other", "match"); + assertEquals( + Arrays.asList( + Arrays.asList("match", "match"), + Arrays.asList("other", "other"), + Arrays.asList("match", "match")), + updates); + } + } + + @Test + public void testAttributeUpdaterDoesNotReuseProjectionAcrossBatches() throws Exception { + Graph projection = new Graph(); + ColumnTransformer input = projection.input(0); + ColumnTransformer caseWhen = projection.caseResult(input); + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + attributeUpdater( + null, null, Collections.emptyList(), projection, updates, caseWhen, input)) { + updateBatch(updater, "match"); + updateBatch(updater, "other", "match"); + assertEquals( + Arrays.asList( + Arrays.asList("match", "match", "match"), + Arrays.asList("other", "other", "other"), + Arrays.asList("match", "match", "match")), + updates); + } + } + + @Test + public void testAttributeUpdaterRecomputesCommonExpressionClearedByCase() throws Exception { + Graph filter = new Graph(); + ColumnTransformer shared = new UpperColumnTransformer(STRING, reference(filter.input(0))); + ColumnTransformer predicate = + filter.caseWhen( + filter.equal(shared, filter.constant("MATCH")), + filter.constant(true), + filter.constant(false)); + Graph projection = new Graph(); + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + attributeUpdater( + filter, + predicate, + Collections.singletonList(shared), + projection, + updates, + projection.input(1))) { + updateBatch(updater, "match"); + updateBatch(updater, "other", "match"); + assertEquals( + Arrays.asList(Arrays.asList("match", "MATCH"), Arrays.asList("match", "MATCH")), updates); + } + } + + @Test + public void testAttributeUpdaterSharedCastAndDiffDoNotWriteNullInEitherOrder() throws Exception { + for (boolean diffFirst : new boolean[] {false, true}) { + Graph filter = new Graph(); + ColumnTransformer input = filter.input(0); + ColumnTransformer value = + new CastFunctionColumnTransformer(DOUBLE, reference(input), ZoneId.of("UTC")); + ColumnTransformer diff = + new DiffColumnTransformer( + DOUBLE, reference(value), reference(filter.equal(input, filter.constant("x")))); + ColumnTransformer zero = + filter.leaf( + new ConstantColumnTransformer( + DOUBLE, new DoubleColumn(1, Optional.empty(), new double[] {0}))); + ColumnTransformer predicate = + filter.caseWhen( + filter.equal(input, filter.constant("n/a")), + filter.constant(false), + new CompareGreaterThanColumnTransformer(BOOLEAN, reference(diff), reference(zero))); + List common = + diffFirst + ? Arrays.asList(reference(diff), reference(value)) + : Arrays.asList(reference(value), reference(diff)); + Graph projection = new Graph(); + ColumnTransformer castInput = + projection.leaf(new IdentityColumnTransformer(DOUBLE, diffFirst ? 2 : 1)); + ColumnTransformer diffInput = + projection.leaf(new IdentityColumnTransformer(DOUBLE, diffFirst ? 1 : 2)); + ColumnTransformer a = + new CastFunctionColumnTransformer(STRING, reference(castInput), ZoneId.of("UTC")); + ColumnTransformer b = + new CastFunctionColumnTransformer(STRING, reference(diffInput), ZoneId.of("UTC")); + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + new DeviceAttributeUpdater( + Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE, TSDataType.DOUBLE), + filter.leaves, + reference(predicate), + common, + COLUMNS, + projection.leaves, + Arrays.asList(reference(a), reference(b)), + (pointer, name) -> null, + (deviceId, pointer, values) -> { + List row = new ArrayList<>(); + row.add(deviceId[0]); + for (Object result : values) { + row.add( + result == null + ? null + : ((Binary) result).getStringValue(TSFileConfig.STRING_CHARSET)); + } + updates.add(row); + }, + Arrays.asList("a", "b"), + "test", + new TsTable("table"))) { + updateBatch(updater, "1", "9", "5"); + assertEquals(Collections.singletonList(Arrays.asList("9", "9.0", "8.0")), updates); + } + } + } + + @Test + public void testAttributeUpdaterPreservesDiffGuardWithMatchingDevice() throws Exception { + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = guardedDiffUpdater(updates, "n/a", false)) { + updateBatch(updater, "1", "5", "n/a"); + assertEquals(Collections.singletonList(Arrays.asList("5", "4.0")), updates); + } + } + + @Test + public void testAttributeUpdaterWritesTheDiffNullSeenByWhere() throws Exception { + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = guardedDiffUpdater(updates, "3", true)) { + updateBatch(updater, "1", "3", "6"); + assertEquals(Arrays.asList(Arrays.asList("1", null), Arrays.asList("6", null)), updates); + } + } + + @Test + public void testAttributeUpdaterDiffDoesNotDependOnBatchBoundaries() throws Exception { + for (boolean split : new boolean[] {false, true}) { + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = guardedDiffUpdater(updates, "0", false)) { + if (split) { + updateBatch(updater, "10", "20", "0"); + updateBatch(updater, "5"); + } else { + updateBatch(updater, "10", "20", "0", "5"); + } + assertEquals(Collections.singletonList(Arrays.asList("20", "10.0")), updates); + } + } + } + + @Test + public void testAttributeUpdaterSkipsGuardedCommonDiffWhenNoDevicesMatch() throws Exception { + Graph filter = new Graph(); + ColumnTransformer input = filter.input(0); + ColumnTransformer value = + new CastFunctionColumnTransformer(DOUBLE, reference(input), ZoneId.of("UTC")); + ColumnTransformer diff = + new DiffColumnTransformer( + DOUBLE, reference(value), reference(filter.equal(input, filter.constant("x")))); + ColumnTransformer zero = + filter.leaf( + new ConstantColumnTransformer( + DOUBLE, new DoubleColumn(1, Optional.empty(), new double[] {0}))); + ColumnTransformer predicate = + filter.caseWhen( + filter.equal(input, filter.constant("n/a")), + filter.constant(false), + new CompareGreaterThanColumnTransformer(BOOLEAN, reference(diff), reference(zero))); + Graph projection = new Graph(); + ColumnTransformer commonInput = projection.leaf(new IdentityColumnTransformer(DOUBLE, 1)); + ColumnTransformer output = + new CastFunctionColumnTransformer(STRING, reference(commonInput), ZoneId.of("UTC")); + List> updates = new ArrayList<>(); + List updatedPointers = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + new DeviceAttributeUpdater( + Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE), + filter.leaves, + reference(predicate), + Collections.singletonList(reference(diff)), + COLUMNS, + projection.leaves, + Collections.singletonList(reference(output)), + (pointer, name) -> null, + (deviceId, pointer, values) -> { + updates.add( + Arrays.asList( + deviceId[0], + ((Binary) values[0]).getStringValue(TSFileConfig.STRING_CHARSET))); + updatedPointers.add(pointer); + }, + Collections.singletonList("a"), + "test", + new TsTable("table"))) { + // The CASE guard excludes the invalid CAST, and the first numeric DIFF is null. + updateBatch(updater, 10, "n/a", "5"); + assertTrue(updates.isEmpty()); + assertFalse(updater.hasComputedResult()); + assertTrue(updater.deviceSchemaBatch.isEmpty()); + + // Retain the filter's DIFF state, but not the skipped batch's attribute pointers. + updateBatch(updater, 20, "10", "20"); + assertEquals(Arrays.asList(Arrays.asList("10", "5.0"), Arrays.asList("20", "10.0")), updates); + assertEquals(Arrays.asList(20, 21), updatedPointers); + + updateBatch(updater, 30, "n/a"); + updateBatch(updater, 40, "30", "40"); + assertEquals( + Arrays.asList( + Arrays.asList("10", "5.0"), Arrays.asList("20", "10.0"), Arrays.asList("40", "10.0")), + updates); + assertEquals(Arrays.asList(20, 21, 41), updatedPointers); + } + } + + @Test + public void testAttributeUpdaterSkipsProjectionWhenNoDevicesMatch() throws Exception { + Graph filter = new Graph(); + ColumnTransformer predicate = filter.equal(filter.input(0), filter.constant("match")); + List> updates = new ArrayList<>(); + try (DeviceAttributeUpdater updater = + attributeUpdater( + filter, + predicate, + Collections.emptyList(), + new Graph(), + updates, + new FailFunctionColumnTransformer( + STRING, "Empty batches must not evaluate projections"))) { + updateBatch(updater, "skip"); + assertTrue(updates.isEmpty()); + assertFalse(updater.hasComputedResult()); + } + } + + private static DeviceAttributeUpdater guardedDiffUpdater( + List> updates, String excluded, boolean selectNull) { + Graph filter = new Graph(); + ColumnTransformer input = filter.input(0); + ColumnTransformer value = + new CastFunctionColumnTransformer(DOUBLE, reference(input), ZoneId.of("UTC")); + ColumnTransformer diff = + new DiffColumnTransformer( + DOUBLE, reference(value), reference(filter.equal(input, filter.constant("x")))); + ColumnTransformer condition = + selectNull + ? new IsNullColumnTransformer(BOOLEAN, reference(diff), false) + : new CompareGreaterThanColumnTransformer( + BOOLEAN, + reference(diff), + reference( + filter.leaf( + new ConstantColumnTransformer( + DOUBLE, new DoubleColumn(1, Optional.empty(), new double[] {0}))))); + ColumnTransformer predicate = + filter.caseWhen( + filter.equal(input, filter.constant(excluded)), filter.constant(false), condition); + Graph projection = new Graph(); + ColumnTransformer common = projection.leaf(new IdentityColumnTransformer(DOUBLE, 1)); + ColumnTransformer output = + new CastFunctionColumnTransformer(STRING, reference(common), ZoneId.of("UTC")); + return new DeviceAttributeUpdater( + Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE), + filter.leaves, + reference(predicate), + Collections.singletonList(reference(diff)), + COLUMNS, + projection.leaves, + Collections.singletonList(reference(output)), + (pointer, name) -> null, + (deviceId, pointer, values) -> + updates.add( + Arrays.asList( + deviceId[0], + values[0] == null + ? null + : ((Binary) values[0]).getStringValue(TSFileConfig.STRING_CHARSET))), + Collections.singletonList("a"), + "test", + new TsTable("table")); + } + + private static DevicePredicateFilter predicateFilter(Graph graph, ColumnTransformer predicate) { + return new DevicePredicateFilter( + graph.leaves, reference(predicate), COLUMNS, "test", new TsTable("table")); + } + + private static List filterBatch(DevicePredicateFilter filter, String... values) { + for (String value : values) { + filter.addBatch(device(value)); + } + filter.prepareBatchResult(); + return consume(filter); + } + + private static List consume(DevicePredicateFilter filter) { + List result = new ArrayList<>(); + while (filter.hasNext()) { + result.add(filter.next().getRawNodes()[3]); + } + return result; + } + + private static DeviceAttributeUpdater attributeUpdater( + Graph filter, + ColumnTransformer predicate, + List common, + Graph projection, + List> updates, + ColumnTransformer... outputs) { + List projectOutputs = new ArrayList<>(); + for (ColumnTransformer output : outputs) { + projectOutputs.add(reference(output)); + } + common.forEach(ColumnTransformer::addReferenceCount); + return new DeviceAttributeUpdater( + Collections.nCopies(1 + common.size(), TSDataType.STRING), + filter == null ? Collections.emptyList() : filter.leaves, + predicate == null ? null : reference(predicate), + common, + COLUMNS, + projection.leaves, + projectOutputs, + (pointer, name) -> null, + (deviceId, pointer, values) -> { + List row = new ArrayList<>(); + row.add(deviceId[0]); + for (Object value : values) { + row.add(((Binary) value).getStringValue(TSFileConfig.STRING_CHARSET)); + } + updates.add(row); + }, + Collections.nCopies(outputs.length, "attribute"), + "test", + new TsTable("table")); + } + + private static void updateBatch(DeviceAttributeUpdater updater, String... values) + throws Exception { + updateBatch(updater, 0, values); + } + + private static void updateBatch( + DeviceAttributeUpdater updater, int firstPointer, String... values) throws Exception { + for (int i = 0; i < values.length; i++) { + updater.handleDeviceNode(node(values[i], firstPointer + i)); + } + updater.prepareBatchResult(); + updater.update(); + } + + private static ShowDevicesResult device(String value) { + return new ShowDevicesResult(null, true, 0, new String[] {"root", "test", "table", value}); + } + + @SuppressWarnings("unchecked") + private static IDeviceMNode node(String value, int pointer) { + IDeviceMNode node = mock(IDeviceMNode.class); + TableDeviceInfo info = new TableDeviceInfo<>(); + info.setAttributePointer(pointer); + when(node.getDeviceInfo()).thenReturn(info); + when(node.getPartialPath()).thenReturn(new PartialPath(device(value).getRawNodes())); + when(node.isAlignedNullable()).thenReturn(true); + return node; + } + + private static ColumnTransformer reference(ColumnTransformer transformer) { + transformer.addReferenceCount(); + return transformer; + } + + private static class Graph { + private final List leaves = new ArrayList<>(); + + private ColumnTransformer input(int index) { + return leaf(new IdentityColumnTransformer(STRING, index)); + } + + private ColumnTransformer constant(String value) { + return leaf( + new ConstantColumnTransformer( + STRING, + new BinaryColumn( + 1, + Optional.empty(), + new Binary[] {new Binary(value, TSFileConfig.STRING_CHARSET)}))); + } + + private ColumnTransformer constant(boolean value) { + return leaf( + new ConstantColumnTransformer( + BOOLEAN, new BooleanColumn(1, Optional.empty(), new boolean[] {value}))); + } + + private ColumnTransformer leaf(LeafColumnTransformer transformer) { + leaves.add(transformer); + return transformer; + } + + private ColumnTransformer equal(ColumnTransformer left, ColumnTransformer right) { + return new CompareEqualToColumnTransformer(BOOLEAN, reference(left), reference(right)); + } + + private ColumnTransformer caseWhen( + ColumnTransformer when, ColumnTransformer then, ColumnTransformer otherwise) { + return new TableCaseWhenThenColumnTransformer( + then.getType(), + Collections.singletonList(reference(when)), + Collections.singletonList(reference(then)), + reference(otherwise)); + } + + private ColumnTransformer caseResult(ColumnTransformer input) { + return caseWhen(equal(input, constant("match")), constant("match"), constant("other")); + } + + private ColumnTransformer matchesCase(ColumnTransformer input) { + return equal(caseResult(input), input); + } + } +}