github-actions[bot] commented on code in PR #66325: URL: https://github.com/apache/doris/pull/66325#discussion_r3689804318
########## 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 = { Review Comment: [P1] Run this heap oracle in an actually isolated suite. This case is tagged only p0,external,paimon, so when the opt-in flag is enabled it stays in the NORMAL pool (default suiteParallel=10) while sampling whole-JVM used heap on every BE. Allocations or GC from concurrent suites can exceed the 192 MiB allowance or inflate the baseline and mask a leak. Add nonConcurrent/dedicated invocation and scope measurements to the BEs/fragments that ran the sink or a writer-owned high-water metric. ########## regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy: ########## @@ -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), Review Comment: [P1] Bound and clean up both workers on every exit from this helper. If readiness times out, this assertion throws before start.countDown(), leaving a started action blocked; if left.get() throws, right.get() is skipped; and after dispatch both get() calls are unbounded, so a commit-lock deadlock can hang the runner. Release the latch in an unconditional path, wait for both workers with a deadline, and cancel/terminate both before rethrowing or starting catalog cleanup. Also add a commit-adjacent synchronization or observable in-flight check if these cases are meant to prove overlapping rather than merely simultaneous SQL dispatch. ########## 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() Review Comment: [P1] Keep controller-owned cleanup until this JDBC worker has actually stopped. cancel() may throw or may not terminate the query within ten seconds, and the connection is only owned inside this non-daemon thread; the following assertion can therefore fail while the worker/query remains alive and races catalog cleanup or hangs the runner. Retain and close the connection/statement from an unconditional cleanup path, preserving cancellation errors, and ensure the worker cannot outlive the suite. ########## 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) Review Comment: [P1] Compare the complex sink with the original OLAP values. Both result sets here read complex_sink, so a writer-side conversion bug can persist the same corrupted value that Spark and Doris later agree on. Capture the internal source rows or use explicit expected values, and compare those against the sink in addition to this cross-reader check. ########## 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 Review Comment: [P1] Make the latest sink generation observably different from the historical source. The baseline and current table keep the same four-column schema, so a sink accidentally reused from the historical FileStoreTable can still append these rows and satisfy every assertion. Evolve the main schema after creating the refs (for example, add a column with a default), then assert both the old reference schema and the latest sink/default behavior. ########## regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy: ########## @@ -0,0 +1,127 @@ +// 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_thread_lifecycle", "p0,external,paimon") { Review Comment: [P1] Do not run this process-wide thread-count oracle alongside normal parallel suites. This tag set is assigned to the NORMAL pool (default suiteParallel=10), while both sampled metrics include all BE/JVM threads, so unrelated suites can make a clean writer fail or mask a leak. At minimum isolate it with nonConcurrent; preferably poll a Paimon-specific thread prefix or lifecycle gauge on the BEs that executed the sinks. ########## 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"), Review Comment: [P1] Reject unexpected writer failures instead of treating every non-OOM error as success. A catalog, connection, analysis, or unsupported-path failure exits early with low heap growth and passes this regression without exercising KEY_DYNAMIC memory. Require either a successful write with a data/snapshot oracle or one specifically whitelisted Doris memory-limit outcome, and distinguish Java heap exhaustion from the tracked native-page error by message/cause rather than the exception class name alone. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
