Skip to content

Commit 710ae55

Browse files
morningmanclaude
morningman
and
claude
committed
[test](fluss) Check from the profile that the heavy JNI readers declare their heap
### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: test_fluss_jni_heap_admission compared every query's rows with enable_jni_heap_admission off and on. That agreement holds just as well if FE never sends jni_heap_bytes or BE never goes through the gate, so the feature could be disconnected while the suite stayed green. Two of the reads it named as declaring did not declare in this environment either: - PK_FULL ranges declare the change log after their bucket's kv snapshot. The environment takes a kv snapshot every ten seconds and nothing writes after init, so by the time the suite runs every snapshot covers its whole log: the PK_FULL reads of pk_basic and pk_part declare nothing (their profile shows JvmHeapDeclaredBytes 0.00). - The union read's lake half is read natively unless forced through JNI, so no paimon split declared. The suite now reads JvmHeapDeclaredBytes from the profile (the per-instance detail profile, fetched through the FE HTTP API as external_table_p0/cache/condition_cache_orc does) and asserts that the union read declares (its PK_TAIL tail), that the same read with force_jni_scanner declares at least a megabyte (a paimon JNI split declares a 1 MB dictionary page for every column of every file, the fluss tail here about a kilobyte), and that a log read declares nothing. The forced-JNI union read is also recorded and compared with the variable off. The header no longer says PK_FULL declares here; those reads stay for the agreement, and FlussJniHeapEstimateTest covers what a PK_FULL range declares. ### Release note None ### Check List (For Author) - Test: Regression test - test_fluss_jni_heap_admission passes against a local cluster whose FE and BE carry this PR's admission code (built from the branch it was split from) and the fluss 1.0.0 docker environment; the .out was regenerated with -forceGenOut, which adds only union_rows_jni, then checked by a normal run - With the variable off both union reads report JvmHeapDeclaredBytes 0.00, so the new assertions fail if the declarations stop reaching BE - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1 parent 1d89558 commit 710ae55

2 files changed

Lines changed: 89 additions & 7 deletions

File tree

‎regression-test/data/external_table_p0/fluss/test_fluss_jni_heap_admission.out‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,17 @@ enable_jni_heap_admission false false 0
2929
-- !union_count --
3030
9
3131

32+
-- !union_rows_jni --
33+
1 m1
34+
10 m10-hot
35+
2 m2-hot
36+
3 m3
37+
4 m4
38+
5 m5-lake
39+
6 m6
40+
8 m8
41+
9 m9
42+
3243
-- !log_rows --
3344
1 alice 10.10
3445
2 bob 20.20

‎regression-test/suites/external_table_p0/fluss/test_fluss_jni_heap_admission.groovy‎

Lines changed: 78 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,23 @@
1515
// specific language governing permissions and limitations
1616
// under the License.
1717

