lxy-9602 commented on code in PR #471: URL: https://github.com/apache/paimon-cpp/pull/471#discussion_r4398684719
########## cmake_modules/vortex-fmm-cache-shards.diff: ########## @@ -0,0 +1,131 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +diff --git a/vortex-layout/src/layouts/struct_/reader.rs b/vortex-layout/src/layouts/struct_/reader.rs +--- a/vortex-layout/src/layouts/struct_/reader.rs ++++ b/vortex-layout/src/layouts/struct_/reader.rs +@@ -115,7 +115,8 @@ + expanded_root_expr, + lazy_children, + field_lookup, +- partitioned_expr_cache: Default::default(), ++ // Each file evaluates few expressions; avoid CPU-count-sized shard allocations. ++ partitioned_expr_cache: DashMap::with_hasher_and_shard_amount(Default::default(), 16), + }) + } + +diff --git a/vortex-file/src/v2/file_stats_reader.rs b/vortex-file/src/v2/file_stats_reader.rs +--- a/vortex-file/src/v2/file_stats_reader.rs ++++ b/vortex-file/src/v2/file_stats_reader.rs +@@ -74,7 +74,8 @@ + file_stats, + struct_fields, + session, +- prune_cache: Default::default(), ++ // File-level pruning reuses only a small set of expressions per reader. ++ prune_cache: DashMap::with_hasher_and_shard_amount(Default::default(), 16), + } + } + +diff --git a/vortex-layout/src/segments/shared.rs b/vortex-layout/src/segments/shared.rs +--- a/vortex-layout/src/segments/shared.rs ++++ b/vortex-layout/src/segments/shared.rs +@@ -32,7 +32,8 @@ + pub fn new(inner: S) -> Self { + Self { + inner, +- in_flight: DashMap::default(), ++ // A per-file request set does not need a shard for every available CPU. ++ in_flight: DashMap::with_hasher_and_shard_amount(Default::default(), 16), + } + } + } +@@ -65,10 +66,16 @@ + + #[cfg(test)] + mod tests { ++ use std::sync::Barrier; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; +- ++ use std::thread; ++ ++ use futures::executor::block_on; ++ use futures::future::join_all; + use vortex_buffer::ByteBuffer; ++ use vortex_error::VortexResult; ++ use vortex_error::vortex_err; + + use super::*; + use crate::segments::SegmentSink; +@@ -119,5 +126,56 @@ + } + + #[tokio::test] ++ async fn test_shared_source_deduplicates_many_segments_across_threads() -> VortexResult<()> { ++ let source = CountingSegmentSource::default(); ++ let mut ids = Vec::new(); ++ let mut expected = Vec::new(); ++ for value in 0..64u8 { ++ let data = ByteBuffer::from(vec![value; 4]); ++ ids.push( ++ source ++ .segments ++ .write(SequenceId::root().downgrade(), vec![data.clone()]) ++ .await?, ++ ); ++ expected.push(data); ++ } ++ ++ let shared_source = SharedSegmentSource::new(source.clone()); ++ let barrier = Barrier::new(8); ++ thread::scope(|scope| -> VortexResult<()> { ++ let mut requests = Vec::new(); ++ for _ in 0..8 { ++ let shared_source = &shared_source; ++ let ids = &ids; ++ let expected = &expected; ++ let barrier = &barrier; ++ requests.push(scope.spawn(move || -> VortexResult<()> { ++ let pending = ids ++ .iter() ++ .map(|id| shared_source.request(*id)) ++ .collect::<Vec<_>>(); ++ // Hold every request until all threads share the same in-flight futures. ++ barrier.wait(); ++ for (actual, expected) in block_on(join_all(pending)).into_iter().zip(expected) ++ { ++ assert_eq!(actual?.unwrap_host(), *expected); ++ } ++ Ok(()) ++ })); ++ } ++ for request in requests { ++ request ++ .join() ++ .map_err(|_| vortex_err!("segment requester panicked"))??; ++ } ++ Ok(()) ++ })?; ++ ++ assert_eq!(source.request_count.load(Ordering::SeqCst), ids.len()); ++ Ok(()) ++ } ++ Review Comment: Could you please describe the real benefits this change will bring? After all, diffs are hard to maintain. ########## crates/vortex_callback_io/callback_io.rs: ########## @@ -615,25 +916,296 @@ mod tests { vx_array_free(vx_array_ptr); vx_dtype_free(vx_dtype_ptr); - let file = File::open(&path).unwrap(); - let size = file.metadata().unwrap().len(); - let context = Box::into_raw(Box::new(FileContext { file })); - let callbacks = vx_input_callbacks { - ctx: context.cast::<c_void>(), - read_at_fn: Some(file_read_at), - release_fn: Some(file_release), + for local in [false, true] { + let file = File::open(&path).unwrap(); + let size = file.metadata().unwrap().len(); + let context = Box::into_raw(Box::new(FileContext { file })); + let callbacks = vx_input_callbacks { + ctx: context.cast::<c_void>(), + read_at_fn: Some(file_read_at), + release_fn: Some(file_release), + }; + + let data_source = if local { + vx_data_source_new_callback_with_coalesce( + session, Review Comment: It seems we only support the local and object store modes right now. We have HDFS scenarios that need to support configurable `max_gap` and `max_size`, so we can’t just hard-code it based on the mode. ########## src/paimon/format/vortex/vortex_reader_builder.h: ########## @@ -24,14 +24,21 @@ #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/format/reader_builder.h" #include "paimon/format/vortex/vortex_file_batch_reader.h" +#include "paimon/format/vortex/vortex_predicate_converter.h" #include "paimon/memory/memory_pool.h" namespace paimon::vortex { class VortexReaderBuilder : public ReaderBuilder { public: - explicit VortexReaderBuilder(int32_t batch_size) - : batch_size_(batch_size), pool_(GetDefaultPool()), arrow_pool_(GetArrowPool(pool_)) {} + explicit VortexReaderBuilder( + int32_t batch_size, const VortexPredicateOptions& predicate_options = {}, + VortexIoCoalesceMode coalesce_mode = VortexIoCoalesceMode::OBJECT_STORAGE) Review Comment: Please avoid default parameters in the production code. ########## cmake_modules/vortex.diff: ########## @@ -7,3 +7,74 @@ diff --git a/vortex-ffi/src/lib.rs b/vortex-ffi/src/lib.rs mod array_iterator; mod binary; mod data_source; +diff --git a/vortex-ffi/src/scan.rs b/vortex-ffi/src/scan.rs +--- a/vortex-ffi/src/scan.rs ++++ b/vortex-ffi/src/scan.rs +@@ -25,8 +25,11 @@ + use vortex::error::VortexResult; + use vortex::error::vortex_bail; + use vortex::error::vortex_ensure; ++use vortex::expr::merge; ++use vortex::expr::pack; + use vortex::expr::root; + use vortex::io::runtime::BlockingRuntime; ++use vortex::layout::layouts::row_idx::row_idx; + use vortex::layout::scan::arrow::RecordBatchIteratorAdapter; + use vortex::scan::DataSourceScan; + use vortex::scan::Partition; +@@ -112,4 +115,7 @@ + pub limit: u64, + /// If true, return in storage order. + pub ordered: bool, ++ /// If true, append a `_row_idx` UInt64 column carrying each returned row's original physical ++ /// row index, stable across `filter` and `selection`. The host strips it before use. ++ pub include_row_idx: bool, + } +@@ -182,5 +188,12 @@ + Ok(ScanRequest { +- projection, ++ // paimon-cpp: when include_row_idx is set, append a hidden `_row_idx` column so the host can ++ // recover each surviving row's original physical index after `filter`/`selection` drop rows. ++ // row_idx() is substituted with the real index array during the scan; the host strips it. ++ projection: if opts.include_row_idx { ++ merge([projection, pack([("_row_idx", row_idx())], false.into())]) ++ } else { ++ projection ++ }, + filter, + row_range, + selection, +@@ -356,3 +369,5 @@ +- let schema = dtype.to_arrow_schema()?; ++ // paimon-cpp: export standard (non-view) string/binary types so the C++ reader skips its ++ // per-row view->standard rebuild; execute_arrow casts to this target internally. ++ let schema = crate::callback_io::normalize_arrow_schema_views(dtype.to_arrow_schema()?); + let schema = Arc::new(schema); + let target = Field::new_struct("", schema.fields().clone(), false); +diff --git a/vortex-ffi/src/dtype.rs b/vortex-ffi/src/dtype.rs +--- a/vortex-ffi/src/dtype.rs ++++ b/vortex-ffi/src/dtype.rs +@@ -344,7 +344,9 @@ + ) -> c_int { + try_or(err, 1, || { Review Comment: **[P2] Bound native batches before converting view arrays to Utf8/Binary** Normalizing the target schema here makes `execute_arrow` materialize ordinary Utf8/Binary arrays for the entire native chunk, before `VortexFileBatchReader::NextBatch()` slices it to the configured `batch_size`. Previously, the view-to-standard conversion happened after that slicing. For dictionary-encoded long strings, this can expand a small encoded chunk into a large contiguous allocation. If a column’s expanded payload exceeds the approximately 2 GiB limit of 32-bit offsets, conversion can fail even though each requested C++ batch would fit. Reducing the C++ `batch_size` cannot prevent this because conversion has already happened. Could we split the native output before `execute_arrow`, or retain view arrays across the FFI and convert after slicing? This would preserve bounded materialization and avoid introducing a larger conversion boundary. ########## src/paimon/format/vortex/vortex_file_batch_reader.h: ########## @@ -50,7 +52,9 @@ class VortexFileBatchReader : public FileBatchReader { public: static Result<std::unique_ptr<VortexFileBatchReader>> Create( const std::shared_ptr<InputStream>& input, int32_t batch_size, - const std::shared_ptr<arrow::MemoryPool>& arrow_pool); + const std::shared_ptr<arrow::MemoryPool>& arrow_pool, + const VortexPredicateOptions& predicate_options = {}, + VortexIoCoalesceMode coalesce_mode = VortexIoCoalesceMode::OBJECT_STORAGE); Review Comment: Please avoid default params and keep pool as the last parameter. ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## @@ -262,14 +144,22 @@ Result<std::unique_ptr<VortexFileBatchReader>> VortexFileBatchReader::Create( auto length = static_cast<uint64_t>(signed_length); auto input_context = std::make_shared<VortexInputContext>(input); - VxSessionPtr session(vx_session_new(), vx_session_free); + VxSessionPtr session(vx_paimon_read_session_new(), vx_session_free); if (session == nullptr) { return Status::IOError("failed to create Vortex session"); } vx_error* error = nullptr; + const bool local_coalesce = coalesce_mode == VortexIoCoalesceMode::LOCAL; + // Smaller gaps limit read amplification for explicitly selected low-latency sources. + constexpr uint64_t kLocalMaxGap = 64ULL << 10; + constexpr uint64_t kLocalMaxSize = 4ULL << 20; VxDataSourcePtr data_source( - vx_data_source_new_callback(session.get(), VortexInputContext::MakeCallbacks(input_context), - length, &error), + local_coalesce + ? vx_data_source_new_callback_with_coalesce( + session.get(), VortexInputContext::MakeCallbacks(input_context), length, + kLocalMaxGap, kLocalMaxSize, &error) + : vx_data_source_new_callback( + session.get(), VortexInputContext::MakeCallbacks(input_context), length, &error), Review Comment: I understand that these two parameters, `max_gap` and `max_size`, can be configured by the user. Would it be possible to remove the default callback `vx_data_source_new_callback`, and keep only `vx_data_source_new_callback_with_coalesce`? ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## @@ -306,20 +193,76 @@ Result<std::unique_ptr<VortexFileBatchReader>> VortexFileBatchReader::Create( // scan may be consumed only once. return std::unique_ptr<VortexFileBatchReader>(new VortexFileBatchReader( input, batch_size, std::move(input_context), std::move(session), std::move(data_source), - VxScanPtr(nullptr, vx_scan_free), file_schema, struct_type, total_rows, arrow_pool)); + VxScanPtr(nullptr, vx_scan_free), file_schema, total_rows, arrow_pool, predicate_options)); } VortexFileBatchReader::~VortexFileBatchReader() { CloseInternal(); } Result<bool> VortexFileBatchReader::OpenNextPartitionStream() { + if (has_selection_ && selected_row_count_ == 0) { + return false; + } if (scan_ == nullptr) { vx_error* error = nullptr; // Scan in storage order: Vortex defaults to ordered=false and may emit chunks out of order, // which would misalign the physical row positions NextBatch assigns. - vx_scan_options options{}; // zero-init: all columns, no filter/row-range/selection/limit + vx_scan_options options{}; options.ordered = true; + // Missing fields are null-filled by AlignToReadSchema and cannot be selected here. + // The scan clones the projection, so these expressions need only outlive scan creation. + VxExpressionPtr root(nullptr, vx_expression_free); + VxExpressionPtr projection(nullptr, vx_expression_free); + std::vector<std::string> names; + std::vector<const char*> name_ptrs; + if (read_schema_ != nullptr) { + names.reserve(read_schema_->num_fields()); + // Native merge requires distinct fields; output alignment restores repeated columns. + std::vector<uint8_t> selected_columns(file_schema_->num_fields(), 0); + for (const std::shared_ptr<arrow::Field>& field : read_schema_->fields()) { + const int32_t field_index = file_schema_->GetFieldIndex(field->name()); + if (field_index >= 0 && selected_columns[field_index] == 0) { + names.push_back(field->name()); + selected_columns[field_index] = 1; Review Comment: Actually, the upstream guarantees that the provided `read_schema` won’t contain any duplicate fields. It also won’t reference fields that don’t exist—if such a case happens, could we directly fail with an error? ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## @@ -448,24 +409,68 @@ Result<std::unique_ptr<::ArrowSchema>> VortexFileBatchReader::GetFileSchema() co Status VortexFileBatchReader::SetReadSchema( ::ArrowSchema* read_schema, const std::shared_ptr<Predicate>& predicate, const std::optional<RoaringBitmap32>& selection_bitmap) { + if (closed_) { + return Status::Invalid("Vortex reader is closed"); + } if (read_schema == nullptr) { return Status::Invalid("Vortex read schema is nullptr"); } - // TODO(zhouyc-ali): push the row selection into the Vortex scan - (void)selection_bitmap; - // No pushdown: the whole file is scanned and reading restarts from row 0, so drop the - // single-use scan; the next read re-creates it. + std::vector<uint64_t> selection_indices; + bool selection_excludes = false; + uint64_t selected_row_count = total_rows_; + if (selection_bitmap) { + const int32_t cardinality = selection_bitmap->Cardinality(); + if (cardinality < 0 || static_cast<uint64_t>(cardinality) > total_rows_) { + return Status::Invalid("Vortex selection cardinality exceeds the file row count"); + } + selected_row_count = static_cast<uint64_t>(cardinality); + selection_excludes = selected_row_count > total_rows_ / 2; + selection_indices.reserve(static_cast<size_t>( + selection_excludes ? total_rows_ - selected_row_count : selected_row_count)); + uint64_t next_row = 0; + const auto end = selection_bitmap->End(); + for (auto iter = selection_bitmap->Begin(); iter != end; ++iter) { + const auto row_id = static_cast<uint64_t>(*iter); + if (row_id >= total_rows_) { + return Status::Invalid( + fmt::format("Vortex selected row {} is out of range {}", row_id, total_rows_)); + } + if (selection_excludes) { + // Enumerate gaps directly instead of first expanding all selected rows. + for (; next_row < row_id; ++next_row) { + selection_indices.push_back(next_row); + } + next_row = row_id + 1; + } else { + selection_indices.push_back(row_id); + } + } + if (selection_excludes) { + for (; next_row < total_rows_; ++next_row) { + selection_indices.push_back(next_row); + } + } + } + // Leave the active scan and its row mapping unchanged if validation or import fails. PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> imported_read_schema, arrow::ImportSchema(read_schema)); + // Recreate the single-use scan on the next read with the new projection and predicates. read_schema_ = std::move(imported_read_schema); predicate_filter_ = std::dynamic_pointer_cast<PredicateFilter>(predicate); + has_selection_ = selection_bitmap.has_value(); + selection_excludes_ = selection_excludes; + selected_row_count_ = selected_row_count; + selection_indices_ = std::move(selection_indices); + metrics_->SetCounter("vortex.selection-index-count", selection_indices_.size()); ReleaseStream(); Review Comment: Please define the metric name in `defs.h` first. Also, could we rename it to `vortex.selection-row-count`? “index” is a bit ambiguous since it could be interpreted as an index. ########## src/paimon/format/vortex/vortex_file_format.cpp: ########## @@ -29,7 +31,33 @@ namespace paimon::vortex { Result<std::unique_ptr<ReaderBuilder>> VortexFileFormat::CreateReaderBuilder( int32_t batch_size) const { - return std::make_unique<VortexReaderBuilder>(batch_size); + VortexPredicateOptions predicate_options; + const auto membership_mode = options_.find("vortex.read.membership-pushdown"); + if (membership_mode != options_.end()) { Review Comment: Similar changes needed: please adjust the parameter naming and their positions as well. ########## docs/source/user_guide/read.rst: ########## @@ -41,6 +41,26 @@ cross-language task scheduling and interaction (e.g., Java and C++), substantial across the two language ecosystems. +Vortex Read Options +------------------- + +For Vortex files, pass these options through ``ReadContextBuilder::AddOption``: + +- ``vortex.read.membership-pushdown`` accepts ``auto`` (default), ``force`` or ``off``. + ``auto`` limits each IN/NOT IN expression to 64 comparison and boolean nodes after + deduplication and integer-range compression. ``force`` bypasses this budget while + retaining type and value safety checks. ``off`` disables only IN/NOT IN pushdown. +- ``vortex.read.io-coalesce`` accepts ``object-storage`` (default) or ``local``. + The default retains Vortex's object-storage coalescing policy. ``local`` uses a + 64 KiB maximum gap and a 4 MiB coalescing limit for low-latency storage. + Choose the mode explicitly; it is not selected from the file path. + +Native pushdown may return candidate rows requiring further filtering. For exact +results, retain the full residual predicate: enable ``EnablePredicateFilter(true)`` +with ``TableRead``, or apply equivalent filtering in the caller. ``force`` does not +bypass this requirement. + Review Comment: Also, why is this documentation placed in `read.rst`? Please check where it should live—it seems intended for external users (the user guide) rather than format-configuration documentation. Please adjust the doc location accordingly. ########## test/inte/scan_and_read_inte_test.cpp: ########## @@ -3727,6 +3727,233 @@ TEST_F(ScanAndReadInteTest, TestMosaicJavaAndPythonCompatibility) { #endif #ifdef PAIMON_ENABLE_VORTEX +namespace { + +Result<std::shared_ptr<arrow::ChunkedArray>> ReadVortexSdkRows( + const std::string& table_path, const std::vector<std::shared_ptr<Split>>& splits, + const std::vector<std::string>& projection, const std::shared_ptr<Predicate>& predicate, + bool exact_filter, bool late_materializing) { + ReadContextBuilder builder(table_path); + builder.SetReadFieldNames(projection) + .SetPredicate(predicate) + .EnablePredicateFilter(exact_filter) + .EnableLateMaterializing(late_materializing) + .EnablePrefetch(true) + .SetPrefetchBatchCount(2) + .SetPrefetchMaxParallelNum(2) + .AddOption(Options::READ_BATCH_SIZE, "2"); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> context, builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> read, TableRead::Create(std::move(context))); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> reader, read->CreateReader(splits)); + return ReadResultCollector::CollectResult(std::move(reader)); +} + +// Append splits need not have a stable file order. Compare full rows as a multiset. +void AssertVortexSdkRows(const std::shared_ptr<arrow::ChunkedArray>& actual, + const std::shared_ptr<arrow::DataType>& type, + const std::string& expected_json, bool exact_filter = true) { + auto expected = arrow::ipc::internal::json::ArrayFromJSON(type, expected_json).ValueOrDie(); + if (!actual) { + ASSERT_EQ(expected->length(), 0); + return; + } + ASSERT_TRUE(actual->type()->Equals(type)) << actual->type()->ToString(); + if (exact_filter) { + ASSERT_EQ(actual->length(), expected->length()) << actual->ToString(); + } else { + ASSERT_GE(actual->length(), expected->length()); + } + arrow::ArrayVector candidates; + for (const auto& chunk : actual->chunks()) { + ASSERT_TRUE(chunk->ValidateFull().ok()); + for (int64_t row = 0; row < chunk->length(); ++row) { + candidates.push_back(chunk->Slice(row, 1)); + } + } + std::vector<bool> used(candidates.size(), false); + for (int64_t row = 0; row < expected->length(); ++row) { + bool found = false; + for (size_t candidate = 0; candidate < candidates.size(); ++candidate) { + if (!used[candidate] && expected->Slice(row, 1)->Equals(candidates[candidate])) { + used[candidate] = true; + found = true; + break; + } + } + ASSERT_TRUE(found) << "Missing row: " << expected->Slice(row, 1)->ToString() + << "\nActual: " << actual->ToString(); + } +} + +} // namespace + +TEST_F(ScanAndReadInteTest, TestVortexSdkPredicatesWithResidualAndLateMaterialization) { + auto dir = UniqueTestDirectory::Create("local"); + ASSERT_NE(dir, nullptr); + const arrow::FieldVector fields = {arrow::field("id", arrow::int32()), + arrow::field("tag", arrow::utf8()), + arrow::field("payload", arrow::utf8())}; + const std::map<std::string, std::string> options = {{Options::FILE_FORMAT, "vortex"}, + {Options::FILE_SYSTEM, "local"}, + {Options::BUCKET, "-1"}, + {Options::WRITE_ONLY, "true"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper, + TestHelper::Create(dir->Str(), arrow::schema(fields), {}, {}, options, + /*is_streaming_mode=*/false)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([ + [0,"skip","zero"],[1,null,"one"],[2,"keep","two"],[3,"drop","three"], + [4,"keep",null],[5,"keep","five"],[6,"skip","six"],[7,"keep","seven"] + ])", + {}, 0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), 0, std::nullopt)); + const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits, + helper->NewScan(StartupMode::LatestFull(), std::nullopt, false)); + const auto result_type = arrow::struct_( + {arrow::field("_VALUE_KIND", arrow::int8()), fields[2], fields[0], fields[1]}); + auto equal = PredicateBuilder::Equal(1, "id", FieldType::INT, Literal(int32_t{4})); + auto greater = PredicateBuilder::GreaterThan(1, "id", FieldType::INT, Literal(int32_t{1})); + auto tag = + PredicateBuilder::Equal(2, "tag", FieldType::STRING, Literal(FieldType::STRING, "keep", 4)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> conjunction, + PredicateBuilder::And({greater, tag})); + // Sparse IN(64) exceeds the auto expression budget and requires the SDK residual filter. + std::vector<Literal> sparse = {Literal(int32_t{2}), Literal(int32_t{5})}; + for (int32_t index = 0; index < 62; ++index) { + sparse.emplace_back(1000 + index * 2); + } + auto membership = PredicateBuilder::In(1, "id", FieldType::INT, sparse); + const std::vector<std::pair<std::shared_ptr<Predicate>, std::string>> cases = { + {equal, R"([[0,null,4,"keep"]])"}, + {conjunction, R"([[0,"two",2,"keep"],[0,null,4,"keep"], + [0,"five",5,"keep"],[0,"seven",7,"keep"]])"}, + {membership, R"([[0,"two",2,"keep"],[0,"five",5,"keep"]])"}}; + for (bool late_materializing : {false, true}) { + SCOPED_TRACE(late_materializing); + for (const auto& [predicate, expected] : cases) { + SCOPED_TRACE(predicate->ToString()); + for (bool exact_filter : {false, true}) { + SCOPED_TRACE(exact_filter); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<arrow::ChunkedArray> actual, + ReadVortexSdkRows(table_path, splits, {"payload", "id", "tag"}, predicate, + exact_filter, late_materializing)); + // A non-exact reader may prune candidates, but it must retain every matching row. + ASSERT_NO_FATAL_FAILURE( + AssertVortexSdkRows(actual, result_type, expected, exact_filter)); + } Review Comment: Please tighten your validation. Enabling either `late_materializing` or `exact_filter` is sufficient to guarantee correct query results and predicate accuracy. ########## docs/source/user_guide/read.rst: ########## @@ -41,6 +41,26 @@ cross-language task scheduling and interaction (e.g., Java and C++), substantial across the two language ecosystems. +Vortex Read Options +------------------- + +For Vortex files, pass these options through ``ReadContextBuilder::AddOption``: + +- ``vortex.read.membership-pushdown`` accepts ``auto`` (default), ``force`` or ``off``. + ``auto`` limits each IN/NOT IN expression to 64 comparison and boolean nodes after + deduplication and integer-range compression. ``force`` bypasses this budget while + retaining type and value safety checks. ``off`` disables only IN/NOT IN pushdown. +- ``vortex.read.io-coalesce`` accepts ``object-storage`` (default) or ``local``. + The default retains Vortex's object-storage coalescing policy. ``local`` uses a + 64 KiB maximum gap and a 4 MiB coalescing limit for low-latency storage. + Choose the mode explicitly; it is not selected from the file path. + +Native pushdown may return candidate rows requiring further filtering. For exact +results, retain the full residual predicate: enable ``EnablePredicateFilter(true)`` +with ``TableRead``, or apply equivalent filtering in the caller. ``force`` does not +bypass this requirement. + Review Comment: Can you confirm whether this configuration is the default for Java? If it isn’t, please use a more flexible configuration instead of an enum that limits choices to a fixed set of categories. ########## docs/source/user_guide/read.rst: ########## @@ -41,6 +41,26 @@ cross-language task scheduling and interaction (e.g., Java and C++), substantial across the two language ecosystems. +Vortex Read Options +------------------- + +For Vortex files, pass these options through ``ReadContextBuilder::AddOption``: + +- ``vortex.read.membership-pushdown`` accepts ``auto`` (default), ``force`` or ``off``. + ``auto`` limits each IN/NOT IN expression to 64 comparison and boolean nodes after + deduplication and integer-range compression. ``force`` bypasses this budget while + retaining type and value safety checks. ``off`` disables only IN/NOT IN pushdown. +- ``vortex.read.io-coalesce`` accepts ``object-storage`` (default) or ``local``. + The default retains Vortex's object-storage coalescing policy. ``local`` uses a + 64 KiB maximum gap and a 4 MiB coalescing limit for low-latency storage. + Choose the mode explicitly; it is not selected from the file path. + +Native pushdown may return candidate rows requiring further filtering. For exact +results, retain the full residual predicate: enable ``EnablePredicateFilter(true)`` +with ``TableRead``, or apply equivalent filtering in the caller. ``force`` does not +bypass this requirement. + Review Comment: Please put these parameters into a `vortex_defs.h` file. ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## @@ -262,14 +144,22 @@ Result<std::unique_ptr<VortexFileBatchReader>> VortexFileBatchReader::Create( auto length = static_cast<uint64_t>(signed_length); auto input_context = std::make_shared<VortexInputContext>(input); - VxSessionPtr session(vx_session_new(), vx_session_free); + VxSessionPtr session(vx_paimon_read_session_new(), vx_session_free); if (session == nullptr) { return Status::IOError("failed to create Vortex session"); } vx_error* error = nullptr; + const bool local_coalesce = coalesce_mode == VortexIoCoalesceMode::LOCAL; + // Smaller gaps limit read amplification for explicitly selected low-latency sources. + constexpr uint64_t kLocalMaxGap = 64ULL << 10; + constexpr uint64_t kLocalMaxSize = 4ULL << 20; Review Comment: For this kind of static variable, could we keep them centralized in the class or in the corresponding `defs` file, rather than defining them inside the function? They’re easy to miss that way. ########## src/paimon/format/vortex/vortex_file_batch_reader.h: ########## @@ -118,8 +118,21 @@ class VortexFileBatchReader : public FileBatchReader { uint64_t previous_first_row_ = std::numeric_limits<uint64_t>::max(); uint64_t previous_batch_row_count_ = 0; + // Keep the smaller side of the bitmap. Dense selections store excluded physical row ids, + // avoiding an O(file rows) index list both here and in the native scan's copy. + bool has_selection_ = false; + bool selection_excludes_ = false; + uint64_t selected_row_count_ = 0; + std::vector<uint64_t> selection_indices_; + + // A pushed filter returns physical row ids in a hidden _row_idx column. Retain the last + // batch's column because emitted offsets no longer identify physical rows after filtering. + bool has_filter_ = false; + std::shared_ptr<arrow::Array> previous_batch_row_idx_; + std::shared_ptr<arrow::MemoryPool> arrow_pool_; - std::shared_ptr<PredicateFilter> predicate_filter_; // from SetReadSchema; not pushed down + VortexPredicateOptions predicate_options_; + std::shared_ptr<PredicateFilter> predicate_filter_; Review Comment: It seems there are quite a lot of member variables. Would it be possible to consider encapsulating some of them? ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## @@ -306,20 +193,76 @@ Result<std::unique_ptr<VortexFileBatchReader>> VortexFileBatchReader::Create( // scan may be consumed only once. return std::unique_ptr<VortexFileBatchReader>(new VortexFileBatchReader( input, batch_size, std::move(input_context), std::move(session), std::move(data_source), - VxScanPtr(nullptr, vx_scan_free), file_schema, struct_type, total_rows, arrow_pool)); + VxScanPtr(nullptr, vx_scan_free), file_schema, total_rows, arrow_pool, predicate_options)); } VortexFileBatchReader::~VortexFileBatchReader() { CloseInternal(); } Result<bool> VortexFileBatchReader::OpenNextPartitionStream() { + if (has_selection_ && selected_row_count_ == 0) { + return false; + } if (scan_ == nullptr) { vx_error* error = nullptr; // Scan in storage order: Vortex defaults to ordered=false and may emit chunks out of order, // which would misalign the physical row positions NextBatch assigns. - vx_scan_options options{}; // zero-init: all columns, no filter/row-range/selection/limit + vx_scan_options options{}; options.ordered = true; + // Missing fields are null-filled by AlignToReadSchema and cannot be selected here. + // The scan clones the projection, so these expressions need only outlive scan creation. + VxExpressionPtr root(nullptr, vx_expression_free); + VxExpressionPtr projection(nullptr, vx_expression_free); + std::vector<std::string> names; + std::vector<const char*> name_ptrs; + if (read_schema_ != nullptr) { + names.reserve(read_schema_->num_fields()); + // Native merge requires distinct fields; output alignment restores repeated columns. + std::vector<uint8_t> selected_columns(file_schema_->num_fields(), 0); + for (const std::shared_ptr<arrow::Field>& field : read_schema_->fields()) { + const int32_t field_index = file_schema_->GetFieldIndex(field->name()); + if (field_index >= 0 && selected_columns[field_index] == 0) { + names.push_back(field->name()); + selected_columns[field_index] = 1; + } + } + if (!names.empty()) { + name_ptrs.reserve(names.size()); + for (const std::string& name : names) { + name_ptrs.push_back(name.c_str()); + } + root = VxExpressionPtr(vx_expression_root(), vx_expression_free); + projection = VxExpressionPtr( + vx_expression_select(name_ptrs.data(), name_ptrs.size(), root.get()), + vx_expression_free); + options.projection = projection.get(); + } + } + // These FFI modes take individual indices. A dense bitmap sends only excluded rows; + // an all-selected bitmap needs no indices and uses the default INCLUDE_ALL mode. + if (has_selection_ && !selection_indices_.empty()) { + options.selection.idx = selection_indices_.data(); + options.selection.idx_len = selection_indices_.size(); + options.selection.include = + selection_excludes_ ? VX_SELECTION_EXCLUDE_RANGE : VX_SELECTION_INCLUDE_RANGE; + } + // Filtering drops rows; include_row_idx preserves their original physical positions. + // The scan clones the filter, so it can be freed after scan creation. + VxExpressionPtr filter(nullptr, vx_expression_free); + // The hidden index must not collide with a file field or a requested missing field. + const bool has_row_idx_column = + file_schema_->GetFieldIndex(kRowIdxColumn) >= 0 || + (read_schema_ != nullptr && !read_schema_->GetAllFieldIndices(kRowIdxColumn).empty()); + if (!has_row_idx_column) { Review Comment: It seems the current code assumes `read_schema_` and the file schema have no relationship. Could we simplify the design accordingly? Assuming the upstream guarantees that the provided `read_schema` won’t contain duplicate fields and won’t reference non-existent fields, we can explicitly validate the conditions in `SetReadSchema()`, and return an error if any requirement isn’t met. ########## src/paimon/format/vortex/vortex_file_batch_reader.cpp: ########## Review Comment: Could you please refactor this part of the code? For example, the creation of `scan_` could be extracted into a separate function. The current function is a bit too long. ########## test/inte/scan_and_read_inte_test.cpp: ########## @@ -3727,6 +3727,233 @@ TEST_F(ScanAndReadInteTest, TestMosaicJavaAndPythonCompatibility) { #endif #ifdef PAIMON_ENABLE_VORTEX +namespace { + +Result<std::shared_ptr<arrow::ChunkedArray>> ReadVortexSdkRows( + const std::string& table_path, const std::vector<std::shared_ptr<Split>>& splits, + const std::vector<std::string>& projection, const std::shared_ptr<Predicate>& predicate, + bool exact_filter, bool late_materializing) { + ReadContextBuilder builder(table_path); + builder.SetReadFieldNames(projection) + .SetPredicate(predicate) + .EnablePredicateFilter(exact_filter) + .EnableLateMaterializing(late_materializing) + .EnablePrefetch(true) + .SetPrefetchBatchCount(2) + .SetPrefetchMaxParallelNum(2) + .AddOption(Options::READ_BATCH_SIZE, "2"); Review Comment: I’m curious whether `vortex` supports prefetch. If it doesn’t, why are the `EnablePrefetch` and related options being configured? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
