[fix](arrow-flight) Fix utf8/large string schema-array mismatch for oversized columns#65789
Open
morningman wants to merge 2 commits into
Open
[fix](arrow-flight) Fix utf8/large string schema-array mismatch for oversized columns#65789morningman wants to merge 2 commits into
morningman wants to merge 2 commits into
Conversation
…versized columns When an Arrow Flight result column's data reaches 2GB within a single batch, FromBlockToRecordBatchConverter silently rebuilt that column's array as large_utf8/large_binary (int64 offsets) while the RecordBatch schema kept utf8/binary (int32 offsets). The batch then violates the Arrow spec: its array offsets (int64) disagree with its declared schema type (int32). - Generic Arrow clients (pyarrow, Flight SQL JDBC, ...) read the int64 offset buffer as int32 -> corrupted string values. - The Doris receiver (read_column_from_arrow, cross-cluster type=doris catalog) derives negative lengths -> out-of-bounds memcpy / crash. Arrow Flight sends one fixed schema for the whole stream, so a per-batch type swap can never be correct. Fix: keep the schema authoritative and never swap array types. - Remove the silent large_utf8/large_binary upgrade in the converter. - Add convert_to_arrow_batches(): an O(1) fast-path check (no cost on the normal path); when a string/binary column would exceed the int32 offset limit, split the block by rows so every emitted batch stays valid utf8. A single value that alone exceeds the limit returns a clear error instead of corrupting or crashing. - The Arrow Flight local/remote readers drain the batches produced from one block. - Validate() the RecordBatch after building it so any future schema/array divergence fails loudly instead of silently corrupting downstream. - Replace the hard-coded 2GB constant with BE config arrow_flight_result_max_utf8_bytes (default 2^31) so tests can exercise the split path without materializing 2GB of data. Columns under the limit are emitted exactly as before (utf8/binary): no wire change on the normal path; only the previously-broken >2GB path changes.
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 29470 ms |
Contributor
TPC-DS: Total hot run time: 178975 ms |
Contributor
ClickBench: Total hot run time: 24.91 s |
Contributor
Author
|
/review |
Contributor
Author
|
This should close #62681 |
Contributor
Author
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Request changes. Keeping the fixed Flight stream schema authoritative is the right direction, and the ordinary top-level String/Varbinary splitter preserves row order and nulls, but four correctness gaps remain in the targeted oversized-result path; each is described inline.
Critical checkpoint conclusions:
- Goal and proof: Partially achieved. The schema/array mismatch is removed for ordinary top-level String/Varbinary, but SerDe-rendered types can bypass accounting (MAIN-2), nested int32-offset descendants are not split (MAIN-3), remote/proxy results can fail before conversion (MAIN-4), and the exact Arrow 17 boundary is off by one (MAIN-1). The added unit tests prove only the plain top-level utf8 helper path.
- Scope and reuse: The change is focused and reuses the existing converter/readers. No unrelated source change was found.
- Concurrency: The new pending deque is used by one serialized RecordBatchStream reader. Producer/consumer buffer state remains under existing locks; no new lock-order, deadlock, or atomic-ordering issue was found.
- Lifecycle and memory: Pending batches have acyclic reader ownership, drain before the next source fetch, and are released on return/destruction. Eager conversion retains one source block's Arrow output, but no separate lifecycle leak was substantiated.
- Configuration: The mutable limit is snapshotted per conversion and subsequent blocks observe updates, but its validated maximum exceeds Arrow 17's actual builder capacity (MAIN-1).
- Compatibility and parallel paths: Keeping utf8/binary schema types is wire-compatible on normal results and fixes the invalid type swap. Both local and remote readers were edited, but the remote protobuf boundary remains before splitting (MAIN-4), and recursive List/Map/Struct paths remain uncovered (MAIN-3). No storage-format or rolling-upgrade protocol change applies.
- Conditions, errors, and data correctness: Statuses are otherwise propagated and the single-value error includes useful field/row/size context. The physical-size assumption is not a valid invariant for all String/Binary SerDes and can silently produce an empty valid value after unchecked int32 narrowing (MAIN-2).
- Tests and results: New tests cover non-null top-level utf8 splitting, no-split, one oversized value, and two top-level columns. They do not cover binary, nullable, transformed/complex SerDes, nested schemas, exact Arrow capacity, reader draining/EOF, or proxy routing. This review was static as required; no local build was attempted. Current CI shows Linux compile/BE UT and formatting green; regression jobs are pending, and macOS BE UT stopped before tests because the runner had JDK 25 instead of required JDK 17.
- Performance and observability: The normal fast path is O(fields), the rare slow path is O(rows times tracked fields), and existing conversion timers/packet logs remain. Correct emitted-size accounting should avoid double rendering where practical. No additional observability blocker was found.
- Transactions, persistence, data writes, and FE-BE variable propagation: Not applicable to this result-serialization change.
- Additional user focus: No additional user-provided focus was supplied; the complete PR was reviewed.
Overall, the PR should not merge until the four inline correctness issues and their missing boundary/integration coverage are addressed.
| // below 2^31 (= INT32_MAX + 1). When a block would exceed this, the Arrow Flight readers split | ||
| // it by rows into multiple batches so each batch stays valid. Default is the hard 2^31 limit; | ||
| // only lower it (e.g. in tests) to exercise the split path without materializing 2GB of data. | ||
| DEFINE_mInt64(arrow_flight_result_max_utf8_bytes, "2147483648"); |
Contributor
There was a problem hiding this comment.
[P2] Match Arrow 17's actual builder capacity
Doris pins Arrow 17, where BaseBinaryBuilder::memory_limit() is INT32_MAX - 1 and ValidateOverflow rejects anything larger. This exclusive cap is INT32_MAX + 1, and the splitter only cuts on >= max_bytes, so an ordinary String/Varbinary payload totaling exactly INT32_MAX is passed through and then fails with CapacityError instead of being split. Please cap the exclusive threshold at INT32_MAX (so emitted data is at most INT32_MAX - 1) and add an exact boundary test.
| // batch -- zero added cost on the normal path, and this never misses an overflow. | ||
| std::vector<int> big_fields; | ||
| for (int idx = 0; idx < num_fields; ++idx) { | ||
| if (is_int32_offset_binary_type(schema->field(idx)->type()) && |
Contributor
There was a problem hiding this comment.
[P1] Measure the bytes that each SerDe actually appends
This gate assumes physical IColumn::byte_size()/get_data_at() matches the Arrow payload, but several types with an utf8/binary schema transform their data. For example, binary JSONB below 2 GiB can render above INT32_MAX after escaping; DataTypeJsonbSerDe then narrows that size with cast_set<int, size_t, false>, producing a non-positive length on supported targets. Arrow 17 treats that as a valid empty value, so Validate() passes and local Flight silently corrupts the result. Variant, formatted fixed-width types, and Bitmap/HLL/QuantileState have the same storage-vs-SerDe mismatch. Please base cuts on the emitted SerDe payload and reject before any int32 narrowing, with JSONB/Variant/complex-type tests.
|
|
||
| // Only utf8/binary use int32 offsets and can overflow at 2^31; fixed-size and large_* cannot. | ||
| bool is_int32_offset_binary_type(const std::shared_ptr<arrow::DataType>& type) { | ||
| return type->id() == arrow::Type::STRING || type->id() == arrow::Type::BINARY; |
Contributor
There was a problem hiding this comment.
[P1] Recurse into nested int32-offset builders
This predicate only recognizes a top-level STRING/BINARY. ARRAY, MAP, and STRUCT can contain shared child String/Binary builders (and List/Map themselves have int32 element offsets), but those fields always take the fast path. For example, two ARRAY<STRING> parent rows with individually representable child data can exceed the child builder together and fail conversion, although emitting one parent row per batch would succeed. Please account for descendant byte/element limits per top-level row, return a clear error when one parent row cannot fit, and add nested nullable collection tests.
| auto st = convert_to_arrow_batch(*_block, _schema, arrow::default_memory_pool(), out, | ||
| _timezone_obj); | ||
| std::vector<std::shared_ptr<arrow::RecordBatch>> batches; | ||
| auto st = convert_to_arrow_batches(*_block, _schema, arrow::default_memory_pool(), &batches, |
Contributor
There was a problem hiding this comment.
[P1] Split before the remote protobuf boundary
The remote reader reaches this helper only after _fetch_data() receives and deserializes one complete PFetchArrowDataResult.block. On the producing BE, GetArrowResultBatchCtx::on_data() rejects any response whose ByteSizeLong() exceeds INT32_MAX and clears the block, so a proxy-routed high-entropy result at the size this PR targets fails before this line even though the local reader can split it. Please chunk on the producing BE before protobuf serialization (or use a streaming transport) and add a proxy-routed boundary test.
Contributor
There was a problem hiding this comment.
Supplemental request for changes. Review 4730854844 already covers the exact Arrow builder boundary, SerDe/storage byte mismatch, nested int32-offset builders, and the remote protobuf boundary. This review adds two distinct blockers inline: the complete Arrow Flight IPC body can still exceed Arrow 17's transport limit even when each value buffer fits, and the new mutable configuration accepts and can persist unchecked invalid values.
Critical checkpoint conclusions:
- Goal and proof: Partially achieved. Keeping the declared utf8/binary schema authoritative fixes the original schema-array mismatch for ordinary top-level strings, but the live review's four issues plus the two additional inline issues leave the oversized-result path incomplete. The added tests prove only the direct plain-ColumnString helper behavior.
- Focus, minimality, and reuse: The change is localized and reuses the existing converter and readers; no unrelated source change was found. The new helper's assumptions are narrower than the supported Doris-to-Arrow type and transport contracts.
- Concurrency: The pending-batch deques are consumed by the serialized stream-reader calls, and the existing producer/result-buffer locking is unchanged. No new race, deadlock, or atomic-ordering issue was substantiated.
- Lifecycle and ownership: Pending batches drain before another source block is fetched and are released with the reader. EOF and error statuses otherwise propagate through the reader; the already-reported remote serialization failure occurs before this lifecycle can help.
- Configuration: The declared range and default are documented, but the exact-capacity problem is already reported and the mutable validator path validates the old value, then commits and optionally persists the unchecked candidate (second inline issue).
- Compatibility and parallel paths: Preserving utf8/binary schema types is wire-compatible for normal results and removes the invalid large_utf8/large_binary swap. Local and remote readers and generic converter call sites were checked; nested, transformed-SerDe, remote-proxy, and complete Flight-payload paths remain incomplete as reported.
- Conditions, errors, and data correctness: The ordinary single-value error is contextual, but supported transformed/nested types can bypass its assumptions, and batches that pass
RecordBatch::Validate()can still be rejected by Flight payload validation (first inline issue). - Tests and results: New tests cover direct top-level ColumnString split/no-split/error cases, but not FlightPayload validation, nested or transformed types, remote no-compression routing, or
set_configinvalid/recovery/persistence behavior. This review was static as required; no local build was attempted. Current CI has Linux compile/BE UT, formatting, external regression, cloud_p0, and vault_p0 green; P0 and NonConcurrent regression remain pending. macOS BE UT stopped before tests because the runner used JDK 25 instead of required JDK 17. - Observability: Existing conversion timers and contextual status messages remain. No additional observability blocker was found beyond the already-reported remote failure occurring before a useful client response.
- Transactions, persistence, data writes, and FE-BE variables: No transaction or storage-data write path changes. BE config persistence is applicable and unsafe under the new mutable declaration (second inline issue); no new FE-BE variable propagation applies.
- Performance: The common fast path remains cheap and the row scan is confined to suspected large columns. Correct full-payload and SerDe-aware sizing should preserve bounded overhead and avoid rendering values twice where possible.
- Additional user focus: No additional focus was supplied; the complete PR and its upstream/downstream paths were reviewed.
The review is complete after three convergence rounds. The PR should not merge until the existing four issues and the two additional inline blockers are addressed with boundary and integration coverage.
| schema->field(big_fields[c])->name(), end, rb, max_bytes); | ||
| } | ||
| row_bytes[c] = rb; | ||
| if (running[c] + rb >= max_bytes) { |
Contributor
There was a problem hiding this comment.
[P1] Bound the complete Flight IPC body, not only value bytes
The exact per-buffer threshold is covered separately; even after that is corrected, Arrow 17's FlightPayload::Validate() rejects an ipc_message.body_length above INT32_MAX, while this cut accounts only for one tracked value buffer. Offsets, validity buffers, padding, fixed-width columns, and other individually sub-limit string columns all contribute to the same body. The existing BigString test is already a counterexample: its 2,147,483,640 value bytes pass this < 2^31 check by 8 bytes, but the offset buffer makes the Flight payload too large even though RecordBatch::Validate() succeeds. Please size/cut for the complete IPC body as well and add a Flight-payload-level test.
| // below 2^31 (= INT32_MAX + 1). When a block would exceed this, the Arrow Flight readers split | ||
| // it by rows into multiple batches so each batch stays valid. Default is the hard 2^31 limit; | ||
| // only lower it (e.g. in tests) to exercise the split path without materializing 2GB of data. | ||
| DEFINE_mInt64(arrow_flight_result_max_utf8_bytes, "2147483648"); |
Contributor
There was a problem hiding this comment.
[P1] Do not expose this validator through the broken mutable path
DEFINE_mInt64 exposes this value through /api/update_config, but UPDATE_FIELD invokes the registered zero-argument validator before assigning new_value, so it validates the old default and then accepts 0, -1, or a value above 2^31. Zero makes every nonempty result hit the row-limit error, while a negative value becomes a huge size_t and disables splitting; after that, a valid recovery update is rejected because the invalid old value is what gets validated. With persist=true, the unchecked value is also written to be_custom.conf and can prevent the next startup. Since the comment says this is lowered only in tests, please make it non-mutable, or fix the update path to validate the candidate before committing it, and add set_config coverage.
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
…type serde size Follow-up on the review of this PR: - Cap the split threshold at INT32_MAX (was 2^31). Arrow's BaseBinaryBuilder rejects a value buffer larger than memory_limit() = INT32_MAX - 1; the splitter cuts before a column reaches the threshold, so INT32_MAX keeps each committed batch at most INT32_MAX - 1 bytes -- exactly the builder limit. Previously 2^31 let a batch total INT32_MAX and then fail with CapacityError. - Guard JSONB -> Arrow against silent corruption. JSONB is stored as binary but rendered to (usually larger) JSON text, so the physical column size does not bound the arrow payload; the old unchecked cast_set<int, size_t, false> of a rendered length > INT32_MAX would wrap to a bogus small/negative length that Arrow silently accepts. Reject with a clear error before narrowing. (Variant already uses a checked cast.) - Scope the row-splitter to raw string/binary types (is_string_type || VARBINARY) whose SerDe appends get_data_at(row) verbatim, so physical byte_size equals the emitted payload and per-row cuts are exact. Transform-serde types that also map to utf8/binary (JSONB, Variant, largeint/date/datetime text) stay on the fast path and rely on their checked size cast plus the Arrow builder capacity check to error cleanly instead of being mis-split (Variant has no get_data_at at all). Tests: exact INT32_MAX-vs-(INT32_MAX-1) boundary; transform type (largeint) whose physical size exceeds a lowered threshold is not split.
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 29662 ms |
Contributor
TPC-DS: Total hot run time: 178572 ms |
Contributor
ClickBench: Total hot run time: 24.97 s |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
When an Arrow Flight result column's data reaches 2GB within a single batch,
FromBlockToRecordBatchConvertersilently rebuilt that column's array aslarge_utf8/large_binary(int64 offsets) while the RecordBatch schema keptutf8/binary(int32 offsets). The batch then violates the Arrow spec — itsarray offsets (int64) disagree with its declared schema type (int32):
buffer as int32 → corrupted string values;
read_column_from_arrow, cross-clustertype=doriscatalog) derives negative lengths → out-of-bounds
memcpy/ crash.Arrow Flight sends one fixed schema for the whole stream, so a per-batch type
swap can never be correct.
Fix
Keep the schema authoritative and never swap array types:
large_utf8/large_binaryupgrade in the converter.convert_to_arrow_batches(): an O(1) fast-path check (no cost on thenormal path); when a string/binary column would exceed the int32 offset
limit, split the block by rows so every emitted batch stays valid
utf8.A single value that alone exceeds the limit returns a clear error instead of
corrupting data or crashing.
produced from one block.
Validate()the RecordBatch after building it, so any future schema/arraydivergence fails loudly here instead of silently corrupting downstream.
arrow_flight_result_max_utf8_bytes(default 2^31) so tests can exercise thesplit path without materializing 2GB of data.
Columns under the limit are emitted exactly as before (
utf8/binary) — nowire change on the normal path; only the previously-broken >2GB path changes.
Testing
DataTypeSerDeArrowTest): split-and-round-trip byte identity,no-split under the limit, single-value-over-limit error, and split driven by
the tightest column.
arrow_flight_sql_p0regression passes unchanged (no wire change on thenormal path).