Fix CASE expression caches leaking across TsBlocks

[ColinLeeo]

Description

Closes #18792.

When a CASE expression and a later consumer share an aggregate or intermediate expression, CASE's short-circuit cleanup can invalidate that shared cache. Recomputing it for the later consumer restores its full reference count, leaving a cached column after the block has been consumed. Reusing the transformer DAG for the next TsBlock or schema-device batch can then produce stale values, incorrect filtering, or invalid column sizes.

Clear the filter, common-expression, and projection DAGs at each new input boundary in FilterAndProjectOperator, DevicePredicateHandler, and DeviceAttributeUpdater. This covers schema predicates used by SHOW/COUNT DEVICES and attribute updates, as well as the shared predicate handler used by device blacklist construction. At construction, each evaluation phase collects an identity-deduplicated flat array of its transformer nodes, including CASE descendants. The filter-phase array includes the filter and common-expression roots; the projection-phase array includes all projection roots. Each new-input boundary only loops over its array and clears each node's own cache before any output is evaluated. DAG traversal, hash lookups, and visited-set allocation occur only during construction.

Each transformer declares its child edges once, used by both DAG collection and evaluation-time cleanup. clearCacheForNewInput() is final and clears only the current node, so looping over the precollected array does not recursively revisit subtrees. The legacy evaluation-time clearCache() remains overridable for existing Mockito-based tests and preserves the CASE boundary: clearing a cached CASE must not invalidate an already-computed sibling column awaiting consumption. The operator copy constructor reuses the precollected arrays, and their retained memory is included in ramBytesUsed(). Schema predicate handlers and attribute updaters use the same construction-time collection.

Also recompute common expressions before reading their columns in DeviceAttributeUpdater, matching FilterAndProjectOperator: CASE may have cleared their caches during filtering within the same batch. Update the comments to distinguish this from new-input invalidation. ColumnCache reference-count semantics are unchanged.

Regression coverage

• 15 operator unit tests cover different-sized, equal-sized, and empty blocks; shared aggregates and intermediates; filter/common expressions; normal and RLE time columns; nested CASE branches; shared THEN results; short-circuit behavior; pending sibling columns; identity-based deduplication across a deep shared DAG and multiple roots; Mappable UDF descendants; and the operator copy constructor. The collection tests assert one traversal at construction and no further traversal while processing consecutive blocks.
• Two cache API tests verify identity-based collection even when different nodes compare equal, and that new-input clearing touches only the current node.
• Six schema unit tests cover full automatic batches, changing batch sizes, device predicate filtering, blacklist construction, attribute-update filtering and projection, and a common expression cleared by CASE. All six failed before the schema fixes with wrong results, column-size errors, or a null common column.
• The original aggregation integration test reproduces the issue across devices and time windows, including CAST, arithmetic, and shared THEN projections.
• The pending-sibling integration test guards against making evaluation-time CASE cleanup recursive. It passes on the original implementation; its expected:<3> but was:<0> failure came from an intermediate, overly recursive fix, not from the original bug.
• Three schema integration tests force one device per batch and exercise SHOW/COUNT DEVICES, attribute projections across batches, and UPDATE with a common expression cleared by CASE. DELETE DEVICES currently rejects CASE predicates during analysis, so blacklist construction is covered directly by its unit test rather than a CASE SQL reproducer.

Validation

• All 87 calc-commons unit tests passed.
• All 249 DataNode tests selected by *ColumnTransformerTest,DevicePredicateHandlerTest passed during a full-reactor English build: 243 tests in 40 existing transformer test classes plus the six schema regression tests. This includes all four BlobLengthColumnTransformerTest cases, which failed when the legacy clearCache() entry point was made final.
• The two CASE methods in IoTDBCaseWhenThenTableIT and all three methods in IoTDBCaseDeviceCacheIT passed with TableSimpleIT (1 ConfigNode + 1 DataNode), including the integration-test dependency reactor build. They were rerun after the flat-array optimization, along with the dependency reactor build.
• Full-reactor English test compilation and mvn -o test-compile -DskipTests -P with-zh-locale passed.
• Spotless was applied to all affected modules; git diff --check passed.

