feat: support Apache Paimon non-direct-file reads - #616
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: yanbinyang The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #616 +/- ##
==========================================
+ Coverage 76.19% 76.58% +0.39%
==========================================
Files 173 173
Lines 17388 17747 +359
Branches 2618 2699 +81
==========================================
+ Hits 13248 13591 +343
- Misses 4140 4156 +16
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Accept Paimon table specifications and propagate snapshot, scan-mode, and object-storage options through external collection planning. Route raw-compatible files through direct-file readers and preserve native DataSplits for scans that require Paimon execution. Keep DataSplits atomic, group them by bucket, and pack them into Milvus segments with the existing row target and a configurable split-count limit. Preserve storage-owned descriptors in StorageV3 manifests and use decoded-memory estimates for load planning. Add C++, Go, and client coverage for routing, manifest round trips, segment packing, deletion vectors, merge-on-read, and snapshot reads. Storage direct-file support: milvus-io/milvus-storage#602. Depends on: milvus-io/milvus-storage#616. Design + tracking: milvus-io#45881. Related tracking: milvus-io#50632. --------- Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
| auto all_tasks = ChunkTask::Build(unique_chunk_indices, get_chunk_info); | ||
| SplitAsyncTasks(all_tasks, std::max<size_t>(parallelism, 1), ChunkTask::SplitTraits{get_chunk_info}, | ||
| GetAsyncTaskSplitStrategy(properties_)); | ||
| auto natural_tasks = ChunkTask::Build(unique_chunk_indices, get_chunk_info); |
There was a problem hiding this comment.
Currently, Paimon does not implement the native async interface. It uses the generic fallback, which executes the blocking read and wraps the result in an async task/future. Given that, I think the sequential-read constraint should remain within Paimon’s own get_chunk semantics instead of changing the common ChunkTask splitting logic.
Also, for the non-direct-read path, the current approach does not fully solve the duplicated-read problem. I would prefer to keep this PR focused on functional correctness first, and address the performance issue later with a more complete design.
There was a problem hiding this comment.
OK. I’ll simplify this PR to focus on functional correctness, remove the cross-call cursor reuse and related test hooks, and leave broader DataSplit read optimization for a separate change.
| // Forward-only readers must receive requested chunks for one file through a | ||
| // shared ascending cursor. This avoids reopening and replaying their source | ||
| // stream for every logical range. | ||
| [[nodiscard]] virtual bool requires_sequential_chunk_reads() const { return false; } |
There was a problem hiding this comment.
I don’t think requires_sequential_chunk_reads is a good API for adapting a forward-only streaming reader to a random-access chunk interface.
There was a problem hiding this comment.
Removed in 4283be8d. The common reader and ChunkTask scheduling are unchanged; Paimon now creates a private forward cursor for each read operation.
|
change commit title to |
| if (metadata.value("read_path", std::string{}) != "direct-file") { | ||
| return arrow::Status::Invalid("Paimon planner returned a non-direct metadata descriptor"); | ||
| } | ||
| read_path = metadata.value("read_path", std::string{}); |
There was a problem hiding this comment.
constexpr "read_path"/direct-file/data-split
and check others
There was a problem hiding this comment.
Updated in 4283be8d. read_path, direct-file, and data-split now use shared constants from paimon_common.h.
| schema: SchemaRef, | ||
| } | ||
|
|
||
| fn classify_stream_error(error: paimon::Error) -> ArrowError { |
There was a problem hiding this comment.
cpp/src/format/bridge/rust/src/paimon_bridgeimpl.rs line:999
Medium ---- This stream-specific mapping turns every non-temporary IoUnexpected (including NotFound and PermissionDenied) into InvalidArgumentError, while temporary failures become a generic Arrow I/O error and lose the StorageTransientThrottling / StorageTransientService details used by the rest of the bridge. A failure after stream open therefore has different retry and refresh semantics from planning/open. Please preserve the same not-found, transient, and plain-I/O classification across the Arrow stream boundary.
There was a problem hiding this comment.
Fixed in c70fa849. Stream errors now preserve the existing Paimon markers through Arrow FFI, so not-found, transient, invalid, not-implemented, and plain I/O errors keep the same handling as planning and open failures. The FFI and C++ status mappings are covered by tests.
| // Keep the handle: every reader created from this cached metadata shares | ||
| // it, so schema and snapshot resolution happen once per descriptor. | ||
| metadata->payload.split_reader_handle = std::move(reader); | ||
| ARROW_ASSIGN_OR_RAISE(metadata->row_group_infos, MakeLogicalRowGroups(parsed.record_count, logical_chunk_rows)); |
There was a problem hiding this comment.
cpp/src/format/paimon/paimon_format_reader.cpp line:557
Medium ---- Planning filters out zero-row DataSplits, but the persisted descriptor reader accepts record_count == 0 and MakeLogicalRowGroups publishes no chunks. Because no stream is then consumed, the EOF validation never compares the declared count with the actual split, so corrupted metadata can silently hide a non-empty split. Please reject zero for data-split descriptors, or otherwise force explicit count validation before exposing the metadata.
There was a problem hiding this comment.
Fixed in e8758f0e. Milvus planning already omits zero-row splits, so a persisted zero-row descriptor violates that contract. This commit rejects it during metadata parsing, preventing an empty chunk list from bypassing stream count validation. The new test covers a non-empty split whose declared count is corrupted to zero; the existing fully-deleted-table test verifies legitimate empty results are omitted during planning.
| std::cerr << std::endl; | ||
| std::cerr << "Creates a demo table with schema (id int64, name string," | ||
| << " value float64)." << std::endl; | ||
| std::cerr << "Scalar schema: id int64, name string, value float64." << std::endl; |
There was a problem hiding this comment.
cpp/tools/loon.cpp line:167
Low ---- This help now applies to both Iceberg and Paimon, but the Paimon scalar fixture declares id as Int32 in both fixture_schema and scalar_fixture_batch, not int64. A user or downstream E2E that follows this text can construct the wrong external schema. Please distinguish the two table types here or document Paimon's actual id type.
There was a problem hiding this comment.
Fixed in 6f3a9a5f. The Paimon scalar fixture now uses BIGINT / Arrow Int64, so the shared Loon schema text is accurate. The dependent C++ and Loon tests were updated
Persist native Paimon DataSplit descriptors for scans that cannot be served safely through direct files, including merge-on-read, row-range, partitioned, and evolved tables. Stream projected Arrow batches without materializing a complete split. Keep forward-only split handling inside the Paimon reader so common sync and async scheduling remain unchanged; each chunk batch, take, and range operation owns its stream. Validate DataSplit bindings and declared row counts, preserve existing direct-file behavior, and cover automatic and forced routing, cross-batch reads, sparse take, and malformed descriptors. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Generate append-only, deletion-vector, and merge-on-read Paimon tables through the existing demo-table command. Return committed snapshot history and accept storage options so external-collection E2E can use local or object-storage fixtures. Use int64 IDs for the scalar fixture to match the existing Iceberg and Lance test schemas. Also verify DataSplit deletion-vector application and merge-on-read values in the C++ integration suite. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Carry the existing bridge error markers through Arrow C streams and translate them after import. Stream-time not-found and transient failures now retain the same retry semantics as planning and open failures, while plain storage failures remain I/O errors. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Reject persisted data-split descriptors with a zero row count because planning never emits them and they otherwise bypass stream count validation. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
123a30c to
e8758f0
Compare
Updated the first commit title to |
|
|
||
| unsafe fn open_stream( | ||
| self: &BlockingPaimonDataSplitReader, | ||
| projected_columns: &CxxVector<CxxString>, |
There was a problem hiding this comment.
Vec<String> ?
should convert vector to rust type in cpp.
There was a problem hiding this comment.
open_stream consumes the projection synchronously and does not retain it, so I used a borrowed CxxVector to avoid copying the C++ strings into an owned Rust vector. Would you still prefer converting it on the C++ side to keep the Rust interface as Vec? I can change it if so.
There was a problem hiding this comment.
for projection field copy, I do think that is fine.
There was a problem hiding this comment.
Understood. I’ll switch the projection argument to an owned Vec and convert it on the C++ side.
| @@ -34,7 +37,38 @@ | |||
| namespace milvus_storage::paimon { | |||
| namespace { | |||
There was a problem hiding this comment.
namespace { No need empty namespace. This is a bad habit of GPT.
There was a problem hiding this comment.
The anonymous namespace contains helpers used only in this translation unit and keeps their symbols file-local, consistent with the Iceberg and Vortex readers in this repository. Would you prefer using static helpers here instead? I can adjust it if that is the intended style.
There was a problem hiding this comment.
ok, just let me clean up empty namespaces across the entire storage project in the future.
| out_schema_ptr: *mut u8, | ||
| ) -> Result<()>; | ||
|
|
||
| unsafe fn open_stream( |
There was a problem hiding this comment.
can open_stream pass with footer or some meta?
Then different projection won't read footer in multi-times.
There was a problem hiding this comment.
notice that: The external table will projects each column individually.
There was a problem hiding this comment.
The current paimon-rust 0.3 API does not expose reusable file metadata across projected readers. I see two options:
-
Keep this PR on the released 0.3 API and accept repeated footer reads for now, then optimize it separately.
-
Add a scan-scoped read context with internal footer metadata caching to paimon-rust, and share it from BlockingPaimonDataSplitReader across projected streams. This is closer to paimon-cpp’s design, but requires upstream API work and likely waiting for paimon-rust 0.4.0, no earlier than the end of this month.
Would you prefer keeping this PR small, or waiting for the upstream API?
There was a problem hiding this comment.
okay, So let's just leave a FIXME(name) then.
| /// same cached metadata. | ||
| class BlockingPaimonDataSplitReader { | ||
| public: | ||
| static arrow::Result<std::unique_ptr<BlockingPaimonDataSplitReader>> Open(const std::string& metadata_json, |
There was a problem hiding this comment.
direct return std::shared_ptr? Then u don't need warp it as shared
| for (auto index : indices) { | ||
| const auto& group = metadata_->row_group_infos[index]; | ||
| ARROW_ASSIGN_OR_RAISE(auto batches, cursor->ReadRange(group.start_offset, group.end_offset)); | ||
| ARROW_ASSIGN_OR_RAISE(auto batch, CombineBatches(batches, output_schema_)); |
There was a problem hiding this comment.
It will make copy.
In fact, the "logical chunk rows" value is merely an indicator rather than a hard requirement. You can observe the approach in vortex: it actually returns a native split to inform the caller how to partition the data so that no resulting RecordBatch exceeds the limit, thereby avoiding data copying.
There was a problem hiding this comment.
We can allow chunks to be smaller, rather than copying them.
There was a problem hiding this comment.
I checked Paimon Java and paimon-cpp. Neither exposes merged DataSplit output boundaries during planning. Both use a pull-based reader (readBatch / NextBatch) and allow each returned batch to have its actual size.
For DataSplit/MOR, output boundaries are determined during merge and deletion processing, so exposing physical file boundaries from paimon-rust would not describe the logical output. Removing CombineBatches cleanly would require a variable-sized streaming contract in milvus-storage, which is broader than this PR.
| if (is_data_split()) { | ||
| ARROW_ASSIGN_OR_RAISE(auto cursor, make_data_split_cursor()); | ||
| ARROW_ASSIGN_OR_RAISE(auto selected_batches, cursor->TakeRows(indices)); | ||
| return arrow::Table::FromRecordBatches(output_schema_, selected_batches); |
There was a problem hiding this comment.
On the data-split path output_schema_ (which is simply read_schema whenever the caller supplies one, see ProjectSchema at :164) is bound to batches that carry Paimon's own stream schema, and nothing reconciles the two. CombineBatches returns the source batches untouched for any non-empty input, so get_chunk/get_chunks emit stream-schema batches and DataSplitRangeReader::schema() (:427) advertises a schema its ReadNext (:436) does not produce, while take() passes output_schema_ to arrow::Table::FromRecordBatches, which rejects any batch whose schema differs in type or nullability. A caller whose read_schema differs from Paimon's Arrow schema only in nullability therefore gets Invalid: Schema at index 0 was different from take() and from arrow::Table::FromRecordBatchReader on the range reader, even though the Paimon direct-file path and the other format readers accept the same input. Pick one rule for this path — project the stream onto output_schema_, or derive output_schema_ from the stream — and apply it to all three entry points. (Note: the original claim that take()/read_with_range() honor read_schema is not accurate; read_with_range uses the same CombineBatches, and take() only validates.)
There was a problem hiding this comment.
Fixed in 6343a578. DataSplit batches are validated and rebound to output_schema_ for chunk, take, and range reads. Incompatible fields and nulls for non-nullable fields return Invalid.
| auto column_groups = std::make_shared<api::ColumnGroups>(); | ||
| column_groups->push_back(std::make_shared<api::ColumnGroup>( | ||
| api::ColumnGroup{.columns = {"id"}, .format = LOON_FORMAT_PAIMON_TABLE, .files = {files.front()}})); | ||
| auto schema = arrow::schema({arrow::field("id", arrow::int64())}); |
There was a problem hiding this comment.
No new data-split test exercises take() or read_with_range() with a non-null read_schema: DataSplitTakeCompactsSparseBatches (:668) and DataSplitSupportsRangeAndCloneReads (:607) both build the reader with FormatReader::create(nullptr, ...), so output_schema_ is derived from Paimon's own stream schema and the schema binding cannot fail. The two tests that do pass a schema use arrow::schema({arrow::field("id", arrow::int64())}), which matches the fixture's nullable BIGINT exactly — and one of them only asserts total_rows() while the other goes through the chunk path. Add take/range cases with a read_schema that differs from Paimon's schema in nullability or type so the mismatch behaviour is pinned down rather than untested.
There was a problem hiding this comment.
Covered in 6343a578. The test uses a non-null read_schema with different nullability for take and range reads, and verifies that an incompatible type is rejected.
| } | ||
| previous = index; | ||
| } | ||
| ARROW_ASSIGN_OR_RAISE(auto cursor, make_data_split_cursor()); |
There was a problem hiding this comment.
Every public read on the data-split path opens a brand-new Paimon merge stream and forward-skips to its start, and the reader keeps no cursor between calls. get_chunk(i) delegates to get_chunks({i}), which builds a fresh cursor, so iterating a split chunk-by-chunk decodes O(N^2) rows: for a 10M-row split at the default 8192-row chunk size that is ~1220 streams and ~6 billion rows decoded instead of 10 million. This is the primary Milvus access pattern — loon_get_chunk (cpp/src/ffi/reader_c.cpp:111) is per-chunk, and api::Reader itself falls back to single-chunk reads whenever a predicate is set (cpp/src/reader.cpp:449-457). Hoisting the cursor to a reader member and reusing it while start >= position_ (falling back to a fresh stream only for backward seeks) would keep the forward-only design and make sequential iteration linear; get_chunks already proves one cursor can serve many chunks.
There was a problem hiding this comment.
Fixed in 81e4d6f. Sequential get_chunk calls reuse a reader-local forward cursor and reopen only for backward reads or after an error. Batch and range reads keep independent streams. The test covers both forward reuse and backward reopening.
There was a problem hiding this comment.
A clarification to my earlier reply: the common reader and ChunkTask changes remain removed. Removed in 4283be8. After the later O(N²) issue was identified, 81e4d6f restores only a Paimon reader-local forward cursor for sequential get_chunk calls. It reopens for backward reads or errors, while batch and range reads use independent streams, so no common scheduling capability was reintroduced.
Parse demo-table row and dimension values without throwing on malformed input, and return the existing command error path instead. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Recover not-found and transient storage classifications when the Paimon Parquet adapter stringifies FileRead errors, while leaving other permanent storage failures as I/O errors. Cover the Arrow stream path with Rust classification cases and a C++ mid-stream not-found test. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Rebind projected batches to the requested Arrow schema when column names and types match, and reject incompatible schemas before they reach take or range consumers. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Keep a reader-local forward cursor for get_chunk, and reopen it only for backward reads or failures. Batch and range operations continue to use independent streams. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Expose the Paimon RecordBatchReader wrapper through the same internal test seam used by Vortex, and cover not-found and throttling status translation with synthetic readers. Delete data files before opening the integration stream so the test validates the final ENOENT contract without depending on eager or lazy open behavior or POSIX unlink semantics. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Document the remaining repeated footer reads across projected DataSplit streams until paimon-rust exposes a reusable scan-scoped read context. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
Project the parent read schema by needed_columns and preserve the requested column order for both direct-file and data-split readers. Cover reordered subset projections for Parquet, Vortex, and MOR data splits. Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>
|
Updated in 31c3bf1. read_schema is now treated as the parent schema and projected by needed_columns in the requested order. Added coverage for reordered subset projections across Parquet/Vortex direct-file and data-split paths, including both get_chunk and take. |
Part of #593
Design + tracking: milvus-io/milvus#45881
Related tracking: milvus-io/milvus#50632
Consumed by: milvus-io/milvus#51897
What changed
Add native Apache Paimon DataSplit reads as a follow-up to #602.
data-splitscan mode and automatic fallback when direct-file reads cannot preserve Paimon semantics.Routing
autokeeps safe raw-convertible Parquet and Vortex splits on the direct-file path.autofalls back to DataSplit for merge-on-read, row ranges, partition columns, schema or data evolution, sidecars, and unsupported direct-file formats.direct-filefails explicitly when a split cannot be read safely.data-splitforces native Paimon execution.The direct-file behavior introduced by #602 remains unchanged.
Compared with the earlier combined #594 implementation, this PR is limited to DataSplit execution, uses the released native split codec, avoids whole-split materialization, coalesces normal eager and lazy loads onto a forward stream, and isolates take, range, and backward reads from the shared load cursor.
Test plan
Signed-off-by: YangYanbin warlock.yyb@alibaba-inc.com