This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 81cb9c85dfc [fix](maxcompute) Avoid merging exchange on dynamic
partition writes (#69060)
81cb9c85dfc is described below
commit 81cb9c85dfce76c7e8f56fd1141f5be7caa8b4b6
Author: daidai <[email protected]>
AuthorDate: Fri Oct 16 11:41:41 2026 +0800
[fix](maxcompute) Avoid merging exchange on dynamic partition writes
(#69060)
### What problem does this PR solve?
Problem Summary:
MaxCompute dynamic-partition writes need each partition's rows to be
contiguous in one writer stream, so the connector sink requested a
partition shuffle plus a local sort. It used `MustLocalSortOrderSpec`
for that, which the enforcer reads as "sort locally BEFORE distributing"
and therefore appends a global sort: an extra merging exchange after the
partition shuffle. Under local-channel backpressure that merging
receiver can wait forever on a sender blocked by another full receiver
queue, hanging `INSERT INTO <dynamic-partition table>` until the
client-side socket times out.
This PR requests a plain `OrderSpec` on top of the hash distribution
instead, so the plan becomes `distribution -> local sort` (no global
sort, no merging exchange) while keeping the same writer-local grouping
contract. The SPI javadoc, the unit tests, and the MaxCompute write
cases (explain shape: `VSORT` before `VEXCHANGE`, no
`VMERGING-EXCHANGE`) are updated accordingly.
### Release note
Fix intermittent hangs when writing dynamic partitions to MaxCompute.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../spi/write/ConnectorWritePlanProvider.java | 11 ++-
.../plans/physical/PhysicalConnectorTableSink.java | 15 +++-
.../properties/RequestPropertyDeriverTest.java | 89 ++++++++++++++++++++++
.../physical/PhysicalConnectorTableSinkTest.java | 24 ++++--
.../write/test_mc_write_large_data.groovy | 25 ++++++
.../write/test_mc_write_static_partitions.groovy | 25 ++++--
6 files changed, 173 insertions(+), 16 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java
index 0336a4f0f54..dd2b2eea03c 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java
@@ -284,7 +284,9 @@ public interface ConnectorWritePlanProvider {
/**
* Whether dynamic-partition writes must be hash-distributed by partition
columns and locally sorted by
- * them before the sink (e.g. MaxCompute Storage API). Formerly a static
{@code ConnectorCapability}
+ * them before the sink (e.g. MaxCompute Storage API). The required shape
is "shuffle by the partition
+ * columns, then sort locally inside each writer", so every partition's
rows end up contiguous in one
+ * writer stream. Formerly a static {@code ConnectorCapability}
* switch; now this per-provider method is the single source of truth. A
connector declaring this must
* also declare {@link #requiresParallelWrite()} and {@link
#requiresFullSchemaWriteOrder()}. Default: no.
*/
@@ -299,8 +301,11 @@ public interface ConnectorWritePlanProvider {
* partition writer as soon as a different partition value appears and
therefore needs the rows grouped by a
* local sort) — the hash distribution alone keeps each partition's rows
on one instance and the output file
* count at ~one-per-partition, with no sort cost. The engine ({@code
PhysicalConnectorTableSink}) picks the
- * matching distribution: {@code requiresPartitionLocalSort} ⇒ hash +
{@code MustLocalSortOrderSpec};
- * {@code requiresPartitionHashWrite} ⇒ hash only. A connector sets at
most one of the two hash arms; a
+ * matching distribution: {@code requiresPartitionLocalSort} ⇒ hash + a
local sort <em>after</em> the
+ * shuffle, expressed as a plain {@code OrderSpec} (deliberately <b>not</b>
+ * {@code MustLocalSortOrderSpec}, which the enforcer reads as "sort
before distributing" and therefore
+ * turns into distribution + a merging exchange); {@code
requiresPartitionHashWrite} ⇒ hash only. A
+ * connector sets at most one of the two hash arms; a
* connector declaring this should also declare {@link
#requiresParallelWrite()} +
* {@link #requiresFullSchemaWriteOrder()}. Default: no.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java
index 749c1bf96b3..d7f2543b613 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java
@@ -28,8 +28,8 @@ import
org.apache.doris.nereids.properties.DistributionSpecExternalTableSinkHash
import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
import
org.apache.doris.nereids.properties.DistributionSpecHiveTableSinkHashPartitioned;
import org.apache.doris.nereids.properties.LogicalProperties;
-import org.apache.doris.nereids.properties.MustLocalSortOrderSpec;
import org.apache.doris.nereids.properties.OrderKey;
+import org.apache.doris.nereids.properties.OrderSpec;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
@@ -431,11 +431,22 @@ public class PhysicalConnectorTableSink<CHILD_TYPE
extends Plan> extends Physica
// Local sort by partition columns so rows for the same
partition are grouped
// together before the streaming partition writer
(MaxCompute Storage API closes a
// partition writer once a different partition value
appears).
+ //
+ // A plain OrderSpec is deliberate, NOT
MustLocalSortOrderSpec: the enforcer
+ //
(EnforceMissingPropertiesHelper#enforceSortAndDistribution) reads
+ // MustLocalSortOrderSpec as "sort locally BEFORE
distributing" and therefore emits
+ // localSort -> distribution -> global sort, i.e. it
appends a merging exchange
+ // (VMERGING-EXCHANGE) after the partition shuffle. That
merging receiver can wait
+ // forever on a sender whose pipeline task is blocked by
another full local-channel
+ // receiver queue, hanging the INSERT until the
client-side socket times out. A plain
+ // OrderSpec on top of the hash distribution instead
yields distribution -> local
+ // sort: every partition's rows land on one instance and
are contiguous in its
+ // writer stream, which is exactly the connector's
contract, with no global sort.
List<OrderKey> orderKeys = columnIdx.stream()
.map(idx -> new OrderKey(child().getOutput().get(
idx + (hasRowOperationColumn() ? 1 : 0)),
true, false))
.collect(Collectors.toList());
- return requiredProperties.withOrderSpec(new
MustLocalSortOrderSpec(orderKeys));
+ return requiredProperties.withOrderSpec(new
OrderSpec(orderKeys));
}
// Partition columns exist but none in cols == all partitions
statically specified;
// fall through to the parallel/gather branch (no sort/shuffle
needed).
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
index 06e02ccfddb..34426c65e82 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
@@ -18,8 +18,11 @@
package org.apache.doris.nereids.properties;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.datasource.ExternalDatabase;
import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.plugin.ConnectorWritePlanContext;
+import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
import org.apache.doris.nereids.hint.DistributeHint;
import org.apache.doris.nereids.jobs.JobContext;
import org.apache.doris.nereids.memo.Group;
@@ -46,9 +49,11 @@ import org.apache.doris.nereids.trees.plans.AggPhase;
import org.apache.doris.nereids.trees.plans.DistributeType;
import org.apache.doris.nereids.trees.plans.GroupPlan;
import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalConnectorTableSink;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalExternalRowLevelMergeSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
@@ -68,6 +73,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.BitSet;
import java.util.List;
import java.util.Optional;
@@ -135,6 +141,89 @@ class RequestPropertyDeriverTest {
return new RequestPropertyDeriver(ctx,
jobContext).getRequestChildrenPropertyList(groupExpression);
}
+ // The connector sink's required physical properties carry a
writer-correctness contract: the MaxCompute
+ // Storage API closes a partition writer as soon as a different partition
value appears, so a
+ // dynamic-partition write must be hash-distributed and locally sorted by
the partition columns.
+ // enable_strict_consistency_dml is an optional replica-consistency knob
(and is forced off in cloud
+ // mode), so it may relax only the generic SINK_RANDOM_PARTITIONED
preference -- never the connector's
+ // own contract. Both polarities are pinned so a future "unify the gates"
change cannot silently drop
+ // the connector's distribution + local sort.
+ @Test
+ void
testConnectorTableSinkKeepsWriterContractWhenStrictConsistencyDmlIsOff() {
+ PhysicalConnectorTableSink<Plan> dynamicPartitionSink =
testConnectorSink(true);
+ PhysicalProperties dynamicPartitionWrite =
dynamicPartitionSink.getRequirePhysicalProperties();
+ Assertions.assertEquals(
+ ImmutableList.of(ImmutableList.of(dynamicPartitionWrite)),
+
requestChildrenPropertiesForConnectorSink(dynamicPartitionSink, false),
+ "a dynamic-partition connector write must keep its
distribution + local sort even with "
+ + "strict-consistency DML off");
+ Assertions.assertEquals(
+ ImmutableList.of(ImmutableList.of(dynamicPartitionWrite)),
+
requestChildrenPropertiesForConnectorSink(dynamicPartitionSink, true),
+ "strict-consistency DML on must not change the connector's
writer contract");
+
+ PhysicalConnectorTableSink<Plan> parallelSink =
testConnectorSink(false);
+ PhysicalProperties parallelWrite =
parallelSink.getRequirePhysicalProperties();
+ Assertions.assertEquals(PhysicalProperties.SINK_RANDOM_PARTITIONED,
parallelWrite,
+ "a non-partitioned parallel write falls back to the generic
random preference");
+ Assertions.assertEquals(
+ ImmutableList.of(ImmutableList.of(PhysicalProperties.ANY)),
+ requestChildrenPropertiesForConnectorSink(parallelSink, false),
+ "with strict-consistency DML off only the generic random
parallel-write preference relaxes");
+ Assertions.assertEquals(
+ ImmutableList.of(ImmutableList.of(parallelWrite)),
+ requestChildrenPropertiesForConnectorSink(parallelSink, true),
+ "with strict-consistency DML on the generic random
parallel-write preference is kept");
+ }
+
+ private List<List<PhysicalProperties>>
requestChildrenPropertiesForConnectorSink(
+ PhysicalConnectorTableSink<Plan> sink, boolean
enableStrictConsistencyDml) {
+ ConnectContext ctx = new ConnectContext();
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableStrictConsistencyDml =
enableStrictConsistencyDml;
+ ctx.setSessionVariable(sessionVariable);
+
+ GroupExpression groupExpression = new GroupExpression(sink);
+ new Group(null, groupExpression, null);
+
+ return new RequestPropertyDeriver(ctx,
jobContext).getRequestChildrenPropertyList(groupExpression);
+ }
+
+ /**
+ * A MaxCompute-like connector sink: a non-empty partition column list
makes the write a
+ * dynamic-partition write (partition hash + local sort), an empty one
keeps the plain parallel write.
+ * Built through the public constructor (a real connector is still mocked
out) so the resulting plan can
+ * be wrapped in a {@link GroupExpression}.
+ */
+ private static PhysicalConnectorTableSink<Plan> testConnectorSink(boolean
dynamicPartition) {
+ Column dataColumn = new Column("data", PrimitiveType.INT);
+ Column partColumn = new Column("part", PrimitiveType.INT);
+ List<Column> fullSchema = ImmutableList.of(dataColumn, partColumn);
+ List<Column> partitionColumns = dynamicPartition ?
ImmutableList.of(partColumn) : ImmutableList.of();
+
+ PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class);
+ ConnectorWritePlanContext writePlanContext =
Mockito.mock(ConnectorWritePlanContext.class);
+
Mockito.when(writePlanContext.requiresParallelWrite()).thenReturn(true);
+
Mockito.when(writePlanContext.requiresPartitionLocalSort()).thenReturn(true);
+
Mockito.when(writePlanContext.requiresPartitionHashWrite()).thenReturn(false);
+
Mockito.when(writePlanContext.getDistribution()).thenReturn(Optional.empty());
+
Mockito.when(table.resolveWritePlanContext()).thenReturn(writePlanContext);
+ Mockito.when(table.getPartitionColumns()).thenReturn(partitionColumns);
+ Mockito.when(table.getFullSchema()).thenReturn(fullSchema);
+
+ Plan child = Mockito.mock(Plan.class);
+ Mockito.when(child.getOutput()).thenReturn(ImmutableList.of(
+ new SlotReference("data", IntegerType.INSTANCE),
+ new SlotReference("part", IntegerType.INSTANCE)));
+ Mockito.when(child.getAllChildrenTypes()).thenReturn(new BitSet());
+ Mockito.when(child.depth()).thenReturn(1);
+
+ return new PhysicalConnectorTableSink<>(
+ Mockito.mock(ExternalDatabase.class), table, fullSchema,
partitionColumns, fullSchema,
+ ImmutableList.of(), Optional.empty(),
+ new LogicalProperties(child::getOutput, () ->
DataTrait.EMPTY_TRAIT), false, child);
+ }
+
@Test
void testNestedLoopJoin() {
PhysicalNestedLoopJoin<GroupPlan, GroupPlan> join = new
PhysicalNestedLoopJoin<>(JoinType.CROSS_JOIN,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSinkTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSinkTest.java
index 4c151cc8e7f..7cbcb25f3ac 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSinkTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSinkTest.java
@@ -33,6 +33,7 @@ import
org.apache.doris.nereids.properties.DistributionSpecHiveTableSinkHashPart
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.MustLocalSortOrderSpec;
import org.apache.doris.nereids.properties.OrderKey;
+import org.apache.doris.nereids.properties.OrderSpec;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
@@ -105,8 +106,14 @@ public class PhysicalConnectorTableSinkTest {
// The hash key is the child slot at 'part's full-schema position
(index 1).
Assertions.assertEquals(ImmutableList.of(partSlot.getExprId()),
dist.getOutputColExprIds(),
"hash key must be the partition-column slot taken at its
full-schema position");
- Assertions.assertTrue(props.getOrderSpec() instanceof
MustLocalSortOrderSpec,
- "dynamic-partition write must require a mandatory local sort
to group partition rows");
+ // A plain OrderSpec, NOT MustLocalSortOrderSpec: the enforcer reads
MustLocalSortOrderSpec as
+ // "sort locally BEFORE distributing", so it would emit localSort ->
distribution -> global sort,
+ // i.e. a VMERGING-EXCHANGE after the partition shuffle. That merging
receiver can deadlock
+ // against local-channel backpressure and hang the write until the
client-side socket times out.
+ Assertions.assertFalse(props.getOrderSpec() instanceof
MustLocalSortOrderSpec,
+ "dynamic-partition write must not use MustLocalSortOrderSpec
(it appends a merging exchange)");
+ Assertions.assertInstanceOf(OrderSpec.class, props.getOrderSpec(),
+ "dynamic-partition write must require a local sort to group
partition rows");
List<OrderKey> orderKeys = props.getOrderSpec().getOrderKeys();
Assertions.assertEquals(1, orderKeys.size(), "exactly one partition
column to sort by");
Assertions.assertEquals(partSlot, orderKeys.get(0).getExpr(),
@@ -202,8 +209,12 @@ public class PhysicalConnectorTableSinkTest {
Assertions.assertEquals(ImmutableList.of(dsSlot.getExprId(),
regionSlot.getExprId()),
dist.getOutputColExprIds(),
"hash keys must be the partition-column slots at their
full-schema positions");
- Assertions.assertTrue(props.getOrderSpec() instanceof
MustLocalSortOrderSpec,
- "partial-static write must require a mandatory local sort");
+ // Same rationale as dynamicPartitionWriteRequiresHashAndLocalSort: a
plain OrderSpec, so the
+ // remaining dynamic partition column is shuffled and then locally
sorted without a merging exchange.
+ Assertions.assertFalse(props.getOrderSpec() instanceof
MustLocalSortOrderSpec,
+ "partial-static write must not use MustLocalSortOrderSpec (it
appends a merging exchange)");
+ Assertions.assertInstanceOf(OrderSpec.class, props.getOrderSpec(),
+ "partial-static write must require a local sort");
List<OrderKey> orderKeys = props.getOrderSpec().getOrderKeys();
Assertions.assertEquals(2, orderKeys.size(), "sort by both partition
columns in full-schema order");
Assertions.assertEquals(dsSlot, orderKeys.get(0).getExpr());
@@ -394,7 +405,10 @@ public class PhysicalConnectorTableSinkTest {
Assertions.assertInstanceOf(DistributionSpecExternalTableSinkHashPartitioned.class,
properties.getDistributionSpec());
- Assertions.assertInstanceOf(MustLocalSortOrderSpec.class,
properties.getOrderSpec());
+ Assertions.assertFalse(properties.getOrderSpec() instanceof
MustLocalSortOrderSpec,
+ "a connector-provided distribution must not be paired with
MustLocalSortOrderSpec, which "
+ + "would make the enforcer append a merging exchange
after the shuffle");
+ Assertions.assertInstanceOf(OrderSpec.class,
properties.getOrderSpec());
Assertions.assertEquals(partSlot,
properties.getOrderSpec().getOrderKeys().get(0).getExpr());
}
diff --git
a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
index 3bd61371461..f952b416b4b 100644
---
a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
+++
b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
@@ -77,6 +77,18 @@ suite("test_mc_write_large_data", "p2,external") {
retry(10, 1000) { sql """use ${db}""" }
try {
+ // Explain prints a parent before its child, so a VSORT index that
precedes the VEXCHANGE index
+ // means the exchange feeds the sort: rows are shuffled by the
partition columns and then sorted
+ // inside each final writer stream. A VMERGING-EXCHANGE would be the
extra global sort that
+ // MaxCompute dynamic-partition writes must not have -- its merging
receiver can deadlock against
+ // local-channel backpressure and hang the INSERT until the
client-side socket times out.
+ def checkPartitionShuffleThenSort = { String explainString ->
+ int sortIndex = explainString.indexOf(":VSORT")
+ int exchangeIndex = explainString.indexOf(":VEXCHANGE")
+ sortIndex >= 0 && exchangeIndex > sortIndex
+ && !explainString.contains("VMERGING-EXCHANGE")
+ }
+
// Non-partition table
String tb1 = "large_no_part_${uuid}"
sql """DROP TABLE IF EXISTS ${tb1}"""
@@ -104,6 +116,10 @@ suite("test_mc_write_large_data", "p2,external") {
// Step 3: Insert from internal table
sql """INSERT INTO ${tb1} SELECT * FROM
internal.${internal_db}.${internal_tb}"""
+ explain {
+ sql("INSERT INTO ${tb2} SELECT * FROM
internal.${internal_db}.${internal_tb}")
+ check { explainString ->
checkPartitionShuffleThenSort(explainString) }
+ }
sql """INSERT INTO ${tb2} SELECT * FROM
internal.${internal_db}.${internal_tb}"""
// Step 4: Verify results
@@ -126,6 +142,15 @@ suite("test_mc_write_large_data", "p2,external") {
PARTITION BY (ds, region)()
"""
+ explain {
+ sql("""
+ INSERT INTO ${tb3}
+ SELECT id, name, val, ds,
+ concat('r', cast((id % 3 + 1) AS STRING)) AS region
+ FROM internal.${internal_db}.${internal_tb}
+ """)
+ check { explainString ->
checkPartitionShuffleThenSort(explainString) }
+ }
sql """
INSERT INTO ${tb3}
SELECT id, name, val, ds,
diff --git
a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_static_partitions.groovy
b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_static_partitions.groovy
index a5f2e7318cd..2e7277f66e1 100644
---
a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_static_partitions.groovy
+++
b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_static_partitions.groovy
@@ -151,6 +151,18 @@ suite("test_mc_write_static_partitions", "p2,external") {
order_qt_overwrite_p1 """ SELECT * FROM ${tb4} WHERE ds = '20250101'
"""
order_qt_overwrite_p2 """ SELECT * FROM ${tb4} WHERE ds = '20250102'
"""
+ // Explain prints a parent before its child, so a VSORT index that
precedes the VEXCHANGE index
+ // means the exchange feeds the sort: rows are shuffled by the
partition columns and then sorted
+ // inside each final writer stream. A VMERGING-EXCHANGE would be the
extra global sort that
+ // MaxCompute dynamic-partition writes must not have -- its merging
receiver can deadlock against
+ // local-channel backpressure and hang the INSERT until the
client-side socket times out.
+ def checkPartitionShuffleThenSort = { String explainString ->
+ int sortIndex = explainString.indexOf(":VSORT")
+ int exchangeIndex = explainString.indexOf(":VEXCHANGE")
+ sortIndex >= 0 && exchangeIndex > sortIndex
+ && !explainString.contains("VMERGING-EXCHANGE")
+ }
+
// Test 5: Dynamic partition regression (ensure not broken)
// Dynamic partition: partition column 'ds' is in the data, SORT node
IS expected
String tb5 = "dynamic_reg_${uuid}"
@@ -165,15 +177,16 @@ suite("test_mc_write_static_partitions", "p2,external") {
sql """INSERT INTO ${tb5} VALUES (1, 'a', '20250101'), (2, 'b',
'20250102')"""
order_qt_dynamic_regression """ SELECT * FROM ${tb5} """
- // Explain: dynamic partition INSERT should HAVE SORT node (partition
col in data)
+ // Dynamic partition writes need a partition shuffle followed by a
local sort, and no merging
+ // exchange (its merging receiver can deadlock against local-channel
backpressure).
explain {
sql("INSERT INTO ${tb5} VALUES (3, 'c', '20250103')")
- contains "SORT"
+ check { explainString ->
checkPartitionShuffleThenSort(explainString) }
}
- // Explain: dynamic partition INSERT INTO SELECT should HAVE SORT node
+ // Explain: dynamic partition INSERT INTO SELECT must use the same
non-merging plan shape.
explain {
sql("INSERT INTO ${tb5} SELECT * FROM ${tb3_src}")
- contains "SORT"
+ check { explainString ->
checkPartitionShuffleThenSort(explainString) }
}
// Test 6: INSERT OVERWRITE non-partitioned table
@@ -213,10 +226,10 @@ suite("test_mc_write_static_partitions", "p2,external") {
region STRING
) PARTITION BY (ds, region)()
"""
- // Explain: partial static partition (ds static, region dynamic)
should HAVE SORT node
+ // A remaining dynamic partition column also requires shuffle followed
by local sort.
explain {
sql("INSERT INTO ${tb7} PARTITION(ds='20250101') VALUES (1, 'v1',
'bj')")
- contains "SORT"
+ check { explainString ->
checkPartitionShuffleThenSort(explainString) }
}
// Explain: all static partition should NOT have SORT node
explain {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
