Repository navigation
[1.3] Fix query partition memory retention and failure diagnostics - #19269
Conversation
JackieTien97
changed the title
JackieTien97
left a comment
JackieTien97
left a comment
There was a problem hiding this comment.
Reviewed at 856d4a0d82 (base 013622816a). I built base, 33277a6cf5, and this head and ran each as a 1C1D cluster (Windows, JDK 21, dn_thrift_max_frame_size=1048576, 64 devices with one point each). The inline numbers come from these clusters and from direct transport probes against the same jars.
Verified on this head
- The oversized partition request (
select s1 from root.db.** where time >= 0 and time < 3000000000000, ≈3.8 MB) now fails in 35 ms withFrame size (1048582) larger than protect max size (1048576)!. At33277a6cf5and on base it took 15.2 s (15 reconnect attempts) and reported "Fail to connect to any config node", even with a 3 s JDBC query timeout. - Compressed transport, two cases. At
33277a6cf5both escaped asIllegalArgumentException, and the flush case left the transport open. Both now raiseTTransportExceptionand close the transport:- a
maxCompressedLengthoverflow inflush()(forced 1.9 GB frame); - a varint length ≥ 2^31 in
readFrame().
- a
- A near-unbounded range (
time >= 0 and time <= 9223372036854775806) now fails with 1707 "Not enough memory while analyzing time partitions … requested this time 340.87 GB …, current free memory 772.99 MB". - The growth fix and the deadline checks look correct.
startTimeis only set byCoordinator, so standalone contexts are unaffected. - All 56 tests in the 7 touched test classes pass locally:
ConfigNodeClientTest2,MPPQueryContextTest7,QueryPartitionMemoryTest5,QueryTimePartitionTest16,LocalExecutionPlannerOperatorsMemoryTest12,AutoResizingBufferTest9,TElasticFramedTransportTest5.
Comments
- [P2]
Coordinator: the new[QueryFailed]WARN logs every analysis/planning failure with a stack trace. That includes user errors thatErrorHandlingUtilskeeps at INFO, and every other failure ends up logged twice. - [P3]
TElasticFramedTransport#write: buffer growth still overshoots the limit by 1.30–1.44×, and the error now reports the size at the crossing write instead of the request size.
Nit: sonar-codecov.yml. codecov has failed on the last five dev/1.3 pushes with the same spotless NoSuchMethodError (e.g. run 43390944744), so this JDK pin fixes the branch, not just this PR. It may deserve its own commit or PR so the fix survives if this PR is reworked.
| } catch (RuntimeException | Error e) { | ||
| // Analysis can fail before the execution is registered, so later error handling cannot look | ||
| // up its SQL in queryExecutionMap. Keep the SQL and query ID in the failure log here. | ||
| LOGGER.warn("[QueryFailed] queryId: {}, sql: {}", globalQueryId, sql, e); |
There was a problem hiding this comment.
[P2] This logs every analysis/planning failure at WARN with a stack trace, including user errors that ErrorHandlingUtils deliberately keeps at INFO
ErrorHandlingUtils#onQueryException already logs these failures:
- user-caused ones (
SQL_PARSE_ERROR,SEMANTIC_ERROR,NO_PERMISSION,QUERY_TIMEOUT, …) at INFO with no stack trace; - everything else with the stack.
On the client path that line already includes the SQL: Query Statement: "<sql>". executeStatement failed because …. This catch runs before that classification. DataNode log on this head, client statements via JDBC:
| statement | existing ErrorHandlingUtils line |
new [QueryFailed] line |
|---|---|---|
select count(s1), s1 from root.db.d0 → 701 |
INFO, no stack | WARN + SemanticException stack (23 frames) |
| oversized partition request → 305 | WARN + stack | WARN + the same stack again |
select s1 from root.db.d0 where time >= 0 and time <= 9223372036854775806 → 1707 |
WARN + stack | WARN + the same stack again |
The consequences:
- A dashboard that keeps sending an invalid query now writes a WARN stack trace on every execution.
- So does a query that hits the new analysis-time deadline check (
QueryTimeoutRuntimeException). - On the client path, every other failure that throws out of
execution()is logged twice.
If the goal is to tie the query ID to the SQL when a failure happens before registration, two options:
- Log this line without the throwable, since callers already log it.
- Skip the classes that
ErrorHandlingUtilstreats as user errors (at leastSemanticExceptionandQueryTimeoutRuntimeException), or log those at DEBUG.
assertFailureLogs in QueryPartitionMemoryTest would need to follow whichever option is chosen.
There was a problem hiding this comment.
Verified fixed at 827763a31d. I ran the same statements, and the DataNode log now shows:
| statement | ErrorHandlingUtils line |
[QueryFailed] line |
|---|---|---|
select count(s1), s1 from root.db.d0 → 701 |
INFO, no stack | INFO, no stack |
| oversized partition request → 305 | WARN + stack (once) | INFO, no stack |
select s1 from root.db.d0 where time >= 0 and time <= 9223372036854775806 → 1707 |
WARN + stack (once) | INFO, no stack |
User errors no longer produce a WARN, and each failure's stack trace is logged only once.
| close(); | ||
| throw new TTransportException( | ||
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); |
There was a problem hiding this comment.
[P3] The reported size is now just past the limit, not the request size
For the same 64-device query:
- base logged
Frame size (3811712) larger than protect max size (1048576)!; - this head logs
Frame size (1048582) larger than protect max size (1048576)!, which is also what the client now sees in the 305 message.
The old number showed how far over the limit the request was, which is what an operator needs in order to size dn_thrift_max_frame_size. The full size isn't known at this point anymore. Consider wording the message as "at least N bytes" and naming the setting.
There was a problem hiding this comment.
The new text fixes both points: it says "at least N" and it names a setting. However, TElasticFramedTransport is the shared service-rpc transport, so the hardcoded (dn_thrift_max_frame_size on DataNode) is wrong outside a DataNode:
- Session/JDBC clients: the limit is the client's own
Session.Builder#thriftMaxFrameSize(default 64 MiB,SessionConfig.DEFAULT_MAX_FRAME_SIZE) or JDBC'sthrift_max_frame_size. The rejection happens on the client, before anything is sent. - ConfigNode: there is no setting at all. The limit is the fixed
RpcUtils.THRIFT_FRAME_MAX_SIZE(536870912), because the ConfigNode never loads the DataNode config. For example, a ConfigNode whoseiotdb-system.propertiessetdn_thrift_max_frame_size=1048576still reportedprotect max size (536870912).
Repro: a plain Java client against this head's cluster (DataNode dn_thrift_max_frame_size=1048576), using new Session.Builder()...thriftMaxFrameSize(4194304) to insert a 1,000,000-row INT64 tablet:
base client jars: IoTDBConnectionException: org.apache.thrift.transport.TTransportException: Frame size (16000109) larger than protect max size (4194304)!
this head's client: IoTDBConnectionException: org.apache.thrift.transport.TTransportException: Serialized frame size is at least 8000077 bytes, exceeding the RPC frame-size limit (dn_thrift_max_frame_size on DataNode) of 4194304 bytes.
The message names a setting that didn't fire, and that setting's actual value here (1048576) doesn't match the number printed. A Session bulk insert over the 64 MiB default would point users to the server config, which can't fix a client-side rejection.
Suggestion: keep the library message generic, e.g. ... exceeding the configured Thrift max frame size of N bytes. If the DataNode hint is wanted, add it where the DataNode handles the failure (e.g. the CORRUPTED_DATA branch in ConfigNodeClient#executeRemoteCallWithRetry), or pass a limit description into the transport factory. The dn_thrift_max_frame_size assertions in TElasticFramedTransportTest and QueryPartitionMemoryTest would follow. My earlier suggestion to name the setting didn't account for the transport being shared.
There was a problem hiding this comment.
Verified fixed at c7003fff63. The message is now generic and names the limit that actually fired:
- DataNode → ConfigNode, oversized partition request (305, 15–42 ms):
Serialized frame size is at least 1048582 bytes, exceeding the configured Thrift max frame size of 1048576 bytes. - Plain Java Session client with
thriftMaxFrameSize(4194304)against a DataNode set to 1048576:Serialized frame size is at least 8000077 bytes, exceeding the configured Thrift max frame size of 4194304 bytes.The number now matches the client's own limit.
The tests now also assert that dn_thrift_max_frame_size is absent from the message.
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); | ||
| } | ||
| writeBuffer.write(buf, off, len); |
There was a problem hiding this comment.
[P3] Growth can still overshoot the limit before the rejection
writeBuffer.write grows the buffer by 1.5× without knowing getMaxWriteBufferSize(). The last growth step before the rejection can therefore allocate well past the limit. Measured on this head with 8-byte writes and a 1024-byte initial buffer:
| limit | capacity at rejection |
|---|---|
| 1 MiB | 1,513,342 (1.44×) |
| 64 MiB (DataNode default for heaps ≥ 4 GiB) | 87,266,902 (1.30×) |
Passing the cap down to the write buffer would keep the array at or below the limit, e.g. min(grow, maxSize) when growing.
There was a problem hiding this comment.
Verified fixed at 827763a31d. Capacity at rejection is now exactly the limit (1,048,576 and 67,108,864, i.e. 1.00×), and testIncrementalGrowthDoesNotExceedFrameLimit covers it.
Optional perf note: these per-write checks sit on Thrift's hottest path. I benchmarked serialization into a null sink, taking the median of 3 alternating JVM runs.
- Small fields: a 100k-slot
TDataPartitionReq, i.e. 4writecalls per 12-byte struct. - Large binary: a 1M-row
TSInsertTabletReq.
| base | this head | this head + fast path below | |
|---|---|---|---|
| small fields, ns/slot | 8.87–9.03 | 9.97–12.10 | 8.03–8.69 |
| large binary, ns/row | 1.69–1.71 | 1.67–1.74 | 1.69–1.73 |
Only small-field-heavy messages pay the cost, about 11% or more; large binary writes are unaffected. This commit makes capacity <= limit an invariant, so a write that fits in the current array can skip every limit check. The checks are only needed when the buffer must grow:
// AutoScalingBufferWriteTransport
boolean tryWriteWithinCapacity(byte[] toWrite, int off, int len) {
byte[] array = buf.array();
if (len > array.length - pos) {
return false;
}
System.arraycopy(toWrite, off, array, pos, len);
pos += len;
return true;
}
// TElasticFramedTransport#write, as the first statement
if (writeBuffer.tryWriteWithinCapacity(buf, off, len)) {
return;
}With only this change, the limit behavior in my probes is unchanged: a write is rejected after exactly limit bytes, capacity equals the limit, and the compressed paths behave as before. The "fast path" column above is this change.
There was a problem hiding this comment.
Verified at c7003fff63. The fast path doesn't change any limit behavior:
- Writes are rejected after exactly
limitbytes (1 MiB and 64 MiB), and capacity stays at 1.00× the limit. - Both compressed paths still raise
TTransportExceptionand close the transport.
Safety reasoning:
write(..., maxSize)is the only writer of the framedwriteBuffer, so capacity ≤ limit holds and a fitting write can't cross the limit.len > array.length - poscan't overflow.testBufferedWriteCannotBypassArrayLimitOnOverflowwould catch apos + lenrewrite.
Same benchmark as before (null sink, median ns over 3 alternating JVM runs):
| base | 827763a31d |
c7003fff63 |
|
|---|---|---|---|
| small fields, ns/slot | 8.80–9.02 | 9.89–10.29 | 8.37–8.50 |
| large binary, ns/row | 1.72–1.76 | 1.70–1.73 | 1.70–1.75 |
The small-field regression is gone; this head is now about 5% faster than base on that path. Large binary writes are unchanged.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## dev/1.3 #19269 +/- ##
=============================================
+ Coverage 45.56% 47.48% +1.92%
- Complexity 198 221 +23
=============================================
Files 3667 3678 +11
Lines 251340 252123 +783
Branches 31156 31360 +204
=============================================
+ Hits 114529 119728 +5199
+ Misses 136811 132395 -4416 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
JackieTien97
left a comment
JackieTien97
left a comment
There was a problem hiding this comment.
Re-checked at 827763a31d. I rebuilt this head, ran the same 1C1D cluster (dn_thrift_max_frame_size=1048576, 64 devices) and transport probes, and the PR's CI is green (19/19).
All three earlier comments are fixed; details are in the thread replies. One follow-up and one optional note:
- Error message: the new wording hardcodes the DataNode setting, but
TElasticFramedTransportis the sharedservice-rpctransport. Session/JDBC clients and the ConfigNode get a message naming a setting that didn't trigger the rejection. Details in the reply on the message thread. - Optional perf note: the per-write checks cost about 11% when serializing small-field-heavy messages. Now that capacity never exceeds the limit, a fast path removes this cost. Details in the reply on the growth thread.
FYI, pre-existing and outside this PR: SessionConnection#callWithRetry retries every TException, including this deterministic CORRUPTED_DATA rejection. With the defaults (maxRetryCount=60, 500 ms interval), the Session probe in the message thread takes 30.7 s to fail. This PR doesn't touch SessionConnection, so base has the same loop. This is the same pattern this PR fixed in ConfigNodeClient, so it could be a follow-up.
| } catch (RuntimeException | Error e) { | ||
| // Analysis can fail before the execution is registered, so later error handling cannot look | ||
| // up its SQL in queryExecutionMap. Keep the SQL and query ID in the failure log here. | ||
| LOGGER.warn("[QueryFailed] queryId: {}, sql: {}", globalQueryId, sql, e); |
There was a problem hiding this comment.
Verified fixed at 827763a31d. I ran the same statements, and the DataNode log now shows:
| statement | ErrorHandlingUtils line |
[QueryFailed] line |
|---|---|---|
select count(s1), s1 from root.db.d0 → 701 |
INFO, no stack | INFO, no stack |
| oversized partition request → 305 | WARN + stack (once) | INFO, no stack |
select s1 from root.db.d0 where time >= 0 and time <= 9223372036854775806 → 1707 |
WARN + stack (once) | INFO, no stack |
User errors no longer produce a WARN, and each failure's stack trace is logged only once.
| close(); | ||
| throw new TTransportException( | ||
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); |
There was a problem hiding this comment.
The new text fixes both points: it says "at least N" and it names a setting. However, TElasticFramedTransport is the shared service-rpc transport, so the hardcoded (dn_thrift_max_frame_size on DataNode) is wrong outside a DataNode:
- Session/JDBC clients: the limit is the client's own
Session.Builder#thriftMaxFrameSize(default 64 MiB,SessionConfig.DEFAULT_MAX_FRAME_SIZE) or JDBC'sthrift_max_frame_size. The rejection happens on the client, before anything is sent. - ConfigNode: there is no setting at all. The limit is the fixed
RpcUtils.THRIFT_FRAME_MAX_SIZE(536870912), because the ConfigNode never loads the DataNode config. For example, a ConfigNode whoseiotdb-system.propertiessetdn_thrift_max_frame_size=1048576still reportedprotect max size (536870912).
Repro: a plain Java client against this head's cluster (DataNode dn_thrift_max_frame_size=1048576), using new Session.Builder()...thriftMaxFrameSize(4194304) to insert a 1,000,000-row INT64 tablet:
base client jars: IoTDBConnectionException: org.apache.thrift.transport.TTransportException: Frame size (16000109) larger than protect max size (4194304)!
this head's client: IoTDBConnectionException: org.apache.thrift.transport.TTransportException: Serialized frame size is at least 8000077 bytes, exceeding the RPC frame-size limit (dn_thrift_max_frame_size on DataNode) of 4194304 bytes.
The message names a setting that didn't fire, and that setting's actual value here (1048576) doesn't match the number printed. A Session bulk insert over the 64 MiB default would point users to the server config, which can't fix a client-side rejection.
Suggestion: keep the library message generic, e.g. ... exceeding the configured Thrift max frame size of N bytes. If the DataNode hint is wanted, add it where the DataNode handles the failure (e.g. the CORRUPTED_DATA branch in ConfigNodeClient#executeRemoteCallWithRetry), or pass a limit description into the transport factory. The dn_thrift_max_frame_size assertions in TElasticFramedTransportTest and QueryPartitionMemoryTest would follow. My earlier suggestion to name the setting didn't account for the transport being shared.
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); | ||
| } | ||
| writeBuffer.write(buf, off, len); |
There was a problem hiding this comment.
Verified fixed at 827763a31d. Capacity at rejection is now exactly the limit (1,048,576 and 67,108,864, i.e. 1.00×), and testIncrementalGrowthDoesNotExceedFrameLimit covers it.
Optional perf note: these per-write checks sit on Thrift's hottest path. I benchmarked serialization into a null sink, taking the median of 3 alternating JVM runs.
- Small fields: a 100k-slot
TDataPartitionReq, i.e. 4writecalls per 12-byte struct. - Large binary: a 1M-row
TSInsertTabletReq.
| base | this head | this head + fast path below | |
|---|---|---|---|
| small fields, ns/slot | 8.87–9.03 | 9.97–12.10 | 8.03–8.69 |
| large binary, ns/row | 1.69–1.71 | 1.67–1.74 | 1.69–1.73 |
Only small-field-heavy messages pay the cost, about 11% or more; large binary writes are unaffected. This commit makes capacity <= limit an invariant, so a write that fits in the current array can skip every limit check. The checks are only needed when the buffer must grow:
// AutoScalingBufferWriteTransport
boolean tryWriteWithinCapacity(byte[] toWrite, int off, int len) {
byte[] array = buf.array();
if (len > array.length - pos) {
return false;
}
System.arraycopy(toWrite, off, array, pos, len);
pos += len;
return true;
}
// TElasticFramedTransport#write, as the first statement
if (writeBuffer.tryWriteWithinCapacity(buf, off, len)) {
return;
}With only this change, the limit behavior in my probes is unchanged: a write is rejected after exactly limit bytes, capacity equals the limit, and the compressed paths behave as before. The "fast path" column above is this change.
|
JackieTien97
left a comment
JackieTien97
left a comment
There was a problem hiding this comment.
Re-checked at c7003fff63 on a rebuilt 1C1D cluster (dn_thrift_max_frame_size=1048576) plus the same client and transport probes. Both round-2 follow-ups are addressed, and the earlier fixes still hold. Details are in the two thread replies. Nothing further from me.
Normal traffic also passed a round trip over the new fast path. A 50,000-row Session insertTablet read back via JDBC as count=50000, sum=1249975000, and all 50,000 rows fetched intact.
AutoResizingBufferTest (9) and TElasticFramedTransportTest (9) pass locally. CI was 18/19 green at posting time, with the auto-create-schema IT still running.
| close(); | ||
| throw new TTransportException( | ||
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); |
There was a problem hiding this comment.
Verified fixed at c7003fff63. The message is now generic and names the limit that actually fired:
- DataNode → ConfigNode, oversized partition request (305, 15–42 ms):
Serialized frame size is at least 1048582 bytes, exceeding the configured Thrift max frame size of 1048576 bytes. - Plain Java Session client with
thriftMaxFrameSize(4194304)against a DataNode set to 1048576:Serialized frame size is at least 8000077 bytes, exceeding the configured Thrift max frame size of 4194304 bytes.The number now matches the client's own limit.
The tests now also assert that dn_thrift_max_frame_size is absent from the message.
| TTransportException.CORRUPTED_DATA, | ||
| "Frame size (" + size + ") larger than protect max size (" + maxSize + ")!"); | ||
| } | ||
| writeBuffer.write(buf, off, len); |
There was a problem hiding this comment.
Verified at c7003fff63. The fast path doesn't change any limit behavior:
- Writes are rejected after exactly
limitbytes (1 MiB and 64 MiB), and capacity stays at 1.00× the limit. - Both compressed paths still raise
TTransportExceptionand close the transport.
Safety reasoning:
write(..., maxSize)is the only writer of the framedwriteBuffer, so capacity ≤ limit holds and a fitting write can't cross the limit.len > array.length - poscan't overflow.testBufferedWriteCannotBypassArrayLimitOnOverflowwould catch apos + lenrewrite.
Same benchmark as before (null sink, median ns over 3 alternating JVM runs):
| base | 827763a31d |
c7003fff63 |
|
|---|---|---|---|
| small fields, ns/slot | 8.80–9.02 | 9.89–10.29 | 8.37–8.50 |
| large binary, ns/row | 1.72–1.76 | 1.70–1.73 | 1.70–1.75 |
The small-field regression is gone; this head is now about 5% faster than base on that path. Large binary writes are unchanged.
Description
Very large time-partition requests can keep query analysis running while retaining reservations from the operator memory pool. The Thrift buffer's 1.5x growth overflowed
intabove roughly 1.33 GiB, causing an entire large array to be copied for every small field write. The uncompressed frame limit was checked only after serialization, atflush().The change makes these failures bounded and diagnosable:
Reproduction and validation
On the base revision
013622816a, a real 1,500,000,000-byte buffer grew by only 8 bytes for each of three successive writes, copying the entire array each time. With the fix, it grows once and subsequent writes reuse the buffer. This large-array check runs separately with a bounded 5 GiB heap; ordinary unit tests do not require multi-gigabyte allocations.Regression tests exercise the real SQL parser, Coordinator, Analyzer, and generated ConfigNode Thrift serializer with test schema/partition fetchers and in-memory transports. Repeated oversized requests and partition-memory failures fully restore earlier operator-pool reservations. Tests verify INFO context records include the SQL and query ID for memory shortages, semantic errors, timeouts, and frame-size failures without an attached throwable, and that normal queries do not produce failure logs. Incremental 8-byte writes prove a 1 MiB cap never allocates beyond 1 MiB; initial capacities above the configured cap are also covered. Further cases exercise position overflow at the capacity fast path and oversized compressed output after reusing a buffer. A simulated OutOfMemoryError test verifies that SQL context is recorded and the original error is rethrown unchanged while earlier reservations are released.
86 distinct tests passed in a clean reactor build on JDK 21 (Java 17 target), covering normal and templated analysis, query retry, memory accounting, diagnostics, ConfigNode retry behavior, compressed/uncompressed frames, and buffer reuse/shrinking. Checkstyle and Spotless passed. The coverage job now explicitly selects JDK 21, matching Sonar, because the runner default JDK caused google-java-format to fail before any coverage tests could run.
A null-sink
TBinaryProtocolmicrobenchmark on aarch64/JDK 21 compared827763a31dwith this change (32 MiB frame limit, 2 s warmup and 3 s measurement, median of 3 alternating JVM runs per variant):TDataPartitionReqTSInsertTabletReqSmall-field serialization time fell about 22.5%; large-binary results remained comparable. Payload lengths, hashes, and buffer capacities matched. These are local serialization measurements, not end-to-end query throughput.
mvn -B clean test -pl iotdb-core/datanode -am \ -Dtest=AutoResizingBufferTest,TElasticFramedTransportTest,RpcUtilsTest,UrlUtilsTest,TSStatusCodeTest,QueryTimePartitionTest,QueryPartitionMemoryTest,AnalyzeTest,QueryExecutionRetryTest,MPPQueryContextTest,LocalExecutionPlannerOperatorsMemoryTest,ConfigNodeClientTest,TemplatedAnalyzeRetryTest \ -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=falseThe local reproductions use controlled inputs and in-memory transports; they do not replay the original production SQL or start a live cluster.