Skip to content

Commit a31421f

Browse files
alex-plekhanovalex-plekhanov
committed
IGNITE-12692 SQL Calcite: Distributed table modify
1 parent 044f62b commit a31421f

File tree

4 files changed

+147
-13
lines changed

4 files changed

+147
-13
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/rel/IgniteTableModify.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.calcite.rel.RelInput;
2525
import org.apache.calcite.rel.RelNode;
2626
import org.apache.calcite.rel.core.TableModify;
27+
import org.apache.calcite.rel.metadata.RelMetadataQuery;
2728
import org.apache.calcite.rex.RexNode;
2829
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
2930

@@ -102,4 +103,9 @@ public IgniteTableModify(RelInput input) {
102103
return new IgniteTableModify(cluster, getTraitSet(), getTable(), sole(inputs),
103104
getOperation(), getUpdateColumnList(), getSourceExpressionList(), isFlattened());
104105
}
106+
107+
/** {@inheritDoc} */
108+
@Override public double estimateRowCount(RelMetadataQuery mq) {
109+
return 1.0D;
110+
}
105111
}

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

Lines changed: 109 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -17,44 +17,140 @@
1717

1818
package org.apache.ignite.internal.processors.query.calcite.rule;
1919

20+
import com.google.common.collect.ImmutableMap;
2021
import org.apache.calcite.plan.RelOptCluster;
21-
import org.apache.calcite.plan.RelOptPlanner;
2222
import org.apache.calcite.plan.RelOptRule;
23+
import org.apache.calcite.plan.RelOptRuleCall;
24+
import org.apache.calcite.plan.RelRule;
2325
import org.apache.calcite.plan.RelTraitSet;
24-
import org.apache.calcite.rel.PhysicalNode;
2526
import org.apache.calcite.rel.RelCollations;
27+
import org.apache.calcite.rel.RelDistribution;
2628
import org.apache.calcite.rel.RelNode;
2729
import org.apache.calcite.rel.logical.LogicalTableModify;
28-
import org.apache.calcite.rel.metadata.RelMetadataQuery;
30+
import org.apache.calcite.rel.type.RelDataTypeField;
31+
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
32+
import org.apache.calcite.tools.RelBuilder;
2933
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
3034
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
35+
import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
36+
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
3137
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
3238
import org.apache.ignite.internal.processors.query.calcite.trait.RewindabilityTrait;
39+
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
40+
import org.immutables.value.Value;
3341