Local integration-test verification used -Drat.skip=true because an unrelated pre-existing, untracked SQL file fails the repository license scan. All added test files include the standard ASF license header.

Performance check for the flat-array optimization

A local JMH 1.37 benchmark compares the previous PR revision (50c38c16dd) with this implementation using the actual FilterAndProjectOperator, one row per block, eight rotating prebuilt input blocks, and output correctness checks. Both revisions use the same dependency classpath, Corretto JDK 17.0.13 on aarch64, a 512 MiB heap, and G1. Each revision has two forks per scenario, with the revision order reversed in the second round; each fork uses three 1-second warmup iterations and four 1-second measurement iterations.

Scenario (no CASE)	Previous PR, µs/block	Flat array, µs/block	Previous PR, B/block	Flat array, B/block
x + 1	0.100	0.073	720	328
16 outputs sharing S = x + 1	0.862	0.643	4824	3640
Filter (x0 + 1) > 0, project 16 columns	1.018	0.961	14672	13888

The allocation reduction was consistent across forks. Timing varied between forks, particularly for the filtered scenario, whose ranges overlap. These are operator microbenchmarks comparing two PR revisions, not an end-to-end SQL benchmark or a comparison against the unfixed merge base. The steady-state reset now performs an O(V) flat loop; the O(V + E) traversal and identity-set allocation happen once during construction.

---

This PR has:

• been self-reviewed.
• added comments explaining the cache lifecycle and intent.
• added unit tests covering the changed code paths.
• added integration tests.
• been tested in a test IoTDB cluster.

---

Key changed/added classes in this PR

• FilterAndProjectOperator
• ColumnTransformer and its composite transformer subclasses
• AbstractCaseWhenThenColumnTransformer
• DevicePredicateHandler and DeviceAttributeUpdater
• FilterAndProjectOperatorTest and DevicePredicateHandlerTest
• IoTDBCaseWhenThenTableIT and IoTDBCaseDeviceCacheIT

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[JackieTien97]

There is a performance concern with the unconditional per-block cache traversal in FilterAndProjectOperator (filter phase, projection phase). Each phase allocates a new identity-based visited set and traverses its transformer DAG. This also adds work to operators with no CASE expressions; an operator performing both filtering and projection does this twice per input block.

I compared the actual operator and changed transformer classes from the merge base fbd5fd22a6c9009dccd7f1a5b84bf7d61e5718e0 against PR head 50c38c16ddf29032a6d1cf1c89b2bcf67627e957, using the same dependencies. The following scenarios contain no CASE expressions. Results below are means across two JVM forks, with one row per input TsBlock:

Operator scenario	Before (µs/block)	PR (µs/block)	Time increase	Additional allocation/block
One addition projection, x + 1	0.084	0.128	~52%	400 B
16 projections sharing s = x + 1, with outputs s + 2 through s + 17	0.762	1.457	~91%	1,440 B
Filter (x0 + 1) > 0, then pass through 16 columns	1.001	1.272	~27%	~772 B

All three scenarios were slower in both forks. Method: JMH 1.37, Zulu JDK 17.0.4.1, one benchmark thread, average-time mode, 3 × 1 s warmup and 4 × 1 s measurement per fork, two forks per revision/scenario with revision order reversed in the second round, a 512 MiB G1 heap, and the GC allocation profiler. The transformer graphs were constructed directly and consumed eight rotating, prebuilt input blocks; setup checked every output against expected values before timing, and each measured invocation returned its result to JMH.

These are operator microbenchmark results, not end-to-end SQL regression percentages. They do not include SQL planning, storage scanning, aggregation, or network costs. The 1024-row variants and the separate 16-independent-additions scenario had substantial variation between forks; a longer-warmup check also showed variation, so I am not drawing a stable, general query-latency regression percentage from those results.

Suggested optimization: once the transformer DAG is fully constructed, collect and identity-deduplicate the nodes once, and retain a flat list for each evaluation phase. The filter-phase list should include the filter and common-expression roots; the projection-phase list should include all projection roots. Both should include CASE descendants. At each new-input boundary, iterate the corresponding list and clear only each node's own cache before initializing the leaves. Keep the existing evaluation-time CASE cleanup semantics unchanged; calling the recursive legacy clearCache() on every list entry would defeat the purpose and could change the CASE boundary behavior.