18+
import groovy.json.JsonSlurper
19+
1820
// enable_jni_heap_admission has the connectors declare, on every range whose JNI reader holds much of
1921
// BE's JVM heap, how much it will hold, and BE opens those readers only while what it admitted fits
2022
// its budget. It is off by default. Turned on it may make a reader wait for room, and nothing else: the
2123
// rows must come back exactly as they do with it off.
2224
//
23-
// The reads that declare are here: fluss primary-key buckets read whole (PK_FULL), and a union read of a
24-
// primary-key table, whose tail is a PK_TAIL range and whose lake half the paimon connector plans. A log
25-
// read declares nothing and is here as the case that must not change either. Each query is recorded
26-
// with the variable on, and compared with the same query with it off; the comparison stays in the code
27-
// because what it asserts is the agreement.
25+
// So two things are checked. Every query here returns with the variable on what it returns with it
26+
// off; the comparison stays in the code because what it asserts is the agreement. And the reads that
27+
// hold much do declare it, as the profile's JvmHeapDeclaredBytes shows: a union read of a primary-key
28+
// table, whose log tail is a PK_TAIL range, and the same read with force_jni_scanner, whose lake half is
29+
// then a paimon JNI split as well; a log read declares nothing.
30+
//
31+
// The primary-key buckets read whole (PK_FULL) declare nothing here. A PK_FULL range declares the change
32+
// log after its bucket's kv snapshot, and this environment snapshots every ten seconds while nothing
33+
// writes after init, so by the time this suite runs every snapshot covers its whole log. Those reads are
34+
// here for the agreement; FlussJniHeapEstimateTest covers what a PK_FULL range declares.
2835
//
2936
// Fixtures come from docker/thirdparties/docker-compose/fluss/sql/init.sql and init-lake-tail.sql, and
3037
// are static: this suite never writes.
@@ -44,8 +51,8 @@ suite("test_fluss_jni_heap_admission", "p0,external") {
4451
qt_default_off """show variables like 'enable_jni_heap_admission'"""
4552

4653
sql """drop catalog if exists ${catalogName}"""
47-
// required: a union read that quietly fell back to fluss alone would never plan a lake split, and
48-
// the paimon half of the declaration would go untested.
54+
// required: a union read that quietly fell back to fluss alone would plan neither a PK_TAIL range
55+
// nor a lake split, and neither declaration would be tested.
4956
sql """
5057
create catalog ${catalogName} properties (
5158
"type" = "fluss",
@@ -71,6 +78,55 @@ suite("test_fluss_jni_heap_admission", "p0,external") {
7178
assertEquals(off, on, "enable_jni_heap_admission changed the rows of: ${query}\noff=${off}\non=${on}")
7279
}
7380

81+
def feGet = { String path ->
82+
def conn = new URL("http://${context.config.feHttpAddress}${path}").openConnection()
83+
conn.setRequestMethod("GET")
84+
def encoding = Base64.getEncoder().encodeToString((context.config.feHttpUser + ":" +
85+
(context.config.feHttpPassword == null ? "" : context.config.feHttpPassword)).getBytes("UTF-8"))
86+
conn.setRequestProperty("Authorization", "Basic ${encoding}")
87+
return conn.getInputStream().getText()
88+
}
89+
long MB = 1024L * 1024
90+
def bytesPerUnit = ["": 1L, "B": 1L, "KB": 1024L, "MB": MB, "GB": 1024L * MB, "TB": 1024L * 1024 * MB]
91+
// The JvmHeapDeclaredBytes of `query`, a select read with profiling on, in bytes: one per scan
92+
// instance that read through JNI.
93+
def declaredHeapOf = { String query ->
94+
assertTrue(query.startsWith("select "), query)
95+
String token = UUID.randomUUID().toString()
96+
sql """set enable_profile = true"""
97+
sql(query.replaceFirst("select ", "select /* ${token} */ "))
98+
sql """set enable_profile = false"""
99+
String profileId = ""
100+
for (int attempt = 0; attempt < 20 && profileId == ""; attempt++) {
101+
def row = new JsonSlurper().parseText(feGet("/rest/v1/query_profile")).data.rows.find {
102+
it["Sql Statement"].toString().contains(token)
103+
}
104+
if (row == null) {
105+
Thread.sleep(300)
106+
} else {
107+
profileId = row["Profile ID"].toString()
108+
}
109+
}
110+
assertTrue(profileId != "", "no profile for: ${query}")
111+
// JNI counters are only in the per-instance DetailProfile, which BE reports a little after the
112+
// statement returns.
113+
List<Long> declared = []
114+
for (int attempt = 0; attempt < 20 && declared.isEmpty(); attempt++) {
115+
declared = feGet("/api/profile/text/?query_id=${profileId}").readLines()
116+
.findAll { it.contains("- JvmHeapDeclaredBytes: ") }
117+
.collect { line ->
118+
def m = line =~ /JvmHeapDeclaredBytes: ([0-9]+[.,][0-9]+) ?([KMGT]?B)?/
119+
assertTrue(m.find(), line)
120+
(long) (Double.parseDouble(m.group(1).replace(',', '.')) * bytesPerUnit[m.group(2) ?: ""])
121+
}
122+
if (declared.isEmpty()) {
123+
Thread.sleep(500)
124+
}
125+
}
126+
assertFalse(declared.isEmpty(), "no JvmHeapDeclaredBytes in the profile of: ${query}")
127+
return declared
128+
}
129+
74130
sql """set enable_jni_heap_admission = true"""
75131

76132
// --- primary-key buckets read whole -------------------------------------
@@ -84,10 +140,25 @@ suite("test_fluss_jni_heap_admission", "p0,external") {
84140
order_qt_union_rows """select id, name from lake_pk_multi"""
85141
order_qt_union_count """select count(*) from lake_pk_multi"""
86142
sameWithAdmissionOff("select id, name from lake_pk_multi order by id")
143+
// The tail is a PK_TAIL range, which declares the rows it will keep.
144+
def unionDeclared = declaredHeapOf("select id, name from lake_pk_multi order by id")
145+
assertTrue(unionDeclared.any { it > 0 }, "the union read declared no heap: ${unionDeclared}")
146+
147+
// --- the same through JNI: the lake half a paimon JNI split -------------
148+
// A paimon JNI split declares at least a dictionary page - a megabyte - for every column of every
149+
// file it reads, far more than this fixture's tail.
150+
sql """set force_jni_scanner = true"""
151+
order_qt_union_rows_jni """select id, name from lake_pk_multi"""
152+
sameWithAdmissionOff("select id, name from lake_pk_multi order by id")
153+
def jniDeclared = declaredHeapOf("select id, name from lake_pk_multi order by id")
154+
assertTrue(jniDeclared.any { it >= MB }, "the paimon JNI split of the union read declared no heap: ${jniDeclared}")
155+
sql """set force_jni_scanner = false"""
87156

88157
// --- a log read declares nothing ----------------------------------------
89158
order_qt_log_rows """select id, name, price from log_basic"""
90159
sameWithAdmissionOff("select id, name, price from log_basic order by id")
160+
def logDeclared = declaredHeapOf("select id, name, price from log_basic order by id")
161+
assertTrue(logDeclared.every { it == 0 }, "the log read declared heap: ${logDeclared}")
91162

92163
sql """set enable_jni_heap_admission = false"""
93164
sql """switch internal"""

0 commit comments

Comments
 (0)