3442
/**
35-
*
43+
* Converts LogicalTableModify to physical relation operators.
44+
* There are two options:
45+
* - Perform table modify on initiator node. In this case IgniteTableModify with single distribution is inserted.
46+
* - Perform table modify on remote nodes. In this case IgniteTableModify with random distribution is inserted and
47+
* sum aggregate on top if this table modify (to aggregate and send to initiator node affected rows count)
3648
*/
37-
public class TableModifyConverterRule extends AbstractIgniteConverterRule<LogicalTableModify> {
49+
@Value.Enclosing
50+
public class TableModifyConverterRule extends RelRule<TableModifyConverterRule.Config> {
3851
/** */
39-
public static final RelOptRule INSTANCE = new TableModifyConverterRule();
52+
public static final RelOptRule INSTANCE = Config.DFLT.toRule();
53+
54+
/** Rule configuration. */
55+
@Value.Immutable
56+
public interface Config extends RelRule.Config {
57+
/** Default config. */
58+
TableModifyConverterRule.Config DFLT = ImmutableTableModifyConverterRule.Config.of()
59+
.withOperandSupplier(b ->
60+
b.operand(LogicalTableModify.class).anyInputs());
61+
62+
/** {@inheritDoc} */
63+
@Override default TableModifyConverterRule toRule() {
64+
return new TableModifyConverterRule(this);
65+
}
66+
}
4067

4168
/**
42-
* Creates a ConverterRule.
69+
* Creates a TableModifyRule.
4370
*/
44-
public TableModifyConverterRule() {
45-
super(LogicalTableModify.class, "TableModifyConverterRule");
71+
public TableModifyConverterRule(Config cfg) {
72+
super(cfg);
4673
}
4774

4875
/** {@inheritDoc} */
49-
@Override protected PhysicalNode convert(RelOptPlanner planner, RelMetadataQuery mq, LogicalTableModify rel) {
76+
@Override public void onMatch(RelOptRuleCall call) {
77+
LogicalTableModify rel = call.rel(0);
78+
79+
RelBuilder relBuilder = relBuilderFactory.create(rel.getCluster(), null);
80+
81+
RelNode singleNodeTableModify = convertTableModify(rel, IgniteDistributions.single(), IgniteDistributions.single());
82+
83+
if (Commons.queryTransactionVersion(call.getPlanner().getContext()) != null) {
84+
// If excplicit transaction is started, table modify can only be executed on initiator node.
85+
call.transformTo(singleNodeTableModify);
86+
87+
return;
88+
}
89+
90+
IgniteTable table = rel.getTable().unwrap(IgniteTable.class);
91+
IgniteDistribution inputDistribution = table.distribution();
92+
93+
switch (rel.getOperation()) {
94+
case MERGE:
95+
// Merge contains insert fields as well as _key field, it's impossible to generate input distribution.
96+
inputDistribution = null;
97+
98+
break;
99+
100+
case INSERT:
101+
case UPDATE:
102+
// Can only safely proceed, if modified values don't affect remote nodes data sources for the same query.
103+
if (inputDistribution.getType() != RelDistribution.Type.HASH_DISTRIBUTED)
104+
inputDistribution = null;
105+
106+
break;
107+
108+
case DELETE:
109+
inputDistribution = IgniteDistributions.random();
110+
111+
break;
112+
113+
default:
114+
throw new IllegalStateException("Unknown operation type: " + rel.getOperation());
115+
}
116+
117+
if (inputDistribution == null) {
118+
call.transformTo(singleNodeTableModify);
119+
120+
return;
121+
}
122+
123+
RelDataTypeField outFld = rel.getRowType().getFieldList().get(0);
124+
125+
relBuilder.push(convertTableModify(rel, IgniteDistributions.random(), inputDistribution));
126+
relBuilder.aggregate(relBuilder.groupKey(),
127+
relBuilder.aggregateCall(SqlStdOperatorTable.SUM0,
128+
relBuilder.field(0)).as(outFld.getName()));
129+
relBuilder.project(relBuilder.cast(relBuilder.fields().get(0), outFld.getType().getSqlTypeName()));
130+
131+
RelNode distributedTableModify = relBuilder.build();
132+
133+
call.transformTo(singleNodeTableModify, ImmutableMap.of(distributedTableModify, rel));
134+
}
135+
136+
/** */
137+
private IgniteTableModify convertTableModify(
138+
LogicalTableModify rel,
139+
IgniteDistribution outputDistribution,
140+
IgniteDistribution inputDistribution
141+
) {
50142
RelOptCluster cluster = rel.getCluster();
143+
51144
RelTraitSet traits = cluster.traitSetOf(IgniteConvention.INSTANCE)
52-
.replace(IgniteDistributions.single())
145+
.replace(outputDistribution)
53146
.replace(RewindabilityTrait.ONE_WAY)
54147
.replace(RelCollations.EMPTY);
55-
RelNode input = convert(rel.getInput(), traits);
148+
149+
RelTraitSet inputTraits = traits.replace(inputDistribution);
150+
151+
RelNode input = convert(rel.getInput(), inputTraits);
56152

57153
return new IgniteTableModify(cluster, traits, rel.getTable(), input,
58-
rel.getOperation(), rel.getUpdateColumnList(), rel.getSourceExpressionList(), rel.isFlattened());
154+
rel.getOperation(), rel.getUpdateColumnList(), rel.getSourceExpressionList(), rel.isFlattened());
59155
}
60156
}

‎modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/TableDmlPlannerTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,4 +190,28 @@ public void testDuplicatedColumnNames() throws Exception {
190190
assertPlan("SELECT CITY.NAME, STREET.NAME FROM STREET JOIN CITY ON STREET.CITY_ID = CITY.ID ORDER BY STREET.ID",
191191
schema, hasColumns("NAME", "NAME1"));
192192
}
193+
194+
/** Tests that table modify can be executed on remote nodes. */
195+
@Test
196+
public void testDistributedTableModify() throws Exception {
197+
IgniteSchema schema = createSchema(
198+
createTable("TEST", IgniteDistributions.affinity(0, "test", "hash"),
199+
"ID", INTEGER, "VAL", INTEGER)
200+
);
201+
202+
assertPlan("UPDATE test SET val = val + 1", schema, hasChildThat(isInstanceOf(IgniteTableModify.class)
203+
.and(hasDistribution(IgniteDistributions.random()))));
204+
}
205+
206+
/** */
207+
@Test
208+
public void testDistributedDelete() throws Exception {
209+
IgniteSchema schema = createSchema(
210+
createTable("TEST", IgniteDistributions.affinity(0, "test", "hash"),
211+
"_KEY", INTEGER, "VAL", INTEGER)
212+
);
213+
214+
assertPlan("DELETE FROM test WHERE val = 10", schema, hasChildThat(isInstanceOf(IgniteTableModify.class)
215+
.and(hasDistribution(IgniteDistributions.random()))));
216+
}
193217
}

0 commit comments

Comments
 (0)