Repository navigation
IGNITE-28899 Calcite. Fix litmus error during join - #13389
Conversation
| continue; | ||
|
|
||
| // Keys can relate to affinity not pk, just assumption here. | ||
| ImmutableIntList distKeys = table.distribution().getKeys(); |
There was a problem hiding this comment.
of course we need to have 'real pk' here, but fix it right now - seems too huge.
There was a problem hiding this comment.
- Check for columns description with
key()is too huge? - For broadcast table and for tables with affinity column defined estimation is wrong. I thing it's a relatively big part of all cases.
There was a problem hiding this comment.
- fixed
- not so clear, seems it partially fixed
| inputRows += mq.getRowCount(input); | ||
|
|
||
| double mem = 0.5 * inputRows * aggregateFieldsCount() * IgniteCost.AVERAGE_FIELD_SIZE; | ||
| double mem = 0.7 * inputRows * aggregateFieldsCount() * IgniteCost.AVERAGE_FIELD_SIZE; |
There was a problem hiding this comment.
empiric constants need to be fixed cause row count estimation is changed for now and some tests (2-3) are fails, all empirics need to be reevaluated during further (i hope) statistics activity changes.
| prepareTables(); | ||
|
|
||
| String sql = "SELECT sum(i.price * i.amount)" + | ||
| String sql = "SELECT /*+ MERGE_JOIN */ sum(i.price * i.amount)" + |
There was a problem hiding this comment.
Sometimes i observe hashJoin here, i really don`t know what algo is more preferably here and i think it need to be discovered more accurate in further activities.
| private static final int DATE_DIM_SIZE = 73_049; | ||
|
|
||
| /** */ | ||
| private static final String SELECT = "SELECT /*+ DISABLE_RULE(" + |
There was a problem hiding this comment.
Of course it work perfectly well without this disablings, i just afraid about some king of flaky...
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Updates Calcite planner metadata/costing and merge-join execution behavior to address a Calcite litmus failure seen during join planning/execution (IGNITE-28899), and adds regression tests to cover the scenario.
Changes:
- Reworked join row-count estimation and adjusted join cost formulas/constants to influence plan selection.
- Modified merge-join runtime node control flow around buffering and early termination.
- Added new planner/integration tests and updated several existing assertions to match the new behavior.
Reviewed changes
Copilot reviewed 23 out of 23 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java | Registers the new join row-count estimation planner test in the suite. |
| modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java | Adds the new litmus regression integration test to the suite. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/hints/JoinTypeHintPlannerTest.java | Relaxes/adjusts hint-related plan assertions (notably around wrong-table hints). |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/UncollectPlannerTest.java | Updates plan shape expectations for UNNEST + join planning. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/PlanSplitterTest.java | Refactors colocation assignments setup to reuse a single assignments instance. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/JoinRowCountEstimationTest.java | New planner tests validating join row-count estimation logic. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/CorrelatedSubqueryPlannerTest.java | Updates plan assertion for table scan condition presence. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/AggregatePlannerTest.java | Removes an early return that skipped part of a test scenario. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/QueryBlockingTaskExecutorIntegrationTest.java | Adds hint to force merge join and simplifies plan matching into a single regex check. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LitmusCheckIntegrationTest.java | New integration test meant to catch Calcite litmus exceptions/regressions. |
| modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DistributedJoinIntegrationTest.java | Forces MERGE_JOIN via hint for deterministic join plan coverage. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java | Adds a helper to build a target mapping from a projected bitset. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/set/IgniteSetOp.java | Adjusts memory estimation coefficient used in set-op costing. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteNestedLoopJoin.java | Revises NLJ cost computation to use metadata row-count and applies an additional penalty factor. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteMergeJoin.java | Revises merge join cost computation to use metadata row-count. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteHashJoin.java | Revises hash join cost computation to use metadata row-count. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteCorrelatedNestedLoopJoin.java | Revises CNLJ cost computation to use metadata row-count. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/cost/IgniteCost.java | Changes the hash lookup cost constant affecting planner decisions. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdSelectivity.java | Updates metadata provider wiring and exposes selectivity constants for tests/consumers. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdRowCount.java | Implements custom join row-count estimation with PK/FK heuristics and new metadata provider wiring. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdColumnOrigins.java | Updates metadata provider wiring and minor indentation cleanup. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java | Reworks buffering/join scheduling and replaces “finishing” cleanup with immediate downstream end conditions. |
| modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessor.java | Removes a previously-added workaround disabling Correlate assertions. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
a6dbda5 to
eea65ee
Compare
| assertPlan(sql, schema, hasFragmentsCount(3).and(hasChildThat(isInstanceOf(IgniteTrimExchange.class)))); | ||
| // First table is replicated and planned with Exchange to colocate data, but set of nodes for partitioned | ||
| // table is differ, so exchange is added after fragments split for colocation. | ||
| assertPlan(sql, schema, hasFragmentsCount(3) |
There was a problem hiding this comment.
Current plan:
IgniteMergeJoin(condition=[=($0, $3)], joinType=[inner]): rowcount = 5.625E9, cumulative cost = IgniteCost [rowCount=5.627075E9, cpu=2.2525784224829422E10, memory=4900000.0, io=0.0, network=4000000.0]
IgniteSort(sort0=[$0], dir0=[ASC]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=4600679.76331665, memory=900000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
IgniteExchange(distribution=[single]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1500000.0, cpu=2.118354506610649E7, memory=4000000.0, io=0.0, network=4000000.0]
IgniteSort(sort0=[$0], dir0=[ASC]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1000000.0, cpu=2.068354506610649E7, memory=4000000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
master branch plan:
IgniteProject(ID=[$2], NAME=[$3], PROJECTID=[$4], ID0=[$0], VER0=[$1]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1672500.0, cpu=8572500.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteExchange(distribution=[single]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1661250.0, cpu=8561250.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteHashJoin(condition=[=($2, $0)], joinType=[inner]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1650000.0, cpu=8550000.0, memory=1170000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
IgniteTrimExchange(distribution=[affinity[identity=hash, cacheId=1355342585][0]]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=2300000.0, memory=0.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
| // fragments split to send data to initiator node. | ||
| assertPlan(sql, schema, hasFragmentsCount(3).and(hasChildThat(isInstanceOf(IgniteTrimExchange.class)))); | ||
| // table is differ, so exchange is added after fragments split for colocation. | ||
| assertPlan(sql, schema, hasFragmentsCount(2) |
There was a problem hiding this comment.
Current plan:
IgniteMergeJoin(condition=[=($2, $3)], joinType=[inner], variablesSet=[[]], leftCollation=[[2]], rightCollation=[[0]]): rowcount = 5.625E9, cumulative cost = IgniteCost [rowCount=5.627075E9, cpu=2.2525784224829422E10, memory=4900000.0, io=0.0, network=4000000.0], id = 3041
IgniteSort(sort0=[$2], dir0=[ASC]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=4600679.76331665, memory=900000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
IgniteExchange(distribution=[single]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1500000.0, cpu=2.118354506610649E7, memory=4000000.0, io=0.0, network=4000000.0]
IgniteSort(sort0=[$0], dir0=[ASC]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1000000.0, cpu=2.068354506610649E7, memory=4000000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
master branch plan:
IgniteProject(ID=[$2], NAME=[$3], PROJECTID=[$4], ID0=[$0], VER0=[$1]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1672500.0, cpu=8572500.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteExchange(distribution=[single]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1661250.0, cpu=8561250.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteHashJoin(condition=[=($4, $0)], joinType=[inner], variablesSet=[[]]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1650000.0, cpu=8550000.0, memory=1170000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
IgniteTrimExchange(distribution=[affinity[identity=hash, cacheId=1355342585][2]]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=2300000.0, memory=0.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
| assertPlan(sql, schema, hasFragmentsCount(3).and(hasChildThat(isInstanceOf(IgniteTrimExchange.class)))); | ||
| // First table is replicated and planned with Exchange to colocate data, but set of nodes for partitioned | ||
| // table is differ, so exchange is added after fragments split for colocation. | ||
| assertPlan(sql, schema, hasFragmentsCount(2) |
There was a problem hiding this comment.
Current plan:
IgniteMergeJoin(condition=[=($0, $3)], joinType=[inner], variablesSet=[[]], leftCollation=[[0]], rightCollation=[[0]]): rowcount = 5.625E9, cumulative cost = IgniteCost [rowCount=5.627075E9, cpu=2.2525784224829422E10, memory=4900000.0, io=0.0, network=4000000.0]
IgniteSort(sort0=[$0], dir0=[ASC]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=4600679.76331665, memory=900000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
IgniteExchange(distribution=[single]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1500000.0, cpu=2.118354506610649E7, memory=4000000.0, io=0.0, network=4000000.0]
IgniteSort(sort0=[$0], dir0=[ASC]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=1000000.0, cpu=2.068354506610649E7, memory=4000000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
master branch plan:
IgniteProject(ID=[$2], NAME=[$3], PROJECTID=[$4], ID0=[$0], VER0=[$1]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1672500.0, cpu=8572500.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteExchange(distribution=[single]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1661250.0, cpu=8561250.0, memory=1170000.0, io=0.0, network=225000.0]
IgniteHashJoin(condition=[=($2, $0)], joinType=[inner], variablesSet=[[]]): rowcount = 11250.0, cumulative cost = IgniteCost [rowCount=1650000.0, cpu=8550000.0, memory=1170000.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 500000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=500000.0, memory=0.0, io=0.0, network=0.0]
IgniteTrimExchange(distribution=[affinity[identity=hash, cacheId=1355342585][0]]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=575000.0, cpu=2300000.0, memory=0.0, io=0.0, network=0.0]
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 75000.0, cumulative cost = IgniteCost [rowCount=500000.0, cpu=2000000.0, memory=0.0, io=0.0, network=0.0]
There was a problem hiding this comment.
Also want to show that previously expected plan still possible but for lower table size, if we try to change
protected static final int DEFAULT_TBL_SIZE = 10;//500_000;
below plan will be raized:
IgniteProject(ID=[$2], NAME=[$3], PROJECTID=[$4], ID0=[$0], VER0=[$1]): rowcount = 2.25, cumulative cost = IgniteCost [rowCount=28.25, cpu=83.0, memory=23.4, io=0.0, network=45.0], id = 4147
IgniteExchange(distribution=[single]): rowcount = 2.25, cumulative cost = IgniteCost [rowCount=26.0, cpu=80.75, memory=23.4, io=0.0, network=45.0], id = 4146
IgniteHashJoin(condition=[=($2, $0)], joinType=[inner], variablesSet=[[]]): rowcount = 2.25, cumulative cost = IgniteCost [rowCount=23.75, cpu=78.5, memory=23.4, io=0.0, network=0.0], id = 4145
IgniteTableScan(table=[[PUBLIC, PROJECT]], requiredColumns=[{0, 2}]): rowcount = 10.0, cumulative cost = IgniteCost [rowCount=10.0, cpu=10.0, memory=0.0, io=0.0, network=0.0], id = 1784
IgniteTrimExchange(distribution=[affinity[identity=hash, cacheId=1355342585][0]]): rowcount = 1.5, cumulative cost = IgniteCost [rowCount=11.5, cpu=46.0, memory=0.0, io=0.0, network=0.0], id = 4144
IgniteTableScan(table=[[PUBLIC, DEVELOPER]], filters=[=(+($t2, 1), ?0)]): rowcount = 1.5, cumulative cost = IgniteCost [rowCount=10.0, cpu=40.0, memory=0.0, io=0.0, network=0.0], id = 1793
Thus i think - if we want a fine tuning here - it need to be done under different issue.
809cdcb to
063753c
Compare
|
Current cut shows: TpchBenchmark, scale = 0.01, q = 5. (H2 just for comparison here, not related to an issue of course) |
|
Perhaps it would be more correct to compare before and after your changes? |
|
@alex-plekhanov Seems I fix all your wishes, one thing not clear for me, this comment : "For broadcast table and for tables with affinity column defined estimation is wrong. I thing it's a relatively big part of all cases." I fix a bit .. but seems we have no appropriate tests? Or I`m wrong ? |
| return false; | ||
|
|
||
| double val = ((Integer)o).doubleValue(); | ||
| // IgniteMdRowCount.EQUI_COEFF with a bit overhead |
There was a problem hiding this comment.
No more EQUI_COEFF in the code
| private static final double NON_EQUI_COEFF = 0.7; | ||
|
|
||
| /** */ | ||
| public static final double INTERSECT_ESTIMATE_COEFF = 0.8; |
|
|
https://issues.apache.org/jira/browse/IGNITE-28899