This should preserve the cross-block correctness fix while removing repeated DAG traversal, identity-set allocation/resizing, and hash lookups from the per-block path. The same approach can be applied to the schema-batch handlers. I recommend addressing this overhead, particularly for workloads producing many small TsBlocks.

[JackieTien97]

Thanks for the quick turnaround on the per-block traversal. The precomputed flat lists in 5e0cb1c address my earlier comment, and the original cross-block repro from #18792 now returns the correct rows.

Requesting changes for two problems in this PR:

1. Regression in the schema UPDATE path: an UPDATE whose CASE guards a common expression now fails where base succeeds (inline on DeviceAttributeUpdater).
2. IoTDBCaseDeviceCacheIT#testAttributeCommonExpressionClearedByCase does not exercise the path it targets: it passes on base (inline).

Same mechanism, not fixed by this PR. Please fix them here or open follow-ups (details inline):

3. Within a single block, CASE clears columns that a sibling operand has already computed but not read yet. Results: silent NULLs, NPEs, and UPDATE … SET a = concat(src, CASE …) overwriting attributes with NULL.
4. MultiColumnTransformer.evaluateWithSelection reads the position count after getColumn(), so CASE … THEN coalesce(x, …) END returns NULL.
5. diff() under a CASE filter is evaluated twice per block. With this PR the multi-block case returns wrong values instead of failing.

Root cause / design suggestion: short-circuit (evaluateWithSelection) results are stored in the shared, ref-counted ColumnCache, so CASE has to clear caches during evaluation. That clearing destroys columns other consumers still need and forces recomputation within the block (3 and 5; combined with the unmasked recompute, also 1). Resetting at block boundaries cannot help inside a block. Keeping selection-mode results out of the shared cache (for example, returning them directly to the CASE), evaluating the first WHEN with tryEvaluate(), and dropping the evaluation-time clearCache() calls would remove that recompute path and leave the block-boundary reset as a safety net. Item 1 would still need the filter mask.

Out of scope (pre-existing, file not touched by this PR): tree-model LogicAndColumnTransformer.evaluateWithSelection caches rightColumn instead of builder.build() (L80). As a result, CASE WHEN s1 > 100 THEN 1 WHEN s1 > 0 AND s2 > 0 THEN 2 ELSE 3 END returns 2 instead of 3 when s1 is NULL and s2 = 1 (same on base). Probably worth a separate issue.

All SQL in this review was run on 1C1D instances built from this PR (50c38c1, key cases re-checked on 5e0cb1c) and from the merge base fbd5fd2.

[JackieTien97]

Regression (must fix): tryEvaluate() recomputes a cleared common expression for every device in the batch, ignoring the filter. If the CASE was guarding that expression, the UPDATE now evaluates it on devices the guard excluded:

CREATE TABLE t (id STRING TAG, src STRING ATTRIBUTE, a STRING ATTRIBUTE, v DOUBLE FIELD);
INSERT INTO t(time, id, src, a, v) VALUES (1, 'd0', 'n/a', 'seed', 1), (2, 'd1', '5', 'seed', 1);
UPDATE t 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;

• base: succeeds; no device matches, so a stays seed
• this PR: 305: internal error. statemachine throws a runtime exception: org.apache.iotdb.commons.exception.SemanticException: Cannot cast n/a to INT32 type

If d1 has src = '50' instead (one device matches), base fails with the NPE this PR targets and this PR fails with the same cast error. The expected result is a = '51' for d1.

The query path has always behaved this way (FilterAndProjectOperator L191-L199): SELECT CAST(s AS INT32) + 1 FROM z4 WHERE CASE WHEN s <> 'n/a' THEN CAST(s AS INT32) > 10 ELSE false END fails with 701: Cannot cast n/a to INT32 type on both base and this PR.

Suggested fix for both call sites: recompute only the rows that passed the filter, for example

// ColumnTransformer
public void tryEvaluateWithSelection(boolean[] selection) {
  if (!columnCache.hasCached()) {
    evaluateWithSelection(selection);
  }
}

