You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2026/06/16 17:29:13 UTC
(impala) 02/02: IMPALA-15085: DML statements should always shuffle when target table is Iceberg V3
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 23b5c1e849cf9f3e223c9554dc15f5c57f515ada
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Tue Jun 9 12:06:01 2026 +0200
IMPALA-15085: DML statements should always shuffle when target table is Iceberg V3
Without this patch the MERGE statement didn't always shuffle
before the MERGE SINK for Iceberg V3 tables. For such tables it is
mandatory to shuffle to guarantee that only a single Deletion Vector
is being created per data file.
DELETE/UPDATE worked correctly even before this patch, but added extra
tests anyway.
Testing
* e2e tests added
Change-Id: Ic7aeed31b832521f0a619c15c116a954c134e4b2
Assisted-by: Claude Opus 4.6
Reviewed-on: http://gerrit.cloudera.org:8080/24425
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/analysis/DmlStatementBase.java | 15 +++++-----
.../java/org/apache/impala/analysis/MergeImpl.java | 13 +++++----
.../org/apache/impala/analysis/ModifyImpl.java | 2 +-
.../apache/impala/planner/DistributedPlanner.java | 32 ++++++++++++++--------
.../queries/PlannerTest/iceberg-merge.test | 4 ++-
.../queries/PlannerTest/iceberg-v2-delete.test | 2 ++
.../queries/PlannerTest/iceberg-v2-update.test | 12 ++++++--
.../queries/QueryTest/iceberg-v3-delete.test | 26 ++++++++++++++++++
.../queries/QueryTest/iceberg-v3-merge.test | 29 ++++++++++++++++++++
.../queries/QueryTest/iceberg-v3-update.test | 30 ++++++++++++++++++++
10 files changed, 135 insertions(+), 30 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java b/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java
index 836bdc4a2..709db293e 100644
--- a/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java
+++ b/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java
@@ -17,6 +17,7 @@
package org.apache.impala.analysis;
+import java.util.Collections;
import java.util.Map;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeKuduTable;
@@ -74,14 +75,14 @@ public abstract class DmlStatementBase extends StatementBase {
*/
abstract public List<Expr> getPartitionKeyExprs();
/**
- * The DML sink requires shuffling on the returned shuffle exprs. In most cases, it
- * is the same as partition key exprs, but in specific cases (e.g. Iceberg V3 tables
- * with Deletion Vectors) we need to shuffle data differently, or we need to shuffle
- * data even if the target table is not partitioned.
- * Another use case is to fan out writers when partition key exprs have low
- * cardinality. In that case, we can add additional shuffle exprs.
+ * Returns the expressions that the exchange node should use to hash-partition rows
+ * before they reach the sink. A non-empty list means the shuffle is required.
+ * E.g. getShuffleExprs() returns a non-empty list when:
+ * 1) shuffling is required for correctness (Iceberg V3 tables with DVs)
+ * (even for non-partitioned tables)
+ * 2) partition exprs have low cardinality, so custom shuffle exprs can fan-out writers
*/
- public List<Expr> getShuffleExprs() { return getPartitionKeyExprs(); }
+ public List<Expr> getShuffleExprs() { return Collections.emptyList(); }
abstract public List<Expr> getSortExprs();
abstract public TSortingOrder getSortingOrder();
diff --git a/fe/src/main/java/org/apache/impala/analysis/MergeImpl.java b/fe/src/main/java/org/apache/impala/analysis/MergeImpl.java
index 3cb48d973..e8cb06938 100644
--- a/fe/src/main/java/org/apache/impala/analysis/MergeImpl.java
+++ b/fe/src/main/java/org/apache/impala/analysis/MergeImpl.java
@@ -24,6 +24,7 @@ import org.apache.impala.planner.PlanNode;
import org.apache.impala.planner.PlannerContext;
import org.apache.impala.thrift.TSortingOrder;
+import java.util.Collections;
import java.util.List;
/**
@@ -65,13 +66,13 @@ public interface MergeImpl {
List<Expr> getPartitionKeyExprs();
/**
- * Returns the expressions used to shuffle (hash-partition) rows across fragment
- * instances before they reach the sink. By default delegates to getPartitionKeyExprs(),
- * but implementations may override this to use a different set of expressions for
- * routing (e.g. routing by input__file__name on unpartitioned V3 tables).
- * @return list of shuffle expressions
+ * Returns the expressions that the exchange node should use to hash-partition rows
+ * before they reach the sink. A non-empty list means the shuffle is required for
+ * correctness (cost-based skip logic is bypassed). An empty list (the default) means
+ * partition key exprs are used as the exchange key but the exchange may be skipped by
+ * cost-based logic if the data is already appropriately distributed.
*/
- default List<Expr> getShuffleExprs() { return getPartitionKeyExprs(); }
+ default List<Expr> getShuffleExprs() { return Collections.emptyList(); }
/**
* Returns sort expressions originated from SORT BY property of the target table.
diff --git a/fe/src/main/java/org/apache/impala/analysis/ModifyImpl.java b/fe/src/main/java/org/apache/impala/analysis/ModifyImpl.java
index 084c2ec9a..27bf1035c 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ModifyImpl.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ModifyImpl.java
@@ -116,7 +116,7 @@ abstract class ModifyImpl {
abstract public List<Expr> getPartitionKeyExprs();
- public List<Expr> getShuffleExprs() { return getPartitionKeyExprs(); }
+ public List<Expr> getShuffleExprs() { return Collections.emptyList(); }
public List<Expr> getSortExprs() {
return Collections.emptyList();
diff --git a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
index f6f6d7580..c9960f1cc 100644
--- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
@@ -210,21 +210,26 @@ public class DistributedPlanner {
if (dmlStmt.hasNoShuffleHint() && !enforceHdfsWriterLimit) return inputFragment;
List<Expr> shuffleExprs = new ArrayList<>(dmlStmt.getShuffleExprs());
+ boolean enforceShuffle = !shuffleExprs.isEmpty();
+ // If shuffle exprs are not set, fall back to partition key exprs for the exchange.
+ List<Expr> exchangeExprs = enforceShuffle
+ ? shuffleExprs
+ : new ArrayList<>(dmlStmt.getPartitionKeyExprs());
// Ignore constants for the sake of partitioning.
- Expr.removeConstants(shuffleExprs);
+ Expr.removeConstants(exchangeExprs);
// Do nothing if the input fragment is already appropriately partitioned. TODO: handle
// Kudu tables here (IMPALA-5254).
DataPartition inputPartition = inputFragment.getDataPartition();
- if (!shuffleExprs.isEmpty()
+ if (!exchangeExprs.isEmpty()
&& analyzer.setsHaveValueTransfer(
- inputPartition.getPartitionExprs(), shuffleExprs, true)
+ inputPartition.getPartitionExprs(), exchangeExprs, true)
&& !(dmlStmt.getTargetTable() instanceof FeKuduTable)
&& !enforceHdfsWriterLimit) {
return inputFragment;
}
- long numPartitions = getNumDistinctValues(shuffleExprs);
+ long numPartitions = getNumDistinctValues(exchangeExprs);
int maxHdfsWriters = analyzer.getQueryOptions().getMax_fs_writers();
// We also consider fragments containing union nodes along with scan fragments
// (leaf fragments) since they are either a part of those scan fragments or are
@@ -246,7 +251,7 @@ public class DistributedPlanner {
inputFragment.getNumNodes(), analyzer.getMaxParallelismPerNode());
int costBasedMaxWriter = minInstances;
- boolean isPartitioned = !shuffleExprs.isEmpty();
+ boolean isPartitioned = !exchangeExprs.isEmpty();
PlanNode root = inputFragment.getPlanRoot();
if (root.getCardinality() > -1 && root.getAvgRowSize() > -1) {
// Both cardinality and avg row size is known.
@@ -291,14 +296,17 @@ public class DistributedPlanner {
}
}
- // Make a cost-based decision only if no user hint was supplied.
- if (!dmlStmt.hasShuffleHint()) {
+ // Make a cost-based decision only if no user hint was supplied and the
+ // shuffle is not enforced (non-empty shuffle exprs mean the statement requires
+ // a shuffle for correctness, e.g. Iceberg V3 tables need to shuffle to guarantee
+ // one Deletion Vector per data file).
+ if (!dmlStmt.hasShuffleHint() && !enforceShuffle) {
if (dmlStmt.getTargetTable() instanceof FeKuduTable) {
// If the table is unpartitioned or all of the partition exprs are constants,
// don't insert the exchange.
// TODO: make a more sophisticated decision here for partitioned tables and when
// we have info about tablet locations.
- if (shuffleExprs.isEmpty()) return inputFragment;
+ if (exchangeExprs.isEmpty()) return inputFragment;
} else if (!enforceHdfsWriterLimit || !hasHdfsScanORUnion
|| (expectedNumInputInstance <= maxHdfsWriters)) {
// Only consider skipping the addition of an exchange node if
@@ -324,10 +332,10 @@ public class DistributedPlanner {
// check if it is distributed across all nodes. If so, don't repartition.
// TODO: If input fragment has ScanNode leave(s), its inputPartitions can be
// empty and numInputPartitions=1. But the scanned table(s) might be a
- // partitioned table(s) that superset of shuffleExprs. Repartition (shuffling)
+ // partitioned table(s) that superset of exchangeExprs. Repartition (shuffling)
// might not needed in that case, but skipping repartition can risk producing
// small files.
- if (Expr.isSubset(inputPartition.getPartitionExprs(), shuffleExprs)) {
+ if (Expr.isSubset(inputPartition.getPartitionExprs(), exchangeExprs)) {
long numInputPartitions =
getNumDistinctValues(inputPartition.getPartitionExprs());
if (numInputPartitions >= inputInstances) { return inputFragment; }
@@ -352,7 +360,7 @@ public class DistributedPlanner {
exchNode.init(analyzer);
Preconditions.checkState(exchNode.hasValidStats());
DataPartition partition;
- if (shuffleExprs.isEmpty()) {
+ if (exchangeExprs.isEmpty()) {
if (enforceHdfsWriterLimit
&& inputFragment.getDataPartition().getType() == TPartitionType.RANDOM) {
// This ensures the parallelism of the writers is maintained while maintaining
@@ -366,7 +374,7 @@ public class DistributedPlanner {
partition = DataPartition.kuduPartitioned(
KuduUtil.createPartitionExpr((InsertStmt)dmlStmt, ctx_.getRootAnalyzer()));
} else {
- partition = DataPartition.hashPartitioned(shuffleExprs);
+ partition = DataPartition.hashPartitioned(exchangeExprs);
}
PlanFragment fragment =
new PlanFragment(ctx_.getNextFragmentId(), exchNode, partition);
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test
index dec3db7da..950f5a53a 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test
@@ -298,10 +298,12 @@ MERGE SINK
|->WRITE TO HDFS [functional_parquet.iceberg_partition_transforms_zorder, OVERWRITE=false, PARTITION-KEYS=(year(target.ts),iceberg_bucket_transform(target.s, 5))]
|->BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_partition_transforms_zorder-POSITION-DELETE]
|
-05:SORT
+06:SORT
| order by: LEXICAL: year(target.ts) ASC NULLS LAST, iceberg_bucket_transform(target.s, 5) ASC NULLS LAST, ZORDER: i, j
| row-size=81B cardinality=1
|
+05:EXCHANGE [HASH(year(target.ts),iceberg_bucket_transform(target.s, 5))]
+|
03:MERGE
| CASE 0: MATCHED
| | result expressions: source.event_time, target.s, target.i, target.j
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-delete.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-delete.test
index de7f81c81..9c19becd1 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-delete.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-delete.test
@@ -237,6 +237,8 @@ BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_v2_partitioned_position
---- DISTRIBUTEDPLAN
BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_v2_partitioned_position_deletes-POSITION-DELETE]
|
+04:EXCHANGE [HASH(functional_parquet.iceberg_v2_partitioned_position_deletes.PARTITION__SPEC__ID,functional_parquet.iceberg_v2_partitioned_position_deletes.ICEBERG__PARTITION__SERIALIZED)]
+|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| row-size=36B cardinality=3
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-update.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-update.test
index a1781b937..f40a1a66e 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-update.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-update.test
@@ -211,10 +211,12 @@ MULTI DATA SINK
|->WRITE TO HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes, OVERWRITE=false, PARTITION-KEYS=(action)]
|->BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_v2_partitioned_position_deletes-POSITION-DELETE]
|
-01:SORT
+02:SORT
| order by: action ASC NULLS LAST
| row-size=76B cardinality=0
|
+01:EXCHANGE [HASH(functional_parquet.iceberg_v2_partitioned_position_deletes.action)]
+|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=0/3 files=0 size=0B
predicates: id IS DISTINCT FROM length(action)
@@ -251,10 +253,12 @@ MULTI DATA SINK
|->WRITE TO HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes, OVERWRITE=false, PARTITION-KEYS=(action)]
|->BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_v2_partitioned_position_deletes-POSITION-DELETE]
|
-04:SORT
+05:SORT
| order by: action ASC NULLS LAST
| row-size=76B cardinality=1
|
+04:EXCHANGE [HASH(functional_parquet.iceberg_v2_partitioned_position_deletes.action)]
+|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| row-size=80B cardinality=1
|
@@ -402,10 +406,12 @@ MULTI DATA SINK
|->WRITE TO HDFS [functional_parquet.iceberg_partition_transforms_zorder, OVERWRITE=false, PARTITION-KEYS=(year(days_add(ts, 10)),iceberg_bucket_transform(functional_parquet.iceberg_partition_transforms_zorder.s, 5))]
|->BUFFERED DELETE FROM ICEBERG [functional_parquet.iceberg_partition_transforms_zorder-POSITION-DELETE]
|
-01:SORT
+02:SORT
| order by: LEXICAL: year(days_add(ts, 10)) ASC NULLS LAST, iceberg_bucket_transform(functional_parquet.iceberg_partition_transforms_zorder.s, 5) ASC NULLS LAST, ZORDER: CAST(i + 1000 AS INT), j
| row-size=80B cardinality=1
|
+01:EXCHANGE [HASH(year(days_add(ts, 10)),iceberg_bucket_transform(functional_parquet.iceberg_partition_transforms_zorder.s, 5))]
+|
00:SCAN HDFS [functional_parquet.iceberg_partition_transforms_zorder]
HDFS partitions=1/1 files=1 size=1.08KB
predicates: ts IS DISTINCT FROM days_add(ts, 10) OR i IS DISTINCT FROM CAST(i + 1000 AS INT)
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
index fd0bc7780..526d6d3e6 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
@@ -537,3 +537,29 @@ BIGINT
SELECT * FROM ice_lineitem WHERE l_orderkey % 5 = 1;
---- RESULTS
====
+---- QUERY
+CREATE TABLE delete_all (s STRING)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO delete_all VALUES ('a'), ('b');
+DELETE FROM delete_all WHERE s='a';
+INSERT INTO delete_all VALUES ('c'), ('d'), ('e');
+---- DML_RESULTS: delete_all
+'b'
+'c'
+'d'
+'e'
+---- TYPES
+STRING
+====
+---- QUERY
+DELETE FROM delete_all where s > '';
+---- DML_RESULTS: delete_all
+---- TYPES
+STRING
+---- RUNTIME_PROFILE
+# Unpartitioned V3 table must shuffle on INPUT__FILE__NAME to guarantee one
+# Deletion Vector per data file.
+row_regex: .*EXCHANGE \[HASH\(.*INPUT__FILE__NAME.*\)\].*
+====
\ No newline at end of file
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-merge.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-merge.test
index 2104c901f..0eec7e07f 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-merge.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-merge.test
@@ -1419,3 +1419,32 @@ select count(1) from $DATABASE.target_part.snapshots
---- TYPES
BIGINT
====
+---- QUERY
+# Merge from/into table with some positional deletes
+CREATE TABLE s23 (s STRING)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO s23 VALUES ('a'), ('b');
+DELETE FROM s23 WHERE s='a';
+INSERT INTO s23 VALUES ('c'), ('d'), ('e');
+---- DML_RESULTS: s23
+'b'
+'c'
+'d'
+'e'
+---- TYPES
+STRING
+====
+---- QUERY
+MERGE INTO s23 AS tgt USING (SELECT * from s23) AS src on src.s = tgt.s
+WHEN MATCHED THEN DELETE
+WHEN NOT MATCHED THEN INSERT (s) VALUES ("what");
+---- DML_RESULTS: s23
+---- TYPES
+STRING
+---- RUNTIME_PROFILE
+# Unpartitioned V3 table must shuffle on INPUT__FILE__NAME to guarantee one
+# Deletion Vector per data file.
+row_regex: .*EXCHANGE \[HASH\(.*INPUT__FILE__NAME.*\)\].*
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-update.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-update.test
index 4960fb3e2..b7aa718f9 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-update.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-update.test
@@ -729,3 +729,33 @@ INT
NumModifiedRows: 2
NumDeletedRows: 2
====
+---- QUERY
+CREATE TABLE update_all (s STRING)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO update_all VALUES ('a'), ('b');
+DELETE FROM update_all WHERE s='a';
+INSERT INTO update_all VALUES ('c'), ('d'), ('e');
+---- DML_RESULTS: update_all
+'b'
+'c'
+'d'
+'e'
+---- TYPES
+STRING
+====
+---- QUERY
+UPDATE update_all set s = 'updated';
+---- DML_RESULTS: update_all
+'updated'
+'updated'
+'updated'
+'updated'
+---- TYPES
+STRING
+---- RUNTIME_PROFILE
+# Unpartitioned V3 table must shuffle on INPUT__FILE__NAME to guarantee one
+# Deletion Vector per data file.
+row_regex: .*EXCHANGE \[HASH\(.*INPUT__FILE__NAME.*\)\].*
+====
\ No newline at end of file