This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a61493093f0 branch-4.1: [fix](paimon) Keep write cache coherent and
extend coverage (#67023)
a61493093f0 is described below
commit a61493093f09cecc5099c2b29756a825f45c0258
Author: Socrates <[email protected]>
AuthorDate: Mon Aug 24 09:38:35 2026 +0800
branch-4.1: [fix](paimon) Keep write cache coherent and extend coverage
(#67023)
### What
- move Paimon write suites and golden outputs under
`external_table_p0/paimon/write`
- extend write coverage for sequence groups, MERGE semantics,
sequence/rowkind, deletion vectors, row tracking/data evolution,
external paths, partition deletes, bucket modes, and JDBC catalog
concurrency
- keep the Doris Paimon metacache as the owner of table instances used
by writes
- invalidate the same table in Paimon's `CachingCatalog` before
reloading it after a Doris cache miss
- add FE unit coverage for cache invalidation and table reload behavior
- validate stable Doris query results through checked-in `.out` files
- keep unsupported behavior disabled with targeted TODO markers
### Why
Doris currently has its own Paimon table metacache on top of Paimon's
`CachingCatalog`. Invalidating only the Doris cache could still make the
loader receive a stale `Table` from the underlying Paimon cache. As a
result, writes after external metadata changes could continue using
outdated bucket, path, or table options.
This change keeps the existing Doris cache boundary, documents why both
cache layers exist, and makes a Doris cache reload invalidate the
corresponding Paimon table entry first.
### Testing
- `source ./custom_env.sh && ./build.sh --fe`
- `source ./custom_env.sh && ./run-fe-ut.sh --run
org.apache.doris.datasource.paimon.PaimonExternalMetaCacheTest,org.apache.doris.datasource.paimon.PaimonExternalTableTest`
- 31 tests, 0 failures, 0 errors
- normal-mode regression execution for the added Paimon write suites and
JDBC catalog concurrency coverage
- focused `.out` verification after the cache fix:
- `test_paimon_write_bucket_modes`
- `test_paimon_write_sequence_group`
- `test_paimon_write_deletion_vector`
- `test_paimon_write_external_paths`
- `git diff --check`
---
.../metacache/paimon/PaimonTableLoader.java | 7 +-
.../datasource/paimon/PaimonExternalCatalog.java | 73 +++++--
.../datasource/paimon/PaimonExternalTable.java | 7 +-
.../paimon/PaimonExternalMetaCacheTest.java | 57 ++++++
.../datasource/paimon/PaimonExternalTableTest.java | 17 ++
.../paimon/test_paimon_jdbc_catalog.out | 16 ++
.../test_paimon_create_ddl_write_properties.out | 0
.../write}/test_paimon_write_append_only.out | 0
.../write}/test_paimon_write_bucket_modes.out | 7 +
.../test_paimon_write_changelog_producer.out | 0
.../paimon/write}/test_paimon_write_compaction.out | 0
.../write}/test_paimon_write_complex_types.out | 0
.../write/test_paimon_write_deletion_vector.out | 38 ++++
.../paimon/write}/test_paimon_write_edge_cases.out | 0
.../write/test_paimon_write_external_paths.out | 77 +++++++
.../paimon/write}/test_paimon_write_failures.out | 0
.../write}/test_paimon_write_merge_engine.out | 0
.../write/test_paimon_write_merge_semantics.out | 8 +
.../write/test_paimon_write_partition_delete.out | 20 ++
.../paimon/write}/test_paimon_write_pk.out | 0
.../write}/test_paimon_write_row_level_dml.out | 0
.../test_paimon_write_row_tracking_evolution.out | 17 ++
.../write}/test_paimon_write_schema_change.out | 0
.../write/test_paimon_write_sequence_group.out | 30 +++
.../write/test_paimon_write_sequence_rowkind.out | 17 ++
.../write}/test_paimon_write_transaction.out | 0
.../paimon/write}/test_paimon_write_types.out | 0
.../paimon/write}/test_paimon_write_variant.out | 0
.../write}/test_paimon_write_variant_dml.out | 0
.../write}/test_paimon_write_variant_errors.out | 0
.../write}/test_paimon_write_variant_nested.out | 0
.../write}/test_paimon_write_variant_shredding.out | 0
.../test_paimon_write_variant_table_modes.out | 0
.../paimon/test_paimon_jdbc_catalog.groovy | 135 ++++++++++--
.../test_paimon_create_ddl_write_properties.groovy | 0
.../write}/test_paimon_write_append_only.groovy | 0
.../write}/test_paimon_write_bucket_modes.groovy | 70 +++++++
.../test_paimon_write_changelog_producer.groovy | 0
.../write}/test_paimon_write_compaction.groovy | 0
.../write}/test_paimon_write_complex_types.groovy | 0
.../write/test_paimon_write_deletion_vector.groovy | 194 ++++++++++++++++++
.../write}/test_paimon_write_edge_cases.groovy | 0
.../write/test_paimon_write_external_paths.groovy | 200 ++++++++++++++++++
.../write}/test_paimon_write_failures.groovy | 0
...paimon_write_key_dynamic_memory_negative.groovy | 0
.../write}/test_paimon_write_merge_engine.groovy | 0
.../write/test_paimon_write_merge_semantics.groovy | 221 ++++++++++++++++++++
.../test_paimon_write_partition_delete.groovy | 150 ++++++++++++++
.../paimon/write}/test_paimon_write_pk.groovy | 0
.../write}/test_paimon_write_row_level_dml.groovy | 0
...test_paimon_write_row_tracking_evolution.groovy | 216 ++++++++++++++++++++
.../write}/test_paimon_write_schema_change.groovy | 0
.../write/test_paimon_write_sequence_group.groovy | 227 +++++++++++++++++++++
.../test_paimon_write_sequence_rowkind.groovy | 200 ++++++++++++++++++
.../write}/test_paimon_write_snapshot_refs.groovy | 0
.../write}/test_paimon_write_source_models.groovy | 0
.../test_paimon_write_thread_lifecycle.groovy | 0
.../write}/test_paimon_write_transaction.groovy | 0
.../paimon/write}/test_paimon_write_types.groovy | 0
.../paimon/write}/test_paimon_write_variant.groovy | 0
.../write}/test_paimon_write_variant_dml.groovy | 0
.../write}/test_paimon_write_variant_errors.groovy | 0
.../write}/test_paimon_write_variant_nested.groovy | 0
.../test_paimon_write_variant_shredding.groovy | 0
.../test_paimon_write_variant_table_modes.groovy | 0
65 files changed, 1962 insertions(+), 42 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonTableLoader.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonTableLoader.java
index 0a134cfd7d7..977d20fbb1e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonTableLoader.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonTableLoader.java
@@ -33,7 +33,12 @@ public final class PaimonTableLoader {
public Table load(NameMapping nameMapping) {
try {
- return catalog(nameMapping).getPaimonTable(nameMapping);
+ // Doris caches Paimon table handles to implement the
external-catalog TTL and REFRESH
+ // lifecycle. Paimon's CachingCatalog has another table cache
because it also owns
+ // lower-level manifest, snapshot and deletion-vector caches. This
method is reached on
+ // a Doris cache miss, so invalidate the Paimon table entry as
well; otherwise Doris can
+ // reload the same stale handle immediately after REFRESH.
+ return catalog(nameMapping).reloadPaimonTable(nameMapping);
} catch (Exception e) {
throw new CacheException("failed to load paimon table %s.%s.%s:
%s",
e, nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName(),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java
index 14dfd290519..40d2621cedc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java
@@ -124,34 +124,65 @@ public class PaimonExternalCatalog extends
ExternalCatalog {
public Table getPaimonTable(NameMapping nameMapping, String branch, String
queryType) {
makeSureInitialized();
+ Identifier identifier = tableIdentifier(nameMapping, branch,
queryType);
+ return loadPaimonTable(nameMapping, queryType, identifier);
+ }
+
+ /**
+ * Invalidates Paimon's catalog-level table cache and reloads the table.
+ *
+ * <p>Doris and Paimon cache table handles for different purposes. Doris
owns the external
+ * metadata lifecycle (TTL, REFRESH and FE-wide invalidation), while
Paimon's CachingCatalog
+ * attaches lower-level manifest, snapshot and deletion-vector caches to a
table handle. A miss
+ * in the Doris cache must therefore invalidate Paimon's table entry
before loading; otherwise a
+ * Doris REFRESH can repopulate its cache with the same stale Paimon table
handle.
+ */
+ public Table reloadPaimonTable(NameMapping nameMapping) {
+ makeSureInitialized();
+ Identifier identifier = tableIdentifier(nameMapping, null, null);
try {
- Identifier identifier;
- if (branch != null && queryType != null) {
- identifier = new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(),
- branch, queryType);
- } else if (branch != null) {
- identifier = new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(),
- branch);
- } else if (queryType != null) {
- identifier = new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(),
- "main", queryType);
- } else {
- identifier = new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName());
- }
return executionAuthenticator.execute(() -> {
- Table table = catalog.getTable(identifier);
- Map<String, String> tableOptions =
paimonProperties.getTableOptionsForCopy();
- // This handle is relation-neutral. Runtime validation and
CPU-local capping belong
- // to the final relation copy, where relation options can
override physical values.
- return tableOptions.isEmpty() ? table :
table.copy(tableOptions);
+ catalog.invalidateTable(identifier);
+ return
copyWithCatalogTableOptions(catalog.getTable(identifier));
});
} catch (Exception e) {
- throw new RuntimeException("Failed to get Paimon table:" +
getName() + "."
- + nameMapping.getRemoteDbName() + "." +
nameMapping.getRemoteTblName() + "$" + queryType
- + ", because " + ExceptionUtils.getRootCauseMessage(e), e);
+ throw tableLoadException(nameMapping, null, e);
+ }
+ }
+
+ private Identifier tableIdentifier(NameMapping nameMapping, String branch,
String queryType) {
+ if (branch != null && queryType != null) {
+ return new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(),
+ branch, queryType);
+ } else if (branch != null) {
+ return new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(), branch);
+ } else if (queryType != null) {
+ return new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName(), "main", queryType);
+ }
+ return new Identifier(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName());
+ }
+
+ private Table loadPaimonTable(NameMapping nameMapping, String queryType,
Identifier identifier) {
+ try {
+ return executionAuthenticator.execute(() ->
copyWithCatalogTableOptions(catalog.getTable(identifier)));
+ } catch (Exception e) {
+ throw tableLoadException(nameMapping, queryType, e);
}
}
+ private Table copyWithCatalogTableOptions(Table table) {
+ Map<String, String> tableOptions =
paimonProperties.getTableOptionsForCopy();
+ // This handle is relation-neutral. Runtime validation and CPU-local
capping belong
+ // to the final relation copy, where relation options can override
physical values.
+ return tableOptions.isEmpty() ? table : table.copy(tableOptions);
+ }
+
+ private RuntimeException tableLoadException(NameMapping nameMapping,
String queryType, Exception e) {
+ return new RuntimeException("Failed to get Paimon table:" + getName()
+ "."
+ + nameMapping.getRemoteDbName() + "." +
nameMapping.getRemoteTblName() + "$" + queryType
+ + ", because " + ExceptionUtils.getRootCauseMessage(e), e);
+ }
+
protected Catalog createCatalog() {
try {
return paimonProperties.initializeCatalog(getName(), new
ArrayList<>(catalogProperty
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
index d0a3c858f4b..b38d0d5a8bb 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
@@ -140,11 +140,12 @@ public class PaimonExternalTable extends ExternalTable
implements MTMVRelatedTab
*
* <p>A statement MVCC snapshot belongs to a read relation. In a
time-travel self-insert the
* same Doris table identity can therefore have a historical source
snapshot registered in
- * StatementContext. Write planning must never reuse that snapshot: the
writer, target schema
- * and partition metadata must all come from the latest remote table
handle.
+ * StatementContext. Write planning must never reuse that snapshot. The
base table cache is the
+ * current write target; REFRESH and cache expiry reload it through {@code
PaimonTableLoader},
+ * which also invalidates Paimon's catalog-level table cache.
*/
public Table getPaimonTableForWrite() {
- return ((PaimonExternalCatalog)
catalog).getPaimonTable(getOrBuildNameMapping());
+ return getBasePaimonTable();
}
private PaimonSnapshotCacheValue
getPaimonSnapshotCacheValue(Optional<TableSnapshot> tableSnapshot,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java
index 228bb112ff0..12052898c25 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java
@@ -55,6 +55,7 @@ import org.junit.Assume;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.mockito.InOrder;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -274,6 +275,62 @@ public class PaimonExternalMetaCacheTest {
}
}
+ @Test
+ public void testReloadInvalidatesPaimonTableCacheBeforeLoad() throws
Exception {
+ java.io.File warehouse = temporaryFolder.newFolder("reload_table");
+ Map<String, String> properties = new HashMap<>();
+ properties.put("type", "paimon");
+ properties.put(PaimonExternalCatalog.PAIMON_CATALOG_TYPE,
+ PaimonExternalCatalog.PAIMON_FILESYSTEM);
+ properties.put("warehouse", warehouse.toURI().toString());
+ PaimonExternalCatalog dorisCatalog = new PaimonExternalCatalog(
+ 92L, "paimon_reload_test", null, properties, "");
+ dorisCatalog.makeSureInitialized();
+ dorisCatalog.catalog.close();
+
+ Catalog paimonCatalog = Mockito.mock(Catalog.class);
+ Table expected = Mockito.mock(Table.class);
+ Identifier identifier = Identifier.create("remote_db", "remote_table");
+ Mockito.when(paimonCatalog.getTable(identifier)).thenReturn(expected);
+ dorisCatalog.catalog = paimonCatalog;
+
+ Table actual = dorisCatalog.reloadPaimonTable(
+ new NameMapping(92L, "local_db", "local_table", "remote_db",
"remote_table"));
+
+ Assert.assertSame(expected, actual);
+ InOrder inOrder = Mockito.inOrder(paimonCatalog);
+ inOrder.verify(paimonCatalog).invalidateTable(identifier);
+ inOrder.verify(paimonCatalog).getTable(identifier);
+ }
+
+ @Test
+ public void testDorisTableCacheMissReloadsPaimonCatalogOnce() throws
Exception {
+ long catalogId = 93L;
+ NameMapping nameMapping = new NameMapping(
+ catalogId, "local_db", "local_table", "remote_db",
"remote_table");
+ Table expected = Mockito.mock(Table.class);
+ PaimonExternalCatalog dorisCatalog =
Mockito.mock(PaimonExternalCatalog.class);
+
Mockito.when(dorisCatalog.reloadPaimonTable(nameMapping)).thenReturn(expected);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.doReturn(dorisCatalog).when(catalogMgr)
+ .getCatalogOrException(Mockito.eq(catalogId), Mockito.any());
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ PaimonExternalMetaCache cache = new
PaimonExternalMetaCache(executor);
+ cache.initCatalog(catalogId, Collections.emptyMap());
+
+ Assert.assertSame(expected, cache.getPaimonTable(nameMapping));
+ Assert.assertSame(expected, cache.getPaimonTable(nameMapping));
+
+ Mockito.verify(dorisCatalog,
Mockito.times(1)).reloadPaimonTable(nameMapping);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Test
public void
testPartitionProjectionRejectsUnsafeEffectiveTableBeforeEnumeration() throws
Exception {
FileStoreTable unsafeTable = newPartitionedTable(
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
index a0f01f25cdf..4399e59b7d1 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
@@ -67,6 +67,23 @@ import java.util.concurrent.atomic.AtomicInteger;
public class PaimonExternalTableTest {
+ @Test
+ public void testWriteUsesBaseTableInsteadOfStatementSnapshot() {
+ PaimonExternalCatalog catalog =
Mockito.mock(PaimonExternalCatalog.class);
+ PaimonExternalDatabase database =
Mockito.mock(PaimonExternalDatabase.class);
+ Table paimonTable = Mockito.mock(Table.class);
+ Mockito.when(catalog.getId()).thenReturn(1L);
+ Mockito.when(database.getCatalog()).thenReturn(catalog);
+ Mockito.when(database.getFullName()).thenReturn("local_db");
+ Mockito.when(database.getRemoteName()).thenReturn("remote_db");
+ PaimonExternalTable externalTable = Mockito.spy(new
PaimonExternalTable(
+ 10L, "local_table", "remote_table", catalog, database));
+ Mockito.doReturn(paimonTable).when(externalTable).getBasePaimonTable();
+
+ Assert.assertSame(paimonTable, externalTable.getPaimonTableForWrite());
+ Mockito.verify(externalTable).getBasePaimonTable();
+ }
+
@Test
public void
testStatementContextDefersPhysicalManifestValidationUntilRelationOptions() {
PaimonExternalCatalog catalog =
Mockito.mock(PaimonExternalCatalog.class);
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_jdbc_catalog.out
b/regression-test/data/external_table_p0/paimon/test_paimon_jdbc_catalog.out
index 0866e5961ba..0dc7742b94d 100644
--- a/regression-test/data/external_table_p0/paimon/test_paimon_jdbc_catalog.out
+++ b/regression-test/data/external_table_p0/paimon/test_paimon_jdbc_catalog.out
@@ -3,3 +3,19 @@
1 alice 2025-01-01
2 bob 2025-01-02
+-- !paimon_jdbc_concurrent_append --
+left 128 128 8128
+right 128 128 136128
+
+-- !paimon_jdbc_same_partition_append --
+128 128 324032
+
+-- !paimon_jdbc_concurrent_aggregation --
+1 30
+
+-- !paimon_jdbc_concurrent_dynamic --
+left 64 64
+right 64 64
+
+-- !paimon_jdbc_merge_during_compact --
+2 merge-during-compact 200
diff --git
a/regression-test/data/paimon_write/test_paimon_create_ddl_write_properties.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_create_ddl_write_properties.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_create_ddl_write_properties.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_create_ddl_write_properties.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_append_only.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_append_only.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_append_only.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_append_only.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_bucket_modes.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_bucket_modes.out
similarity index 88%
rename from regression-test/data/paimon_write/test_paimon_write_bucket_modes.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_bucket_modes.out
index 425fdf51f4c..55d921313e1 100644
--- a/regression-test/data/paimon_write/test_paimon_write_bucket_modes.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_bucket_modes.out
@@ -2,6 +2,13 @@
-- !bucket_hash_fixed --
16 0 15 2
+-- !bucket_rescale_partial --
+p1 1 p1-old-1
+p1 10 p1-after-rescale
+p1 2 p1-old-2
+p2 3 p2-old-3
+p2 4 p2-old-4
+
-- !bucket_hash_dynamic --
p1 1 v1_updated
p1 2 v2
diff --git
a/regression-test/data/paimon_write/test_paimon_write_changelog_producer.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_changelog_producer.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_changelog_producer.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_changelog_producer.out
diff --git a/regression-test/data/paimon_write/test_paimon_write_compaction.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_compaction.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_compaction.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_compaction.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_complex_types.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_complex_types.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_complex_types.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_complex_types.out
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_deletion_vector.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_deletion_vector.out
new file mode 100644
index 00000000000..a13f2ca16fc
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_deletion_vector.out
@@ -0,0 +1,38 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !dv_before_compact_jni --
+1 merged-1 12
+4 insert-4 40
+5 inserted-5 50
+
+-- !dv_before_compact_native --
+1 merged-1 12
+4 insert-4 40
+5 inserted-5 50
+
+-- !dv_after_compact_jni --
+1 merged-1 12
+4 insert-4 40
+5 inserted-5 50
+
+-- !dv_after_compact_native --
+1 merged-1 12
+4 insert-4 40
+5 inserted-5 50
+
+-- !dv_post_compact_write_jni --
+1 post-compact-1 13
+4 insert-4 40
+5 inserted-5 50
+6 post-compact-6 60
+
+-- !dv_post_compact_write_native --
+1 post-compact-1 13
+4 insert-4 40
+5 inserted-5 50
+6 post-compact-6 60
+
+-- !dv_enabled_after_mor_jni --
+1 mow-new-1
+
+-- !dv_enabled_after_mor_native --
+1 mow-new-1
diff --git a/regression-test/data/paimon_write/test_paimon_write_edge_cases.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_edge_cases.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_edge_cases.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_edge_cases.out
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_external_paths.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_external_paths.out
new file mode 100644
index 00000000000..e5763fe6c94
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_external_paths.out
@@ -0,0 +1,77 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !external_round_robin_initial --
+p-bulk 100 2048
+p-bulk 101 2048
+p-bulk 102 2048
+p-bulk 103 2048
+p-bulk 104 2048
+p-bulk 105 2048
+p-bulk 106 2048
+p-bulk 107 2048
+p-bulk 108 2048
+p-bulk 109 2048
+p-bulk 110 2048
+p-bulk 111 2048
+p-bulk 112 2048
+p-bulk 113 2048
+p-bulk 114 2048
+p-bulk 115 2048
+p1 1 3
+p1 2 3
+p2 3 5
+p2 4 4
+
+-- !external_round_robin_changed --
+p-bulk 100 2048
+p-bulk 101 2048
+p-bulk 102 2048
+p-bulk 103 2048
+p-bulk 104 2048
+p-bulk 105 2048
+p-bulk 106 2048
+p-bulk 107 2048
+p-bulk 108 2048
+p-bulk 109 2048
+p-bulk 110 2048
+p-bulk 111 2048
+p-bulk 112 2048
+p-bulk 113 2048
+p-bulk 114 2048
+p-bulk 115 2048
+p-new-bulk 200 2048
+p-new-bulk 201 2048
+p-new-bulk 202 2048
+p-new-bulk 203 2048
+p-new-bulk 204 2048
+p-new-bulk 205 2048
+p-new-bulk 206 2048
+p-new-bulk 207 2048
+p-new-bulk 208 2048
+p-new-bulk 209 2048
+p-new-bulk 210 2048
+p-new-bulk 211 2048
+p-new-bulk 212 2048
+p-new-bulk 213 2048
+p-new-bulk 214 2048
+p-new-bulk 215 2048
+p1 1 3
+p1 2 3
+p2 3 5
+p2 4 4
+p3 5 4
+p3 6 3
+
+-- !external_weight_robin --
+1 weight-1
+2 weight-2
+3 weight-3
+4 weight-4
+5 weight-5
+6 weight-6
+
+-- !external_specific_fs --
+1 specific-1
+2 specific-2
+
+-- !external_default_path --
+1 default-path
diff --git a/regression-test/data/paimon_write/test_paimon_write_failures.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_failures.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_failures.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_failures.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_merge_engine.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_merge_engine.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_merge_engine.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_merge_engine.out
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_merge_semantics.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_merge_semantics.out
new file mode 100644
index 00000000000..5db9108a548
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_merge_semantics.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !merge_semantics_result --
+1 15 11 base-1-source-1 updated required-1-new
+3 30 3 base-3 stable required-3
+4 40 44 source-4 inserted required-4
+
+-- !merge_semantics_recovered --
+5 recovered
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_partition_delete.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_partition_delete.out
new file mode 100644
index 00000000000..92f03fca8c8
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_partition_delete.out
@@ -0,0 +1,20 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !partition_delete_full_partition --
+\N 6 60 default-partition
+p2 3 30 p2-a
+p2 4 40 p2-b
+p3 5 50 p3-a
+
+-- !partition_delete_partial_partition --
+\N 6 60 default-partition
+p2 3 30 p2-a
+p3 5 50 p3-a
+
+-- !partition_delete_expression --
+\N 6 60 default-partition
+p3 5 50 p3-a
+
+-- !partition_delete_not_exists --
+\N 6 60 default-partition
+
+-- !partition_delete_default_partition --
diff --git a/regression-test/data/paimon_write/test_paimon_write_pk.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_pk.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_pk.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_pk.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_row_level_dml.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_row_level_dml.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_row_level_dml.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_row_level_dml.out
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.out
new file mode 100644
index 00000000000..1f71d511f7d
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.out
@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !row_tracking_after_spark_changes --
+1 one-merged
+2 two-updated
+4 four
+
+-- !row_tracking_after_compact_write --
+1 one-merged
+2 two-updated
+4 four
+5 five-after-compact
+
+-- !data_evolution_after_spark_merge --
+1 11 100
+2 22 200
+3 30 \N
+4 44 444
diff --git
a/regression-test/data/paimon_write/test_paimon_write_schema_change.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_schema_change.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_schema_change.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_schema_change.out
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_group.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_group.out
new file mode 100644
index 00000000000..1c72d08a0cc
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_group.out
@@ -0,0 +1,30 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !sequence_group_stale_profile --
+1 alice shanghai 10 12 90 11 base
+
+-- !sequence_group_new_profile --
+1 alice-new shenzhen 12 112 99 11 new-note
+
+-- !sequence_group_null_sequence --
+1 alice-new shenzhen 12 115 99 13 new-note
+
+-- !remove_on_delete_empty --
+0
+
+-- !remove_on_delete_partial --
+1 old-a new-b
+
+-- !remove_on_delete_complete --
+1 new-a new-b
+
+-- !group_remove_on_delete_unchanged --
+1 old-a 100 old-b 100
+
+-- !group_remove_on_delete_sequence --
+1 high-a 101 old-b 100
+
+-- !property_change_still_writable --
+1 still-writable 10
+
+-- !property_change_sequence_group --
+1 high 20
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.out
new file mode 100644
index 00000000000..5b1cf8b4496
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.out
@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !sequence_ascending --
+1 10 21 new-second-field
+2 1 0 nonnull-wins
+
+-- !sequence_equal --
+3 second-equal
+
+-- !sequence_descending --
+1 5 smaller-wins
+
+-- !rowkind_changelog --
+1 new-1
+3 new-3
+
+-- !rowkind_recovered --
+4 recovered
diff --git
a/regression-test/data/paimon_write/test_paimon_write_transaction.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_transaction.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_transaction.out
diff --git a/regression-test/data/paimon_write/test_paimon_write_types.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_types.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
diff --git a/regression-test/data/paimon_write/test_paimon_write_variant.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_variant.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_variant_dml.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_dml.out
similarity index 100%
rename from regression-test/data/paimon_write/test_paimon_write_variant_dml.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_dml.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_variant_errors.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_variant_errors.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_variant_nested.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_nested.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_variant_nested.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_nested.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_variant_shredding.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_shredding.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_variant_shredding.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_shredding.out
diff --git
a/regression-test/data/paimon_write/test_paimon_write_variant_table_modes.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_table_modes.out
similarity index 100%
rename from
regression-test/data/paimon_write/test_paimon_write_variant_table_modes.out
rename to
regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_table_modes.out
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy
index 28a71862c0e..b57bd35617b 100644
---
a/regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy
@@ -68,6 +68,11 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
String sparkSeedCatalogName = "${catalogName}_seed"
// Reuse the fixture-wide Docker command so local and CI permission models
behave identically.
String dockerCommand =
context.config.otherConfigs.get("externalDockerCommand") ?: "docker"
+ String sparkMaster =
context.config.otherConfigs.get("paimon_jdbc_spark_master")
+ String sparkExternalEnvIp =
context.config.otherConfigs.get("paimon_jdbc_spark_external_env_ip")
+ if (sparkExternalEnvIp == null || sparkExternalEnvIp.isEmpty()) {
+ sparkExternalEnvIp = externalEnvIp
+ }
assertTrue(jdbcDriversDir != null && !jdbcDriversDir.isEmpty(),
"jdbc_drivers_dir must be configured")
@@ -168,7 +173,10 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
}
executeCommand("${dockerCommand} cp ${localDriverPath}
${sparkContainerName}:${sparkDriverPath}", true, 60)
- String sparkMinioEndpoint = "http://${externalEnvIp}:${minioPort}";
+ String sparkMinioEndpoint =
context.config.otherConfigs.get("paimon_jdbc_spark_minio_endpoint")
+ if (sparkMinioEndpoint == null || sparkMinioEndpoint.isEmpty()) {
+ sparkMinioEndpoint = "http://${sparkExternalEnvIp}:${minioPort}";
+ }
if (sparkContainerName.contains("spark-iceberg")) {
String sparkMinioContainerName =
sparkContainerName.replaceFirst("spark-iceberg", "minio")
String resolvedSparkMinioContainer = executeCommand(
@@ -182,10 +190,14 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
}
}
logger.info("spark seed minio endpoint: ${sparkMinioEndpoint}")
+ if (sparkMaster == null || sparkMaster.isEmpty()) {
+ sparkMaster = "spark://${sparkContainerName}:7077"
+ }
+ logger.info("spark seed master: ${sparkMaster}")
def sparkPaimonJdbc = { String sqlText ->
String escapedSql = sqlText.replaceAll('"', '\\\\"')
- String command = """${dockerCommand} exec ${sparkContainerName}
spark-sql --master spark://${sparkContainerName}:7077 \
+ String command = """${dockerCommand} exec ${sparkContainerName}
spark-sql --master ${sparkMaster} \
--jars ${sparkDriverPath} \
--driver-class-path ${sparkDriverPath} \
--conf spark.driver.extraClassPath=${sparkDriverPath} \
@@ -194,7 +206,7 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
--conf
spark.sql.catalog.${sparkSeedCatalogName}=org.apache.paimon.spark.SparkCatalog \
--conf
spark.sql.catalog.${sparkSeedCatalogName}.warehouse=s3://${warehouseBucket}/paimon_jdbc_catalog/
\
--conf spark.sql.catalog.${sparkSeedCatalogName}.metastore=jdbc \
---conf
spark.sql.catalog.${sparkSeedCatalogName}.uri=jdbc:postgresql://${externalEnvIp}:${jdbcPort}/postgres
\
+--conf
spark.sql.catalog.${sparkSeedCatalogName}.uri=jdbc:postgresql://${sparkExternalEnvIp}:${jdbcPort}/postgres
\
--conf spark.sql.catalog.${sparkSeedCatalogName}.catalog-key=${catalogName} \
--conf spark.sql.catalog.${sparkSeedCatalogName}.jdbc.user=postgres \
--conf spark.sql.catalog.${sparkSeedCatalogName}.jdbc.password=123456 \
@@ -408,15 +420,12 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
"""
})
sql """REFRESH TABLE paimon_jdbc_concurrent_append"""
- assertEquals([
- ["left", 128L, 128L, 8128L],
- ["right", 128L, 128L, 136128L]
- ], sql("""
+ order_qt_paimon_jdbc_concurrent_append """
SELECT pt, COUNT(*), COUNT(DISTINCT id), SUM(id)
FROM paimon_jdbc_concurrent_append
GROUP BY pt
ORDER BY pt
- """))
+ """
assertEquals(appendSnapshots + 2L, (sql """
SELECT COUNT(*) FROM paimon_jdbc_concurrent_append\$snapshots
""")[0][0] as long)
@@ -435,11 +444,11 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
"""
})
sql """REFRESH TABLE paimon_jdbc_concurrent_append"""
- assertEquals([[128L, 128L, 324032L]], sql("""
+ qt_paimon_jdbc_same_partition_append """
SELECT COUNT(*), COUNT(DISTINCT id), SUM(id)
FROM paimon_jdbc_concurrent_append
WHERE pt = 'same'
- """))
+ """
assertEquals(appendSnapshots + 4L, (sql """
SELECT COUNT(*) FROM paimon_jdbc_concurrent_append\$snapshots
""")[0][0] as long)
@@ -509,9 +518,10 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
"""
})
sql """REFRESH TABLE paimon_jdbc_concurrent_aggregation"""
- assertEquals([[1, 30L]], sql("""
+ order_qt_paimon_jdbc_concurrent_aggregation """
SELECT id, total FROM paimon_jdbc_concurrent_aggregation
- """))
+ ORDER BY id
+ """
// Dynamic bucket only permits multiple jobs when they own disjoint
partitions.
sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_dynamic"""
@@ -543,18 +553,109 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
"""
})
sql """REFRESH TABLE paimon_jdbc_concurrent_dynamic"""
- assertEquals([
- ["left", 64L, 64L],
- ["right", 64L, 64L]
- ], sql("""
+ order_qt_paimon_jdbc_concurrent_dynamic """
SELECT pt, COUNT(*), COUNT(DISTINCT id)
FROM paimon_jdbc_concurrent_dynamic
GROUP BY pt
ORDER BY pt
- """))
+ """
+
+ // P12: Row-level writers use the same catalog lock and snapshot commit
+ // protocol as INSERT. Concurrent MERGEs on one key may expose either
+ // last value, but both transactions must commit without duplicating
it.
+ sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_merge"""
+ sql """
+ CREATE TABLE ${dbName}.paimon_jdbc_concurrent_merge (
+ id INT,
+ payload STRING,
+ score INT
+ ) ENGINE=paimon
+ PROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'deduplicate',
+ 'num-sorted-run.compaction-trigger' = '100'
+ )
+ """
+ sql """INSERT INTO paimon_jdbc_concurrent_merge VALUES
+ (1, 'base-1', 0), (2, 'base-2', 0)
+ """
+ long mergeSnapshots = (sql """
+ SELECT COUNT(*) FROM paimon_jdbc_concurrent_merge\$snapshots
+ """)[0][0] as long
+
+ runConcurrent("paimon-jdbc-merge-left", {
+ sql """
+ MERGE INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_merge t
+ USING (SELECT 1 AS id, 'left' AS payload, 10 AS score) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET
+ payload = s.payload, score = s.score
+ """
+ }, "paimon-jdbc-merge-right", {
+ sql """
+ MERGE INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_merge t
+ USING (SELECT 1 AS id, 'right' AS payload, 20 AS score) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET
+ payload = s.payload, score = s.score
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_merge"""
+ def sameKeyMergeRows = sql """
+ SELECT id, payload, score
+ FROM paimon_jdbc_concurrent_merge
+ WHERE id = 1
+ """
+ assertEquals(1, sameKeyMergeRows.size())
+ assertTrue([
+ [1, "left", 10],
+ [1, "right", 20]
+ ].contains(sameKeyMergeRows[0]))
+ assertEquals(mergeSnapshots + 2L, (sql """
+ SELECT COUNT(*) FROM paimon_jdbc_concurrent_merge\$snapshots
+ """)[0][0] as long)
+
+ // Interleave a Doris MERGE with a Spark full compaction. Whichever
+ // operation obtains the catalog lock first, the MERGE result must
remain
+ // visible and compaction must not resurrect the pre-update value.
+ runConcurrent("paimon-jdbc-merge-compact", {
+ sql """
+ MERGE INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_merge t
+ USING (SELECT 2 AS id, 'merge-during-compact' AS payload,
+ 200 AS score) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET
+ payload = s.payload, score = s.score
+ """
+ }, "paimon-jdbc-spark-compact", {
+ sparkPaimonJdbc """
+ CALL ${sparkSeedCatalogName}.sys.compact(
+ table => '${dbName}.paimon_jdbc_concurrent_merge',
+ compact_strategy => 'full')
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_merge"""
+ order_qt_paimon_jdbc_merge_during_compact """
+ SELECT id, payload, score
+ FROM paimon_jdbc_concurrent_merge
+ WHERE id = 2
+ ORDER BY id
+ """
+
+ sparkPaimonJdbc """
+ SELECT assert_true(
+ COUNT(*) = 2 AND
+ SUM(CASE WHEN id = 2
+ AND payload = 'merge-during-compact'
+ AND score = 200
+ THEN 1 ELSE 0 END) = 1)
+ FROM ${sparkSeedCatalogName}.${dbName}.paimon_jdbc_concurrent_merge
+ """
} finally {
try {
sql """SWITCH ${catalogName}"""
+ sql """DROP TABLE IF EXISTS
${dbName}.paimon_jdbc_concurrent_merge"""
sql """DROP TABLE IF EXISTS
${dbName}.paimon_jdbc_concurrent_dynamic"""
sql """DROP TABLE IF EXISTS
${dbName}.paimon_jdbc_concurrent_aggregation"""
sql """DROP TABLE IF EXISTS ${dbName}.paimon_jdbc_concurrent_pk"""
diff --git
a/regression-test/suites/paimon_write/test_paimon_create_ddl_write_properties.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_create_ddl_write_properties.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_create_ddl_write_properties.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_create_ddl_write_properties.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_append_only.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_append_only.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_append_only.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_append_only.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_bucket_modes.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_bucket_modes.groovy
similarity index 87%
rename from
regression-test/suites/paimon_write/test_paimon_write_bucket_modes.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_bucket_modes.groovy
index 08e9fd85b04..f8ad1af6586 100644
--- a/regression-test/suites/paimon_write/test_paimon_write_bucket_modes.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_bucket_modes.groovy
@@ -42,6 +42,17 @@ suite("test_paimon_write_bucket_modes",
"p0,external,paimon") {
'bucket-key' = 'id'
);
+ DROP TABLE IF EXISTS paimon.${dbName}.t_rescale;
+ CREATE TABLE paimon.${dbName}.t_rescale (
+ pt STRING, id INT, name STRING
+ ) USING paimon
+ PARTITIONED BY (pt)
+ TBLPROPERTIES (
+ 'primary-key' = 'pt,id',
+ 'bucket' = '2',
+ 'bucket-key' = 'id'
+ );
+
DROP TABLE IF EXISTS paimon.${dbName}.t_hash_dynamic;
CREATE TABLE paimon.${dbName}.t_hash_dynamic (
pt STRING, id INT, name STRING
@@ -235,6 +246,65 @@ suite("test_paimon_write_bucket_modes",
"p0,external,paimon") {
def fixedBuckets = assertBucketsInRange("t_hash_fixed", 0, 3)
assertTrue(fixedBuckets.size() > 1)
+ // P04: ALTER only changes the configured bucket count. Existing
+ // partitions must be rewritten before a new writer can use bucket=4.
+ sql """INSERT INTO t_rescale VALUES
+ ('p1', 1, 'p1-old-1'),
+ ('p1', 2, 'p1-old-2'),
+ ('p2', 3, 'p2-old-3'),
+ ('p2', 4, 'p2-old-4')
+ """
+ spark_paimon """
+ ALTER TABLE paimon.${dbName}.t_rescale
+ SET TBLPROPERTIES ('bucket' = '4')
+ """
+ sql """refresh table t_rescale"""
+ long rescaleSnapshot = (sql """
+ SELECT max(snapshot_id) FROM t_rescale\$snapshots
+ """)[0][0] as long
+ long rescaleFiles = (sql """
+ SELECT count(*) FROM t_rescale\$files
+ """)[0][0] as long
+ boolean rejectedBeforeRescale = false
+ try {
+ sql """INSERT INTO t_rescale VALUES ('p1', 10,
'must-fail-before-rescale')"""
+ } catch (Exception ignored) {
+ rejectedBeforeRescale = true
+ }
+ assertTrue(rejectedBeforeRescale)
+ assertEquals(rescaleSnapshot, (sql """
+ SELECT max(snapshot_id) FROM t_rescale\$snapshots
+ """)[0][0] as long)
+ assertEquals(rescaleFiles, (sql """
+ SELECT count(*) FROM t_rescale\$files
+ """)[0][0] as long)
+
+ // Rescale only p1. The rewritten partition accepts new writes, while
p2
+ // remains readable with its old layout and still rejects bucket=4.
+ sql """
+ INSERT OVERWRITE TABLE t_rescale PARTITION (pt = 'p1')
+ SELECT id, name FROM t_rescale WHERE pt = 'p1'
+ """
+ sql """INSERT INTO t_rescale VALUES ('p1', 10, 'p1-after-rescale')"""
+ boolean unreformedPartitionRejected = false
+ try {
+ sql """INSERT INTO t_rescale VALUES ('p2', 20,
'p2-must-still-fail')"""
+ } catch (Exception ignored) {
+ unreformedPartitionRejected = true
+ }
+ assertTrue(unreformedPartitionRejected)
+ order_qt_bucket_rescale_partial """
+ SELECT * FROM t_rescale ORDER BY pt, id
+ """
+
+ sql """
+ INSERT OVERWRITE TABLE t_rescale PARTITION (pt = 'p2')
+ SELECT id, name FROM t_rescale WHERE pt = 'p2'
+ """
+ sql """INSERT INTO t_rescale VALUES ('p2', 20, 'p2-after-rescale')"""
+ assertTableEquals("t_rescale", "ORDER BY pt, id")
+ assertBucketsInRange("t_rescale", 0, 3)
+
// HASH_DYNAMIC: new keys expand buckets independently per partition.
sql """INSERT INTO t_hash_dynamic VALUES
('p1', 1, 'v1'),
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_changelog_producer.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_changelog_producer.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_changelog_producer.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_changelog_producer.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_compaction.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_compaction.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_compaction.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_compaction.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_complex_types.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_complex_types.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_complex_types.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_complex_types.groovy
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_deletion_vector.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_deletion_vector.groovy
new file mode 100644
index 00000000000..62462d1a3b9
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_deletion_vector.groovy
@@ -0,0 +1,194 @@
+// 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_paimon_write_deletion_vector", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_dv_catalog"
+ String dbName = "test_pw_dv_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_dv;
+ CREATE TABLE paimon.${dbName}.t_dv (
+ id INT, payload STRING, score INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'bucket-key' = 'id',
+ 'deletion-vectors.enabled' = 'true'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_enable_dv;
+ CREATE TABLE paimon.${dbName}.t_enable_dv (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'bucket-key' = 'id',
+ 'deletion-vectors.modifiable' = 'true'
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ sql """create database if not exists internal.${dbName}"""
+ sql """drop table if exists internal.${dbName}.dv_source"""
+ sql """
+ create table internal.${dbName}.dv_source (
+ id int, payload string, score int, action string
+ ) distributed by hash(id) buckets 1
+ properties ('replication_num' = '1')
+ """
+
+ try {
+ def deletionVectorEntries = { String tableName ->
+ def rows = spark_paimon """
+ SELECT coalesce(sum(row_count), 0)
+ FROM paimon.${dbName}.`${tableName}\$table_indexes`
+ WHERE index_type = 'DELETION_VECTORS'
+ """
+ return rows[0][0].toString().toLong()
+ }
+ def assertReaders = { String tag, String tableName, String columns,
+ String orderBy ->
+ [true, false].each { boolean forceJni ->
+ sql """set force_jni_scanner = ${forceJni}"""
+ String reader = forceJni ? "jni" : "native"
+ "order_qt_${tag}_${reader}" """
+ SELECT ${columns} FROM ${tableName} ${orderBy}
+ """
+ }
+ def sparkRows = spark_paimon """
+ SELECT ${columns} FROM paimon.${dbName}.${tableName} ${orderBy}
+ """
+ sql """set force_jni_scanner = false"""
+ def dorisRows = sql """SELECT ${columns} FROM ${tableName}
${orderBy}"""
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ // Start in MOW mode. Do not set write-only=true here: Paimon
implements
+ // primary-key deletion vectors during lookup compaction, while
write-only
+ // deliberately disables that compaction and leaves new level-0 files
+ // invisible to the DV-optimized reader until a dedicated compaction
runs.
+ sql """INSERT INTO t_dv VALUES
+ (1, 'old-1', 10),
+ (2, 'delete-2', 20),
+ (3, 'delete-by-merge-3', 30)
+ """
+ spark_paimon_multi """
+ CALL paimon.sys.compact(
+ table => '${dbName}.t_dv',
+ compact_strategy => 'full');
+ """
+ sql """REFRESH CATALOG ${catalogName}"""
+ sql """USE ${dbName}"""
+ sql """INSERT INTO t_dv VALUES
+ (1, 'upsert-1', 11),
+ (4, 'insert-4', 40)
+ """
+ sql """DELETE FROM t_dv WHERE id = 2"""
+ boolean dvProducedBeforeCompact = deletionVectorEntries("t_dv") > 0L
+
+ sql """INSERT INTO internal.${dbName}.dv_source VALUES
+ (1, 'merged-1', 12, 'U'),
+ (3, 'unused-3', 0, 'D'),
+ (5, 'inserted-5', 50, 'I')
+ """
+ sql """
+ MERGE INTO t_dv t
+ USING internal.${dbName}.dv_source s ON t.id = s.id
+ WHEN MATCHED AND s.action = 'D' THEN DELETE
+ WHEN MATCHED THEN UPDATE SET payload = s.payload, score = s.score
+ WHEN NOT MATCHED THEN INSERT (id, payload, score)
+ VALUES (s.id, s.payload, s.score)
+ """
+ assertReaders("dv_before_compact", "t_dv", "id, payload, score",
"ORDER BY id")
+ long dvEntriesBeforeCompact = deletionVectorEntries("t_dv")
+
+ // Full compaction must preserve the logical rows while materializing
at
+ // least part of the accumulated deletion-vector state.
+ spark_paimon_multi """
+ CALL paimon.sys.compact(
+ table => '${dbName}.t_dv',
+ compact_strategy => 'full'
+ );
+ """
+ sql """refresh table t_dv"""
+ assertReaders("dv_after_compact", "t_dv", "id, payload, score", "ORDER
BY id")
+ assertTrue(deletionVectorEntries("t_dv") <= dvEntriesBeforeCompact,
+ "Full compaction must not increase retained deletion-vector
entries")
+
+ // A writer opened after compaction must restore the current index and
+ // continue to hide the previous physical row for the same key.
+ sql """INSERT INTO t_dv VALUES
+ (1, 'post-compact-1', 13),
+ (6, 'post-compact-6', 60)
+ """
+ assertReaders("dv_post_compact_write", "t_dv", "id, payload, score",
"ORDER BY id")
+
+ // P08/P11 transition: enable MOW after MOR files already exist. The
+ // next Doris statement must reload the changed table options.
+ sql """INSERT INTO t_enable_dv VALUES
+ (1, 'mor-old-1'),
+ (2, 'mor-delete-2')
+ """
+ spark_paimon_multi """
+ CALL paimon.sys.compact(
+ table => '${dbName}.t_enable_dv',
+ compact_strategy => 'full');
+ ALTER TABLE paimon.${dbName}.t_enable_dv SET TBLPROPERTIES (
+ 'deletion-vectors.enabled' = 'true')
+ """
+ sql """refresh catalog ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """INSERT INTO t_enable_dv VALUES (1, 'mow-new-1')"""
+ sql """DELETE FROM t_enable_dv WHERE id = 2"""
+ // The first MOR-to-MOW lookup compaction may rewrite the old files
instead
+ // of retaining a non-empty DV index, so validate the stable contract
here:
+ // both readers must expose the converted update/delete result.
+ assertReaders("dv_enabled_after_mor", "t_enable_dv", "id, payload",
"ORDER BY id")
+ assertTrue(dvProducedBeforeCompact,
+ "Doris UPDATE/DELETE must leave a physical deletion vector
before compaction")
+ } finally {
+ sql """set force_jni_scanner = false"""
+ sql """drop catalog if exists ${catalogName}"""
+ sql """drop table if exists internal.${dbName}.dv_source"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_edge_cases.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_edge_cases.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_edge_cases.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_edge_cases.groovy
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
new file mode 100644
index 00000000000..1ce16faf5ce
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
@@ -0,0 +1,200 @@
+// 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_paimon_write_external_paths", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_external_paths_catalog"
+ String dbName = "test_pw_external_paths_db"
+ String pathRoot = "s3://warehouse/paimon-external-paths/${dbName}"
+ String filesTableSuffix = '$files'
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_round_robin;
+ CREATE TABLE paimon.${dbName}.t_round_robin (
+ pt STRING, id INT, payload STRING
+ ) USING paimon
+ PARTITIONED BY (pt)
+ TBLPROPERTIES (
+ 'primary-key' = 'pt,id',
+ 'bucket' = '1',
+ 'write-only' = 'true',
+ 'target-file-size' = '1 kb',
+ 'data-file.external-paths' =
'${pathRoot}/round-a,${pathRoot}/round-b',
+ 'data-file.external-paths.strategy' = 'round-robin'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_weight_robin;
+ CREATE TABLE paimon.${dbName}.t_weight_robin (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'write-only' = 'true',
+ 'target-file-size' = '1 kb',
+ 'data-file.external-paths' =
'${pathRoot}/weight-a,${pathRoot}/weight-b',
+ 'data-file.external-paths.strategy' = 'weight-robin',
+ 'data-file.external-paths.weights' = '1,1'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_specific_fs;
+ CREATE TABLE paimon.${dbName}.t_specific_fs (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'write-only' = 'true',
+ 'data-file.external-paths' =
'${pathRoot}/specific-a,${pathRoot}/specific-b',
+ 'data-file.external-paths.strategy' = 'specific-fs',
+ 'data-file.external-paths.specific-fs' = 's3'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_none;
+ CREATE TABLE paimon.${dbName}.t_none (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'write-only' = 'true',
+ 'data-file.external-paths' =
'${pathRoot}/unused-a,${pathRoot}/unused-b',
+ 'data-file.external-paths.strategy' = 'none'
+ );
+ """
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """SWITCH ${catalogName}"""
+ sql """USE ${dbName}"""
+
+ try {
+ def dataFiles = { String tableName ->
+ String query = """
+ SELECT file_path
+ FROM paimon.${dbName}.`${tableName}${filesTableSuffix}`
+ ORDER BY file_path
+ """
+ return spark_paimon(query).collect { row -> row[0].toString() }
+ }
+ def assertDorisSparkRows = { String tag, String tableName,
+ String columns, String orderBy ->
+ def sparkRows = spark_paimon """
+ SELECT ${columns} FROM paimon.${dbName}.${tableName} ${orderBy}
+ """
+ "order_qt_${tag}" """
+ SELECT ${columns} FROM ${tableName} ${orderBy}
+ """
+ def dorisRows = sql """SELECT ${columns} FROM ${tableName}
${orderBy}"""
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ // Separate statements force separate writer lifecycles. The selected
+ // path must therefore be refreshed for every Doris commit.
+ sql """INSERT INTO t_round_robin VALUES ('p1', 1, 'one')"""
+ sql """INSERT INTO t_round_robin VALUES ('p1', 2, 'two')"""
+ sql """INSERT INTO t_round_robin VALUES ('p2', 3, 'three')"""
+ sql """INSERT INTO t_round_robin VALUES ('p2', 4, 'four')"""
+ sql """
+ INSERT INTO t_round_robin
+ SELECT 'p-bulk', CAST(number + 100 AS INT), repeat('x', 2048)
+ FROM numbers("number" = "16")
+ """
+ def oldRoundFiles = dataFiles("t_round_robin")
+ assertTrue(oldRoundFiles.any { it.startsWith("${pathRoot}/round-a/") })
+ assertTrue(oldRoundFiles.any { it.startsWith("${pathRoot}/round-b/") })
+ assertDorisSparkRows("external_round_robin_initial", "t_round_robin",
+ "pt, id, length(payload)", "ORDER BY pt, id")
+
+ spark_paimon """
+ ALTER TABLE paimon.${dbName}.t_round_robin SET TBLPROPERTIES (
+ 'data-file.external-paths' =
'${pathRoot}/round-c,${pathRoot}/round-d'
+ )
+ """
+ sql """REFRESH CATALOG ${catalogName}"""
+ sql """USE ${dbName}"""
+ sql """INSERT INTO t_round_robin VALUES ('p3', 5, 'five')"""
+ sql """INSERT INTO t_round_robin VALUES ('p3', 6, 'six')"""
+ sql """
+ INSERT INTO t_round_robin
+ SELECT 'p-new-bulk', CAST(number + 200 AS INT), repeat('y', 2048)
+ FROM numbers("number" = "16")
+ """
+ def changedRoundFiles = dataFiles("t_round_robin")
+ boolean oldRoundFilesRetained =
changedRoundFiles.containsAll(oldRoundFiles)
+ def newRoundFiles = changedRoundFiles - oldRoundFiles
+ assertTrue(oldRoundFilesRetained)
+ assertFalse(newRoundFiles.isEmpty())
+ // Round-robin selection is scoped to a writer lifecycle, so a small
+ // number of independent Doris statements need not hit both paths. The
+ // stable contract is that every new file uses the refreshed path set.
+ assertTrue(newRoundFiles.every {
+ it.startsWith("${pathRoot}/round-c/") ||
+ it.startsWith("${pathRoot}/round-d/")
+ })
+ assertDorisSparkRows("external_round_robin_changed", "t_round_robin",
+ "pt, id, length(payload)", "ORDER BY pt, id")
+
+ (1..6).each { id ->
+ sql """INSERT INTO t_weight_robin VALUES (${id}, 'weight-${id}')"""
+ }
+ def weightedFiles = dataFiles("t_weight_robin")
+ assertFalse(weightedFiles.isEmpty())
+ assertTrue(weightedFiles.every {
+ it.startsWith("${pathRoot}/weight-a/") ||
+ it.startsWith("${pathRoot}/weight-b/")
+ })
+ assertDorisSparkRows("external_weight_robin", "t_weight_robin",
+ "id, payload", "ORDER BY id")
+
+ sql """INSERT INTO t_specific_fs VALUES (1, 'specific-1')"""
+ sql """INSERT INTO t_specific_fs VALUES (2, 'specific-2')"""
+ def specificFiles = dataFiles("t_specific_fs")
+ assertFalse(specificFiles.isEmpty())
+ assertTrue(specificFiles.every {
it.startsWith("${pathRoot}/specific-") })
+ assertDorisSparkRows("external_specific_fs", "t_specific_fs",
+ "id, payload", "ORDER BY id")
+
+ sql """INSERT INTO t_none VALUES (1, 'default-path')"""
+ def defaultFiles = dataFiles("t_none")
+ assertFalse(defaultFiles.isEmpty())
+ assertTrue(defaultFiles.every { !it.startsWith(pathRoot) })
+ assertDorisSparkRows("external_default_path", "t_none",
+ "id, payload", "ORDER BY id")
+ } finally {
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_failures.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_failures.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_failures.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_failures.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_key_dynamic_memory_negative.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_key_dynamic_memory_negative.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_key_dynamic_memory_negative.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_key_dynamic_memory_negative.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_merge_engine.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_merge_engine.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_merge_engine.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_merge_engine.groovy
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_merge_semantics.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_merge_semantics.groovy
new file mode 100644
index 00000000000..2fe05e130df
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_merge_semantics.groovy
@@ -0,0 +1,221 @@
+// 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_paimon_write_merge_semantics", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_merge_semantics_catalog"
+ String dbName = "test_pw_merge_semantics_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.t_merge;
+ CREATE TABLE paimon.${dbName}.t_merge (
+ id INT,
+ score INT,
+ payload STRUCT<x: INT, y: STRING>,
+ status STRING,
+ required_value STRING NOT NULL
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '2',
+ 'bucket-key' = 'id',
+ 'num-sorted-run.compaction-trigger' = '100'
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ sql """create database if not exists internal.${dbName}"""
+ sql """drop table if exists internal.${dbName}.merge_source"""
+ sql """
+ create table internal.${dbName}.merge_source (
+ id int,
+ delta int,
+ new_x int,
+ new_y string,
+ action string,
+ required_value string
+ ) distributed by hash(id) buckets 1
+ properties ('replication_num' = '1')
+ """
+
+ try {
+ def latestSnapshotId = {
+ def rows = spark_paimon """
+ SELECT max(snapshot_id)
+ FROM paimon.${dbName}.`t_merge\$snapshots`
+ """
+ return rows[0][0] == null ? 0L : rows[0][0].toString().toLong()
+ }
+ def activeFileCount = {
+ def rows = spark_paimon """
+ SELECT count(*) FROM paimon.${dbName}.`t_merge\$files`
+ """
+ return rows[0][0].toString().toLong()
+ }
+ def assertCrossEngine = {
+ def sparkRows = spark_paimon """
+ SELECT id, score, payload.x, payload.y, status, required_value
+ FROM paimon.${dbName}.t_merge ORDER BY id
+ """
+ def dorisRows = sql """
+ SELECT id, score, payload.x, payload.y, status, required_value
+ FROM t_merge ORDER BY id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ sql """INSERT INTO t_merge VALUES
+ (1, 10, named_struct('x', 1, 'y', 'base-1'), 'old', 'required-1'),
+ (2, 20, named_struct('x', 2, 'y', 'base-2'), 'old', 'required-2'),
+ (3, 30, named_struct('x', 3, 'y', 'base-3'), 'stable',
'required-3')
+ """
+ sql """INSERT INTO internal.${dbName}.merge_source VALUES
+ (1, 5, 11, 'source-1', 'U', 'required-1-new'),
+ (2, 0, 22, 'source-2', 'D', 'required-2-new'),
+ (4, 40, 44, 'source-4', 'I', 'required-4')
+ """
+
+ long beforeMerge = latestSnapshotId()
+ sql """
+ MERGE INTO t_merge t
+ USING internal.${dbName}.merge_source s
+ ON t.id = s.id
+ WHEN MATCHED AND s.action = 'U' THEN UPDATE SET
+ score = t.score + s.delta,
+ payload = named_struct(
+ 'x', s.new_x,
+ 'y', concat(t.payload.y, '-', s.new_y)),
+ status = 'updated',
+ required_value = s.required_value
+ WHEN MATCHED THEN DELETE
+ WHEN NOT MATCHED AND s.action = 'I' THEN INSERT
+ (required_value, status, payload, score, id)
+ VALUES (
+ s.required_value,
+ 'inserted',
+ named_struct('x', s.new_x, 'y', s.new_y),
+ s.delta,
+ s.id)
+ """
+ assertEquals(beforeMerge + 1L, latestSnapshotId())
+ order_qt_merge_semantics_result """
+ SELECT id, score, payload.x, payload.y, status, required_value
+ FROM t_merge ORDER BY id
+ """
+ assertCrossEngine()
+
+ // An empty source is a true no-op: it must not publish an empty Paimon
+ // snapshot or alter the active file set.
+ sql """TRUNCATE TABLE internal.${dbName}.merge_source"""
+ long beforeEmptySnapshot = latestSnapshotId()
+ long beforeEmptyFiles = activeFileCount()
+ sql """
+ MERGE INTO t_merge t
+ USING internal.${dbName}.merge_source s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET score = t.score + s.delta
+ WHEN NOT MATCHED THEN INSERT
+ (id, score, payload, status, required_value)
+ VALUES (s.id, s.delta,
+ named_struct('x', s.new_x, 'y', s.new_y),
+ 'inserted', s.required_value)
+ """
+ assertEquals(beforeEmptySnapshot, latestSnapshotId())
+ assertEquals(beforeEmptyFiles, activeFileCount())
+
+ // P09 failure atomicity: a forbidden key update and a missing required
+ // insert column both fail before a snapshot or file becomes visible.
+ sql """INSERT INTO internal.${dbName}.merge_source VALUES
+ (1, 1, 100, 'invalid-key-update', 'U', 'still-required')
+ """
+ long beforeFailureSnapshot = latestSnapshotId()
+ long beforeFailureFiles = activeFileCount()
+ test {
+ sql """
+ MERGE INTO t_merge t
+ USING internal.${dbName}.merge_source s ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET id = s.id + 100
+ """
+ exception "primary-key"
+ }
+ assertEquals(beforeFailureSnapshot, latestSnapshotId())
+ assertEquals(beforeFailureFiles, activeFileCount())
+
+ sql """TRUNCATE TABLE internal.${dbName}.merge_source"""
+ sql """INSERT INTO internal.${dbName}.merge_source VALUES
+ (9, 9, 9, 'missing-required', 'I', NULL)
+ """
+ test {
+ sql """
+ MERGE INTO t_merge t
+ USING internal.${dbName}.merge_source s ON t.id = s.id
+ WHEN NOT MATCHED THEN INSERT (id, score, payload, status)
+ VALUES (s.id, s.delta,
+ named_struct('x', s.new_x, 'y', s.new_y), 'invalid')
+ """
+ exception "requires values for every table column"
+ }
+ assertEquals(beforeFailureSnapshot, latestSnapshotId())
+ assertEquals(beforeFailureFiles, activeFileCount())
+
+ // A legal MERGE after both failures proves that no stale writer or
+ // transaction state poisoned the table.
+ sql """TRUNCATE TABLE internal.${dbName}.merge_source"""
+ sql """INSERT INTO internal.${dbName}.merge_source VALUES
+ (5, 50, 55, 'recovery', 'I', 'required-5')
+ """
+ sql """
+ MERGE INTO t_merge t
+ USING internal.${dbName}.merge_source s ON t.id = s.id
+ WHEN NOT MATCHED THEN INSERT
+ (id, score, payload, status, required_value)
+ VALUES (s.id, s.delta,
+ named_struct('x', s.new_x, 'y', s.new_y),
+ 'recovered', s.required_value)
+ """
+ assertEquals(beforeFailureSnapshot + 1L, latestSnapshotId())
+ order_qt_merge_semantics_recovered """
+ SELECT id, status FROM t_merge WHERE id = 5 ORDER BY id
+ """
+ assertCrossEngine()
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ sql """drop table if exists internal.${dbName}.merge_source"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_partition_delete.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_partition_delete.groovy
new file mode 100644
index 00000000000..3334a6cf005
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_partition_delete.groovy
@@ -0,0 +1,150 @@
+// 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_paimon_write_partition_delete", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_partition_delete_catalog"
+ String dbName = "test_pw_partition_delete_db"
+ String snapshotsTableSuffix = '$snapshots'
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.t_partition_delete;
+ CREATE TABLE paimon.${dbName}.t_partition_delete (
+ pt STRING, id INT, score INT, payload STRING
+ ) USING paimon
+ PARTITIONED BY (pt)
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '-1',
+ 'dynamic-bucket.target-row-num' = '2',
+ 'write-only' = 'true'
+ );
+ """
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """SWITCH ${catalogName}"""
+ sql """USE ${dbName}"""
+
+ try {
+ def latestSnapshot = {
+ String query = """
+ SELECT snapshot_id, commit_kind
+ FROM
paimon.${dbName}.`t_partition_delete${snapshotsTableSuffix}`
+ ORDER BY snapshot_id DESC LIMIT 1
+ """
+ def rows = spark_paimon(query)
+ return rows.isEmpty() ? [0L, null] : [
+ rows[0][0].toString().toLong(),
rows[0][1].toString().toUpperCase()]
+ }
+ def assertRows = { String tag ->
+ "order_qt_${tag}" """
+ SELECT pt, id, score, payload
+ FROM t_partition_delete ORDER BY id
+ """
+ def dorisRows = sql """
+ SELECT pt, id, score, payload
+ FROM t_partition_delete ORDER BY id
+ """
+ def sparkRows = spark_paimon """
+ SELECT pt, id, score, payload
+ FROM paimon.${dbName}.t_partition_delete ORDER BY id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+ def assertDeleteCommit = { List<Object> before ->
+ def after = latestSnapshot()
+ assertEquals(before[0] + 1L, after[0])
+ assertEquals("APPEND", after[1])
+ return after
+ }
+
+ sql """INSERT INTO t_partition_delete VALUES
+ ('p1', 1, 10, 'p1-a'),
+ ('p1', 2, 20, 'p1-b'),
+ ('p2', 3, 30, 'p2-a'),
+ ('p2', 4, 40, 'p2-b'),
+ ('p3', 5, 50, 'p3-a'),
+ (NULL, 6, 60, 'default-partition')
+ """
+
+ // A predicate covering the complete partition must not affect any
+ // other partition, including the default partition represented by
NULL.
+ def snapshot = latestSnapshot()
+ sql """DELETE FROM t_partition_delete WHERE pt = 'p1'"""
+ snapshot = assertDeleteCommit(snapshot)
+ assertRows("partition_delete_full_partition")
+
+ // A partial-partition predicate is evaluated row by row.
+ sql """DELETE FROM t_partition_delete WHERE pt = 'p2' AND score >=
40"""
+ snapshot = assertDeleteCommit(snapshot)
+ assertRows("partition_delete_partial_partition")
+
+ // A non-convertible partition expression must retain its exact SQL
+ // semantics instead of expanding into a full-partition delete.
+ sql """DELETE FROM t_partition_delete WHERE upper(pt) = 'P2' AND id =
3"""
+ snapshot = assertDeleteCommit(snapshot)
+ assertRows("partition_delete_expression")
+
+ // UNKNOWN predicates match no rows and must not create empty commits.
+ sql """DELETE FROM t_partition_delete WHERE pt = NULL"""
+ assertEquals(snapshot, latestSnapshot())
+ sql """DELETE FROM t_partition_delete WHERE pt NOT IN ('p3', NULL)"""
+ assertEquals(snapshot, latestSnapshot())
+ sql """DELETE FROM t_partition_delete WHERE EXISTS (SELECT 1 WHERE
FALSE)"""
+ assertEquals(snapshot, latestSnapshot())
+
+ // NOT EXISTS is true here and is combined with a target predicate so
+ // only the intended row is removed.
+ sql """
+ DELETE FROM t_partition_delete
+ WHERE id = 5 AND NOT EXISTS (SELECT 1 WHERE FALSE)
+ """
+ snapshot = assertDeleteCommit(snapshot)
+ assertRows("partition_delete_not_exists")
+
+ // IS NULL addresses the default partition explicitly.
+ sql """DELETE FROM t_partition_delete WHERE pt IS NULL"""
+ snapshot = assertDeleteCommit(snapshot)
+ assertRows("partition_delete_default_partition")
+
+ // A second no-match delete proves the empty-table path also avoids a
+ // metadata-only snapshot.
+ sql """DELETE FROM t_partition_delete WHERE id = 999"""
+ assertEquals(snapshot, latestSnapshot())
+ } finally {
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ }
+}
diff --git a/regression-test/suites/paimon_write/test_paimon_write_pk.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_pk.groovy
similarity index 100%
rename from regression-test/suites/paimon_write/test_paimon_write_pk.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_pk.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_row_level_dml.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_row_level_dml.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_row_level_dml.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_row_level_dml.groovy
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.groovy
new file mode 100644
index 00000000000..cdea849f37a
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_row_tracking_evolution.groovy
@@ -0,0 +1,216 @@
+// 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_paimon_write_row_tracking_evolution", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_row_tracking_catalog"
+ String dbName = "test_pw_row_tracking_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_tracking;
+ CREATE TABLE paimon.${dbName}.t_tracking (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'bucket' = '-1',
+ 'row-tracking.enabled' = 'true',
+ 'compaction.min.file-num' = '2'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_evolution;
+ CREATE TABLE paimon.${dbName}.t_evolution (
+ id INT, b INT, c INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'bucket' = '-1',
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_evolution_source;
+ CREATE TABLE paimon.${dbName}.t_evolution_source (
+ id INT, b INT, c INT
+ ) USING paimon;
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ try {
+ def trackingRows = { String tableName ->
+ return sql("""
+ SELECT id, _ROW_ID, _SEQUENCE_NUMBER
+ FROM `${tableName}\$row_tracking`
+ ORDER BY id
+ """)
+ }
+ def latestSnapshotId = { String tableName ->
+ def rows = spark_paimon """
+ SELECT max(snapshot_id)
+ FROM paimon.${dbName}.`${tableName}\$snapshots`
+ """
+ return rows[0][0] == null ? 0L : rows[0][0].toString().toLong()
+ }
+
+ // Doris assigns row ids through the Paimon committer. Their exact
+ // values are not assumed, but they must be unique and stable.
+ sql """INSERT INTO t_tracking VALUES
+ (1, 'one'), (2, 'two'), (3, 'three')
+ """
+ def initialTracking = trackingRows("t_tracking")
+ assertEquals(3, initialTracking.size())
+ assertEquals(3, initialTracking.collect { it[1] }.toSet().size())
+ Map<Integer, Long> initialRowIds = initialTracking.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[1].toString().toLong()]
+ }
+ Map<Integer, Long> initialSequences = initialTracking.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[2].toString().toLong()]
+ }
+
+ // Spark performs row-level changes which Doris does not expose for an
+ // append table. This verifies that rows originally written by Doris
have
+ // valid tracking metadata for every upstream operation.
+ spark_paimon_multi """
+ UPDATE paimon.${dbName}.t_tracking
+ SET payload = 'two-updated' WHERE id = 2;
+ DELETE FROM paimon.${dbName}.t_tracking WHERE id = 3;
+ MERGE INTO paimon.${dbName}.t_tracking t
+ USING (SELECT 1 AS id, 'one-merged' AS payload
+ UNION ALL
+ SELECT 4 AS id, 'four' AS payload) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET payload = s.payload
+ WHEN NOT MATCHED THEN INSERT (id, payload) VALUES (s.id,
s.payload);
+ """
+ sql """refresh table t_tracking"""
+ order_qt_row_tracking_after_spark_changes """
+ SELECT id, payload FROM t_tracking ORDER BY id
+ """
+ def changedTracking = trackingRows("t_tracking")
+ Map<Integer, Long> changedRowIds = changedTracking.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[1].toString().toLong()]
+ }
+ Map<Integer, Long> changedSequences = changedTracking.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[2].toString().toLong()]
+ }
+ assertEquals(initialRowIds[1], changedRowIds[1])
+ assertEquals(initialRowIds[2], changedRowIds[2])
+ assertFalse(initialRowIds.values().contains(changedRowIds[4]))
+ assertTrue(changedSequences[1] > initialSequences[1])
+ assertTrue(changedSequences[2] > initialSequences[2])
+
+ spark_paimon """
+ CALL paimon.sys.compact(
+ table => '${dbName}.t_tracking',
+ compact_strategy => 'full')
+ """
+ sql """refresh table t_tracking"""
+ def compactedTracking = trackingRows("t_tracking")
+ Map<Integer, Long> compactedRowIds = compactedTracking.collectEntries
{ row ->
+ [(row[0].toString().toInteger()): row[1].toString().toLong()]
+ }
+ assertEquals(changedRowIds, compactedRowIds)
+
+ sql """INSERT INTO t_tracking VALUES (5, 'five-after-compact')"""
+ order_qt_row_tracking_after_compact_write """
+ SELECT id, payload FROM t_tracking ORDER BY id
+ """
+ def afterDorisReopen = trackingRows("t_tracking")
+ assertEquals(4, afterDorisReopen.collect { it[1] }.toSet().size())
+ assertFalse(compactedRowIds.values().contains(
+ afterDorisReopen.find { it[0].toString().toInteger() == 5 }[1]
+ .toString().toLong()))
+
+ // Data evolution accepts Doris full and partial INSERTs. Spark MERGE
+ // then updates only selected columns and keeps the original row ids.
+ sql """INSERT INTO t_evolution VALUES (1, 10, 100), (2, 20, 200)"""
+ sql """INSERT INTO t_evolution (id, b) VALUES (3, 30)"""
+ def evolutionBefore = trackingRows("t_evolution")
+ Map<Integer, Long> evolutionRowIds = evolutionBefore.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[1].toString().toLong()]
+ }
+ spark_paimon_multi """
+ INSERT INTO paimon.${dbName}.t_evolution_source VALUES
+ (1, 11, 111), (2, 22, 222), (4, 44, 444);
+ MERGE INTO paimon.${dbName}.t_evolution t
+ USING paimon.${dbName}.t_evolution_source s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET b = s.b
+ WHEN NOT MATCHED THEN INSERT (id, b, c)
+ VALUES (s.id, s.b, s.c);
+ """
+ sql """refresh table t_evolution"""
+ order_qt_data_evolution_after_spark_merge """
+ SELECT * FROM t_evolution ORDER BY id
+ """
+ def evolutionAfter = trackingRows("t_evolution")
+ Map<Integer, Long> evolutionAfterIds = evolutionAfter.collectEntries {
row ->
+ [(row[0].toString().toInteger()): row[1].toString().toLong()]
+ }
+ assertEquals(evolutionRowIds[1], evolutionAfterIds[1])
+ assertEquals(evolutionRowIds[2], evolutionAfterIds[2])
+ assertEquals(evolutionRowIds[3], evolutionAfterIds[3])
+ assertFalse(evolutionRowIds.values().contains(evolutionAfterIds[4]))
+
+ // Paimon 1.4.2 does not support ordinary UPDATE/DELETE on a data
+ // evolution table. Doris currently rejects them at its append-table
+ // boundary; either way no Paimon snapshot may be committed.
+ long evolutionSnapshot = latestSnapshotId("t_evolution")
+ test {
+ sql """UPDATE t_evolution SET b = 999 WHERE id = 1"""
+ exception "primary-key table"
+ }
+ assertEquals(evolutionSnapshot, latestSnapshotId("t_evolution"))
+ test {
+ sql """DELETE FROM t_evolution WHERE id = 1"""
+ exception "primary-key table"
+ }
+ assertEquals(evolutionSnapshot, latestSnapshotId("t_evolution"))
+ test {
+ sql """
+ MERGE INTO t_evolution t
+ USING (SELECT 1 AS id, 999 AS b) s ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET b = s.b
+ """
+ exception "primary-key table"
+ }
+ assertEquals(evolutionSnapshot, latestSnapshotId("t_evolution"))
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_schema_change.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_schema_change.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_schema_change.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_schema_change.groovy
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_group.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_group.groovy
new file mode 100644
index 00000000000..1da8ddca25a
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_group.groovy
@@ -0,0 +1,227 @@
+// 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_paimon_write_sequence_group", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_sequence_group_catalog"
+ String dbName = "test_pw_sequence_group_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_multi_group;
+ CREATE TABLE paimon.${dbName}.t_multi_group (
+ id INT,
+ profile_name STRING,
+ profile_city STRING,
+ profile_seq INT,
+ total BIGINT,
+ peak INT,
+ metric_seq INT,
+ note STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update',
+ 'fields.profile_seq.sequence-group' = 'profile_name,profile_city',
+ 'fields.metric_seq.sequence-group' = 'total,peak',
+ 'fields.total.aggregate-function' = 'sum',
+ 'fields.peak.aggregate-function' = 'max',
+ 'fields.note.aggregate-function' = 'last_non_null_value'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_remove_on_delete;
+ CREATE TABLE paimon.${dbName}.t_remove_on_delete (
+ id INT, a STRING, b STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update',
+ 'partial-update.remove-record-on-delete' = 'true'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_group_remove_on_delete;
+ CREATE TABLE paimon.${dbName}.t_group_remove_on_delete (
+ id INT,
+ a STRING,
+ seq_a INT,
+ b STRING,
+ seq_b INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update',
+ 'fields.seq_a.sequence-group' = 'a',
+ 'fields.seq_b.sequence-group' = 'b',
+ 'partial-update.remove-record-on-sequence-group' = 'seq_a'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_property_change;
+ CREATE TABLE paimon.${dbName}.t_property_change (
+ id INT, a STRING, seq INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update'
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ try {
+ def assertSparkEquals = { String tableName, String columns, String
orderBy ->
+ def sparkRows = spark_paimon """
+ SELECT ${columns} FROM paimon.${dbName}.${tableName} ${orderBy}
+ """
+ def dorisRows = sql """SELECT ${columns} FROM ${tableName}
${orderBy}"""
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ // P01: Each sequence group advances independently. Aggregations only
+ // consume a row when the sequence of their own group is accepted.
+ sql """INSERT INTO t_multi_group VALUES
+ (1, 'alice', 'shanghai', 10, 5, 80, 10, 'base')
+ """
+ sql """INSERT INTO t_multi_group VALUES
+ (1, 'stale-profile', 'beijing', 9, 7, 90, 11, NULL)
+ """
+ order_qt_sequence_group_stale_profile """
+ SELECT * FROM t_multi_group ORDER BY id
+ """
+
+ sql """INSERT INTO t_multi_group VALUES
+ (1, 'alice-new', 'shenzhen', 12, 100, 99, 10, 'new-note')
+ """
+ order_qt_sequence_group_new_profile """
+ SELECT * FROM t_multi_group ORDER BY id
+ """
+
+ // A NULL sequence does not advance its group. A different group in the
+ // same row can still advance and apply its aggregate functions.
+ sql """INSERT INTO t_multi_group VALUES
+ (1, 'null-sequence', 'hangzhou', NULL, 3, 88, 13, NULL)
+ """
+ order_qt_sequence_group_null_sequence """
+ SELECT * FROM t_multi_group ORDER BY id
+ """
+ assertSparkEquals("t_multi_group", "*", "ORDER BY id")
+
+ // P02: remove-record-on-delete must discard the old partial row. A
+ // later partial insert creates a new row and must not revive old
fields.
+ sql """INSERT INTO t_remove_on_delete VALUES (1, 'old-a', 'old-b')"""
+ sql """DELETE FROM t_remove_on_delete WHERE id = 1"""
+ qt_remove_on_delete_empty """SELECT count(*) FROM t_remove_on_delete"""
+ sql """INSERT INTO t_remove_on_delete (id, b) VALUES (1, 'new-b')"""
+ order_qt_remove_on_delete_partial """
+ SELECT id, a, b FROM t_remove_on_delete ORDER BY id
+ """
+ sql """INSERT INTO t_remove_on_delete (id, a) VALUES (1, 'new-a')"""
+ order_qt_remove_on_delete_complete """
+ SELECT id, a, b FROM t_remove_on_delete ORDER BY id
+ """
+ assertSparkEquals("t_remove_on_delete", "*", "ORDER BY id")
+
+ // Paimon makes remove-record-on-delete and sequence groups mutually
+ // exclusive. Doris must reject a whole-row DELETE on this legal
+ // sequence-group configuration without changing its accumulated row.
+ sql """INSERT INTO t_group_remove_on_delete VALUES
+ (1, 'old-a', 100, 'old-b', 100)
+ """
+ test {
+ sql """DELETE FROM t_group_remove_on_delete WHERE id = 1"""
+ exception "partial-update.remove-record-on-delete=true"
+ }
+ order_qt_group_remove_on_delete_unchanged """
+ SELECT * FROM t_group_remove_on_delete ORDER BY id
+ """
+ sql """INSERT INTO t_group_remove_on_delete VALUES
+ (1, 'low-a', 1, 'low-b', 1)
+ """
+ sql """INSERT INTO t_group_remove_on_delete (id, a, seq_a) VALUES
+ (1, 'high-a', 101)
+ """
+ order_qt_group_remove_on_delete_sequence """
+ SELECT * FROM t_group_remove_on_delete ORDER BY id
+ """
+ assertSparkEquals("t_group_remove_on_delete", "*", "ORDER BY id")
+
+ // P03: Invalid writer properties fail as metadata changes and must not
+ // poison the last valid schema. A legal sequence group takes effect
for
+ // the next writer without recreating the catalog.
+ sql """INSERT INTO t_property_change VALUES (1, 'base', 10)"""
+ long propertySnapshot = (sql """
+ SELECT max(snapshot_id) FROM t_property_change\$snapshots
+ """)[0][0] as long
+ String invalidPropertyError = null
+ try {
+ spark_paimon """
+ ALTER TABLE paimon.${dbName}.t_property_change SET
TBLPROPERTIES (
+ 'fields.missing.sequence-group' = 'a')
+ """
+ } catch (Exception e) {
+ invalidPropertyError = e.getMessage()
+ }
+ assertNotNull(invalidPropertyError)
+ assertTrue(invalidPropertyError.toLowerCase().contains("missing"))
+ assertEquals(propertySnapshot, (sql """
+ SELECT max(snapshot_id) FROM t_property_change\$snapshots
+ """)[0][0] as long)
+ sql """INSERT INTO t_property_change (id, a) VALUES (1,
'still-writable')"""
+ order_qt_property_change_still_writable """
+ SELECT * FROM t_property_change ORDER BY id
+ """
+
+ spark_paimon """
+ ALTER TABLE paimon.${dbName}.t_property_change SET TBLPROPERTIES (
+ 'fields.seq.sequence-group' = 'a')
+ """
+ sql """REFRESH CATALOG ${catalogName}"""
+ sql """USE ${dbName}"""
+ sql """INSERT INTO t_property_change VALUES (1, 'high', 20)"""
+ sql """INSERT INTO t_property_change VALUES (1, 'low-must-lose', 15)"""
+ order_qt_property_change_sequence_group """
+ SELECT * FROM t_property_change ORDER BY id
+ """
+ assertSparkEquals("t_property_change", "*", "ORDER BY id")
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.groovy
new file mode 100644
index 00000000000..b33c981e716
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_sequence_rowkind.groovy
@@ -0,0 +1,200 @@
+// 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_paimon_write_sequence_rowkind", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_sequence_rowkind_catalog"
+ String dbName = "test_pw_sequence_rowkind_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_sequence_asc;
+ CREATE TABLE paimon.${dbName}.t_sequence_asc (
+ id INT, seq1 INT, seq2 INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '2',
+ 'sequence.field' = 'seq1,seq2',
+ 'sequence.field.sort-order' = 'ascending'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_sequence_desc;
+ CREATE TABLE paimon.${dbName}.t_sequence_desc (
+ id INT, seq INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'sequence.field' = 'seq',
+ 'sequence.field.sort-order' = 'descending'
+ );
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_rowkind;
+ CREATE TABLE paimon.${dbName}.t_rowkind (
+ id INT, row_kind STRING, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'rowkind.field' = 'row_kind',
+ 'changelog-producer' = 'input'
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ try {
+ def latestSnapshotId = { String tableName ->
+ def rows = spark_paimon """
+ SELECT max(snapshot_id)
+ FROM paimon.${dbName}.`${tableName}\$snapshots`
+ """
+ return rows[0][0] == null ? 0L : rows[0][0].toString().toLong()
+ }
+ def activeFileCount = { String tableName ->
+ def rows = spark_paimon """
+ SELECT count(*) FROM paimon.${dbName}.`${tableName}\$files`
+ """
+ return rows[0][0].toString().toLong()
+ }
+ def assertSparkEquals = { String tableName, String columns, String
orderBy ->
+ def sparkRows = spark_paimon """
+ SELECT ${columns} FROM paimon.${dbName}.${tableName} ${orderBy}
+ """
+ def dorisRows = sql """SELECT ${columns} FROM ${tableName}
${orderBy}"""
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ // Multi-column ascending sequences compare lexicographically across
+ // Doris commits. A NULL tuple loses once a concrete sequence arrives.
+ sql """INSERT INTO t_sequence_asc VALUES
+ (1, 10, 20, 'base'),
+ (2, NULL, NULL, 'null-base')
+ """
+ sql """INSERT INTO t_sequence_asc VALUES
+ (1, 9, 99, 'stale-first-field'),
+ (2, 1, 0, 'nonnull-wins')
+ """
+ sql """INSERT INTO t_sequence_asc VALUES
+ (1, 10, 21, 'new-second-field'),
+ (2, NULL, NULL, 'null-must-not-return')
+ """
+ order_qt_sequence_ascending """
+ SELECT * FROM t_sequence_asc ORDER BY id
+ """
+
+ // Equal sequences fall back to input order. Keep the writer
single-task
+ // so this oracle checks Paimon's tie rule instead of scheduler
ordering.
+ sql """set parallel_pipeline_task_num = 1"""
+ sql """INSERT INTO t_sequence_asc VALUES
+ (3, 7, 7, 'first-equal'),
+ (3, 7, 7, 'second-equal')
+ """
+ order_qt_sequence_equal """
+ SELECT id, payload FROM t_sequence_asc WHERE id = 3 ORDER BY id
+ """
+ assertSparkEquals("t_sequence_asc", "*", "ORDER BY id")
+
+ // Descending order reverses priority: a smaller sequence supersedes
the
+ // current row while a larger sequence is ignored.
+ sql """INSERT INTO t_sequence_desc VALUES (1, 10, 'base')"""
+ sql """INSERT INTO t_sequence_desc VALUES (1, 20, 'larger-is-stale')"""
+ sql """INSERT INTO t_sequence_desc VALUES (1, 5, 'smaller-wins')"""
+ order_qt_sequence_descending """
+ SELECT * FROM t_sequence_desc ORDER BY id
+ """
+ assertSparkEquals("t_sequence_desc", "*", "ORDER BY id")
+
+ // rowkind.field turns ordinary INSERT rows into an input changelog.
+ sql """INSERT INTO t_rowkind VALUES
+ (1, '+I', 'old-1'),
+ (2, '+I', 'old-2')
+ """
+ long rowkindBefore = latestSnapshotId("t_rowkind")
+ sql """INSERT INTO t_rowkind VALUES
+ (1, '+U', 'new-1'),
+ (2, '-D', 'old-2'),
+ (3, '+I', 'new-3')
+ """
+ long rowkindAfter = latestSnapshotId("t_rowkind")
+ order_qt_rowkind_changelog """
+ SELECT id, payload FROM t_rowkind ORDER BY id
+ """
+
+ def auditRows = spark_paimon """
+ SELECT rowkind, id, payload
+ FROM paimon_incremental_query(
+ 'paimon.${dbName}.`t_rowkind\$audit_log`',
+ '${rowkindBefore}', '${rowkindAfter}')
+ ORDER BY id
+ """
+ assertEquals([
+ ["+U", 1, "new-1"],
+ ["-D", 2, "old-2"],
+ ["+I", 3, "new-3"]
+ ], auditRows)
+ assertSparkEquals("t_rowkind", "id, payload", "ORDER BY id")
+
+ // Invalid or omitted row kinds fail atomically. The following valid
+ // changelog record must still be accepted by a newly opened writer.
+ long beforeInvalidSnapshot = latestSnapshotId("t_rowkind")
+ long beforeInvalidFiles = activeFileCount("t_rowkind")
+ test {
+ sql """INSERT INTO t_rowkind VALUES (9, 'XX', 'invalid')"""
+ exception "row kind"
+ }
+ assertEquals(beforeInvalidSnapshot, latestSnapshotId("t_rowkind"))
+ assertEquals(beforeInvalidFiles, activeFileCount("t_rowkind"))
+
+ test {
+ sql """INSERT INTO t_rowkind VALUES (9, NULL, 'missing-kind')"""
+ exception "cannot be null"
+ }
+ assertEquals(beforeInvalidSnapshot, latestSnapshotId("t_rowkind"))
+ assertEquals(beforeInvalidFiles, activeFileCount("t_rowkind"))
+
+ sql """INSERT INTO t_rowkind VALUES (4, '+I', 'recovered')"""
+ order_qt_rowkind_recovered """
+ SELECT id, payload FROM t_rowkind WHERE id = 4 ORDER BY id
+ """
+ assertSparkEquals("t_rowkind", "id, payload", "ORDER BY id")
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_snapshot_refs.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_snapshot_refs.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_snapshot_refs.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_snapshot_refs.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_source_models.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_source_models.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_source_models.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_source_models.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_transaction.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_transaction.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
diff --git a/regression-test/suites/paimon_write/test_paimon_write_types.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
similarity index 100%
rename from regression-test/suites/paimon_write/test_paimon_write_types.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant.groovy
similarity index 100%
rename from regression-test/suites/paimon_write/test_paimon_write_variant.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant_dml.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_dml.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_variant_dml.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_dml.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant_errors.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_variant_errors.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant_nested.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_nested.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_variant_nested.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_nested.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant_shredding.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_shredding.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_variant_shredding.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_shredding.groovy
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_variant_table_modes.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_table_modes.groovy
similarity index 100%
rename from
regression-test/suites/paimon_write/test_paimon_write_variant_table_modes.groovy
rename to
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_table_modes.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
