feat(vortex): add predicate pushdown and optimize selective reads by zhouyc-ali · Pull Request #471 · apache/paimon-cpp

zhouyc-ali · 2026-10-27T10:03:35Z

Purpose

Linked issue: #319.

Push supported predicates, projections and bitmap selections into native Vortex scans while preserving physical file row IDs and Paimon's residual-filter semantics.

Support comparisons, NULL checks, IN and NOT IN with a configurable list-size limit. Normalize decimal literals only when the value is unchanged; fall back for lossy timestamps, floating-point comparisons and legal temporal constants outside the native range. Type mismatches and genuine construction errors are reported.
Reduce selective-read overhead by reusing default callback-reader sessions and lazily allocating dictionary filter caches. Add independent I/O coalescing byte limits while retaining the object-storage defaults and Vortex's default cache sharding.
Bound view-to-standard Arrow materialization by splitting native batches first and trimming nested list children to referenced ranges. Validate pinned native patch states and extend the benchmark runner to support Vortex. Normalize new LTZ file metadata to UTC without changing epoch values.

Related Java NaN fix: apache/paimon#10985.

Tests

Added and updated tests covering:

Predicate conversion and native scan results for comparisons, NULL checks,
IN and NOT IN, including duplicates, integer boundaries, configurable list limits,
and combined AND/OR predicates.
Safe fallback for unsupported conditions, unbound columns, lossy timestamps,
floating-point semantics, and membership size limits. Type mismatches and
malformed/native construction errors are checked separately; legal temporal
constants outside the native range fall back. Fixed-seed differential tests
check native results before SDK residual filtering.
Lossless decimal normalization to the file's precision/scale for comparisons
and IN/NOT IN, with fallback for rounding or overflow.
Reordered, nested and empty projections, schema metadata preservation,
duplicate/missing-field and incompatible-type rejection, and SDK NULL filling
for evolved fields. Empty-projection tests inspect native schemas and
callback reads to verify omitted payload columns are skipped.
Bounded export of encoded strings/binary and nested List/ListView/Struct arrays,
including empty/NULL nested children and shared child arrays; unreferenced
elements must not be materialized.
Sparse, dense, full, and empty bitmap selections combined with predicates
and projection, preserving physical row IDs across batches. The single-deletion
regression verifies one exclusion index instead of expanding all surviving rows.
Large full selections and single deletions cover compressed bounds checks,
physical row mapping and signed 32-bit row-ID boundaries without modifying
the input bitmap.
Reader reset and invalid reconfiguration, ensuring failed resets preserve the
active scan and returned batches remain valid after reset or reader destruction.
Closing a retained reader releases selection storage and hidden row-index
arrays while separately retained output batches remain valid.
Arrow stream schema failures preserve native error details, prefer callback
errors and handle null native messages before releasing the stream.
Independent I/O coalescing limits preserving values and row IDs, injected
read/write failures, waiting for in-flight callbacks on abort, and deferred
stream closure.
C++ SDK reads with both output APIs, residual filtering and late materialization
enabled/disabled. Complete append probe/payload passes and enabled residual
filtering require exact rows; other candidate paths must retain all matches.
Primary-key updates/deletes check merged-row integrity before value filtering,
and schema evolution covers missing columns and a user column named _row_idx.

Local validation passed 130 Vortex C++ tests and 33 SDK/ORC/Parquet regressions.
The subsequent declaration-style fixes passed the affected regression and
clang-tidy checks. The unchanged native library previously passed 106 Rust
tests; TSAN was not rerun locally for this revision.

API and Format

No changes to public C++ API signatures or the table protocol. The stored format remains Vortex; new LTZ files use UTC metadata without changing epoch values. Add the runtime reader options:

vortex.read.in-pushdown.max-values: default 32; 0 disables IN/NOT IN pushdown. The original list length, including duplicates, is counted; oversized lists fall back without truncation.
vortex.read.io-coalesce.max-gap: default 1048576 bytes (1 MiB).
vortex.read.io-coalesce.max-size: default 16777216 bytes (16 MiB).

