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 640a61b7f6d [test](paimon) Add P0 write correctness and snapshot
reference coverage (#66325)
640a61b7f6d is described below
commit 640a61b7f6de0680118d804df5197660000e4df4
Author: Gabriel <[email protected]>
AuthorDate: Mon Aug 3 09:27:45 2026 +0800
[test](paimon) Add P0 write correctness and snapshot reference coverage
(#66325)
## Proposed changes
- Enable the Paimon JDBC catalog lock in the existing object-storage
fixture so concurrent snapshot commits use a supported configuration.
- Add synchronized two-session write coverage for append tables across
different and identical partitions, fixed-bucket deduplication,
aggregation lost-update detection, and dynamic buckets with disjoint
partitions.
- Add write-to-read closure coverage for snapshots, tags, and branches
created from Doris commits.
- Keep the unsupported Paimon branch sink boundary explicit and verify
rejected INSERT and INSERT OVERWRITE statements do not mutate main, tag,
branch, or historical snapshot state.
- Add INSERT SELECT coverage from Doris Duplicate, Unique MOW, Unique
MOR, and Aggregate source tables with LIST/RANGE/unpartitioned layouts
and RANDOM/HASH/AUTO buckets.
- Add cross-engine validation for nested ARRAY, MAP, STRUCT, DECIMAL,
BOOLEAN, DATE, and TIMESTAMP_NTZ values written from an internal OLAP
source into Paimon ORC.
## Motivation
The existing P0 suites cover broad serial write behavior, table models,
merge engines, scalar and nested types, schema evolution, transactions,
and failure paths. They did not exercise independent Doris transactions
committing concurrently to one Paimon table, verify time travel and
references rooted in snapshots written by Doris, or validate every Doris
OLAP source model and recursive complex values through INSERT SELECT.
## Validation
- test_paimon_jdbc_catalog: PASS with Paimon 1.3.1, JDBC catalog
locking, and object storage
- test_paimon_write_snapshot_refs: PASS with cross-engine Spark and
Doris result comparison
- test_paimon_write_source_models: PASS with four Doris source models
and recursive complex types compared by Spark and Doris
- git diff --check: PASS
---
.../paimon/test_paimon_jdbc_catalog.groovy | 211 ++++++++++++++-
...paimon_write_key_dynamic_memory_negative.groovy | 160 +++++++++++
.../test_paimon_write_snapshot_refs.groovy | 192 ++++++++++++++
.../test_paimon_write_source_models.groovy | 294 +++++++++++++++++++++
.../test_paimon_write_thread_lifecycle.groovy | 154 +++++++++++
5 files changed, 1010 insertions(+), 1 deletion(-)
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 cbb3174ea5f..28a71862c0e 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
@@ -15,6 +15,9 @@
// specific language governing permissions and limitations
// under the License.
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
suite("test_paimon_jdbc_catalog", "p0,external") {
String enabled = context.config.otherConfigs.get("enablePaimonTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
@@ -91,6 +94,27 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
}
}
+ def runConcurrent = { String leftName, Closure leftAction,
+ String rightName, Closure rightAction ->
+ CountDownLatch ready = new CountDownLatch(2)
+ CountDownLatch start = new CountDownLatch(1)
+ def left = thread(leftName) {
+ ready.countDown()
+ start.await()
+ leftAction()
+ }
+ def right = thread(rightName) {
+ ready.countDown()
+ start.await()
+ rightAction()
+ }
+ assertTrue(ready.await(30, TimeUnit.SECONDS),
+ "Both Paimon writers must reach the dispatch barrier")
+ start.countDown()
+ left.get()
+ right.get()
+ }
+
executeCommand("mkdir -p ${localDriverDir}", false, 60)
if (!new File(localDriverPath).exists()) {
executeCommand("/usr/bin/curl --max-time 600 ${driverDownloadUrl}
--output ${localDriverPath}", true, 660)
@@ -174,7 +198,7 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
--conf spark.sql.catalog.${sparkSeedCatalogName}.catalog-key=${catalogName} \
--conf spark.sql.catalog.${sparkSeedCatalogName}.jdbc.user=postgres \
--conf spark.sql.catalog.${sparkSeedCatalogName}.jdbc.password=123456 \
---conf spark.sql.catalog.${sparkSeedCatalogName}.lock.enabled=false \
+--conf spark.sql.catalog.${sparkSeedCatalogName}.lock.enabled=true \
--conf
spark.sql.catalog.${sparkSeedCatalogName}.s3.endpoint=${sparkMinioEndpoint} \
--conf spark.sql.catalog.${sparkSeedCatalogName}.s3.access-key=${minioAk} \
--conf spark.sql.catalog.${sparkSeedCatalogName}.s3.secret-key=${minioSk} \
@@ -203,6 +227,7 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
try {
sql """switch internal"""
sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ // Paimon requires a catalog lock for safe concurrent snapshot commits
on object storage.
sql """
CREATE CATALOG ${catalogName} PROPERTIES (
'type' = 'paimon',
@@ -214,6 +239,7 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
'paimon.jdbc.driver_class' = 'org.postgresql.Driver',
'paimon.jdbc.user' = 'postgres',
'paimon.jdbc.password' = '123456',
+ 'paimon.lock.enabled' = 'true',
's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.access_key' = '${minioAk}',
's3.secret_key' = '${minioSk}',
@@ -347,9 +373,192 @@ suite("test_paimon_jdbc_catalog", "p0,external") {
["_ROW_ID", "_SEQUENCE_NUMBER"],
1
)
+
+ // Append writers cover both independent partitions and
snapshot-isolated writes
+ // to the same partition. Every successful transaction must publish
one snapshot.
+ sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_append"""
+ sql """
+ CREATE TABLE ${dbName}.paimon_jdbc_concurrent_append (
+ id BIGINT,
+ writer_id INT,
+ payload STRING,
+ pt STRING
+ ) ENGINE=paimon
+ PARTITION BY (pt) ()
+ PROPERTIES (
+ 'bucket' = '-1',
+ 'write-only' = 'true'
+ )
+ """
+
+ long appendSnapshots = (sql """
+ SELECT COUNT(*) FROM paimon_jdbc_concurrent_append\$snapshots
+ """)[0][0] as long
+ runConcurrent("paimon-jdbc-append-left", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_append
+ SELECT number, 1, concat('left-', number), 'left'
+ FROM numbers('number' = '128')
+ """
+ }, "paimon-jdbc-append-right", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_append
+ SELECT number + 1000, 2, concat('right-', number), 'right'
+ FROM numbers('number' = '128')
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_append"""
+ assertEquals([
+ ["left", 128L, 128L, 8128L],
+ ["right", 128L, 128L, 136128L]
+ ], sql("""
+ 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)
+
+ runConcurrent("paimon-jdbc-same-partition-left", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_append
+ SELECT number + 2000, 3, concat('same-left-', number), 'same'
+ FROM numbers('number' = '64')
+ """
+ }, "paimon-jdbc-same-partition-right", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_append
+ SELECT number + 3000, 4, concat('same-right-', number), 'same'
+ FROM numbers('number' = '64')
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_append"""
+ assertEquals([[128L, 128L, 324032L]], sql("""
+ 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)
+
+ // Fixed-bucket deduplication may expose either value, but it must
preserve
+ // primary-key uniqueness and publish both successful transactions.
+ sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_pk"""
+ sql """
+ CREATE TABLE ${dbName}.paimon_jdbc_concurrent_pk (
+ id INT,
+ payload STRING
+ ) ENGINE=paimon
+ PROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'deduplicate'
+ )
+ """
+
+ long pkSnapshots = (sql """
+ SELECT COUNT(*) FROM paimon_jdbc_concurrent_pk\$snapshots
+ """)[0][0] as long
+ runConcurrent("paimon-jdbc-pk-left", {
+ sql """
+ INSERT INTO ${catalogName}.${dbName}.paimon_jdbc_concurrent_pk
+ VALUES (1, 'left')
+ """
+ }, "paimon-jdbc-pk-right", {
+ sql """
+ INSERT INTO ${catalogName}.${dbName}.paimon_jdbc_concurrent_pk
+ VALUES (1, 'right')
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_pk"""
+ def pkRows = sql """SELECT id, payload FROM
paimon_jdbc_concurrent_pk"""
+ assertEquals(1, pkRows.size())
+ assertEquals(1, pkRows[0][0] as int)
+ assertTrue(["left", "right"].contains(pkRows[0][1].toString()))
+ assertEquals(pkSnapshots + 2L, (sql """
+ SELECT COUNT(*) FROM paimon_jdbc_concurrent_pk\$snapshots
+ """)[0][0] as long)
+
+ // Aggregation is a lost-update oracle because both deltas must remain
visible.
+ sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_aggregation"""
+ sql """
+ CREATE TABLE ${dbName}.paimon_jdbc_concurrent_aggregation (
+ id INT,
+ total BIGINT
+ ) ENGINE=paimon
+ PROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'merge-engine' = 'aggregation',
+ 'fields.total.aggregate-function' = 'sum'
+ )
+ """
+
+ runConcurrent("paimon-jdbc-aggregation-left", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_aggregation
+ VALUES (1, 10)
+ """
+ }, "paimon-jdbc-aggregation-right", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_aggregation
+ VALUES (1, 20)
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_aggregation"""
+ assertEquals([[1, 30L]], sql("""
+ SELECT id, total FROM paimon_jdbc_concurrent_aggregation
+ """))
+
+ // Dynamic bucket only permits multiple jobs when they own disjoint
partitions.
+ sql """DROP TABLE IF EXISTS paimon_jdbc_concurrent_dynamic"""
+ sql """
+ CREATE TABLE ${dbName}.paimon_jdbc_concurrent_dynamic (
+ id INT,
+ pt STRING,
+ payload STRING
+ ) ENGINE=paimon
+ PARTITION BY (pt) ()
+ PROPERTIES (
+ 'primary-key' = 'id,pt',
+ 'bucket' = '-1',
+ 'dynamic-bucket.target-row-num' = '32'
+ )
+ """
+
+ runConcurrent("paimon-jdbc-dynamic-left", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_dynamic
+ SELECT number, 'left', concat('left-', number)
+ FROM numbers('number' = '64')
+ """
+ }, "paimon-jdbc-dynamic-right", {
+ sql """
+ INSERT INTO
${catalogName}.${dbName}.paimon_jdbc_concurrent_dynamic
+ SELECT number + 1000, 'right', concat('right-', number)
+ FROM numbers('number' = '64')
+ """
+ })
+ sql """REFRESH TABLE paimon_jdbc_concurrent_dynamic"""
+ assertEquals([
+ ["left", 64L, 64L],
+ ["right", 64L, 64L]
+ ], sql("""
+ SELECT pt, COUNT(*), COUNT(DISTINCT id)
+ FROM paimon_jdbc_concurrent_dynamic
+ GROUP BY pt
+ ORDER BY pt
+ """))
} finally {
try {
sql """SWITCH ${catalogName}"""
+ 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"""
+ sql """DROP TABLE IF EXISTS
${dbName}.paimon_jdbc_concurrent_append"""
sql """DROP TABLE IF EXISTS
${dbName}.paimon_jdbc_row_tracking_tbl"""
sql """DROP TABLE IF EXISTS ${dbName}.paimon_jdbc_tbl"""
sql """DROP DATABASE IF EXISTS ${dbName} FORCE"""
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_key_dynamic_memory_negative.groovy
b/regression-test/suites/paimon_write/test_paimon_write_key_dynamic_memory_negative.groovy
new file mode 100644
index 00000000000..222562d73f8
--- /dev/null
+++
b/regression-test/suites/paimon_write/test_paimon_write_key_dynamic_memory_negative.groovy
@@ -0,0 +1,160 @@
+// 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.
+
+import java.sql.DriverManager
+import java.util.concurrent.atomic.AtomicReference
+
+suite("test_paimon_write_key_dynamic_memory_negative", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ // This opt-in case intentionally puts sustained pressure on the embedded
JVM.
+ String knownBugTestEnabled =
context.config.otherConfigs.get("enablePaimonKnownBugTest")
+ if (knownBugTestEnabled == null ||
!knownBugTestEnabled.equalsIgnoreCase("true")) {
+ logger.info("skip isolated Paimon known-bug resource regression")
+ return
+ }
+
+ long stressRows =
(context.config.otherConfigs.get("paimonKeyDynamicStressRows")
+ ?: "4000000").toLong()
+ long queryMemoryLimit = 128L * 1024 * 1024
+ long allowedJvmGrowth = queryMemoryLimit + 64L * 1024 * 1024
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_key_dynamic_memory_catalog"
+ String dbName = "test_pw_key_dynamic_memory_db"
+
+ def backendIdToIp = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToIp, backendIdToHttpPort)
+ def backendEndpoints = backendIdToIp.collectEntries { backendId, ip ->
+ [(backendId): [ip.toString(),
backendIdToHttpPort[backendId].toString()]]
+ }
+ assertFalse(backendEndpoints.isEmpty())
+ def heapUsed = {
+ backendEndpoints.collectEntries { backendId, endpoint ->
+ [(backendId): (get_be_metric(endpoint[0], endpoint[1],
+ "jvm_heap_size_bytes", "used") as long)]
+ }
+ }
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.t_key_dynamic_memory;
+ CREATE TABLE paimon.${dbName}.t_key_dynamic_memory (
+ pt STRING, id STRING, payload STRING
+ ) USING paimon
+ PARTITIONED BY (pt)
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '-1',
+ 'dynamic-bucket.target-row-num' = '10000',
+ 'dynamic-bucket.max-buckets' = '64',
+ 'write-buffer-size' = '16 mb',
+ 'page-size' = '64 kb',
+ 'write-buffer-spillable' = '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 {
+ sql """INSERT INTO t_key_dynamic_memory VALUES ('warmup', 'warmup',
'warmup')"""
+ sleep(3000)
+ def baseline = heapUsed()
+ def peak = new LinkedHashMap(baseline)
+ def writeFailure = new AtomicReference<Throwable>()
+ def activeStatement = new AtomicReference<java.sql.Statement>()
+
+ Thread writerThread = Thread.start("paimon-key-dynamic-memory-writer")
{
+ try (def connection =
DriverManager.getConnection(context.config.jdbcUrl,
+ context.config.jdbcUser, context.config.jdbcPassword);
+ def statement = connection.createStatement()) {
+ activeStatement.set(statement)
+ statement.execute("SET exec_mem_limit = ${queryMemoryLimit}")
+ statement.execute("SWITCH ${catalogName}")
+ statement.execute("USE ${dbName}")
+ statement.execute("""
+ INSERT INTO t_key_dynamic_memory
+ SELECT concat('p', CAST(number % 64 AS STRING)),
+ concat(lpad(CAST(number AS STRING), 20, '0'),
repeat('k', 76)),
+ repeat('v', 32)
+ FROM numbers("number" = "${stressRows}")
+ """)
+ } catch (Throwable t) {
+ writeFailure.set(t)
+ } finally {
+ activeStatement.set(null)
+ }
+ }
+
+ long deadline = System.currentTimeMillis() + 20L * 60 * 1000
+ while (writerThread.isAlive() && System.currentTimeMillis() <
deadline) {
+ sleep(1000)
+ heapUsed().each { backendId, used ->
+ peak[backendId] = Math.max(peak[backendId], used)
+ }
+ }
+ writerThread.join(10000)
+ if (writerThread.isAlive()) {
+ // Cancel the stress query before failing so a timeout cannot
leave its
+ // JDBC writer running after the regression suite has already
finished.
+ activeStatement.get()?.cancel()
+ writerThread.join(10000)
+ }
+ assertFalse(writerThread.isAlive(), "KEY_DYNAMIC stress insert did not
finish within 20 minutes")
+
+ def growth = peak.collectEntries { backendId, used ->
+ [(backendId): used - baseline[backendId]]
+ }
+ def failureMessages = []
+ Throwable failure = writeFailure.get()
+ while (failure != null &&
!failureMessages.contains(failure.toString())) {
+ failureMessages.add(failure.toString())
+ failure = failure.getCause()
+ }
+ String failureMessage = failureMessages.join(" caused by ")
+ logger.info("Paimon KEY_DYNAMIC memory result: rows=${stressRows},
baseline=${baseline}, "
+ + "peak=${peak}, growth=${growth}, failure=${failureMessage}")
+
+ // A valid query may be rejected by a memory limit, but the embedded
JVM
+ // must not be the component that exhausts memory outside Doris
accounting.
+ assertFalse(failureMessage.contains("OutOfMemoryError"),
+ "KEY_DYNAMIC write exhausted the embedded JVM:
${failureMessage}")
+ assertTrue(growth.values().every { delta -> delta <= allowedJvmGrowth
},
+ "KEY_DYNAMIC Java heap growth escaped the query memory limit: "
+ + "limit=${queryMemoryLimit},
allowed=${allowedJvmGrowth}, growth=${growth}")
+ } 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/paimon_write/test_paimon_write_snapshot_refs.groovy
new file mode 100644
index 00000000000..1184895f2ad
--- /dev/null
+++ b/regression-test/suites/paimon_write/test_paimon_write_snapshot_refs.groovy
@@ -0,0 +1,192 @@
+// 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_snapshot_refs", "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_snapshot_refs_catalog"
+ String dbName = "test_pw_snapshot_refs_db"
+ String tableName = "t_refs"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.${tableName};
+ CREATE TABLE paimon.${dbName}.${tableName} (
+ id INT,
+ payload STRING,
+ amount DECIMAL(18, 2),
+ event_time TIMESTAMP_NTZ
+ ) USING paimon
+ TBLPROPERTIES (
+ 'bucket' = '-1',
+ 'write-only' = 'true',
+ 'file.format' = 'parquet'
+ );
+ """
+
+ 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',
+ 'meta.cache.paimon.table.ttl-second' = '0'
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ try {
+ // Doris creates the snapshot which becomes the immutable tag and
branch base.
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (1, 'base', 10.25, '2026-07-01 10:11:12.123456')
+ """
+ long baselineSnapshot = (sql """
+ SELECT MAX(snapshot_id) FROM ${tableName}\$snapshots
+ """)[0][0] as long
+
+ spark_paimon """REFRESH TABLE paimon.${dbName}.${tableName}"""
+ spark_paimon_multi """
+ CALL paimon.sys.create_tag(
+ table => '${dbName}.${tableName}',
+ tag => 'baseline_tag',
+ snapshot => ${baselineSnapshot}
+ );
+ CALL paimon.sys.create_branch(
+ '${dbName}.${tableName}',
+ 'audit_branch',
+ 'baseline_tag'
+ );
+ """
+
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (2, 'latest', 20.50, '2026-07-02 10:11:12.654321')
+ """
+ sql """refresh table ${tableName}"""
+
+ def baseline = [[1, "base", "10.25", "2026-07-01 10:11:12.123456"]]
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName} FOR VERSION AS OF ${baselineSnapshot}
+ ORDER BY id
+ """))
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName}@tag(baseline_tag)
+ ORDER BY id
+ """))
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName}@branch(audit_branch)
+ ORDER BY id
+ """))
+
+ // A historical source relation must keep its own schema/snapshot while
+ // the sink is rebound to the latest writable table generation.
+ sql """
+ INSERT INTO ${tableName}
+ SELECT id + 100, concat(payload, '-snapshot-copy'), amount + 1,
event_time
+ FROM ${tableName} FOR VERSION AS OF ${baselineSnapshot}
+ """
+ sql """refresh table ${tableName}"""
+ assertEquals([[1, "base"], [2, "latest"], [101,
"base-snapshot-copy"]], sql("""
+ SELECT id, payload FROM ${tableName} ORDER BY id
+ """))
+
+ // Branches are valid sources even though Doris does not currently
expose
+ // a Paimon branch sink.
+ sql """
+ INSERT INTO ${tableName}
+ SELECT id + 200, concat(payload, '-branch-copy'), amount + 2,
event_time
+ FROM ${tableName}@branch(audit_branch)
+ """
+ sql """refresh table ${tableName}"""
+ assertEquals([[1, "base"], [2, "latest"], [101, "base-snapshot-copy"],
+ [201, "base-branch-copy"]], sql("""
+ SELECT id, payload FROM ${tableName} ORDER BY id
+ """))
+
+ // Keep this unsupported boundary explicit. A rejected branch sink must
+ // not fall back to the main branch or mutate the referenced branch.
+ test {
+ sql """
+ INSERT INTO ${tableName}@branch(audit_branch)
+ VALUES (9, 'branch-write', 9.00, '2026-07-09 00:00:00')
+ """
+ exception "Only support insert data into iceberg table's branch"
+ }
+ test {
+ sql """
+ INSERT OVERWRITE TABLE ${tableName}@branch(audit_branch)
+ VALUES (9, 'branch-overwrite', 9.00, '2026-07-09 00:00:00')
+ """
+ exception "Only support insert overwrite into iceberg table's
branch"
+ }
+
+ assertEquals(4L, (sql """SELECT COUNT(*) FROM ${tableName}""")[0][0]
as long)
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName}@tag(baseline_tag)
+ ORDER BY id
+ """))
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName}@branch(audit_branch)
+ ORDER BY id
+ """))
+ assertEquals(baseline, sql("""
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName} FOR VERSION AS OF ${baselineSnapshot}
+ ORDER BY id
+ """))
+
+ spark_paimon """REFRESH TABLE paimon.${dbName}.${tableName}"""
+ def sparkRows = spark_paimon """
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, 'yyyy-MM-dd HH:mm:ss.SSSSSS')
+ FROM paimon.${dbName}.${tableName}
+ ORDER BY id
+ """
+ def dorisRows = sql """
+ SELECT id, payload, CAST(amount AS STRING),
+ DATE_FORMAT(event_time, '%Y-%m-%d %H:%i:%s.%f')
+ FROM ${tableName}
+ ORDER BY id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_source_models.groovy
b/regression-test/suites/paimon_write/test_paimon_write_source_models.groovy
new file mode 100644
index 00000000000..64f4272c1c3
--- /dev/null
+++ b/regression-test/suites/paimon_write/test_paimon_write_source_models.groovy
@@ -0,0 +1,294 @@
+// 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_source_models", "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_source_models_catalog"
+ String dbName = "test_pw_source_models_db"
+ String internalDb = "test_pw_source_models_internal_db"
+
+ sql """drop database if exists internal.${internalDb} force"""
+ sql """create database internal.${internalDb}"""
+
+ // Keep the source layouts deliberately different. The sink must consume
the
+ // source query result, not raw source rows hidden by each OLAP table
model.
+ sql """
+ create table internal.${internalDb}.source_duplicate (
+ id int,
+ category varchar(20),
+ amount bigint
+ )
+ duplicate key(id)
+ distributed by random buckets 3
+ properties ("replication_num" = "1")
+ """
+ sql """
+ insert into internal.${internalDb}.source_duplicate values
+ (1, 'A', 10),
+ (1, 'A', 11),
+ (2, null, 20)
+ """
+
+ sql """
+ create table internal.${internalDb}.source_unique_mow (
+ id int,
+ category varchar(20),
+ amount bigint
+ )
+ unique key(id, category)
+ partition by list(category) (
+ partition p_ab values in ('A', 'B'),
+ partition p_null values in (null)
+ )
+ distributed by hash(id) buckets auto
+ properties (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true"
+ )
+ """
+ sql """
+ insert into internal.${internalDb}.source_unique_mow values
+ (10, 'A', 100), (11, null, 110)
+ """
+ sql """
+ insert into internal.${internalDb}.source_unique_mow values
+ (10, 'A', 101)
+ """
+
+ sql """
+ create table internal.${internalDb}.source_unique_mor (
+ id int,
+ category varchar(20),
+ amount bigint
+ )
+ unique key(id)
+ partition by range(id) (
+ partition p_lt_20 values less than (20),
+ partition p_max values less than maxvalue
+ )
+ distributed by hash(id) buckets 2
+ properties (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false"
+ )
+ """
+ sql """
+ insert into internal.${internalDb}.source_unique_mor values
+ (20, 'C', 200), (21, 'D', 210)
+ """
+ sql """
+ insert into internal.${internalDb}.source_unique_mor values
+ (20, 'C', 201)
+ """
+
+ sql """
+ create table internal.${internalDb}.source_aggregate (
+ id int,
+ category varchar(20),
+ amount bigint sum
+ )
+ aggregate key(id, category)
+ partition by range(id) (
+ partition p_lt_40 values less than (40),
+ partition p_max values less than maxvalue
+ )
+ distributed by hash(id, category) buckets 4
+ properties ("replication_num" = "1")
+ """
+ sql """
+ insert into internal.${internalDb}.source_aggregate values
+ (30, 'E', 300),
+ (30, 'E', 3),
+ (31, 'F', 310)
+ """
+
+ sql """
+ create table internal.${internalDb}.source_complex (
+ id int,
+ metrics array<decimal(10, 2)>,
+ attributes map<string, int>,
+ profile struct<name:string, active:boolean>,
+ flags array<boolean>,
+ nested_payload map<string, array<struct<score:int, label:string>>>,
+ event_date date,
+ event_time datetime(6)
+ )
+ duplicate key(id)
+ distributed by hash(id) buckets 3
+ properties ("replication_num" = "1")
+ """
+ sql """
+ insert into internal.${internalDb}.source_complex values
+ (
+ 1,
+ array(cast(1.25 as decimal(10, 2)), cast(null as decimal(10,
2))),
+ map('alpha', 10, 'nullable', null),
+ named_struct('name', 'alice', 'active', true),
+ array(true, false, cast(null as boolean)),
+ map('term', array(
+ named_struct('score', 90, 'label', 'good'),
+ named_struct('score', cast(null as int), 'label', null)
+ )),
+ date '2024-02-29',
+ timestamp '2024-02-29 12:34:56.123456'
+ ),
+ (
+ 2,
+ array(),
+ map(),
+ named_struct('name', cast(null as string),
+ 'active', cast(null as boolean)),
+ array(),
+ map('empty', array()),
+ date '1970-01-01',
+ timestamp '1970-01-01 00:00:00.000001'
+ ),
+ (3, null, null, null, null, null, null, null)
+ """
+
+ spark_paimon_multi """
+ SET spark.sql.timestampType=TIMESTAMP_NTZ;
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+
+ DROP TABLE IF EXISTS paimon.${dbName}.source_model_sink;
+ CREATE TABLE paimon.${dbName}.source_model_sink (
+ source_model STRING NOT NULL,
+ id INT,
+ category STRING,
+ amount BIGINT
+ ) USING paimon
+ PARTITIONED BY (source_model)
+ TBLPROPERTIES ('file.format' = 'parquet');
+
+ DROP TABLE IF EXISTS paimon.${dbName}.complex_sink;
+ CREATE TABLE paimon.${dbName}.complex_sink (
+ id INT,
+ metrics ARRAY<DECIMAL(10, 2)>,
+ attributes MAP<STRING, INT>,
+ profile STRUCT<name:STRING, active:BOOLEAN>,
+ flags ARRAY<BOOLEAN>,
+ nested_payload MAP<STRING, ARRAY<STRUCT<score:INT, label:STRING>>>,
+ event_date DATE,
+ event_time TIMESTAMP_NTZ
+ ) USING paimon
+ TBLPROPERTIES ('file.format' = 'orc');
+ """
+
+ 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 {
+ sql """
+ insert into source_model_sink
+ select 'duplicate', id, category, amount
+ from internal.${internalDb}.source_duplicate
+ """
+ sql """
+ insert into source_model_sink
+ select 'unique_mow', id, category, amount
+ from internal.${internalDb}.source_unique_mow
+ """
+ sql """
+ insert into source_model_sink
+ select 'unique_mor', id, category, amount
+ from internal.${internalDb}.source_unique_mor
+ """
+ sql """
+ insert into source_model_sink
+ select 'aggregate', id, category, amount
+ from internal.${internalDb}.source_aggregate
+ """
+
+ def sourceRows = sql """
+ select 'duplicate', id, category, amount
+ from internal.${internalDb}.source_duplicate
+ union all
+ select 'unique_mow', id, category, amount
+ from internal.${internalDb}.source_unique_mow
+ union all
+ select 'unique_mor', id, category, amount
+ from internal.${internalDb}.source_unique_mor
+ union all
+ select 'aggregate', id, category, amount
+ from internal.${internalDb}.source_aggregate
+ order by 1, 2, 3, 4
+ """
+ def sinkRows = sql """
+ select source_model, id, category, amount
+ from source_model_sink
+ order by 1, 2, 3, 4
+ """
+ assertEquals(sourceRows, sinkRows)
+ assertEquals(4L,
+ (sql """select count(*) from
source_model_sink\$snapshots""")[0][0] as long)
+
+ def sparkModelRows = spark_paimon """
+ select source_model, id, category, amount
+ from paimon.${dbName}.source_model_sink
+ order by source_model, id, category, amount
+ """
+ assertSparkDorisResultEquals(sparkModelRows, sinkRows)
+
+ // Complex values now cross the OLAP scanner and an INSERT SELECT
+ // projection before reaching the Paimon Arrow writer.
+ sql """
+ insert into complex_sink
+ select id, metrics, attributes, profile, flags, nested_payload,
+ event_date, event_time
+ from internal.${internalDb}.source_complex
+ """
+ def complexRows = sql """
+ select id, metrics, attributes, profile, flags, nested_payload,
+ event_date, event_time
+ from complex_sink
+ order by id
+ """
+ def sparkComplexRows = spark_paimon """
+ select id, metrics, attributes, profile, flags, nested_payload,
+ event_date, event_time
+ from paimon.${dbName}.complex_sink
+ order by id
+ """
+ assertSparkDorisResultEquals(sparkComplexRows, complexRows)
+ assertEquals(3L, complexRows.size() as long)
+ assertEquals(1L,
+ (sql """select count(*) from complex_sink\$snapshots""")[0][0]
as long)
+ } finally {
+ sql """switch internal"""
+ sql """drop catalog if exists ${catalogName}"""
+ sql """drop database if exists internal.${internalDb} force"""
+ }
+}
diff --git
a/regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy
b/regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy
new file mode 100644
index 00000000000..b13fdd317e6
--- /dev/null
+++
b/regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy
@@ -0,0 +1,154 @@
+// 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.
+
+// Http is a framework utility class, not an injected Suite DSL property.
+import org.apache.doris.regression.util.Http
+
+suite("test_paimon_write_thread_lifecycle", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ // Keep the reproducer opt-in until attached JNI writer threads are
released.
+ String knownBugTestEnabled =
context.config.otherConfigs.get("enablePaimonKnownBugTest")
+ if (knownBugTestEnabled == null ||
!knownBugTestEnabled.equalsIgnoreCase("true")) {
+ logger.info("skip isolated Paimon known-bug thread regression")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_thread_lifecycle_catalog"
+ String dbName = "test_pw_thread_lifecycle_db"
+
+ def backendIdToIp = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToIp, backendIdToHttpPort)
+ def backendEndpoints = backendIdToIp.collectEntries { backendId, ip ->
+ [(backendId): [ip.toString(),
backendIdToHttpPort[backendId].toString()]]
+ }
+ assertFalse(backendEndpoints.isEmpty())
+
+ def jvmThreadCounts = {
+ backendEndpoints.collectEntries { backendId, endpoint ->
+ [(backendId): (get_be_metric(endpoint[0], endpoint[1],
"jvm_thread", "count") as long)]
+ }
+ }
+ def processThreadCounts = {
+ backendEndpoints.collectEntries { backendId, endpoint ->
+ def body =
Http.GET("http://${endpoint[0]}:${endpoint[1]}/api/be_process_thread_num";,
+ false, false).toString()
+ def item = parseJson(body).find { row -> row[0].toString() ==
"total_thread_count" }
+ assertNotNull(item)
+ [(backendId): item[1].toString().toLong()]
+ }
+ }
+ def minimumThreadCounts = { counter ->
+ def minimums = null
+ for (int sample = 0; sample < 5; sample++) {
+ def counts = counter()
+ minimums = minimums == null ? counts : counts.collectEntries {
backendId, count ->
+ [(backendId): Math.min(minimums[backendId], count)]
+ }
+ sleep(1000)
+ }
+ minimums
+ }
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.t_thread_lifecycle;
+ CREATE TABLE paimon.${dbName}.t_thread_lifecycle (
+ id BIGINT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '3',
+ 'bucket-key' = 'id',
+ 'num-sorted-run.compaction-trigger' = '2',
+ 'target-file-size' = '1 gb'
+ );
+ """
+
+ 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 {
+ // Warm all writer and metrics paths before taking the baseline. This
keeps
+ // one-time JVM attachment and SDK class initialization out of the
leak oracle.
+ for (int round = 0; round < 2; round++) {
+ sql """
+ INSERT INTO t_thread_lifecycle
+ SELECT number + ${round * 1000}, repeat('w', 32)
+ FROM numbers("number" = "1000")
+ """
+ }
+ sleep(3000)
+
+ def jvmBefore = minimumThreadCounts(jvmThreadCounts)
+ def processBefore = minimumThreadCounts(processThreadCounts)
+ logger.info("Paimon thread baseline: jvm=${jvmBefore},
process=${processBefore}")
+
+ def writePhase = { int firstRound ->
+ for (int round = firstRound; round < firstRound + 12; round++) {
+ sql """
+ INSERT INTO t_thread_lifecycle
+ SELECT number + ${round * 1000}, repeat('x', 32)
+ FROM numbers("number" = "1000")
+ """
+ }
+ }
+
+ def jvmPhases = []
+ def processPhases = []
+ for (int phase = 0; phase < 4; phase++) {
+ writePhase(2 + phase * 12)
+ sleep(5000)
+ jvmPhases.add(minimumThreadCounts(jvmThreadCounts))
+ processPhases.add(minimumThreadCounts(processThreadCounts))
+ logger.info("Paimon thread phase ${phase + 1}:
jvm=${jvmPhases[-1]}, "
+ + "process=${processPhases[-1]}")
+ }
+
+ assertEquals(50000L,
+ (sql """SELECT COUNT(*) FROM t_thread_lifecycle""")[0][0] as
long)
+
+ backendEndpoints.keySet().each { backendId ->
+ // Equal steady-state phases must reuse or detach JNI writer
threads.
+ // Comparing later phases excludes the cold shared writer-pool
expansion.
+ assertTrue(jvmPhases[-1][backendId] <= jvmPhases[0][backendId] + 4,
+ "JVM threads kept growing on backend ${backendId}: phases="
+ + jvmPhases.collect { counts -> counts[backendId]
})
+ }
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