// here (in FilterAndProjectOperator: filterColumn / positionCount)
boolean[] selection = new boolean[deviceSchemaBatch.size()];
for (int i = 0; i < selection.length; i++) {
  selection[i] = satisfy(curFilterColumn, i);
}
for (ColumnTransformer transformer : commonTransformerList) {
  transformer.tryEvaluateWithSelection(selection);
  filterResultColumns.add(transformer.getColumn());
}

With this change, both UPDATEs and the SELECT return the expected results in my local run, and the calc-commons tests plus the datanode *ColumnTransformerTest / DevicePredicateHandlerTest still pass.

This block also duplicates FilterAndProjectOperator.getFilterTsBlock. The two copies have already drifted once: base had tryEvaluate() there but not here, which is the NPE this PR fixes. A single helper next to constructFilteredTsBlock would keep the fix in one place.

[JackieTien97]

Recomputing here also re-runs stateful functions on the same block after CASE has cleared them. Before this PR the multi-block case below failed; now it silently returns wrong values:

CREATE TABLE dt2 (time TIMESTAMP TIME, dev STRING TAG, x DOUBLE FIELD);
INSERT INTO dt2(time, dev, x) VALUES (1, 'd0', 1.0), (2, 'd0', 3.0), (3, 'd0', 6.0);
FLUSH;
INSERT INTO dt2(time, dev, x) VALUES (4, 'd0', 10.0), (5, 'd0', 11.0);
FLUSH;
SELECT time, x, diff(x) AS d FROM dt2 WHERE CASE WHEN diff(x) > 1.0 THEN 1.0 ELSE 0.0 END > 0.0;

(1C1D with max_tsblock_line_number=1, so [1, 3, 6] and [10, 11] reach this operator as separate blocks)

• expected: d = 2.0, 3.0, 4.0 at time 2, 3, 4
• base: 301: Index 2 out of bounds for length 2
• this PR: time 4 returns d = -1.0. The first WHEN already computed diff for [10, 11] and advanced its state to 11. CASE then clears that column, and this line runs diff again on the same block (10 - 11).
• With ELSE 1.0 (every row passes), time 1 returns -5.0 instead of NULL.

A selection mask (see the DeviceAttributeUpdater comment) doesn't help here, because DIFF evaluates every row anyway. The column computed by the first WHEN was complete and valid; it is lost only because CASE clears its WHEN subtree (see the root-cause note in the review summary).

[JackieTien97]

This preserves a pending sibling column only when the shared node is itself a CASE. CASE's own cleanup (L222-L227) still clears its WHEN/THEN/ELSE subtrees recursively. Any non-CASE node shared with an operand that the parent has already computed but not read yet is therefore wiped within the same block. All of these fail the same way on base and on this PR, with a single input block:

SQL	expected	actual
SELECT s1 + CASE WHEN s1 > 0 THEN 2 ELSE 3 END FROM p (s1 = 1, -2)	3.0, 1.0	NULL, NULL (tree model too)
SELECT time, s1 FROM p WHERE s1 + CASE WHEN s1 > 0 THEN 2 ELSE 3 END > 0	2 rows	empty set
SELECT s1 > CASE WHEN s1 > 0 THEN 2 ELSE -3 END FROM p	false, true	301: … "leftColumn" is null
SHOW DEVICES / COUNT DEVICES … WHERE src = CASE WHEN src = 'match' THEN 'match' ELSE 'other' END	all devices	301: … "leftColumn" is null
UPDATE t SET a = concat(src, CASE WHEN src = 'match' THEN '-m' ELSE '-o' END)	match-m / other-o	succeeds, but every a becomes NULL (old value lost)
greatest(s1, CASE …), s1 BETWEEN -5 AND CASE …, substring(s, 1, CASE …), CASE WHEN s IS NOT NULL THEN substring(s, 1, length(s) - 1) END	values	NULL

Cause: BinaryColumnTransformer.evaluate(), TernaryColumnTransformer.evaluate() / evaluateWithSelection(), MultiColumnTransformer.evaluate() and MappableUDFColumnTransformer.evaluate() / evaluateWithSelection() evaluate all children first and call getColumn() afterwards. When a later child is (or contains) a CASE, its cleanup clears the shared node, and the earlier child's read returns null / 0 positions.