The private Vortex FFI is extended; its patched header and native library must be rebuilt together. Readers continue to require SDK residual filtering (EnablePredicateFilter(true)) or equivalent filtering by the caller for exact results. The shared read path and the meanings of its filtering flags are unchanged.

Documentation

Document the numeric Vortex reader options, defaults, concrete fallback rules, residual-filter requirement and LTZ metadata compatibility in docs/source/api/file_format.rst. Remove the format-specific additions from docs/source/user_guide/read.rst.

Generative AI tooling

Generated-by: OpenAI Codex 0.163.0

Retain the smaller side of bitmap selections and preserve physical row mapping. Add regression coverage for dense, sparse, and reconfigured selections. Use callback I/O in native-scan test fixtures to avoid the Rust async filesystem path reported by TSAN, without adding suppressions. Add Apache license headers to both Vortex patch files.

lxy-9602 · 2026-10-28T09:04:52Z

+        assert_eq!(source.request_count.load(Ordering::SeqCst), ids.len());
+        Ok(())
+    }
+


Could you please describe the real benefits this change will bring? After all, diffs are hard to maintain.

Removed the fixed-shard patch; benchmarks across CPU counts and concurrency levels did not show consistent gains.

lxy-9602 · 2026-10-28T09:41:22Z

+++ b/vortex-ffi/src/dtype.rs
+@@ -344,7 +344,9 @@
+ ) -> c_int {
+     try_or(err, 1, || {


[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.

Split native arrays before Arrow conversion and trim nested children to referenced ranges. Added encoded-string and empty/NULL-list regressions.

lxy-9602 · 2026-10-28T09:55:48Z

+
+                let data_source = if local {
+                    vx_data_source_new_callback_with_coalesce(
+                        session,


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.

Replaced preset modes with independent vortex.read.io-coalesce.max-gap and vortex.read.io-coalesce.max-size byte options.

lxy-9602 · 2026-10-28T10:00:25Z

-        : 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)


Please avoid default parameters in the production code.

Removed the constructor's default arguments; callers pass the options explicitly.

lxy-9602 · 2026-10-28T10:05:48Z

+results, retain the full residual predicate: enable ``EnablePredicateFilter(true)``
+with ``TableRead``, or apply equivalent filtering in the caller. ``force`` does not
+bypass this requirement.
+


Please put these parameters into a vortex_defs.h file.

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.

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.

Moved configuration keys, defaults and option structs into vortex_defs.h.

The 1 MiB/16 MiB I/O defaults match Java/JNI's object-store path. I/O limits and the IN/NOT IN limit are now numeric options.

Moved Vortex options to api/file_format.rst and removed all Vortex additions from read.rst.

lxy-9602 · 2026-10-28T13:07:22Z

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.

Extracted scan creation into CreateScan(), leaving partition and stream advancement in OpenNextPartitionStream().

lxy-9602 · 2026-10-28T13:26:04Z

+    selected_row_count_ = selected_row_count;
+    selection_indices_ = std::move(selection_indices);
+    metrics_->SetCounter("vortex.selection-index-count", selection_indices_.size());
    ReleaseStream();


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.

Centralized metric names in vortex_defs.h and renamed the counter to vortex.selection-row-count.

lxy-9602 · 2026-10-28T13:41:54Z

-    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()) {


Similar changes needed: please adjust the parameter naming and their positions as well.

Centralized option definitions in vortex_defs.h and made argument order consistent, with the pool last.

lxy-9602 · 2026-10-28T13:45:05Z

+        .EnablePrefetch(true)
+        .SetPrefetchBatchCount(2)
+        .SetPrefetchMaxParallelNum(2)
+        .AddOption(Options::READ_BATCH_SIZE, "2");


I’m curious whether vortex supports prefetch. If it doesn’t, why are the EnablePrefetch and related options being configured?

Vortex does not use framework prefetch; removed those test settings.

lxy-9602 · 2026-10-28T13:56:54Z

+                // 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));
+            }


Please tighten your validation. Enabling either late_materializing or exact_filter is sufficient to guarantee correct query results and predicate accuracy.

Updated this append test to require exact results when either mode is enabled, since every case has both probe and payload columns. The shared read path is unchanged.

zjw1111 · 2026-10-29T04:02:02Z

@@ -0,0 +1,131 @@
+# Licensed to the Apache Software Foundation (ASF) under one


The "fmm" is from the db name? It's not a public name, maybe need rename it.

Removed the cache-sharding patch, so this filename is no longer used.

lxy-9602 · 2026-10-29T01:10:05Z

+    /// @return An owned expression, or nullptr if no condition can be pushed.
+    static VxExpressionPtr Convert(const std::shared_ptr<Predicate>& predicate,
+                                   const std::shared_ptr<arrow::Schema>& file_schema,
+                                   const VortexPredicateOptions& options = {});


Similar to before: in formal code, please avoid using default parameters as much as possible.

Removed Convert()'s default argument; callers already pass the options explicitly.

lxy-9602 · 2026-10-29T01:31:11Z

+    if (!timestamp_type.timezone().empty() && timestamp_type.timezone() != "UTC") {
+        return false;
+    }
+    const auto timestamp = literal.GetValue<Timestamp>();


Please carefully review the read/write paths and predicates related to tz, and ensure the behavior is consistent with Java.

New LTZ files use UTC metadata without changing epoch values; C++/Java compatibility tests pass. Legacy non-UTC predicates are not pushed down in C++.

lxy-9602 · 2026-10-29T01:36:30Z

+    *result = value + fraction;
+    *result_unit = unit;
+    return true;
+}


This function seems to be primarily handling timestamp precision conversion. Could you please try to use the utility functions from DateTimeUtils instead?

Reused DateTimeUtils::TimestampToInteger() and GetTimeTypeFromArrowType(), retaining overflow and lossless-conversion checks.

lxy-9602 · 2026-10-29T01:47:56Z

+        case FieldType::FLOAT:
+        case FieldType::DOUBLE:
+            // Native totalOrder distinguishes NaN signs and payloads; Paimon treats all NaNs alike.
+            return scalar;


I checked the Java logic— it looks like they push down floating-point values. For correctness around NaN, could you please open a PR to fix it in Java and keep both sides consistent?

Opened apache/paimon#10985 for the Java NaN fix, keeping FLOAT/DOUBLE fallback consistent with C++.

lxy-9602 · 2026-10-29T02:57:29Z

+    if (literal.IsNull() || literal.GetType() != field_type || arrow_type == nullptr) {
+        return false;
+    }
+    switch (field_type) {


Similarly, you can assume the predicate types match the corresponding field types in the file. If there’s a mismatch, just error out—there’s no need for this kind of fallback.

Type mismatches now return errors, including IN/NOT IN lists over the configured limit.

lxy-9602 · 2026-10-29T03:10:44Z

+        nodes += lower == upper ? 1 : 3;
+    }
+    return nodes;
+}


I think these IN-expression optimizations should be handled by the query engine itself. Do we really need to introduce these code changes specifically for this? It seems to add a lot of complexity, and other formats don’t appear to benefit from the same optimization.

Also, pushing down deduplication seems more appropriate to be done in the upstream layer.

Removed deduplication, sorting and integer-range rewrites from the converter. Added vortex.read.in-pushdown.max-values, defaulting to 32.

Removed deduplication from the Vortex converter; it is left to the query engine.

lxy-9602 · 2026-10-29T05:13:27Z

+            const auto& decimal_type = checked_cast<const arrow::Decimal128Type&>(*arrow_type);
+            const auto decimal = literal.GetValue<Decimal>();
+            return decimal.Precision() == decimal_type.precision() &&
+                   decimal.Scale() == decimal_type.scale();


If the decimal precisions don’t match, can we do a precision conversion? Or does vortex actually support this kind of pushdown?

Added lossless decimal literal conversion to the file's precision/scale before pushdown; inexact values fall back.

lxy-9602 · 2026-10-29T05:37:19Z

+    if (type == Function::Type::AND || type == Function::Type::OR) {
+        const auto compound = std::dynamic_pointer_cast<CompoundPredicate>(predicate);
+        if (compound == nullptr) {
+            return NullExpr();


For compound, besides and/or, are there any other type? It looks like most predicate logic is swallowed and falls back to a “don’t push down” behavior on error. We should distinguish between cases where we silently avoid pushdown because it’s unsupported, versus cases where there’s a real issue that should result in an error.

Only AND/OR are compound functions. Unsupported cases fall back; malformed predicates and native construction errors are propagated.

Make membership and I/O limits configurable, simplify reader state, and bound native Arrow materialization, including nested arrays. Preserve safe temporal and decimal fallbacks, propagate type errors, and extend native and SDK regressions without changing shared readers.

SteNicholas · 2026-11-04T12:11:31Z

+    SelectionState selection_;
+    RowPositionState row_position_;


Release the new selection and row-position state in CloseInternal(). These members now own selection_.indices_ and row_position_.previous_batch_row_idx_, but CloseInternal() only resets the stream, current batch, scan, data source and input. Calling Close() while retaining the reader therefore retains these allocations until destruction. For example, selecting half of a 10-million-row file leaves roughly 40 MB in the host index vector alone, even though BatchReader::Close() promises to release associated resources.

Please reset both states during close, releasing the vector's capacity rather than only clearing its elements. A regression should release the returned batches, close the reader while keeping the reader object alive, and verify that its internal resources are released; separately retained output batches must remain valid.

Reset both states in CloseInternal(), releasing vector capacity and the row-index array. Added cleanup and output batch lifetime tests.

SteNicholas · 2026-11-04T12:11:31Z

+            state.excludes_ ? total_rows - state.selected_row_count_ : state.selected_row_count_));
+        uint64_t next_row = 0;
+        const auto end = bitmap->End();
+        for (auto iter = bitmap->Begin(); iter != end; ++iter) {


Dense selections still enumerate every surviving row before the first batch. Storing only exclusion indices fixes the retained-memory expansion, but this loop always walks the original selection. A file with one deleted row still visits almost every physical row, and an all-selected bitmap also incurs the full traversal before any pushed predicate can run.

In an isolated -O2 microbenchmark using this repository's bundled Roaring implementation on local arm64, constructing the exclusion vector for 100 million rows with one deletion took about 133 ms with this loop, versus about 0.2 ms when copying/flipping the compressed bitmap and enumerating the excluded side. This measures selection construction, not an end-to-end scan.

Please validate the bounds and then enumerate a compressed complement for dense selections, with an appropriate all-selected fast path. Large all-selected and single-deletion benchmarks would guard this first-batch latency.

Bounds are checked first; full selections return early and dense selections enumerate the compressed complement. Added boundary tests and benchmarked large full selections and single deletions.

SteNicholas · 2026-11-04T12:11:31Z

+    VxExpressionPtr root(nullptr, vx_expression_free);
+    VxExpressionPtr projection(nullptr, vx_expression_free);
+    std::vector<const char*> name_ptrs;
+    if (read_schema_ != nullptr && read_schema_->num_fields() > 0) {


An explicitly empty read schema currently falls back to a full-column native scan. When num_fields() == 0, options.projection stays null, which Vortex interprets as root(). All file columns can therefore be read and converted to standard Arrow arrays before AlignToReadSchema() discards them. Count-only/empty-projection reads can still expand large string payloads unnecessarily.

Please distinguish an unset read schema from an explicitly empty projection and construct an empty native projection for the latter. EmptyProjectionPreservesRowCountsAndPhysicalIds verifies output shape and row mapping, but it would also pass with a full payload scan; add an I/O or export/materialization assertion showing that omitted payload columns are not consumed.

Explicitly empty schemas now produce an empty native projection. Added native schema and I/O assertions verifying omitted payload columns are skipped.

SteNicholas · 2026-11-04T12:11:31Z

+    if (current_stream_.get_schema(&current_stream_, &ffi_schema) != 0) {
+        Status callback_status = input_context_->GetCallbackStatus();
+        ReleaseStream();


Preserve the native Arrow stream error before releasing the stream. If get_schema() fails without a filesystem callback error, the current path returns only Vortex Arrow stream get_schema failed, losing the actual schema/export failure detail.

Please copy get_last_error() before ReleaseStream() and include it in the returned Status, following the existing ReadNextArray() pattern. Keep the callback status as the preferred error when present, and handle a null native message. Returning the detailed status is sufficient; an additional duplicate log is unnecessary.

Native error messages are now copied before stream release and included in the status. Added tests for callback error precedence and null messages.

SteNicholas · 2026-11-04T12:11:32Z

+}
+
+// Type mismatches violate the predicate contract; valid but unsafe conversions fall back.
+Result<bool> LiteralMatchesColumn(const Literal& literal,


The helper names conflate type validity with pushdown eligibility. LiteralMatchesColumn() returns false for fully matching FLOAT/DOUBLE types and for legacy non-UTC timestamp columns. Consequently, ValidateMembershipLiteralTypes() can set valid = false for a perfectly legal predicate that simply needs residual evaluation.

Please use names that express pushdown eligibility, such as CheckLiteralPushdownEligibility() and all_literals_pushable, or otherwise make the result semantics explicit. Keeping unsupported-but-valid predicates distinct from the type mismatches that return Status::Invalid will make these fallback branches easier to maintain.

Renamed the helpers and flags to distinguish pushdown eligibility from type validity. Type mismatches still return Status::Invalid.

SteNicholas · 2026-11-04T12:11:32Z

+A positive value permits lists of up to that length, subject to type and value
+safety checks. ``0`` disables only IN/NOT IN pushdown. A list exceeding the limit


Document the concrete fallback conditions behind “type and value safety checks.” The implementation declines floating-point comparisons/membership, timestamps with lossy precision or unsupported stored timezones, and all native predicate pushdown when the file contains a user column named _row_idx. Increasing the membership limit does not change those conditions, so users need this information to understand when the optimization applies.

Please also document the new LTZ storage-metadata behavior introduced by NormalizeTimestampTimezones(): newly written non-empty timezones are normalized to UTC while epoch values and timestamp units stay unchanged. This is observable through the file schema and deserves a user-facing compatibility note in addition to the PR description.

Documented the floating-point, timestamp and _row_idx fallback rules. Added the LTZ compatibility note: non-empty timezones are normalized to UTC for new writes, preserving epoch values and units.

Release selection and row-index state on close, build dense selections from compressed complements, and push empty projections into native scans. Preserve Arrow schema error details, clarify pushdown eligibility and document fallback and LTZ metadata rules. Add focused regressions and fix cast declarations reported by clang-tidy.

Preserve configurable read coalescing and both integration-test includes while adopting upstream comments and the shared timezone utility.

feat(vortex): add predicate pushdown and optimize selective reads

37dce2a

SteNicholas reviewed Oct 27, 2026

View reviewed changes

Comment thread src/paimon/format/vortex/vortex_file_batch_reader.cpp Resolved

SteNicholas reviewed Oct 27, 2026

View reviewed changes

Comment thread cmake_modules/vortex-fmm-cache-shards.diff Resolved

lxy-9602 reviewed Oct 29, 2026

View reviewed changes

zjw1111 reviewed Oct 29, 2026

View reviewed changes

lxy-9602 reviewed Oct 29, 2026

View reviewed changes

zhouyc-ali requested review from lxy-9602 and zjw1111 November 4, 2026 09:34

SteNicholas reviewed Nov 4, 2026

View reviewed changes

zhouyc-ali requested a review from SteNicholas November 5, 2026 06:31

chore(vortex): merge main and resolve pushdown conflicts

a2631bb

Preserve configurable read coalescing and both integration-test includes while adopting upstream comments and the shared timezone utility.

		@@ -0,0 +1,131 @@
		# Licensed to the Apache Software Foundation (ASF) under one

		A positive value permits lists of up to that length, subject to type and value
		safety checks. ``0`` disables only IN/NOT IN pushdown. A list exceeding the limit

Conversation

zhouyc-ali commented Oct 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Tests

API and Format

Documentation

Generative AI tooling

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

zhouyc-ali commented Oct 27, 2026 •

edited

Loading