This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new acc5a3ddc06 branch-4.2: [feature](iceberg) Add
iceberg_partition_overwrite_mode session variable (#69010) (#69841)
acc5a3ddc06 is described below
commit acc5a3ddc067d5abe8ba88742a2450ac4a5cd8d6
Author: daidai <[email protected]>
AuthorDate: Thu Nov 5 16:48:52 2026 +0800
branch-4.2: [feature](iceberg) Add iceberg_partition_overwrite_mode session
variable (#69010) (#69841)
pick #69010
---
.../datasource/iceberg/IcebergTransaction.java | 37 +++++-
.../commands/insert/IcebergInsertExecutor.java | 7 +-
.../java/org/apache/doris/qe/SessionVariable.java | 37 ++++++
.../datasource/iceberg/IcebergTransactionTest.java | 131 ++++++++++++++++++++-
.../commands/insert/IcebergInsertExecutorTest.java | 83 +++++++++++++
.../org/apache/doris/qe/SessionVariablesTest.java | 36 ++++++
.../test_iceberg_partition_overwrite_mode.out | 22 ++++
.../iceberg/iceberg_branch_insert_data.groovy | 3 +
.../iceberg/test_iceberg_jdbc_catalog.groovy | 3 +
.../iceberg/test_iceberg_timestamp_ns.groovy | 3 +
.../write/test_iceberg_insert_overwrite.groovy | 3 +
...ceberg_insert_overwrite_with_empty_table.groovy | 3 +
.../test_iceberg_partition_overwrite_mode.groovy | 94 +++++++++++++++
.../test_iceberg_write_branch_dml_boundary.groovy | 3 +
.../write/test_iceberg_write_evolution_refs.groovy | 3 +
.../test_iceberg_write_overwrite_atomicity.groovy | 3 +
...est_iceberg_write_overwrite_delete_files.groovy | 3 +
.../test_iceberg_write_overwrite_evolution.groovy | 3 +
.../test_s3tables_glue_insert_overwrite.groovy | 3 +
.../iceberg/test_s3tables_insert_overwrite.groovy | 3 +
20 files changed, 478 insertions(+), 5 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java
index 3c2b40ae609..e2291105ef2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java
@@ -27,6 +27,7 @@ import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.iceberg.helper.IcebergWriterHelper;
import
org.apache.doris.nereids.trees.plans.commands.insert.IcebergInsertCommandContext;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext;
+import org.apache.doris.qe.SessionVariable;
import org.apache.doris.thrift.TFileContent;
import org.apache.doris.thrift.TIcebergCommitData;
import org.apache.doris.thrift.TUpdateMode;
@@ -96,6 +97,7 @@ public class IcebergTransaction implements Transaction {
private Optional<Expression> conflictDetectionFilter = Optional.empty();
private IcebergInsertCommandContext insertCtx;
+ private boolean wholeTableOverwrite;
private Optional<IcebergWriteSchemaContext> writeSchemaContext =
Optional.empty();
private String branchName;
private Long baseSnapshotId;
@@ -147,7 +149,15 @@ public class IcebergTransaction implements Transaction {
/** Begin an insert against the metadata generation retained during sink
binding. */
public void beginInsert(ExternalTable dorisTable, Table targetTable,
Optional<InsertCommandContext> ctx) throws UserException {
+ beginInsert(dorisTable, targetTable, ctx,
SessionVariable.ICEBERG_PARTITION_OVERWRITE_MODE_STATIC);
+ }
+
+ /** Capture the session's overwrite scope before executing the write. */
+ public void beginInsert(ExternalTable dorisTable, Table targetTable,
+ Optional<InsertCommandContext> ctx, String partitionOverwriteMode)
throws UserException {
this.insertCtx = ctx.map(c -> (IcebergInsertCommandContext)
c).orElse(null);
+ this.wholeTableOverwrite =
SessionVariable.ICEBERG_PARTITION_OVERWRITE_MODE_STATIC
+ .equalsIgnoreCase(partitionOverwriteMode);
this.writeSchemaContext = insertCtx == null
? Optional.empty() : insertCtx.getWriteSchemaContext();
try {
@@ -158,7 +168,7 @@ public class IcebergTransaction implements Transaction {
this.table = createTransactionTable(dorisTable, targetTable);
this.branchName = null;
this.isRewriteMode = false;
- this.baseSnapshotId = null;
+ this.baseSnapshotId = getSnapshotIdIfPresent(table);
this.hasStagedUpdates = false;
// check branch
if (insertCtx != null &&
insertCtx.getBranchName().isPresent()) {
@@ -171,6 +181,7 @@ public class IcebergTransaction implements Transaction {
branchName
+ " is a tag, not a branch. Tags
cannot be targets for producing snapshots");
}
+ this.baseSnapshotId = branchRef.snapshotId();
}
if (writeSchemaContext.isPresent()) {
writeSchemaContext.get().validateCurrentSchema(
@@ -664,6 +675,8 @@ public class IcebergTransaction implements Transaction {
// Check if this is a static partition overwrite
if (insertCtx != null && insertCtx.isStaticPartitionOverwrite()) {
commitStaticPartitionOverwrite(pendingResults);
+ } else if (wholeTableOverwrite) {
+ commitWholeTableOverwrite(pendingResults);
} else {
commitReplaceTxn(pendingResults);
}
@@ -1135,6 +1148,28 @@ public class IcebergTransaction implements Transaction {
return referencedDataFiles;
}
+ /** A row filter clears every partition spec, even when the write produces
no files. */
+ private void commitWholeTableOverwrite(List<WriteResult> pendingResults) {
+ OverwriteFiles overwriteFiles = transaction.newOverwrite();
+ if (branchName != null) {
+ overwriteFiles = overwriteFiles.toBranch(branchName);
+ }
+ overwriteFiles =
overwriteFiles.scanManifestsWith(ops.getThreadPoolWithPreAuth())
+ .overwriteByRowFilter(Expressions.alwaysTrue())
+ .conflictDetectionFilter(Expressions.alwaysTrue());
+ if (baseSnapshotId != null) {
+ overwriteFiles =
overwriteFiles.validateFromSnapshot(baseSnapshotId);
+ }
+ overwriteFiles =
overwriteFiles.validateNoConflictingData().validateNoConflictingDeletes();
+ for (WriteResult result : pendingResults) {
+ Preconditions.checkState(result.referencedDataFiles().length == 0,
+ "Should have no referenced data files for a whole-table
overwrite.");
+ Arrays.stream(result.dataFiles()).forEach(overwriteFiles::addFile);
+ }
+ overwriteFiles.commit();
+ hasStagedUpdates = true;
+ }
+
private void commitReplaceTxn(List<WriteResult> pendingResults) {
if (pendingResults.isEmpty()) {
// such as : insert overwrite table `dst_tb` select * from
`empty_tb`
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutor.java
index 59b0fdd8e4f..5130e2f0180 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutor.java
@@ -54,7 +54,8 @@ public class IcebergInsertExecutor extends
BaseExternalTableInsertExecutor {
@Override
protected void beforeExec() throws UserException {
IcebergTransaction transaction = (IcebergTransaction)
transactionManager.getTransaction(txnId);
- transaction.beginInsert((IcebergExternalTable) table,
targetIcebergTable, insertCtx);
+ transaction.beginInsert((IcebergExternalTable) table,
targetIcebergTable, insertCtx,
+ ctx.getSessionVariable().getIcebergPartitionOverwriteMode());
}
@Override
@@ -75,10 +76,12 @@ public class IcebergInsertExecutor extends
BaseExternalTableInsertExecutor {
@Override
public boolean shouldExecuteEmptyInsert() {
+ // An overwrite still has side effects with no source rows. The
transaction decides whether
+ // to clear the whole table, delete explicit partitions, or preserve a
partitioned dynamic table.
return emptyInsert
&&
insertCtx.filter(IcebergInsertCommandContext.class::isInstance)
.map(IcebergInsertCommandContext.class::cast)
-
.map(IcebergInsertCommandContext::isStaticPartitionOverwrite)
+ .map(IcebergInsertCommandContext::isOverwrite)
.orElse(false);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 0c21e481a86..ff315d8674f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -4003,6 +4003,19 @@ public class SessionVariable implements Serializable,
Writable {
"Enable merge partitioning for Iceberg UPDATE/DELETE
(INSERT by partition columns, "
+ "DELETE by row_id)."})
public boolean enableIcebergMergePartitioning = true;
+
+ public static final String ICEBERG_PARTITION_OVERWRITE_MODE =
"iceberg_partition_overwrite_mode";
+ public static final String ICEBERG_PARTITION_OVERWRITE_MODE_DYNAMIC =
"dynamic";
+ public static final String ICEBERG_PARTITION_OVERWRITE_MODE_STATIC =
"static";
+ @VariableMgr.VarAttr(name = ICEBERG_PARTITION_OVERWRITE_MODE, needForward
= true,
+ checker = "checkIcebergPartitionOverwriteMode", setter =
"setIcebergPartitionOverwriteMode",
+ description = {"Scope of INSERT OVERWRITE on an Iceberg table when
no PARTITION clause is given. "
+ + "'static' (default, matching Spark's
partitionOverwriteMode=static) overwrites the whole "
+ + "table, including data written under older partition
specs. 'dynamic' replaces only "
+ + "the partitions the new data lands in, preserving the
previous Doris behavior."},
+ options = {ICEBERG_PARTITION_OVERWRITE_MODE_STATIC,
ICEBERG_PARTITION_OVERWRITE_MODE_DYNAMIC})
+ public String icebergPartitionOverwriteMode =
ICEBERG_PARTITION_OVERWRITE_MODE_STATIC;
+
// If this fe is in fuzzy mode, then will use initFuzzyModeVariables to
generate some variables,
// not the default value set in the code.
@@ -5714,6 +5727,30 @@ public class SessionVariable implements Serializable,
Writable {
parseInsertVisibleTimeoutReturnMode(mode);
}
+ public String getIcebergPartitionOverwriteMode() {
+ return icebergPartitionOverwriteMode;
+ }
+
+ public void checkIcebergPartitionOverwriteMode(String mode) {
+ parseIcebergPartitionOverwriteMode(mode);
+ }
+
+ public void setIcebergPartitionOverwriteMode(String mode) {
+ icebergPartitionOverwriteMode =
parseIcebergPartitionOverwriteMode(mode);
+ }
+
+ private static String parseIcebergPartitionOverwriteMode(String mode) {
+ if (mode != null) {
+ String normalized = mode.trim().toLowerCase(Locale.ROOT);
+ if (ICEBERG_PARTITION_OVERWRITE_MODE_DYNAMIC.equals(normalized)
+ ||
ICEBERG_PARTITION_OVERWRITE_MODE_STATIC.equals(normalized)) {
+ return normalized;
+ }
+ }
+ throw new UnsupportedOperationException(
+ "iceberg_partition_overwrite_mode must be 'dynamic' or
'static', but was: " + mode);
+ }
+
// Parse the stored string case-insensitively and expose the enum only to
business logic.
private InsertVisibleTimeoutReturnMode
parseInsertVisibleTimeoutReturnMode(String mode) {
if (StringUtils.isEmpty(mode)) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java
index 73765a64927..0bf0ce98424 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java
@@ -46,6 +46,7 @@ import org.apache.iceberg.RowDelta;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.CommitFailedException;
@@ -576,7 +577,7 @@ public class IcebergTransactionTest {
try (MockedStatic<IcebergUtils> mockedUtils =
Mockito.mockStatic(IcebergUtils.class,
Mockito.CALLS_REAL_METHODS)) {
mockWritableTable(mockedUtils, table);
- txn.beginInsert(dorisTable, Optional.of(insertContext));
+ txn.beginInsert(dorisTable, table, Optional.of(insertContext),
"dynamic");
if (hasOutputFile) {
txn.finishInsert(NameMapping.createForTest(dbName, tableName));
}
@@ -854,6 +855,124 @@ public class IcebergTransactionTest {
Assert.assertEquals("2024-12-11", numToDay(dt));
}
+ @Test
+ public void testPartitionOverwriteModes() throws Exception {
+ for (String mode : new String[] {null, "static", "dynamic"}) {
+ verifyPartitionOverwriteMode(mode, false, false, false);
+ verifyPartitionOverwriteMode(mode, true, false, false);
+ }
+ }
+
+ @Test
+ public void testWholeTableOverwriteClearsOlderPartitionSpecs() throws
Exception {
+ verifyPartitionOverwriteMode("static", false, true, false);
+ verifyPartitionOverwriteMode("static", true, true, false);
+ }
+
+ @Test
+ public void testWholeTableOverwriteTargetsOnlyBranch() throws Exception {
+ verifyPartitionOverwriteMode("static", false, false, true);
+ verifyPartitionOverwriteMode("static", true, false, true);
+ }
+
+ private void verifyPartitionOverwriteMode(String mode, boolean emptySource,
+ boolean evolveSpec, boolean targetBranch) throws Exception {
+ String tableName = "overwrite_" +
UUID.randomUUID().toString().replace("-", "");
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.required(2, "p", Types.StringType.get()));
+ PartitionSpec spec = evolveSpec ? PartitionSpec.unpartitioned()
+ : PartitionSpec.builderFor(schema).identity("p").build();
+ Table table = ops.getCatalog().createTable(TableIdentifier.of(dbName,
tableName), schema, spec);
+ DataFile seed = DataFiles.builder(spec).withPath(table.location() +
"/seed.parquet")
+ .withPartitionPath(evolveSpec ? "" :
"p=us").withFileSizeInBytes(1).withRecordCount(1).build();
+ table.newAppend().appendFile(seed).commit();
+ if (evolveSpec) {
+ table.updateSpec().addField("p").commit();
+ }
+ IcebergInsertCommandContext context = new
IcebergInsertCommandContext();
+ context.setOverwrite(true);
+ if (targetBranch) {
+ table.manageSnapshots().createBranch("audit").commit();
+ context.setBranchName(Optional.of("audit"));
+ // Main advances independently, so the conflict base must be the
audit branch's head.
+ table.newAppend().appendFile(DataFiles.builder(table.spec())
+ .withPath(table.location() +
"/main.parquet").withPartitionPath("p=ap")
+
.withFileSizeInBytes(1).withRecordCount(1).build()).commit();
+ }
+ IcebergExternalTable dorisTable =
Mockito.mock(IcebergExternalTable.class);
+ Mockito.when(dorisTable.getName()).thenReturn(tableName);
+ try (MockedStatic<IcebergUtils> mockedUtils =
+ Mockito.mockStatic(IcebergUtils.class,
Mockito.CALLS_REAL_METHODS)) {
+ mockWritableTable(mockedUtils, table);
+ IcebergTransaction txn = getTxn();
+ if (mode == null) {
+ txn.beginInsert(dorisTable, table, Optional.of(context));
+ } else {
+ txn.beginInsert(dorisTable, table, Optional.of(context), mode);
+ }
+ if (!emptySource) {
+ TIcebergCommitData data = new TIcebergCommitData();
+ data.setFilePath(table.location() + "/output.parquet");
+ data.setPartitionValues(Collections.singletonList("eu"));
+ data.setPartitionSpecId(table.spec().specId());
+ data.setFileContent(TFileContent.DATA);
+ data.setRowCount(1);
+ data.setFileSize(1);
+ txn.updateIcebergCommitData(Collections.singletonList(data));
+ }
+ txn.finishInsert(NameMapping.createForTest(dbName, tableName));
+ txn.commit();
+ }
+ table.refresh();
+ TableScan scan = targetBranch ? table.newScan().useRef("audit") :
table.newScan();
+ List<String> files = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ tasks.forEach(task -> files.add(task.file().path().toString()));
+ }
+ Assert.assertEquals(("dynamic".equals(mode) ? 1 : 0) + (emptySource ?
0 : 1), files.size());
+ Assert.assertEquals("dynamic".equals(mode),
files.contains(seed.path().toString()));
+ if (!emptySource) {
+ Assert.assertTrue(files.contains(table.location() +
"/output.parquet"));
+ }
+ if (targetBranch) {
+ checkPushDownByPartition(table, Expressions.alwaysTrue(), 2);
+ }
+ }
+
+ @Test
+ public void testWholeTableOverwriteRejectsConcurrentDataAndDeletes()
throws Exception {
+ for (boolean concurrentDelete : new boolean[] {false, true}) {
+ String tableName = "overwrite_conflict_" + concurrentDelete;
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ Table table =
ops.getCatalog().createTable(TableIdentifier.of(dbName, tableName), schema,
+ PartitionSpec.unpartitioned(),
ImmutableMap.of("format-version", "2"));
+ table.newAppend().appendFile(DataFiles.builder(table.spec())
+ .withPath(table.location() +
"/seed.parquet").withFileSizeInBytes(1)
+ .withRecordCount(1).build()).commit();
+ IcebergExternalTable dorisTable =
Mockito.mock(IcebergExternalTable.class);
+ Mockito.when(dorisTable.getName()).thenReturn(tableName);
+ IcebergInsertCommandContext context = new
IcebergInsertCommandContext();
+ context.setOverwrite(true);
+ try (MockedStatic<IcebergUtils> mockedUtils =
+ Mockito.mockStatic(IcebergUtils.class,
Mockito.CALLS_REAL_METHODS)) {
+ mockWritableTable(mockedUtils, table);
+ IcebergTransaction txn = getTxn();
+ txn.beginInsert(dorisTable, table, Optional.of(context));
+ txn.finishInsert(NameMapping.createForTest(dbName, tableName));
+ if (concurrentDelete) {
+
table.newRowDelta().addDeletes(FileMetadata.deleteFileBuilder(table.spec())
+ .ofEqualityDeletes(1).withPath(table.location() +
"/delete.parquet")
+
.withFileSizeInBytes(1).withRecordCount(1).build()).commit();
+ } else {
+
table.newAppend().appendFile(DataFiles.builder(table.spec())
+ .withPath(table.location() +
"/concurrent.parquet").withFileSizeInBytes(1)
+ .withRecordCount(1).build()).commit();
+ }
+
Assert.assertThrows(org.apache.iceberg.exceptions.ValidationException.class,
txn::commit);
+ }
+ }
+ }
+
@Test
public void testUnPartitionedTableOverwriteWithData() throws UserException
{
@@ -904,7 +1023,15 @@ public class IcebergTransactionTest {
@Test
public void testUnpartitionedTableOverwriteWithoutData() throws
UserException {
+ verifyUnpartitionedEmptyOverwrite("static");
+ }
+ @Test
+ public void testUnpartitionedDynamicOverwriteWithoutData() throws
UserException {
+ verifyUnpartitionedEmptyOverwrite("dynamic");
+ }
+
+ private void verifyUnpartitionedEmptyOverwrite(String mode) throws
UserException {
testUnPartitionedTableOverwriteWithData();
Table table = ops.getCatalog().loadTable(TableIdentifier.of(dbName,
tbWithoutPartition));
@@ -917,7 +1044,7 @@ public class IcebergTransactionTest {
IcebergTransaction txn = getTxn();
IcebergInsertCommandContext ctx = new
IcebergInsertCommandContext();
- txn.beginInsert(icebergExternalTable, table, Optional.of(ctx));
+ txn.beginInsert(icebergExternalTable, table, Optional.of(ctx),
mode);
ctx.setOverwrite(true);
txn.finishInsert(NameMapping.createForTest(dbName,
tbWithPartition));
txn.commit();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutorTest.java
new file mode 100644
index 00000000000..4516a5fc849
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergInsertExecutorTest.java
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.commands.insert;
+
+import org.apache.doris.catalog.EnvFactory;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.datasource.iceberg.IcebergTransaction;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.Coordinator;
+import org.apache.doris.transaction.TransactionManager;
+
+import org.apache.iceberg.Table;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Optional;
+
+class IcebergInsertExecutorTest {
+ @AfterEach
+ void tearDown() {
+ ConnectContext.remove();
+ }
+
+ @Test
+ void testEmptyOverwriteExecutesAndPassesSessionMode() throws Exception {
+ ConnectContext ctx = new ConnectContext();
+ ctx.setThreadLocalInfo();
+ IcebergExternalCatalog catalog =
Mockito.mock(IcebergExternalCatalog.class);
+ IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+ Mockito.when(table.getCatalog()).thenReturn(catalog);
+
Mockito.when(table.getDatabase()).thenReturn(Mockito.mock(ExternalDatabase.class));
+ TransactionManager manager = Mockito.mock(TransactionManager.class);
+ Mockito.when(catalog.getTransactionManager()).thenReturn(manager);
+ IcebergTransaction transaction =
Mockito.mock(IcebergTransaction.class);
+
Mockito.when(manager.getTransaction(Mockito.anyLong())).thenReturn(transaction);
+ Table targetTable = Mockito.mock(Table.class);
+ EnvFactory factory = Mockito.mock(EnvFactory.class);
+ Mockito.when(factory.createCoordinator(Mockito.any(), Mockito.any(),
Mockito.any(), Mockito.anyLong()))
+ .thenReturn(Mockito.mock(Coordinator.class));
+ try (MockedStatic<EnvFactory> mockedFactory =
Mockito.mockStatic(EnvFactory.class)) {
+ mockedFactory.when(EnvFactory::getInstance).thenReturn(factory);
+ IcebergInsertCommandContext context = new
IcebergInsertCommandContext();
+ context.setOverwrite(true);
+ for (String mode : new String[] {"static", "dynamic"}) {
+
ctx.getSessionVariable().setIcebergPartitionOverwriteMode(mode);
+ IcebergInsertExecutor executor = new
IcebergInsertExecutor(ctx, table, targetTable,
+ "empty_overwrite", Mockito.mock(NereidsPlanner.class),
Optional.of(context), true, 0L);
+ Assertions.assertTrue(executor.shouldExecuteEmptyInsert());
+ executor.beforeExec();
+ Mockito.verify(transaction).beginInsert(table, targetTable,
Optional.of(context), mode);
+ }
+ context.setOverwrite(false);
+ IcebergInsertExecutor append = new IcebergInsertExecutor(ctx,
table, targetTable,
+ "empty_append", Mockito.mock(NereidsPlanner.class),
Optional.of(context), true, 0L);
+ Assertions.assertFalse(append.shouldExecuteEmptyInsert());
+ context.setOverwrite(true);
+ IcebergInsertExecutor nonEmpty = new IcebergInsertExecutor(ctx,
table, targetTable,
+ "non_empty", Mockito.mock(NereidsPlanner.class),
Optional.of(context), false, 0L);
+ Assertions.assertFalse(nonEmpty.shouldExecuteEmptyInsert());
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index dc984606dcb..1ec21e7e961 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -248,6 +248,42 @@ public class SessionVariablesTest extends
TestWithFeService {
Assertions.assertEquals(Action.NOPUSH,
restored.getForceEagerAggHintMap().get("sum:t2.b"));
}
+ @Test
+ public void testIcebergPartitionOverwriteMode() throws Exception {
+ SessionVariable sessionVar = new SessionVariable();
+ Assertions.assertEquals("static",
sessionVar.getIcebergPartitionOverwriteMode());
+ VariableMgr.setVar(sessionVar, new SetVar(SetType.SESSION,
+ SessionVariable.ICEBERG_PARTITION_OVERWRITE_MODE, new
StringLiteral(" DyNaMiC ")));
+ Assertions.assertEquals("dynamic",
sessionVar.getIcebergPartitionOverwriteMode());
+ Assertions.assertEquals("dynamic",
+
sessionVar.getForwardVariables().get(SessionVariable.ICEBERG_PARTITION_OVERWRITE_MODE));
+ SessionVariable forwarded = new SessionVariable();
+
forwarded.setForwardedSessionVariables(sessionVar.getForwardVariables());
+ Assertions.assertEquals("dynamic",
forwarded.getIcebergPartitionOverwriteMode());
+ sessionVar.setIcebergPartitionOverwriteMode("StAtIc");
+ Assertions.assertEquals("static",
sessionVar.getIcebergPartitionOverwriteMode());
+ for (String invalid : new String[] {null, "", "replace"}) {
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () ->
sessionVar.checkIcebergPartitionOverwriteMode(invalid));
+ }
+ ExceptionChecker.expectThrowsWithMsg(DdlException.class,
+ "iceberg_partition_overwrite_mode must be 'dynamic' or
'static'",
+ () -> VariableMgr.setVar(sessionVar, new
SetVar(SetType.SESSION,
+ SessionVariable.ICEBERG_PARTITION_OVERWRITE_MODE, new
StringLiteral("replace"))));
+ }
+
+ @Test
+ public void testIcebergPartitionOverwriteModeInHint() {
+ connectContext.setThreadLocalInfo();
+ new NereidsParser().parseSQL("select /*+
set_var(iceberg_partition_overwrite_mode='DyNaMiC') */ "
+ + "* from test_t1");
+ Assertions.assertEquals("dynamic",
connectContext.getSessionVariable().getIcebergPartitionOverwriteMode());
+ RuntimeException exception =
Assertions.assertThrows(RuntimeException.class,
+ () -> new NereidsParser().parseSQL(
+ "select /*+
set_var(iceberg_partition_overwrite_mode='replace') */ * from test_t1"));
+
Assertions.assertTrue(exception.getMessage().contains("iceberg_partition_overwrite_mode"));
+ }
+
@Test
public void testSetVarInHint() {
String sql = "insert into test_t1 select /*+
set_var(enable_nereids_dml_with_pipeline=false)*/ * from test_t1 where
enable_nereids_dml_with_pipeline=true";
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.out
new file mode 100644
index 00000000000..60587a9e8ae
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.out
@@ -0,0 +1,22 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before --
+1 1
+1 2
+1 3
+
+-- !default_static_data --
+7 3
+
+-- !default_static_empty --
+
+-- !dynamic_data --
+1 1
+1 3
+9 2
+
+-- !dynamic_empty --
+1 1
+1 3
+9 2
+
+-- !static_optimized_empty --
diff --git
a/regression-test/suites/external_table_p0/iceberg/iceberg_branch_insert_data.groovy
b/regression-test/suites/external_table_p0/iceberg/iceberg_branch_insert_data.groovy
index 68146796d0a..b1088fca73a 100644
---
a/regression-test/suites/external_table_p0/iceberg/iceberg_branch_insert_data.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/iceberg_branch_insert_data.groovy
@@ -22,6 +22,9 @@ suite("iceberg_branch_insert_data",
"p0,external,doris,external_docker,external_
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String rest_port = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
index 368fccc008b..d7db63d301b 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
@@ -22,6 +22,9 @@ suite("test_iceberg_jdbc_catalog",
"p0,external,iceberg,external_docker,external
return;
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String enabledJdbc = context.config.otherConfigs.get("enableJdbcTest")
if (enabledJdbc == null || !enabledJdbc.equalsIgnoreCase("true")) {
logger.info("Iceberg JDBC catalog test requires enableJdbcTest, skip
this test")
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_timestamp_ns.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_timestamp_ns.groovy
index 5815bad6219..085de0036e5 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_timestamp_ns.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_timestamp_ns.groovy
@@ -19,6 +19,9 @@ suite("test_iceberg_timestamp_ns", "p0,external,iceberg") {
if
(!context.config.otherConfigs.get("enableIcebergTest")?.equalsIgnoreCase("true"))
{
return
}
+
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
String host = context.config.otherConfigs.get("externalEnvIp")
// Use the existing HDFS fixture so metadata defaults are not rescaled by
the older REST server.
String hdfsPort = context.config.otherConfigs.get("hive2HdfsPort")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite.groovy
index ce2415f9634..2e37dfc1e2b 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite.groovy
@@ -16,6 +16,9 @@
// under the License.
suite("test_iceberg_insert_overwrite",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
def format_compressions = ["parquet_zstd", "orc_zlib"]
def q01 = { String format_compression, String catalog_name ->
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.groovy
index 625a836ff30..00b90457fcc 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.groovy
@@ -22,6 +22,9 @@ suite("test_iceberg_insert_overwrite_with_empty_table",
"p0,external,doris,exter
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String rest_port = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.groovy
new file mode 100644
index 00000000000..0f9f9a7f3bb
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_partition_overwrite_mode.groovy
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_partition_overwrite_mode",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test.")
+ return
+ }
+
+ String rest_port = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalog_name = "test_iceberg_partition_overwrite_mode"
+
+ sql """drop catalog if exists ${catalog_name}"""
+ sql """
+ CREATE CATALOG ${catalog_name} PROPERTIES (
+ 'type'='iceberg',
+ 'iceberg.catalog.type'='rest',
+ 'uri' = 'http://${externalEnvIp}:${rest_port}',
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minio_port}";,
+ "s3.region" = "us-east-1"
+ );"""
+
+ sql """ switch ${catalog_name} """
+
+ String db1 = catalog_name + "_db1"
+ String tb1 = db1 + "_tb1"
+ String tb_empty = db1 + "_tb_empty"
+
+ sql """ drop database if exists ${db1} force"""
+ sql """ create database ${db1} """
+ sql """ create table ${db1}.${tb1} (id int, val int) partition by list
(val)() """
+ sql """ create table ${db1}.${tb_empty} (id int, val int) """
+
+ sql """ use ${db1} """
+ sql """ insert into ${tb1} values (1,1), (1,2), (1,3) """
+
+ order_qt_before """ select * from ${tb1} order by val """
+
+ // The default (no SET) is static, matching Spark's default
partitionOverwriteMode=static: a bare
+ // INSERT OVERWRITE clears the whole table first, so only the written row
survives.
+ sql """ insert overwrite table ${tb1} values (7,3) """
+
+ order_qt_default_static_data """ select * from ${tb1} order by val """
+
+ // ... and an empty source empties the table, even though it is
partitioned.
+ sql """ insert overwrite table ${tb1} select * from ${tb_empty} """
+
+ order_qt_default_static_empty """ select * from ${tb1} order by val """
+
+ // dynamic (Hive-compatible, Spark's partitionOverwriteMode=dynamic): only
the written partition (val=2)
+ // is replaced, the others survive.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+ sql """ insert into ${tb1} values (1,1), (1,2), (1,3) """
+ sql """ insert overwrite table ${tb1} values (9,2) """
+
+ order_qt_dynamic_data """ select * from ${tb1} order by val """
+
+ // ... and an empty source is a no-op for a partitioned table.
+ sql """ insert overwrite table ${tb1} select * from ${tb_empty} """
+
+ order_qt_dynamic_empty """ select * from ${tb1} order by val """
+
+ // A planner-proven empty source (WHERE FALSE) must still clear the table
in static mode. With strict DML
+ // consistency disabled (the cloud default) the empty relation reaches the
sink directly, so the insert
+ // must still run its transaction instead of taking the empty-insert
shortcut.
+ sql """ set enable_strict_consistency_dml = false """
+ sql """ set iceberg_partition_overwrite_mode = 'static' """
+ sql """ insert into ${tb1} values (1,1), (1,2), (1,3) """
+ sql """ insert overwrite table ${tb1} select * from ${tb1} where false """
+
+ order_qt_static_optimized_empty """ select * from ${tb1} order by val """
+
+ sql """ set enable_strict_consistency_dml = true """
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_branch_dml_boundary.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_branch_dml_boundary.groovy
index f5f8ef5bb9c..28cf859f65d 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_branch_dml_boundary.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_branch_dml_boundary.groovy
@@ -23,6 +23,9 @@ suite("test_iceberg_write_branch_dml_boundary",
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy
index 69fe59d2b72..fa8dc38ffb1 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy
@@ -23,6 +23,9 @@ suite("test_iceberg_write_evolution_refs",
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_atomicity.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_atomicity.groovy
index fe6d8f2cc37..14ca8dd4587 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_atomicity.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_atomicity.groovy
@@ -23,6 +23,9 @@ suite("test_iceberg_write_overwrite_atomicity",
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_delete_files.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_delete_files.groovy
index d787962094a..6877e54610c 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_delete_files.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_delete_files.groovy
@@ -23,6 +23,9 @@ suite("test_iceberg_write_overwrite_delete_files",
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_evolution.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_evolution.groovy
index 07fa302ba52..b642e53f900 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_evolution.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_overwrite_evolution.groovy
@@ -23,6 +23,9 @@ suite("test_iceberg_write_overwrite_evolution",
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
diff --git
a/regression-test/suites/external_table_p2/iceberg/test_s3tables_glue_insert_overwrite.groovy
b/regression-test/suites/external_table_p2/iceberg/test_s3tables_glue_insert_overwrite.groovy
index 4fa8d1e0291..6c77331a7e3 100644
---
a/regression-test/suites/external_table_p2/iceberg/test_s3tables_glue_insert_overwrite.groovy
+++
b/regression-test/suites/external_table_p2/iceberg/test_s3tables_glue_insert_overwrite.groovy
@@ -18,6 +18,9 @@
import java.util.concurrent.ThreadLocalRandom
suite("test_s3tables_glue_insert_overwrite",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
def format_compressions = ["parquet_zstd", "orc_zlib"]
def q01 = { String format_compression, String catalog_name ->
diff --git
a/regression-test/suites/external_table_p2/iceberg/test_s3tables_insert_overwrite.groovy
b/regression-test/suites/external_table_p2/iceberg/test_s3tables_insert_overwrite.groovy
index 32d95fb2f04..c7738340723 100644
---
a/regression-test/suites/external_table_p2/iceberg/test_s3tables_insert_overwrite.groovy
+++
b/regression-test/suites/external_table_p2/iceberg/test_s3tables_insert_overwrite.groovy
@@ -631,6 +631,9 @@ suite("test_s3tables_insert_overwrite",
"p0,external,iceberg,external_docker,ext
return
}
+ // Preserve this suite's dynamic overwrite expectations after the default
changes to static.
+ sql """ set iceberg_partition_overwrite_mode = 'dynamic' """
+
String catalog_name = "test_s3tables_insert_overwrite"
String props = context.config.otherConfigs.get("icebergS3TablesCatalog")
sql """drop catalog if exists ${catalog_name};"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
