zhouyc-ali commented on code in PR #471: URL: https://github.com/apache/paimon-cpp/pull/471#discussion_r4465883299
########## 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: Removed the fixed-shard patch; benchmarks across CPU counts and concurrency levels did not show consistent gains. ########## 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: Split native arrays before Arrow conversion and trim nested children to referenced ranges. Added encoded-string and empty/NULL-list regressions. ########## 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: Replaced preset modes with independent `vortex.read.io-coalesce.max-gap` and `vortex.read.io-coalesce.max-size` byte options. ########## 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: Removed the constructor's default arguments; callers pass the options explicitly. -- 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]