BinaryColumnTransformer.evaluateWithSelection() already avoids this by reading the left column before evaluating the right one (see its comment). I applied that rule everywhere: read each child's column and position count right after evaluating it, before evaluating the next sibling. In my local run that fixes every row above, and the calc-commons tests and datanode transformer tests still pass.

The same argument applies to this opt-out itself: "preserve pending sibling columns" holds for every node type, not just CASE. The new tests only use the CASE … = src orientation, which happens to avoid the problem (see the IT comment).

[JackieTien97]

Not changed by this PR, but same mechanism, same class: evaluateWithSelection() reads getColumnCachePositionCount() at L72 after getColumn() has already released the first child's column (when that child has a single reference), so the result has 0 rows. Any multi-argument function under THEN/ELSE whose first argument is referenced only once returns NULL, on base and on this PR:

SELECT time, CASE WHEN time > 0 THEN coalesce(s1, 0.0) END FROM p  -- expected 1.0, -2.0; actual NULL, NULL
SELECT time, CASE WHEN time > 0 THEN greatest(s1, 0.5) END FROM p  -- expected 1.0, 0.5;  actual NULL, NULL

8bf2294 fixed this exact pattern in LogicAnd / LogicOr / LogicalAndMulti / LogicalOrMulti / MappableUDF but missed this class. Reading the position count before getColumn(), as evaluate() does at L50-L51, fixes it.

[JackieTien97]

Nit: CASE now keeps the same child list in three places: here, the clear loop in doTransform() (L222-L227) and close() (L255-L263). Since the goal is a single child-edge definition, consider forEachChild(ColumnTransformer::clearCache) in doTransform() and forEachChild(ColumnTransformer::close) in close(). The same applies to the close() methods of Unary/Binary/Ternary/Multi/MappableUDF. That way a new child edge cannot be added to one list and missed in another.

[JackieTien97]

Optional (same as base, not blocking): the reset runs when the next block arrives. Columns left over from the previous block therefore stay reachable while the driver is blocked and after the last block; an example is a shared expression recomputed after CASE, which keeps an unconsumed reference. Resetting at the end of next() (in a finally) would release them as soon as the output block is built.

[JackieTien97]

This test does not exercise the common-expression path it is named after. src is a bare column reference: ColumnTransformerBuilder.visitSymbolReference turns it into a leaf without consulting hasSeen, so commonTransformerList stays empty. The same UPDATE already returns the correct result on base (checked with max_tsblock_line_number=1 and one schema region). The unit test at DevicePredicateHandlerTest L167 likewise uses a leaf as the common transformer, which the planner never produces.

A statement that does go through the common-expression path:

UPDATE common_cache SET a = upper(src) WHERE CASE WHEN upper(src) = 'MATCH' THEN true ELSE false END

• base: 305: internal error. statemachine throws a runtime exception: java.lang.NullPointerException: Cannot invoke "org.apache.tsfile.block.column.Column.isNull(int)" because "curColumn" is null
• this PR: d0/d1 get MATCH, d2/d3 stay NULL

Please also cover the guarded-cast UPDATE from the DeviceAttributeUpdater comment.

[JackieTien97]

All new schema tests (and DevicePredicateHandlerTest) put the CASE on the left: CASE … = src. The mirrored predicate src = CASE … END already fails in the first batch on both base and this PR (301: … "leftColumn" is null; see the comment on shouldClearChildrenCache()). Please cover both orientations once that is fixed.

[JackieTien97]

Nit: this duplicates the production ValuesOperator in the same package. new ValuesOperator(new TestOperatorContext(), Arrays.asList(blocks)) should do the same job, and it won't need updating when the Operator interface changes.

[JackieTien97]

filter.input(0) as a common transformer cannot come from the planner (bare column references never become common expressions). This test therefore exercises the recompute mechanism, but not a DAG the planner builds. A shared non-leaf node (for example, a function over input) would match what UPDATE … SET a = upper(src) WHERE CASE WHEN upper(src) = 'MATCH' … produces.