Skip to content

Commit 8dd63f8

Browse files
alex-plekhanovalex-plekhanov
committed
IGNITE-12692 SQL Calcite: Distributed table modify - Fixes #12593.
Signed-off-by: Aleksey Plekhanov <[email protected]>
1 parent 93394ab commit 8dd63f8

File tree

12 files changed

+651
-212
lines changed

12 files changed

+651
-212
lines changed

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdRowCount.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
3737
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteLimit;
3838
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSortedIndexSpool;
39+
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
3940
import org.apache.ignite.internal.util.typedef.F;
4041
import org.jetbrains.annotations.Nullable;
4142

@@ -149,4 +150,11 @@ public double getRowCount(IgniteAggregate rel, RelMetadataQuery mq) {
149150
public double getRowCount(IgniteLimit rel, RelMetadataQuery mq) {
150151
return rel.estimateRowCount(mq);
151152
}
153+
154+
/**
155+
* Estimation of row count for Table modify operator.
156+
*/
157+
public double getRowCount(IgniteTableModify rel, RelMetadataQuery mq) {
158+
return rel.estimateRowCount(mq);
159+
}
152160
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,8 @@
6464
import org.apache.ignite.internal.processors.query.calcite.rule.SortAggregateConverterRule;
6565
import org.apache.ignite.internal.processors.query.calcite.rule.SortConverterRule;
6666
import org.apache.ignite.internal.processors.query.calcite.rule.TableFunctionScanConverterRule;
67-
import org.apache.ignite.internal.processors.query.calcite.rule.TableModifyConverterRule;
67+
import org.apache.ignite.internal.processors.query.calcite.rule.TableModifyDistributedConverterRule;
68+
import org.apache.ignite.internal.processors.query.calcite.rule.TableModifySingleNodeConverterRule;
6869
import org.apache.ignite.internal.processors.query.calcite.rule.UncollectConverterRule;
6970
import org.apache.ignite.internal.processors.query.calcite.rule.UnionConverterRule;
7071
import org.apache.ignite.internal.processors.query.calcite.rule.ValuesConverterRule;
@@ -292,7 +293,8 @@ public enum PlannerPhase {
292293
SetOpConverterRule.MAP_REDUCE_INTERSECT,
293294
ProjectConverterRule.INSTANCE,
294295
FilterConverterRule.INSTANCE,
295-
TableModifyConverterRule.INSTANCE,
296+
TableModifySingleNodeConverterRule.INSTANCE,
297+
TableModifyDistributedConverterRule.INSTANCE,
296298
UnionConverterRule.INSTANCE,
297299
SortConverterRule.INSTANCE,
298300
TableFunctionScanConverterRule.INSTANCE

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableModify.java‎

Lines changed: 49 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,17 +18,25 @@
1818
package org.apache.ignite.internal.processors.query.calcite.rel;
1919

2020
import java.util.List;
21+
import com.google.common.collect.ImmutableList;
2122
import org.apache.calcite.plan.RelOptCluster;
2223
import org.apache.calcite.plan.RelOptTable;
2324
import org.apache.calcite.plan.RelTraitSet;
2425
import org.apache.calcite.rel.RelInput;
2526
import org.apache.calcite.rel.RelNode;
2627
import org.apache.calcite.rel.core.TableModify;
28+
import org.apache.calcite.rel.metadata.RelMetadataQuery;
2729
import org.apache.calcite.rex.RexNode;
30+
import org.apache.calcite.util.Pair;
31+
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
32+
import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils;
2833
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
2934

3035
/** */
3136
public class IgniteTableModify extends TableModify implements IgniteRel {
37+
/** If table modify can affect data source. */
38+
private final boolean affectsSrc;
39+
3240
/**
3341
* Creates a {@code TableModify}.
3442
*
@@ -43,8 +51,9 @@ public class IgniteTableModify extends TableModify implements IgniteRel {
4351
* @param input Sub-query or filter condition.
4452
* @param operation Modify operation (INSERT, UPDATE, DELETE, MERGE).
4553
* @param updateColumnList List of column identifiers to be updated (e.g. ident1, ident2); null if not UPDATE.
46-
* @param sourceExpressionList List of value expressions to be set (e.g. exp1, exp2); null if not UPDATE.
54+
* @param srcExpressionList List of value expressions to be set (e.g. exp1, exp2); null if not UPDATE.
4755
* @param flattened Whether set flattens the input row type.
56+
* @param affectsSrc If table modify can affect data source.
4857
*/
4958
public IgniteTableModify(
5059
RelOptCluster cluster,
@@ -53,12 +62,15 @@ public IgniteTableModify(
5362
RelNode input,
5463
Operation operation,
5564
List<String> updateColumnList,
56-
List<RexNode> sourceExpressionList,
57-
boolean flattened
65+
List<RexNode> srcExpressionList,
66+
boolean flattened,
67+
boolean affectsSrc
5868
) {
5969
super(cluster, traitSet, table, Commons.context(cluster).catalogReader(),
6070
input, operation, updateColumnList,
61-
sourceExpressionList, flattened);
71+
srcExpressionList, flattened);
72+
73+
this.affectsSrc = affectsSrc;
6274
}
6375

6476
/**
@@ -75,7 +87,8 @@ public IgniteTableModify(RelInput input) {
7587
input.getEnum("operation", Operation.class),
7688
input.getStringList("updateColumnList"),
7789
input.getExpressionList("sourceExpressionList"),
78-
input.getBoolean("flattened", true)
90+
input.getBoolean("flattened", true),
91+
false // Field only for planning.
7992
);
8093
}
8194

@@ -89,7 +102,8 @@ public IgniteTableModify(RelInput input) {
89102
getOperation(),
90103
getUpdateColumnList(),
91104
getSourceExpressionList(),
92-
isFlattened());
105+
isFlattened(),
106+
affectsSrc);
93107
}
94108

95109
/** {@inheritDoc} */
@@ -100,6 +114,34 @@ public IgniteTableModify(RelInput input) {
100114
/** {@inheritDoc} */
101115
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel> inputs) {
102116
return new IgniteTableModify(cluster, getTraitSet(), getTable(), sole(inputs),
103-
getOperation(), getUpdateColumnList(), getSourceExpressionList(), isFlattened());
117+
getOperation(), getUpdateColumnList(), getSourceExpressionList(), isFlattened(), affectsSrc);
118+
}
119+
120+
/** {@inheritDoc} */
121+
@Override public double estimateRowCount(RelMetadataQuery mq) {
122+
return 1.0D;
123+
}
124+
125+
/** {@inheritDoc} */
126+
@Override public Pair<RelTraitSet, List<RelTraitSet>> deriveTraits(RelTraitSet childTraits, int childId) {
127+
// Don't derive traits for single-node table modify.
128+
if (TraitUtils.distribution(traitSet) == IgniteDistributions.single())
129+
return null;
130+
131+
assert childId == 0;
132+
133+
if (childTraits.getConvention() != IgniteConvention.INSTANCE)
134+
return null;
135+
136+
// If modify can affect data source (for example, INSERT contains self table as source) only
137+
// modified table affinity distibution is possible, otherwise inconsistency is possible on remote nodes.
138+
if (affectsSrc)
139+
return null;
140+
141+
// Any distributed (random/hash) trait is accepted if data source is not affected by modify.
142+
if (!TraitUtils.distribution(childTraits).satisfies(IgniteDistributions.random()))
143+
return null;
144+
145+
return Pair.of(traitSet, ImmutableList.of(childTraits));
104146
}
105147
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AbstractIgniteConverterRule.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.calcite.rel.convert.ConverterRule;
2525
import org.apache.calcite.rel.metadata.RelMetadataQuery;
2626
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
27+
import org.jetbrains.annotations.Nullable;
2728

2829
/** */
2930
public abstract class AbstractIgniteConverterRule<T extends RelNode> extends ConverterRule {
@@ -39,7 +40,7 @@ protected AbstractIgniteConverterRule(Class<T> clazz, String descriptionPrefix)
3940
}
4041

4142
/** {@inheritDoc} */
42-
@Override public final RelNode convert(RelNode rel) {
43+
@Override public final @Nullable RelNode convert(RelNode rel) {
4344
return convert(rel.getCluster().getPlanner(), rel.getCluster().getMetadataQuery(), (T)rel);
4445
}
4546

@@ -51,5 +52,5 @@ protected AbstractIgniteConverterRule(Class<T> clazz, String descriptionPrefix)
5152
* @param rel Rel node.
5253
* @return Physical rel.
5354
*/
54-
protected abstract PhysicalNode convert(RelOptPlanner planner, RelMetadataQuery mq, T rel);
55+
protected abstract @Nullable PhysicalNode convert(RelOptPlanner planner, RelMetadataQuery mq, T rel);
5556
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/HashAggregateConverterRule.java‎

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,10 @@
1919

2020
import org.apache.calcite.plan.RelOptCluster;
2121
import org.apache.calcite.plan.RelOptPlanner;
22-
import org.apache.calcite.plan.RelOptRule;
2322
import org.apache.calcite.plan.RelTraitSet;
2423
import org.apache.calcite.rel.PhysicalNode;
2524
import org.apache.calcite.rel.RelNode;
25+
import org.apache.calcite.rel.convert.ConverterRule;
2626
import org.apache.calcite.rel.logical.LogicalAggregate;
2727
import org.apache.calcite.rel.metadata.RelMetadataQuery;
2828
import org.apache.ignite.internal.processors.query.calcite.hint.HintUtils;
@@ -31,16 +31,17 @@
3131
import org.apache.ignite.internal.processors.query.calcite.rel.agg.IgniteMapHashAggregate;
3232
import org.apache.ignite.internal.processors.query.calcite.rel.agg.IgniteReduceHashAggregate;
3333
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
34+
import org.jetbrains.annotations.Nullable;
3435

3536
/**
3637
*
3738
*/
3839
public class HashAggregateConverterRule {
3940
/** */
40-
public static final RelOptRule COLOCATED = new ColocatedHashAggregateConverterRule();
41+
public static final ConverterRule COLOCATED = new ColocatedHashAggregateConverterRule();
4142

4243
/** */
43-
public static final RelOptRule MAP_REDUCE = new MapReduceHashAggregateConverterRule();
44+
public static final ConverterRule MAP_REDUCE = new MapReduceHashAggregateConverterRule();
4445

4546
/** */
4647
private HashAggregateConverterRule() {
@@ -55,8 +56,11 @@ private static class ColocatedHashAggregateConverterRule extends AbstractIgniteC
5556
}
5657

5758
/** {@inheritDoc} */
58-
@Override protected PhysicalNode convert(RelOptPlanner planner, RelMetadataQuery mq,
59-
LogicalAggregate agg) {
59+
@Override protected @Nullable PhysicalNode convert(
60+
RelOptPlanner planner,
61+
RelMetadataQuery mq,
62+
LogicalAggregate agg
63+
) {
6064
if (HintUtils.isExpandDistinctAggregate(agg))
6165
return null;
6266

@@ -84,8 +88,11 @@ private static class MapReduceHashAggregateConverterRule extends AbstractIgniteC
8488
}
8589

8690
/** {@inheritDoc} */
87-
@Override protected PhysicalNode convert(RelOptPlanner planner, RelMetadataQuery mq,
88-
LogicalAggregate agg) {
91+
@Override protected @Nullable PhysicalNode convert(
92+
RelOptPlanner planner,
93+
RelMetadataQuery mq,
94+
LogicalAggregate agg
95+
) {
8996
if (HintUtils.isExpandDistinctAggregate(agg))
9097
return null;
9198

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ProjectConverterRule.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,10 @@
2020
import java.util.Set;
2121
import org.apache.calcite.plan.RelOptCluster;
2222
import org.apache.calcite.plan.RelOptPlanner;
23-
import org.apache.calcite.plan.RelOptRule;
2423
import org.apache.calcite.plan.RelTraitSet;
2524
import org.apache.calcite.rel.PhysicalNode;
2625
import org.apache.calcite.rel.RelNode;
26+
import org.apache.calcite.rel.convert.ConverterRule;
2727
import org.apache.calcite.rel.core.CorrelationId;
2828
import org.apache.calcite.rel.logical.LogicalProject;
2929
import org.apache.calcite.rel.metadata.RelMetadataQuery;
@@ -39,7 +39,7 @@
3939
*/
4040
public class ProjectConverterRule extends AbstractIgniteConverterRule<LogicalProject> {
4141
/** */
42-
public static final RelOptRule INSTANCE = new ProjectConverterRule();
42+
public static final ConverterRule INSTANCE = new ProjectConverterRule();
4343

4444
/** */
4545
public ProjectConverterRule() {

0 commit comments

Comments
 (0)