This is an automated email from the ASF dual-hosted git repository.
morningman 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 2fc5248d55c [test](external) Cover connector INSERT OVERWRITE of a
source the planner proves empty (#69291)
2fc5248d55c is described below
commit 2fc5248d55cc5813c77f06453d5c216cc7d1908e
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Oct 28 06:15:43 2026 +0000
[test](external) Cover connector INSERT OVERWRITE of a source the planner
proves empty (#69291)
### What problem does this PR solve?
Issue Number: None
Related PR: #69010, #62344
Problem Summary:
**In short.** This ports the test of branch-4.1 #62344 to master and
adds the tests that master's broader fix, #69010, does not have: an
`INSERT OVERWRITE` whose source the planner proves empty must still
replace its target with nothing. Master behaves correctly since #69010,
but none of its tests reaches the code path #69010 changed, so taking
that change back out keeps every test green. The tests added here fail
without it.
**Background.** When the planner folds the source of an `INSERT
OVERWRITE` to an empty relation (`LIMIT 0`, `WHERE false`, a
contradiction) and the sink needs no exchange below it, the sink sits
directly on `PhysicalEmptyRelation`. `InsertIntoTableCommand` then
treats the insert as empty, and no backend runs it. An empty `INSERT`
has nothing to commit, but an empty overwrite still has to replace the
table, or the static partition, with nothing. Before #69010,
`requiresTransaction()` was false for every empty insert, so the
overwrite never reached the connector: the statement returned OK and
every row stayed. branch-4.1 #62344 fixed this for an Iceberg static
partition overwrite only
(`IcebergInsertExecutor.shouldExecuteEmptyInsert`); its master pick
#65105 changes `IcebergInsertExecutor`, which master no longer has.
#69010 made `PluginDrivenInsertExecutor.requiresTransaction()` true for
every connector overwrite, which covers #62344's case and the others.
Both connectors already commit an overwrite that receives no data:
Paimon commits `withOverwrite(staticPartition)` with no messages, and
Iceberg overwrites the matching files with no new ones.
**The problem, and what it cost.** No test checks any of this.
`PluginDrivenInsertExecutorTest` has no case for
`requiresTransaction()`, and the empty overwrites in the regression
suites never put the sink on the empty relation, so they pass with or
without #69010's change:
| Existing case | Plan on master | Without #69010's
`requiresTransaction` override |
|---|---|---|
| `test_iceberg_partition_overwrite_mode`, `static_optimized_empty`
(partitioned table, `WHERE false`, strict DML off) | hash exchange for
the partition routing, the writers run | clears the table anyway |
| `test_paimon_write_transaction`, `txn_empty_overwrite_limit_zero`
(append-only table, `LIMIT 0`) | `ExecutionAny` exchange, the writers
run | clears the table anyway |
On master the sink sits directly on the empty relation for a Paimon
primary-key table without a fixed bucket count, which is Paimon's
default and is written by a single writer, and for an unpartitioned
Iceberg table when `enable_strict_consistency_dml` is false, the default
in cloud mode. On a local cluster with an FE built without the override,
that is, master before #69010:
| Statement | Without the override | master |
|---|---|---|
| `INSERT OVERWRITE TABLE t SELECT 1, 1 LIMIT 0`, Paimon primary-key
table without `bucket` | returns OK, both rows stay, no snapshot | table
cleared, OVERWRITE snapshot |
| `INSERT OVERWRITE TABLE t PARTITION (p = 'a') SELECT 1, 1 LIMIT 0`,
the same kind of table, partitioned | returns OK, both partitions stay |
only partition `a` cleared |
| `INSERT OVERWRITE TABLE t SELECT * FROM t LIMIT 0`, unpartitioned
Iceberg table, strict DML off | returns OK, all three rows stay | table
cleared |
So a change that brought the shortcut back, or a connector that returned
early when it receives no data, would silently skip such overwrites
again, and no test would notice. branch-4.2 still skips the first and
the third (only an Iceberg static partition overwrite runs its
transaction there); that is reported to its maintainers separately.
**How this PR fixes it.** Tests only:
- `PluginDrivenInsertExecutorTest`: an empty insert that overwrites the
whole table or a static partition requires its transaction; an empty
`INSERT` does not.
- `PaimonConnectorTransactionTest`: an overwrite that receives no commit
data replaces its static partition with nothing and keeps the other
partition, and a whole-table overwrite clears the table, both with an
OVERWRITE snapshot. The table is a real on-disk Paimon table written
through the Paimon API.
- `IcebergConnectorTransactionTest`: an Iceberg static partition
overwrite with no data deletes only the file of the matching partition,
ported from the branch-4.1 test of #62344.
- `test_paimon_write_transaction`: `LIMIT 0` overwrites of a primary-key
table without `bucket` and of a static partition of one. `EXPLAIN SHAPE
PLAN` pins the sink on `PhysicalEmptyRelation`; the suite then checks
the rows, the snapshots, and what Spark reads. The comment of the
existing append-only `LIMIT 0` case said "A direct
PhysicalEmptyRelation"; it now says that the table is written through an
exchange.
- `test_iceberg_insert_overwrite_with_empty_table`: a `LIMIT 0`
overwrite of an unpartitioned table with strict DML consistency off,
with its plan.
**Results.**
- FE UT: `PluginDrivenInsertExecutorTest` passes (`run-fe-ut.sh`, 11
cases). Removing the override turns the new overwrite case red;
requiring a transaction for every empty insert turns the new INSERT case
red. The new Paimon cases fail when the connector also returns early for
an overwrite with no commit data, and the Iceberg case fails when a
static partition overwrite with no data returns early. All 826 tests of
fe-connector-paimon and all 1611 of fe-connector-iceberg pass.
- Regression: both suites pass on a local cluster against the iceberg
docker environment, and fail with the override removed (at
`txn_dynamic_bucket_empty_overwrite` and `q4`). The regenerated `.out`
files differ from master's only by the new blocks.
Original author: @suxiaogang223 (the Iceberg static partition overwrite
test, from #62344)
---
.../iceberg/IcebergConnectorTransactionTest.java | 24 +++++
.../paimon/PaimonConnectorTransactionTest.java | 108 ++++++++++++++++++++-
.../insert/PluginDrivenInsertExecutorTest.java | 38 ++++++++
...t_iceberg_insert_overwrite_with_empty_table.out | 6 ++
.../paimon/write/test_paimon_write_transaction.out | 22 +++++
...ceberg_insert_overwrite_with_empty_table.groovy | 10 ++
.../write/test_paimon_write_transaction.groovy | 50 +++++++++-
7 files changed, 256 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
index acc8400b2bc..ba5ce1cc586 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
@@ -1135,6 +1135,30 @@ public class IcebergConnectorTransactionTest {
"the overwritten partition must be added: " + remaining);
}
+ @Test
+ public void overwriteEmptyStaticPartitionClearsOnlyThatPartition() {
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ PartitionSpec spec =
PartitionSpec.builderFor(PART_SCHEMA).identity("region").build();
+ Table table = catalog.createTable(id, PART_SCHEMA, spec,
props("write.format.default", "parquet"));
+ table.newAppend()
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=us/seed.parquet", 1L, "region=us"))
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=eu/seed.parquet", 1L, "region=eu"))
+ .commit();
+
+ // INSERT OVERWRITE ... PARTITION(region='us') of a source with no
rows: no commit data, and the 'us'
+ // partition is still replaced with nothing while the 'eu' file stays.
+ IcebergConnectorTransaction txn =
txnFor(opsReturning(catalog.loadTable(id)), new RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1",
+ overwriteStaticCtx(catalog.loadTable(id),
Collections.singletonMap("region", "us")));
+ txn.commit();
+
+ List<DataFile> files = currentDataFiles(catalog.loadTable(id));
+ Assertions.assertEquals(1, files.size(), files.toString());
+
Assertions.assertTrue(files.get(0).path().toString().contains("region=eu"),
+ files.get(0).path().toString());
+ }
+
@Test
public void promotedDateStaticOverwriteRejectsUnsafeRetainedBounds() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
index 45f5cf2ceb1..89263a7964c 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
@@ -24,22 +24,31 @@ import
org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
import org.apache.doris.thrift.TPaimonCommitMessage;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.consumer.ConsumerManager;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.thrift.TSerializer;
@@ -64,7 +73,8 @@ import java.util.Map;
* {@code PaimonTransactionTest}: a commit whose outcome was lost is
reconciled against the snapshots
* Paimon actually published, a transaction whose outcome stays unknown is
never aborted, a failure
* before the commit started stays abortable, and the commit payloads the BEs
report are
- * de-duplicated by their exact content.
+ * de-duplicated by their exact content. An overwrite whose writers report no
commit data still
+ * commits: it replaces its static partition, or the whole table, with nothing.
*
* <p>No mocking framework (this module has none by convention). The table is
a real on-disk Paimon
* table, so reconciliation reads real snapshot files; only the committer it
hands out is scripted,
@@ -188,6 +198,35 @@ public class PaimonConnectorTransactionTest {
Assertions.assertEquals(2, target.committedMessages.size());
}
+ @Test
+ public void emptyStaticPartitionOverwriteReplacesThePartitionWithNothing()
throws Exception {
+ // INSERT OVERWRITE ... PARTITION (p = 'a') of a source with no rows:
no writer reports commit data, and
+ // the commit still replaces partition 'a' with nothing. (Mutation:
return early on empty commit data for
+ // an overwrite too -> no snapshot is committed and both rows stay.)
+ FileStoreTable partitioned = partitionedTableWithRows();
+ PaimonConnectorTransaction transaction =
+ overwriteTransaction(partitioned,
Collections.singletonMap("p", "a"));
+
+ transaction.commit();
+
+ Assertions.assertEquals(Snapshot.CommitKind.OVERWRITE,
+ partitioned.snapshotManager().latestSnapshot().commitKind());
+ Assertions.assertEquals(Collections.singletonList("2:b"),
rows(partitioned),
+ "only the overwritten partition is cleared");
+ }
+
+ @Test
+ public void emptyOverwriteReplacesTheTableWithNothing() throws Exception {
+ FileStoreTable partitioned = partitionedTableWithRows();
+ PaimonConnectorTransaction transaction =
overwriteTransaction(partitioned, Collections.emptyMap());
+
+ transaction.commit();
+
+ Assertions.assertEquals(Snapshot.CommitKind.OVERWRITE,
+ partitioned.snapshotManager().latestSnapshot().commitKind());
+ Assertions.assertEquals(Collections.emptyList(), rows(partitioned));
+ }
+
@Test
public void commitUserIsNamespacedByDorisCluster() {
Assertions.assertEquals(
@@ -211,6 +250,73 @@ public class PaimonConnectorTransactionTest {
return transaction;
}
+ /** A table {@code (id INT, p STRING)} partitioned by {@code p}, holding
the rows (1, 'a') and (2, 'b'). */
+ private FileStoreTable partitionedTableWithRows() throws Exception {
+ org.apache.paimon.fs.Path location =
+ new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/partitioned");
+ LocalFileIO fileIO = LocalFileIO.create();
+ new FileSystemSchemaManager(fileIO, location).createTable(new Schema(
+ Arrays.asList(ID, new DataField(1, "p", DataTypes.STRING())),
Collections.singletonList("p"),
+ Collections.emptyList(), Collections.emptyMap(), ""));
+ FileStoreTable partitioned = FileStoreTableFactory.create(fileIO,
location);
+ BatchWriteBuilder writeBuilder = partitioned.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite()) {
+ write.write(GenericRow.of(1, BinaryString.fromString("a")));
+ write.write(GenericRow.of(2, BinaryString.fromString("b")));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+ return partitioned;
+ }
+
+ private static PaimonConnectorTransaction
overwriteTransaction(FileStoreTable target,
+ Map<String, String> staticPartition) {
+ PaimonConnectorTransaction transaction =
+ new PaimonConnectorTransaction(TRANSACTION_ID, new
RecordingConnectorContext());
+ transaction.bind(PaimonWriteBinding.create(
+ new PaimonTableHandle("db", "partitioned",
target.partitionKeys(), Collections.emptyList()),
+ target, Collections.emptyMap(), overwrite(staticPartition),
true));
+ return transaction;
+ }
+
+ /** The rows of the table as {@code id:p}, ordered by id. */
+ private static List<String> rows(FileStoreTable target) throws Exception {
+ ReadBuilder readBuilder = target.newReadBuilder();
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row -> rows.add(row.getInt(0) + ":" +
row.getString(1)));
+ }
+ rows.sort(null);
+ return rows;
+ }
+
+ /** An INSERT OVERWRITE, of the given static partition or, when it is
empty, of the whole table. */
+ private static ConnectorWriteHandle overwrite(Map<String, String>
staticPartition) {
+ return new ConnectorWriteHandle() {
+ @Override
+ public ConnectorTableHandle getTableHandle() {
+ return null;
+ }
+
+ @Override
+ public List<ConnectorColumn> getColumns() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isOverwrite() {
+ return true;
+ }
+
+ @Override
+ public Map<String, String> getStaticPartitionSpec() {
+ return staticPartition;
+ }
+ };
+ }
+
/** An INSERT without a PARTITION clause. */
private static ConnectorWriteHandle plainInsert() {
return new ConnectorWriteHandle() {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertExecutorTest.java
index 1c1f98a73a7..c2fb27541d0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertExecutorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertExecutorTest.java
@@ -32,6 +32,7 @@ import org.apache.doris.connector.spi.handle.WriteOperation;
import org.apache.doris.connector.spi.write.ConnectorSinkPlan;
import org.apache.doris.connector.spi.write.ConnectorWritePlanProvider;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
import org.apache.doris.planner.PluginDrivenTableSink;
import org.apache.doris.thrift.TDataSink;
import org.apache.doris.transaction.PluginDrivenTransactionManager;
@@ -188,6 +189,33 @@ public class PluginDrivenInsertExecutorTest {
"with no connector transaction (empty insert), transactionType
is UNKNOWN");
}
+ @Test
+ public void overwriteOfASourceProvenEmptyStillRunsItsTransaction() {
+ // A sink planned directly on an empty relation (a single-writer plan
of a source folded away, such as
+ // LIMIT 0) makes the write an empty insert. An overwrite of nothing
still replaces the table or the static
+ // partition with nothing, so it must reach the connector's commit
instead of skipping it and reporting
+ // success. (Mutation: drop the requiresTransaction override -> both
overwrites skip their transaction.)
+ PluginDrivenInsertCommandContext wholeTable = new
PluginDrivenInsertCommandContext();
+ wholeTable.setOverwrite(true);
+
Assertions.assertTrue(emptyInsertExecutor(Optional.of(wholeTable)).requiresTransaction(),
+ "an empty overwrite of the whole table must still clear the
table");
+
+ PluginDrivenInsertCommandContext staticPartition = new
PluginDrivenInsertCommandContext();
+ staticPartition.setOverwrite(true);
+ staticPartition.setStaticPartitionSpecFromExpressions(
+ Collections.singletonMap("p", new StringLiteral("a")));
+
Assertions.assertTrue(emptyInsertExecutor(Optional.of(staticPartition)).requiresTransaction(),
+ "an empty overwrite of a static partition must still clear the
partition");
+ }
+
+ @Test
+ public void insertOfASourceProvenEmptySkipsItsTransaction() {
+ // An INSERT of no rows has nothing to commit.
+ Assertions.assertFalse(emptyInsertExecutor(Optional.of(new
PluginDrivenInsertCommandContext()))
+ .requiresTransaction());
+
Assertions.assertFalse(emptyInsertExecutor(Optional.empty()).requiresTransaction());
+ }
+
@Test
public void doBeforeCommitBackfillsLoadedRowsWhenTransactionReportsCount()
throws UserException {
// Transaction model with a real row count (e.g. maxcompute): BE
reports rows only through
@@ -239,6 +267,16 @@ public class PluginDrivenInsertExecutorTest {
return Mockito.mock(PluginDrivenInsertExecutor.class,
Mockito.CALLS_REAL_METHODS);
}
+ /** An executor whose sink sits directly on an empty relation, as initPlan
builds it for such a plan. */
+ private static PluginDrivenInsertExecutor
emptyInsertExecutor(Optional<InsertCommandContext> insertCtx) {
+ PluginDrivenInsertExecutor exec = newUnconstructedExecutor();
+ Deencapsulation.setField(exec, "emptyInsert", true);
+ // Field initializers do not run without the constructor.
+ Deencapsulation.setField(exec, "streamUpdateInfos",
Collections.emptyList());
+ Deencapsulation.setField(exec, "insertCtx", insertCtx);
+ return exec;
+ }
+
/** Write ops that hand back a fixed connector transaction and record the
handle the executor threads in. */
private static final class FakeTxnWriteOps implements ConnectorWriteOps {
private final ConnectorTransaction txn;
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.out
index 042af1d473a..721545afdc4 100644
---
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.out
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_insert_overwrite_with_empty_table.out
@@ -16,3 +16,9 @@
-- !q3 --
+-- !q4_plan --
+PhysicalConnectorTableSink
+--PhysicalEmptyRelation
+
+-- !q4 --
+
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
index b2e4a106999..4f94c306780 100644
---
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
@@ -37,6 +37,28 @@
-- !txn_empty_overwrite_limit_zero --
0
+-- !txn_dynamic_bucket_empty_overwrite_plan --
+PhysicalConnectorTableSink
+--PhysicalEmptyRelation
+
+-- !txn_dynamic_bucket_empty_overwrite --
+0
+
+-- !txn_dynamic_bucket_empty_overwrite_snapshots --
+1 APPEND
+2 OVERWRITE
+
+-- !txn_dynamic_bucket_empty_static_overwrite_plan --
+PhysicalConnectorTableSink
+--PhysicalEmptyRelation
+
+-- !txn_dynamic_bucket_empty_static_overwrite --
+2 west_old west
+
+-- !txn_dynamic_bucket_empty_static_overwrite_snapshots --
+1 APPEND
+2 OVERWRITE
+
-- !txn_static_partition --
10 east_new east
2 west_old west
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 96e603a90ae..af9bcedb932 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
@@ -69,4 +69,14 @@ suite("test_iceberg_insert_overwrite_with_empty_table",
"p0,external") {
order_qt_q2 """ select * from ${tb1} """ // should have 3 records
order_qt_q3 """ select * from ${tb2} """ // should have no records
+ // With strict DML consistency off (the cloud default), the sink of an
unpartitioned table sits directly on
+ // the empty relation of a source the planner proves empty, and no writer
runs. Such an insert of nothing
+ // skips its transaction, but an overwrite still clears the table.
+ sql """ insert into ${tb2} values (1,1), (1,2), (1,3) """
+ sql """ set enable_strict_consistency_dml = false """
+ qt_q4_plan """ explain shape plan insert overwrite table ${tb2} select *
from ${tb2} limit 0 """
+ sql """ insert overwrite table ${tb2} select * from ${tb2} limit 0 """
+ sql """ set enable_strict_consistency_dml = true """
+ order_qt_q4 """ select * from ${tb2} """ // should have no records
+
}
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
index a4dcfec00b5..209b6a6a6e4 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
@@ -47,6 +47,23 @@ suite("test_paimon_write_transaction", "p0,external,paimon")
{
id INT, name STRING
) USING paimon;
+ DROP TABLE IF EXISTS paimon.${dbName}.t_overwrite_dynamic_bucket;
+ CREATE TABLE paimon.${dbName}.t_overwrite_dynamic_bucket (
+ id INT, name STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_overwrite_dynamic_bucket_part;
+ CREATE TABLE paimon.${dbName}.t_overwrite_dynamic_bucket_part (
+ id INT, name STRING, region STRING
+ ) USING paimon
+ PARTITIONED BY (region)
+ TBLPROPERTIES (
+ 'primary-key' = 'id,region'
+ );
+
DROP TABLE IF EXISTS paimon.${dbName}.t_overwrite_part;
CREATE TABLE paimon.${dbName}.t_overwrite_part (
id INT, name STRING, region STRING
@@ -208,12 +225,43 @@ suite("test_paimon_write_transaction",
"p0,external,paimon") {
qt_txn_empty_overwrite """SELECT COUNT(*) FROM t_overwrite"""
assertTableEquals("t_overwrite", "ORDER BY id")
- // A direct PhysicalEmptyRelation must still publish the empty
overwrite snapshot.
+ // A source that LIMIT 0 folds to an empty relation must still publish
the empty overwrite snapshot. An
+ // append-only table is written through an exchange, so the writers
run and report nothing.
sql """INSERT INTO t_overwrite VALUES (30, 'old_for_limit_zero')"""
sql """INSERT OVERWRITE TABLE t_overwrite SELECT 1, 'unused' LIMIT 0"""
qt_txn_empty_overwrite_limit_zero """SELECT COUNT(*) FROM
t_overwrite"""
assertTableEquals("t_overwrite", "ORDER BY id")
+ // A primary-key table without a fixed bucket count has a single
writer, so the sink of an overwrite
+ // whose source the planner proves empty sits directly on the empty
relation, and no writer runs. Such an
+ // insert of nothing skips its transaction, but an overwrite still
replaces the table, or the static
+ // partition, with nothing.
+ sql """INSERT INTO t_overwrite_dynamic_bucket VALUES (1, 'old1'), (2,
'old2')"""
+ qt_txn_dynamic_bucket_empty_overwrite_plan """
+ EXPLAIN SHAPE PLAN INSERT OVERWRITE TABLE
t_overwrite_dynamic_bucket SELECT 1, 'unused' LIMIT 0
+ """
+ sql """INSERT OVERWRITE TABLE t_overwrite_dynamic_bucket SELECT 1,
'unused' LIMIT 0"""
+ qt_txn_dynamic_bucket_empty_overwrite """SELECT COUNT(*) FROM
t_overwrite_dynamic_bucket"""
+ order_qt_txn_dynamic_bucket_empty_overwrite_snapshots """
+ SELECT snapshot_id, commit_kind FROM
t_overwrite_dynamic_bucket\$snapshots
+ """
+ assertTableEquals("t_overwrite_dynamic_bucket", "ORDER BY id")
+
+ sql """INSERT INTO t_overwrite_dynamic_bucket_part VALUES
+ (1, 'east_old', 'east'), (2, 'west_old', 'west')"""
+ qt_txn_dynamic_bucket_empty_static_overwrite_plan """
+ EXPLAIN SHAPE PLAN INSERT OVERWRITE TABLE
t_overwrite_dynamic_bucket_part
+ PARTITION (region = 'east') SELECT 1, 'unused' LIMIT 0
+ """
+ sql """INSERT OVERWRITE TABLE t_overwrite_dynamic_bucket_part
+ PARTITION (region = 'east') SELECT 1, 'unused' LIMIT 0"""
+ order_qt_txn_dynamic_bucket_empty_static_overwrite """SELECT id, name,
region
+ FROM t_overwrite_dynamic_bucket_part ORDER BY id"""
+ order_qt_txn_dynamic_bucket_empty_static_overwrite_snapshots """
+ SELECT snapshot_id, commit_kind FROM
t_overwrite_dynamic_bucket_part\$snapshots
+ """
+ assertTableEquals("t_overwrite_dynamic_bucket_part", "ORDER BY id")
+
// FT-015: Static partition overwrite replaces only the requested
partition.
sql """INSERT INTO t_overwrite_part VALUES
(1, 'east_old', 'east'), (2, 'west_old', 'west')"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
