From 26f7af4f5cf9e678ad27da62e11c9f4490cae946 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 15 Oct 2026 16:23:28 +0800 Subject: [PATCH 1/2] Reset CASE caches at input block boundaries Collect phase DAG nodes once and clear each cached column before a new block or device batch. Rebuild missing schema UPDATE common expressions with the filter mask while preserving existing selection evaluation. --- .../it/db/it/IoTDBCaseWhenThenTableIT.java | 50 ++ .../it/schema/IoTDBCaseDeviceCacheIT.java | 167 ++++++ .../process/FilterAndProjectOperator.java | 30 +- ...AbstractCaseWhenThenColumnTransformer.java | 11 + .../dag/column/ColumnTransformer.java | 40 ++ .../column/FailFunctionColumnTransformer.java | 7 + .../binary/BinaryColumnTransformer.java | 8 + .../column/leaf/LeafColumnTransformer.java | 7 + .../multi/MappableUDFColumnTransformer.java | 9 + .../column/multi/MultiColumnTransformer.java | 9 +- .../ternary/TernaryColumnTransformer.java | 9 + .../column/unary/UnaryColumnTransformer.java | 7 + .../process/FilterAndProjectOperatorTest.java | 501 ++++++++++++++++++ .../dag/column/ColumnTransformerTest.java | 110 ++++ .../schema/source/DeviceAttributeUpdater.java | 22 +- .../schema/source/DevicePredicateHandler.java | 28 + .../operator/schema/source/DeviceUpdater.java | 21 + .../source/DevicePredicateHandlerTest.java | 324 +++++++++++ 18 files changed, 1354 insertions(+), 6 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBCaseDeviceCacheIT.java create mode 100644 iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/process/FilterAndProjectOperatorTest.java create mode 100644 iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformerTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandlerTest.java 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..169ebd3084ae3 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 @@ -695,4 +695,54 @@ 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); + } } 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..a6d3276a037b9 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBCaseDeviceCacheIT.java @@ -0,0 +1,167 @@ +/* + * 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 { + // Keep devices in one schema region and exercise both mixed-row and repeated batches. + EnvFactory.getEnv().getConfig().getCommonConfig().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"}) { + 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 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,"))); + } + } + + 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..54a9ec2c0266c 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,13 @@ public void tryEvaluate() { } } + /** Reuses a cached result, or recomputes only selected rows when it is missing. */ + public void tryEvaluateWithSelection(boolean[] selection) { + if (!columnCache.hasCached()) { + evaluateWithSelection(selection); + } + } + public Column getColumn() { return columnCache.getColumn(); } @@ -115,4 +129,30 @@ 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(); + } + + /** 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..eeeb168a15b77 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; @@ -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..240ab51f5664e 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; @@ -100,4 +102,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..c7bd2df7d0dd5 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,6 +27,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; import java.util.stream.Collectors; public abstract class MultiColumnTransformer extends ColumnTransformer { @@ -68,7 +69,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 +106,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..f413481212b9e 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; @@ -118,4 +120,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/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/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..70087c89b6926 --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/ColumnTransformerTest.java @@ -0,0 +1,110 @@ +/* + * 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.leaf.IdentityColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.multi.CoalesceColumnTransformer; + +import org.apache.tsfile.block.column.Column; +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.TimeColumn; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Optional; + +import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +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); + } + + 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/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..c4721675bf96b 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 @@ -46,11 +46,13 @@ import java.util.stream.Collectors; import static org.apache.iotdb.calc.execution.operator.process.FilterAndProjectOperator.constructFilteredTsBlock; +import static org.apache.iotdb.calc.execution.operator.process.FilterAndProjectOperator.satisfy; public class DeviceAttributeUpdater extends DeviceUpdater { private final List commonTransformerList; 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,6 +77,7 @@ public DeviceAttributeUpdater( super( filterLeafColumnTransformerList, filterOutputTransformer, + commonTransformerList, columnSchemaList, attributeProvider, database, @@ -82,6 +85,8 @@ public DeviceAttributeUpdater( this.commonTransformerList = commonTransformerList; this.projectLeafColumnTransformerList = projectLeafColumnTransformerList; this.projectOutputTransformerList = projectOutputTransformerList; + this.projectCacheTransformers = + ColumnTransformer.collectTransformers(projectOutputTransformerList); this.attributeUpdater = attributeUpdater; this.filterTsBlockBuilder = new TsBlockBuilder(filterOutputDataTypes); this.attributeNames = attributeNames; @@ -97,6 +102,9 @@ public void handleDeviceNode(final IDeviceMNode node) throws Metadata protected void update() throws MetadataException { final TsBlock block = getFilterTsBlock(); + for (ColumnTransformer transformer : projectCacheTransformers) { + transformer.clearCacheForNewInput(); + } projectLeafColumnTransformerList.forEach( leafColumnTransformer -> leafColumnTransformer.initFromTsBlock(block)); @@ -145,9 +153,17 @@ 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 = new boolean[deviceSchemaBatch.size()]; + for (int i = 0; i < selection.length; i++) { + selection[i] = satisfy(curFilterColumn, i); + } + // CASE may clear a common expression. Recompute only rows whose filter guard passed. + for (ColumnTransformer transformer : commonTransformerList) { + transformer.tryEvaluateWithSelection(selection); + 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..b3cf75f120cc6 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; + // Retain the original constructor descriptor for precompiled subclasses without common roots. 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..c6690634e91cc 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; + // Retain the original constructor descriptor for precompiled subclasses without common roots. 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..149062fa53e48 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandlerTest.java @@ -0,0 +1,324 @@ +/* + * 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.TableCaseWhenThenColumnTransformer; +import org.apache.iotdb.calc.transformation.dag.column.binary.CompareEqualToColumnTransformer; +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.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.utils.Binary; +import org.junit.Test; + +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.StringType.STRING; +import static org.junit.Assert.assertEquals; +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); + } + } + + 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 { + for (int i = 0; i < values.length; i++) { + updater.handleDeviceNode(node(values[i], 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); + } + } +} From 770814c0f73da051e94f307da66c9590bcaa67c0 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 15 Oct 2026 17:50:50 +0800 Subject: [PATCH 2/2] Preserve DIFF input state during CASE reevaluation --- .../it/db/it/IoTDBCaseWhenThenTableIT.java | 164 +++++ .../dag/column/ColumnTransformer.java | 6 + .../unary/scalar/DiffColumnTransformer.java | 12 + .../scalar/DiffFunctionColumnTransformer.java | 12 + .../dag/column/DiffCacheLifecycleTest.java | 600 ++++++++++++++++++ 5 files changed, 794 insertions(+) create mode 100644 iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/DiffCacheLifecycleTest.java 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 169ebd3084ae3..7b6bb41ea4713 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 @@ -745,4 +745,168 @@ public void testReusingCasePreservesPendingSiblingColumn() { 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() { + 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); + tableResultSetEqualTest( + "SELECT 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"}, + new String[] { + "100.0,null,null,", + "100.0,2.0,2.0,", + "100.0,null,null,", + "null,200.0,7.0,", + "100.0,null,null,", + "null,200.0,null,", + "5.0,200.0,5.0,", + "100.0,null,null,", + "null,200.0,5.0," + }, + DATABASE); + tableResultSetEqualTest( + "SELECT " + + 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"}, + new String[] {"2.0,", "5.0,"}, + DATABASE); + } + + @Test + public void testLocalDiffPreservesMasterSelectionsAndCaseGuards() { + 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)"}) { + tableResultSetEqualTest( + "SELECT CASE WHEN x > 0 THEN x ELSE " + + diff + + " END AS a, " + + diff + + " AS d FROM case_partial_diff ORDER BY time", + new String[] {"a", "d"}, + diff.equals("diff(x)") + ? new String[] {"1.0,null,", "-3.0,-3.0,", "3.0,5.0,", "-7.0,-7.0,"} + : new String[] {"1.0,null,", "null,-3.0,", "3.0,5.0,", "null,-7.0,"}, + DATABASE); + } + 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() { + 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" + }); + tableResultSetEqualTest( + "SELECT 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"}, + new String[] {"null,", "null,", "3.0,", "null,", "null,", "4.0,"}, + DATABASE); + // 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); + } } 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 54a9ec2c0266c..60aaa43337a6f 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 @@ -133,6 +133,12 @@ public void clearCache() { /** 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. */ 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..83be67a1184cd 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 @@ -34,11 +34,21 @@ 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; + public DiffColumnTransformer( Type returnType, ColumnTransformer leftTransformer, ColumnTransformer rightTransformer) { super(returnType, leftTransformer, rightTransformer); } + @Override + protected void resetForNewInput() { + lastValueBeforeCurrentInput = lastValue; + lastValueIsNullBeforeCurrentInput = lastValueIsNull; + } + @Override protected void checkType() { // do nothing @@ -47,6 +57,8 @@ protected void checkType() { @Override protected void doTransform( Column leftColumn, Column rightColumn, ColumnBuilder builder, int positionCount) { + lastValue = lastValueBeforeCurrentInput; + lastValueIsNull = lastValueIsNullBeforeCurrentInput; Type leftType = leftTransformer.getType(); Type rightType = rightTransformer.getType(); for (int i = 0; i < positionCount; i++) { 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/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..786ad9eeea4e0 --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/transformation/dag/column/DiffCacheLifecycleTest.java @@ -0,0 +1,600 @@ +/* + * 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.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.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 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; + } + } +}