[
https://issues.apache.org/jira/browse/CALCITE-7924?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated CALCITE-7924:
------------------------------------
Labels: pull-request-available (was: )
> RelMdExpressionLineage Cartesian expansion over UNION ALL causes excessive
> memory use and GC
> --------------------------------------------------------------------------------------------
>
> Key: CALCITE-7924
> URL: https://issues.apache.org/jira/browse/CALCITE-7924
> Project: Calcite
> Issue Type: Bug
> Components: core
> Reporter: huangsheng
> Priority: Major
> Labels: pull-request-available
>
> h2. Summary
> Expression-lineage inference can enumerate the Cartesian product of possible
> column origins across UNION ALL inputs. For an expression referencing K
> columns over N branches, this can construct N^K candidate expressions, with
> no work budget in the expansion helper.
> This is reproducible using Calcite alone, without Feldera classes, data
> ingestion, or a running database. Both window expressions (RexOver in
> Project/Calc) and ordinary scalar expressions are affected. The resulting
> allocations can dominate planning time and memory.
> This issue was discovered while compiling SQL in Feldera and was subsequently
> reproduced in Calcite alone.
> Related issue: https://github.com/feldera/feldera/issues/7870
> h2. Parameterized reproducer
> The following storage-free helper generates SQL with adjustable branch count,
> referenced-column count, and scalar/window expression type. Put it in
> core/src/test/java/org/apache/calcite/rel/metadata/ExpressionLineageProgram.java
> (with the normal ASF source header).
> {code:java}
> package org.apache.calcite.rel.metadata;
> import org.apache.calcite.rel.RelNode;
> import org.apache.calcite.rel.type.RelDataType;
> import org.apache.calcite.rel.type.RelDataTypeFactory;
> import org.apache.calcite.schema.SchemaPlus;
> import org.apache.calcite.schema.impl.AbstractTable;
> import org.apache.calcite.sql.SqlNode;
> import org.apache.calcite.sql.type.SqlTypeName;
> import org.apache.calcite.tools.Frameworks;
> import org.apache.calcite.tools.Planner;
> import java.util.stream.Collectors;
> import java.util.stream.IntStream;
> import static com.google.common.base.Preconditions.checkArgument;
> final class ExpressionLineageProgram {
> private ExpressionLineageProgram() {
> }
> static String sql(int branches, int columns, boolean window) {
> checkArgument(branches >= 1, "branches must be positive");
> checkArgument(columns >= 1, "columns must be positive");
> String fields = IntStream.range(0, columns).mapToObj(i -> "c" + i)
> .collect(Collectors.joining(", "));
> String expression;
> if (window) {
> String partition = columns == 1 ? "" : "partition by "
> + IntStream.range(0, columns - 1).mapToObj(i -> "c" + i)
> .collect(Collectors.joining(", ")) + " ";
> expression = "row_number() over (" + partition + "order by c" +
> (columns - 1) + ")";
> } else {
> expression = IntStream.range(0, columns).mapToObj(i -> "c" + i)
> .collect(Collectors.joining(" + "));
> }
> String inputs = IntStream.range(0, branches)
> .mapToObj(i -> "select " + fields + " from events")
> .collect(Collectors.joining(" union all "));
> return "select " + fields + ", " + expression + " as computed_value from
> (" + inputs + ") u";
> }
> static RelNode rel(int branches, int columns, boolean window) throws
> Exception {
> String sql = sql(branches, columns, window);
> SchemaPlus schema = Frameworks.createRootSchema(true);
> schema.add("EVENTS", new AbstractTable() {
> @Override public RelDataType getRowType(RelDataTypeFactory typeFactory)
> {
> RelDataTypeFactory.Builder builder = typeFactory.builder();
> for (int i = 0; i < columns; i++) {
> builder.add("C" + i, SqlTypeName.INTEGER);
> }
> return builder.build();
> }
> });
> try (Planner planner = Frameworks.getPlanner(Frameworks.newConfigBuilder()
> .defaultSchema(schema).build())) {
> SqlNode validated = planner.validate(planner.parse(sql));
> return planner.rel(validated).project();
> }
> }
> }
> {code}
> A bounded regression test can be placed in the same package as
> RelMdExpressionLineageTest.java:
> {code:java}
> package org.apache.calcite.rel.metadata;
> import org.apache.calcite.rel.RelNode;
> import org.apache.calcite.rex.RexInputRef;
> import org.apache.calcite.rex.RexNode;
> import org.junit.jupiter.api.Test;
> import java.util.Set;
> import static org.junit.jupiter.api.Assertions.assertTrue;
> class RelMdExpressionLineageTest {
> @Test void testGeneratedProgramExceedsBudget() throws Exception {
> RelNode rel = ExpressionLineageProgram.rel(16, 4, true);
> Set<RexNode> lineage =
> rel.getCluster().getMetadataQuery().getExpressionLineage(rel,
> RexInputRef.of(4, rel.getRowType().getFieldList()));
> assertTrue(lineage == null, () -> "Expected unknown lineage, not "
> + (lineage == null ? "null" : lineage.size()) + " candidates");
> }
> }
> {code}
> {code}
> ./gradlew -PjdkBuildVersion=21 :core:test --tests
> org.apache.calcite.rel.metadata.RelMdExpressionLineageTest
> {code}
> The assertion above expresses the proposed bounded behavior with a
> 10,000-candidate prototype budget; it is not a claim that this budget is
> already an upstream contract. On unchanged upstream it fails with 65,536
> candidates; with the prototype it passes. There is no need to trigger an
> actual OOM in the regression test.
> Observed upstream candidate counts for four referenced columns:
> ||UNION ALL branches||Candidates||
> |2|16|
> |4|256|
> |8|4,096|
> |16|65,536|
> |32|1,048,576|
> The exponent comes from the referenced columns, not UNION nesting depth.
> Setting window=false demonstrates that the expansion problem is not exclusive
> to window functions.
> h2. Root-cause analysis
> The relevant path is:
> {code}
> ExpandDisjunctionForTableRule.apply
> -> RelMetadataQuery.getExpressionLineage
> -> Project/Calc lineage
> -> Union lineage collects candidate origins per referenced column
> -> RelMdExpressionLineage.createAllPossibleExpressions
> -> createExpressions copies an expression for each combination
> {code}
> The common helper recursively forms the Cartesian product of replacement sets
> and retains candidate RexNodes in sets. The bloat check in
> ExpandDisjunctionForTableRule occurs after the lineage request, so it does
> not bound this earlier work.
> There is also a representation difference: the relational Window handler
> already returns unknown for window-result fields, while RexOver inside
> Project/Calc can enter the expansion path. However, a window-only guard would
> not cover the scalar-expression case shown above.
> h2. Proposed budget prototype (for discussion)
> Rather than rejecting all window lineage, the prototype keeps existing
> behavior within a configurable per-expansion budget and returns unknown when
> the candidate-construction work would exceed it. It never returns a truncated
> set as complete lineage.
> In createAllPossibleExpressions, after extracting referenced fields and
> handling the no-input-reference case, the candidate product can be checked
> before copying RexNodes:
> {code:java}
> if (maxExpansions >= 0) {
> int remaining = maxExpansions;
> for (Map.Entry<RexInputRef, Set<RexNode>> entry : mapping.entrySet()) {
> if (predFieldsUsed.get(entry.getKey().getIndex())) {
> int count = entry.getValue().size();
> if (count > remaining) {
> return null;
> }
> if (count > 0) {
> remaining /= count;
> }
> }
> }
> }
> {code}
> This counts construction attempts before deduplication and avoids integer
> overflow. Repeated occurrences of the same input reference share a
> substitution; unused mapping keys do not contribute a dimension.
> The local prototype uses calcite.metadata.expression.lineage.max.expansions
> with default 10,000, -1 for unlimited, and 0 to disable expansions involving
> input references. The configuration mechanism, default, and appropriate scope
> are open for discussion. This is a per-expansion work bound, not a query-wide
> memory limit or a bound on the size of an individual expression.
> AllPredicates metadata also uses this helper and therefore needs regression
> coverage.
> h2. Pure Calcite measurements
> Median of three runs per variant. Time measures the lineage metadata call;
> memory is peak process RSS.
> ||Case||Upstream metadata time||Budget prototype||Upstream peak
> RSS||Prototype peak RSS||
> |8 branches x 4 columns, window|22.82 ms|22.84 ms|164.00 MiB|164.28 MiB|
> |16 branches x 4 columns, window|203.93 ms|5.77 ms|241.55 MiB|164.64 MiB|
> |32 branches x 4 columns, window|2026.93 ms|10.97 ms|879.31 MiB|169.55 MiB|
> |32 branches x 4 columns, scalar|1809.44 ms|11.78 ms|927.88 MiB|167.16 MiB|
> The 8-branch window case preserves all 4,096 candidates in both variants.
> Larger cases return unknown in the prototype, so the savings are from bounded
> optional inference, not from computing the complete large candidate set
> faster.
> For the 32-branch window case, total heap used after an explicit GC while
> retaining the plan/result was approximately 436.18 MiB versus 9.15 MiB. These
> are total live-heap observations, not a retained-size attribution to one
> object.
> h2. Downstream confirmation (separate from the pure Calcite benchmark)
> The full anonymized SQL is already public at:
> https://github.com/user-attachments/files/34479263/oom-anonymized.sql.txt
> For the same full anonymized SQL, changing only the Calcite budget
> implementation:
> ||Metric||Unmodified||Calcite budget prototype||
> |SQL-to-Rust generation|402.32 s|4.83 s|
> |Peak process RSS|8.35 GiB|522.48 MiB|
> |Full GC count|142|0|
> |Exit code|0|0|
> These timings cover SQL-to-Rust generation, not native Rust compilation.
> Would a per-expansion budget in this common helper be an appropriate starting
> point, or should the budget be accounted at a broader metadata-query scope?
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
