huangsheng created CALCITE-7924:
-----------------------------------
Summary: 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
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)
