This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 775638dc28f [fix](regression) Scope Hive split and Paimon writer
thread checks (#69230)
775638dc28f is described below
commit 775638dc28fcdf46476205c3faad0799a42fe6d0
Author: Gabriel <[email protected]>
AuthorDate: Mon Oct 19 12:23:27 2026 +0800
[fix](regression) Scope Hive split and Paimon writer thread checks (#69230)
### What problem does this PR solve?
`test_hive_compress_type_large_data` compares exact file and split
counts against streaming EXPLAIN estimates. A cold catalog can report
`(approximate)inputSplitNum=1024` even though the fixture contains 16
files. Use statement-local
`SET_VAR(enable_external_table_batch_mode=false)` for both exact plan
checks. Keep the synchronous and explicitly forced streaming result
checks.
`test_paimon_write_thread_lifecycle` treats growth in all BE JVM threads
as a writer leak. Existing native blocking workers attach to the JVM on
their first JNI call and remain attached until the worker exits. A
stable global baseline is therefore not guaranteed, even when every
writer closes its executor correctly. Check the writer-owned
`doris-paimon-` threads through the existing process-thread endpoint
instead, accounting for Linux's 15-byte thread-name limit. Require them
to disappear within 30 seconds after each of ten write phases, and
retain total process-thread counts in diagnostics. This detects even one
surviving writer thread without increasing a tolerance or moving a
baseline.
### Release note
None
### Check List (For Author)
- Test
- [x] Regression test
- [x] Manual test (details below)
- Behavior changed:
- [x] No production behavior changes; regression assertions now measure
the intended resources.
- Does this need documentation?
- [x] No.
Local validation used the original DEBUG/ASAN build
`6c27d4d8b0b4649f6517e15792fba672a54e2935`, an isolated FE/BE, Hive
3/HDFS, Spark 4/Paimon, and MinIO:
- **Before:** ran both unmodified suites. Hive failed the exact split
assertion with the approximate 1024 estimate. With
`parallel_pipeline_task_num=4`, Paimon failed bounded warm-up with JVM
counts `451, 497, 525, 587, 606, 676`. Thread dumps mapped the growing
attached threads to `p_normal_blocki`; writer compaction threads had
exited. The failure reproduces the invalid global-count assumption,
through the warm-up assertion rather than the later baseline assertion.
- **After:** restarted the BE, kept parallelism at 4, and ran both fixed
suites through `run-regression-test.sh`; **2 suites passed, 0 failed, 0
skipped**. Paimon executed 120 inserts, verified 120,000 rows, and
passed all ten writer-thread checks. Its attached blocking-worker count
still grew from 54 to 368, so passing does not depend on a pre-warmed
JVM. BE logs confirm the compaction executor actually ran.
- Hive used the original fixture: 16 files, 734,675,596 bytes, eight
partitions. Both plan checks and all synchronous/forced-streaming result
checks passed.
- Verified Paimon's actual `ExecutorThreadFactory` name truncates to
`doris-paimon-co` on JDK 17/Linux. `git diff --check` passed.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../test_paimon_write_thread_lifecycle.groovy | 103 ++++++---------------
.../hive/test_hive_compress_type_large_data.groovy | 5 +-
2 files changed, 29 insertions(+), 79 deletions(-)
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
index daa60ccefcb..5e361e060ee 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
@@ -38,30 +38,21 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon,nonConcurrent")
}
assertFalse(backendEndpoints.isEmpty())
- def jvmThreadCounts = {
- backendEndpoints.collectEntries { backendId, endpoint ->
- [(backendId): (get_be_metric(endpoint[0], endpoint[1],
"jvm_thread", "count") as long)]
- }
- }
- def processThreadCounts = {
+ // Native execution workers remain attached to the JVM until they exit, so
total JVM
+ // thread growth does not identify a writer leak. Track the writer-owned
executor instead.
+ // Linux /proc comm names are truncated to 15 bytes; this prefix also
matches
+ // PaimonJniWriter's "doris-paimon-compaction-thread-N" after truncation.
+ def writerThreadCounts = {
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)
+ def rows = parseJson(body)
+ def total = rows.find { row -> row[0].toString() ==
"total_thread_count" }
+ assertNotNull(total)
+ def writers = rows.findAll { row ->
row[0].toString().startsWith("doris-paimon-") }
+ .collectEntries { row -> [(row[0].toString()):
row[1].toString().toLong()] }
+ [(backendId): [total: total[1].toString().toLong(), writers:
writers]]
}
- minimums
}
spark_paimon_multi """
@@ -95,74 +86,32 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon,nonConcurrent")
sql """use ${dbName}"""
try {
- def jvmPhases = []
- def processPhases = []
- int nextRound = 0
- def writeAndSamplePhase = {
- for (int round = nextRound; round < nextRound + 12; round++) {
+ int roundsPerPhase = 12
+ int phaseCount = 10
+ for (int phase = 0; phase < phaseCount; phase++) {
+ for (int round = phase * roundsPerPhase; round < (phase + 1) *
roundsPerPhase; round++) {
sql """
INSERT INTO t_thread_lifecycle
SELECT number + ${round * 1000}, repeat('x', 32)
FROM numbers("number" = "1000")
"""
}
- nextRound += 12
- sleep(5000)
- jvmPhases.add(minimumThreadCounts(jvmThreadCounts))
- processPhases.add(minimumThreadCounts(processThreadCounts))
- logger.info("Paimon thread phase ${jvmPhases.size()}:
jvm=${jvmPhases[-1]}, "
- + "process=${processPhases[-1]}")
- }
- // Native workers can initialize after the first writes. Require a
stable window instead
- // of treating a single increase followed by a plateau as a
continuously leaking writer.
- def stableWindow = { phases, long tolerance ->
- backendEndpoints.keySet().every { backendId ->
- def counts = phases.takeRight(3).collect { sample ->
sample[backendId] as long }
- counts.max() - counts.min() <= tolerance
+ // Each writer owns its compaction executor and closes it with the
sink. Require all
+ // of those threads to exit after every phase, without a baseline
that can absorb leaks.
+ def counts = [:]
+ try {
+ awaitUntil(30) {
+ counts = writerThreadCounts()
+ counts.values().every { sample ->
sample.writers.values().every { count -> count == 0 } }
+ }
+ } finally {
+ logger.info("Paimon writer threads after phase ${phase + 1}:
${counts}")
}
}
- boolean warmedUp = false
- for (int phase = 0; phase < 6; phase++) {
- writeAndSamplePhase()
- if (jvmPhases.size() >= 3 && stableWindow(jvmPhases, 2L)
- && stableWindow(processPhases, 4L)) {
- warmedUp = true
- break
- }
- }
- assertTrue(warmedUp, "Thread counts did not stabilize during bounded
warm-up: "
- + "jvm=${jvmPhases}, process=${processPhases}")
- int warmupPhases = jvmPhases.size()
- def baseline = { phases ->
- backendEndpoints.keySet().collectEntries { backendId ->
- [(backendId): phases.takeRight(3).collect { sample ->
sample[backendId] as long }.min()]
- }
- }
- def jvmBaseline = baseline(jvmPhases)
- def processBaseline = baseline(processPhases)
- // Freeze the low-water marks before additional writes. Moving the
baseline on each phase
- // would hide slow, persistent leaks that fit within the per-window
sampling tolerance.
- for (int phase = 0; phase < 6; phase++) {
- writeAndSamplePhase()
- }
-
- assertEquals(nextRound * 1000L,
+ assertEquals(phaseCount * roundsPerPhase * 1000L,
(sql """SELECT COUNT(*) FROM t_thread_lifecycle""")[0][0] as
long)
-
- // Use the final two low-water marks to tolerate one isolated
background-thread spike.
- // Six validation phases also expose one new process thread per phase
starting after warm-up.
- backendEndpoints.keySet().each { backendId ->
- def jvmCounts = jvmPhases.collect { sample -> sample[backendId] as
long }
- def processCounts = processPhases.collect { sample ->
sample[backendId] as long }
- assertTrue(jvmCounts.drop(warmupPhases).takeRight(2).min() <=
jvmBaseline[backendId] + 2,
- "JVM threads grew after warm-up on backend ${backendId}: "
- + "baseline=${jvmBaseline[backendId]},
phases=${jvmCounts}")
- assertTrue(processCounts.drop(warmupPhases).takeRight(2).min() <=
processBaseline[backendId] + 4,
- "Process threads grew after warm-up on backend
${backendId}: "
- + "baseline=${processBaseline[backendId]},
phases=${processCounts}")
- }
} finally {
sql """drop catalog if exists ${catalogName}"""
}
diff --git
a/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
b/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
index 042b5439608..e445aada5a2 100644
---
a/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
+++
b/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
@@ -38,7 +38,8 @@ suite("test_hive_compress_type_large_data", "p2,external") {
// All 16 fixture files fit in that size, so the default must preserve
one range per file.
sql """set file_split_size=0"""
explain {
- sql("select count(*) from test_compress_partitioned")
+ // Streaming EXPLAIN reports estimates before file enumeration,
not exact split counts.
+ sql("select /*+ SET_VAR(enable_external_table_batch_mode=false) */
count(*) from test_compress_partitioned")
contains "inputSplitNum=16, totalFileSize=734675596, scanRanges=16"
contains "partition=8/8"
}
@@ -52,7 +53,7 @@ suite("test_hive_compress_type_large_data", "p2,external") {
sql """set file_split_size=8388608"""
explain {
- sql("select count(*) from test_compress_partitioned")
+ sql("select /*+ SET_VAR(enable_external_table_batch_mode=false) */
count(*) from test_compress_partitioned")
// Compression determines which files can split; require actual
splitting without
// coupling this correctness test to the legacy planner's exact
split count.
check { result ->
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
