Fix table UDF result block splitting - #18333
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #18333 +/- ##
============================================
+ Coverage 43.41% 43.48% +0.06%
Complexity 374 374
============================================
Files 5366 5394 +28
Lines 382956 385142 +2186
Branches 49809 50088 +279
============================================
+ Hits 166259 167474 +1215
- Misses 216697 217668 +971 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
There was a problem hiding this comment.
Pull request overview
This PR addresses oversized TsBlock outputs from table-model UDFs by enforcing configured TsBlock size/row limits in TableFunctionOperator, preventing large blocks from propagating through the query pipeline and risking RPC frame overflows.
Changes:
- Add final-result TsBlock splitting in
TableFunctionOperator(after pass-through columns are appended) based on configured TsBlock constraints. - Add a unit test covering variable-width outputs with/without pass-through columns.
- Add an integration-test UDF plus an IT validating large UDF results are returned without loss/duplication and with pass-through alignment under a small TsBlock limit.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java | Adds unit coverage for splitting variable-width UDF output blocks with optional pass-through columns. |
| iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java | Implements final-result TsBlock splitting and adjusts operator completion/return-size reporting. |
| integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedTableFunctionIT.java | Adds end-to-end validation that large table-UDF results are split without data loss and keep pass-through columns aligned. |
| integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/relational/LargeResultTableFunction.java | Introduces a table UDF used by the integration test to generate large variable-width outputs. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| @Override | ||
| public long calculateMaxReturnSize() { | ||
| return Math.max(DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES, properBlockBuilder.getRetainedSizeInBytes()); | ||
| return maxTsBlockSizeInBytes; | ||
| } |
| /** | ||
| * Splits the final result using the same logical in-memory size accounting as {@link | ||
| * TsBlockBuilder}. | ||
| * | ||
| * <p>Serializing candidate regions to find their exact sizes would write every value into | ||
| * temporary buffers, only for the exchange layer to serialize the selected regions again. | ||
| * Rebuilding the result with a size-tracking {@link TsBlockBuilder} would avoid that temporary | ||
| * serialization, but it would turn the UDF's batched column output into row-by-row, | ||
| * column-by-column copies. | ||
| * | ||
| * <p>Instead, fixed-width values are accounted for directly from their data types, while only the | ||
| * retained sizes of variable-width values are inspected. This deliberately estimates the | ||
| * in-memory TsBlock size rather than its serialized size because the two representations are not | ||
| * equivalent. The resulting regions are views over the original columns and do not copy their |
| @@ -62,14 +64,15 @@ public class TableFunctionOperator implements ProcessOperator { | |||
| private static final long INSTANCE_SIZE = | |||
| RamUsageEstimator.shallowSizeOfInstance(AggregationMergeSortOperator.class); | |||
There was a problem hiding this comment.
| RamUsageEstimator.shallowSizeOfInstance(AggregationMergeSortOperator.class); | |
| RamUsageEstimator.shallowSizeOfInstance(TableFunctionOperator.class); |
There was a problem hiding this comment.
make TableFunctionOperator extends AbstractOperator, an reuse the functions in that class just like AbstractTableScanOperator.
c380a82 to
03b2432
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (2)
iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java:298
- calculateMaxReturnSize() now returns only maxReturnSize, but calculateMaxPeekMemory() explicitly accounts for Math.max(maxReturnSize, properBlockBuilder.getRetainedSizeInBytes()). This makes the operator’s return-size estimate inconsistent and can underestimate peak output size when the UDF builds wide/variable-width columns (and AbstractOperator may still return a single-row TsBlock larger than maxReturnSize when oneTupleSize > maxReturnSize). Consider restoring the previous max() logic here so memory accounting remains conservative.
public long calculateMaxPeekMemory() {
return inputOperator.calculateMaxPeekMemory()
+ Math.max(maxReturnSize, properBlockBuilder.getRetainedSizeInBytes());
}
@Override
public long calculateMaxReturnSize() {
return maxReturnSize;
}
iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java:200
- Splitting via AbstractOperator.checkTsBlockSizeAndGetResult() is based on an average per-row size computed from the whole TsBlock. For variable-width columns with skewed row sizes, this can still produce regions whose retained/serialized size significantly exceeds max_tsblock_size_in_bytes (e.g., one very large row in a block with many small rows lowers the average). This doesn’t match the PR description’s per-position accounting for variable-width columns and may not reliably prevent oversized downstream blocks.
private TsBlock getNextResultTsBlock() {
if (retainedTsBlock != null) {
return getResultFromRetainedTsBlock();
}
resultTsBlock = resultTsBlocks.poll();
return resultTsBlock == null ? null : checkTsBlockSizeAndGetResult();
}
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (1)
iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java:301
calculateMaxReturnSize()now returnsmaxReturnSizeonly, which can be smaller thanproperBlockBuilder.getRetainedSizeInBytes()(e.g., many output columns). This can underestimate the operator’s maximum returned block footprint for memory planning, and it regresses the previous behavior that guarded against this by takingMath.max(...).
@Override
public long calculateMaxReturnSize() {
return maxReturnSize;
}




Description
Problem
A table-model UDF with a table argument can accumulate a large result while processing a partition, especially when one device contains many rows or the UDF expands each input row into multiple output rows. Previously,
TableFunctionOperatorreturned each generatedTsBlockdirectly. A sufficiently large result could therefore reach the exchange or RPC layer as one block and exceed its frame-size limit.Pass-through columns are appended only after the UDF has produced its proper columns, so the final block must be split after that step to keep all output columns aligned.
Design
TableFunctionOperatornow extendsAbstractOperatorand reuses its existing lazy result-splitting mechanism.The final block is first built with any pass-through columns appended and then queued. Before a queued block is returned,
checkTsBlockSizeAndGetResult()derives a maximum position count from the logical average row size of the first non-empty result block. If the block exceeds that position count,AbstractOperatorretains it and returns ordered regions over subsequentnext()calls. These regions are views over the original columns and do not rebuild, copy, or serialize values.This design avoids both expensive alternatives:
TsBlockBuilderwhile tracking its size.The configured
max_tsblock_size_in_bytesis therefore used as a low-cost logical target rather than an exact serialized-size guarantee. The mechanism does not inspect every variable-width payload, and one indivisible oversized row or highly skewed binary values may still exceed the target. This PR is intended to prevent large, ordinary partition results from being returned as a single block while keeping the UDF execution path inexpensive.No new configuration is introduced. This change deliberately does not add a separate
max_tsblock_line_numberconstraint; it follows the existingAbstractOperatorbehavior.Operator lifecycle
Pending result blocks are drained before another partition is processed, and
hasNext()/isFinished()continue to report pending retained or queued results. Closing the operator clears the partition cache, queued results, and retained result references before releasing the child operator and UDF resources.Tests
The unit tests cover:
process(), with and without pass-through columns;finish(), with and without pass-through columns;The integration test configures a small TsBlock target and uses a table UDF that expands four input rows into 256 variable-width rows. It verifies that every expected row is returned exactly once, payload values are complete, and pass-through columns including null values remain aligned.
Targeted unit test:
mvn test -pl iotdb-core/datanode -am \ -Dtest=TableFunctionOperatorTest \ -DfailIfNoTests=false \ -Dsurefire.failIfNoSpecifiedTests=falseTargeted integration test:
This PR has:
Key changed/added classes
TableFunctionOperatorAbstractOperatorto lazily split final table-UDF results after pass-through columns are appended.TableFunctionOperatorTestLargeResultTableFunctionIoTDBUserDefinedTableFunctionIT