SteNicholas commented on code in PR #471:
URL: https://github.com/apache/paimon-cpp/pull/471#discussion_r4467266386
##########
src/paimon/format/vortex/vortex_file_batch_reader.cpp:
##########
@@ -232,26 +149,152 @@ std::shared_ptr<arrow::ArrayData>
AlignTimestampTimezones(
} // namespace
+Result<VortexFileBatchReader::SelectionState>
VortexFileBatchReader::SelectionState::Create(
+ const std::optional<RoaringBitmap32>& bitmap, uint64_t total_rows) {
+ SelectionState state;
+ state.has_selection_ = bitmap.has_value();
+ state.selected_row_count_ = total_rows;
+ if (bitmap) {
+ const int32_t cardinality = bitmap->Cardinality();
+ if (cardinality < 0 || static_cast<uint64_t>(cardinality) >
total_rows) {
+ return Status::Invalid("Vortex selection cardinality exceeds the
file row count");
+ }
+ state.selected_row_count_ = static_cast<uint64_t>(cardinality);
+ state.excludes_ = state.selected_row_count_ > total_rows / 2;
+ state.indices_.reserve(static_cast<size_t>(
+ 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) {
Review Comment:
**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.
##########
src/paimon/format/vortex/vortex_file_batch_reader.h:
##########
@@ -96,30 +155,26 @@ class VortexFileBatchReader : public FileBatchReader {
VxSessionPtr session_;
VxDataSourcePtr data_source_;
VxScanPtr scan_;
- // The file schema with Vortex's string/binary view types normalized to
the standard types; this
- // is what GetFileSchema() reports to paimon.
+ // Full file schema with standard Arrow types, as reported by
GetFileSchema().
std::shared_ptr<arrow::Schema> file_schema_;
- // The RAW view-typed struct that Vortex's Arrow export produces.
NextBatch imports each scanned
- // batch as this type (so it must match the C export's buffer layout),
then NormalizeViewArray
- // rewrites it to file_schema_'s types. Keeping this view-typed is
load-bearing: normalizing it
- // would make ImportArray misread the buffers.
- std::shared_ptr<arrow::DataType> struct_type_;
// The schema NextBatch must output, set by SetReadSchema (the projected
read schema). Null
// until SetReadSchema is called, in which case the full file schema is
returned.
std::shared_ptr<arrow::Schema> read_schema_;
uint64_t total_rows_ = 0;
::ArrowArrayStream current_stream_{};
+ // Actual scan output schema, including projection and any hidden physical
row index.
+ std::shared_ptr<arrow::Schema> current_stream_schema_;
bool stream_active_ = false;
std::shared_ptr<arrow::Array> current_batch_;
int64_t current_batch_offset_ = 0;
- uint64_t rows_emitted_ = 0;
- uint64_t previous_first_row_ = std::numeric_limits<uint64_t>::max();
- uint64_t previous_batch_row_count_ = 0;
+ SelectionState selection_;
+ RowPositionState row_position_;
Review Comment:
**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.
##########
src/paimon/format/vortex/vortex_file_batch_reader.cpp:
##########
@@ -306,28 +347,67 @@ 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,
predicate_options, arrow_pool));
}
VortexFileBatchReader::~VortexFileBatchReader() {
CloseInternal();
}
+Status VortexFileBatchReader::CreateScan() {
+ 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{};
+ options.ordered = true;
+ // 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<const char*> name_ptrs;
+ if (read_schema_ != nullptr && read_schema_->num_fields() > 0) {
Review Comment:
**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.
##########
src/paimon/format/vortex/vortex_file_batch_reader.cpp:
##########
@@ -338,15 +418,29 @@ Result<bool>
VortexFileBatchReader::OpenNextPartitionStream() {
if (partition == nullptr) {
return false; // Scan exhausted.
}
- // vx_partition_scan_arrow consumes the partition (even on error), so
release ownership here.
+ // Arrow stream export consumes the partition (even on error), so release
ownership here.
::ArrowArrayStream stream{};
error = nullptr;
- if (vx_partition_scan_arrow(session_.get(), partition.release(), &stream,
&error) != 0) {
+ if (vx_partition_scan_arrow_with_batch_size(session_.get(),
partition.release(),
+
static_cast<size_t>(batch_size_), &stream,
+ &error) != 0) {
return VortexCallbackError("scan Vortex partition to Arrow", error,
input_context_->GetCallbackStatus());
}
current_stream_ = stream;
stream_active_ = true;
+ // Import batches with the scan's actual output schema (projected columns,
standard types).
+ ::ArrowSchema ffi_schema = {};
+ if (current_stream_.get_schema(¤t_stream_, &ffi_schema) != 0) {
+ Status callback_status = input_context_->GetCallbackStatus();
+ ReleaseStream();
Review Comment:
**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.
##########
docs/source/api/file_format.rst:
##########
@@ -55,3 +55,59 @@ Interface
.. doxygenclass:: paimon::WrittenFileStatsProvider
:members:
:undoc-members:
+
+
+.. _cpp-api-vortex-read-options:
+
+Vortex Read Options
+===================
+
+Pass Vortex read options through ``ReadContextBuilder::AddOption`` at runtime.
+Read options override table options for each key; a key absent from both uses
+its default. Changes take effect when a new reader is created and do not change
+existing readers.
+
+Predicate Pushdown
+------------------
+
+``vortex.read.in-pushdown.max-values`` sets the maximum list length eligible
for
+native IN/NOT IN pushdown. It accepts a decimal integer from ``0`` to
+``4294967295`` (``UINT32_MAX``), with a default of ``32``. Negative, malformed
or
+out-of-range values are rejected. The limit applies independently to each IN or
+NOT IN predicate and counts the original list entries, including duplicates.
+
+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
Review Comment:
**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.
##########
src/paimon/format/vortex/vortex_predicate_converter.cpp:
##########
@@ -0,0 +1,541 @@
+/*
+ * 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.
+ */
+
+#include "paimon/format/vortex/vortex_predicate_converter.h"
+
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/type.h"
+#include "arrow/util/decimal.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/field_type_utils.h"
+#include "paimon/core/casting/decimal_to_decimal_cast_executor.h"
+#include "paimon/data/decimal.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/defs.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/function.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate.h"
+
+namespace paimon::vortex {
+namespace {
+
+VxExpressionPtr NullExpr() {
+ return VxExpressionPtr(nullptr, vx_expression_free);
+}
+
+Status CheckNativeResult(const std::string& operation, const void* handle,
+ vx_error* error = nullptr) {
+ PAIMON_RETURN_NOT_OK(VortexFfiError(operation, error));
+ if (handle == nullptr) {
+ return Status::Invalid(operation, ": Vortex returned a null handle");
+ }
+ return Status::OK();
+}
+
+// vx_expression_not returns an owned handle typed const; reclaim it as an
owned non-const pointer.
+Result<VxExpressionPtr> MakeNotExpr(const vx_expression* child) {
+ VxExpressionPtr
expression(const_cast<vx_expression*>(vx_expression_not(child)),
+ vx_expression_free);
+ PAIMON_RETURN_NOT_OK(CheckNativeResult("create Vortex NOT expression",
expression.get()));
+ return expression;
+}
+
+// Reject precision loss and overflow before constructing a native timestamp
scalar.
+bool GetTimestampValue(const Literal& literal, const
std::shared_ptr<arrow::DataType>& arrow_type,
+ int64_t* result, vx_time_unit* result_unit) {
+ if (arrow_type == nullptr || arrow_type->id() != arrow::Type::TIMESTAMP) {
+ return false;
+ }
+ const auto timestamp_type =
checked_pointer_cast<arrow::TimestampType>(arrow_type);
+ if (!timestamp_type->timezone().empty() && timestamp_type->timezone() !=
"UTC") {
+ return false;
+ }
+ const auto timestamp = literal.GetValue<Timestamp>();
+ int64_t value = timestamp.GetMillisecond();
+ const int32_t nanos = timestamp.GetNanoOfMillisecond();
+ vx_time_unit unit = VX_TIME_UNIT_MILLISECOND;
+ int64_t factor = 1;
+ int64_t fraction = 0;
+ switch (timestamp_type->unit()) {
+ case arrow::TimeUnit::SECOND:
+ if (value % 1000 != 0 || nanos != 0) {
+ return false;
+ }
+ unit = VX_TIME_UNIT_SECOND;
+ break;
+ case arrow::TimeUnit::MILLI:
+ if (nanos != 0) {
+ return false;
+ }
+ break;
+ case arrow::TimeUnit::MICRO:
+ if (nanos % 1000 != 0) {
+ return false;
+ }
+ factor = 1000;
+ fraction = nanos / 1000;
+ unit = VX_TIME_UNIT_MICROSECOND;
+ break;
+ case arrow::TimeUnit::NANO:
+ factor = 1000000;
+ fraction = nanos;
+ unit = VX_TIME_UNIT_NANOSECOND;
+ break;
+ default:
+ return false;
+ }
+ if (value < std::numeric_limits<int64_t>::min() / factor ||
+ value > std::numeric_limits<int64_t>::max() / factor) {
+ return false;
+ }
+ value *= factor;
+ if (value > std::numeric_limits<int64_t>::max() - fraction) {
+ return false;
+ }
+ *result = DateTimeUtils::TimestampToInteger(
+ timestamp, DateTimeUtils::GetTimeTypeFromArrowType(timestamp_type));
+ *result_unit = unit;
+ return true;
+}
+
+// Normalize constants only; shared CAST may round, so accept it only if the
value is unchanged.
+Result<std::optional<Decimal>> NormalizeDecimalLiteral(
+ const Literal& literal, const std::shared_ptr<arrow::DataType>&
arrow_type) {
+ const auto decimal = literal.GetValue<Decimal>();
+ const auto& target_type = checked_cast<const
arrow::Decimal128Type&>(*arrow_type);
+ // Keep literals within Paimon's decimal range and the shared scale table.
+ if (decimal.Precision() < Decimal::MIN_PRECISION ||
+ decimal.Precision() > Decimal::MAX_PRECISION || decimal.Scale() < 0 ||
+ decimal.Scale() > decimal.Precision() || target_type.precision() <
Decimal::MIN_PRECISION ||
+ target_type.precision() > Decimal::MAX_PRECISION ||
target_type.scale() < 0 ||
+ target_type.scale() > target_type.precision()) {
+ return std::optional<Decimal>();
+ }
+ if (!arrow::Decimal128(decimal.HighBits(), decimal.LowBits())
+ .FitsInPrecision(decimal.Precision())) {
+ return std::optional<Decimal>();
+ }
+ if (decimal.Precision() == target_type.precision() && decimal.Scale() ==
target_type.scale()) {
+ return std::make_optional(decimal);
+ }
+ const DecimalToDecimalCastExecutor cast_executor;
+ PAIMON_ASSIGN_OR_RAISE(Literal converted, cast_executor.Cast(literal,
arrow_type));
+ if (converted.IsNull()) {
+ return std::optional<Decimal>();
+ }
+ const auto normalized = converted.GetValue<Decimal>();
+ if (decimal.CompareTo(normalized) != 0) {
+ return std::optional<Decimal>();
+ }
+ return std::make_optional(normalized);
+}
+
+Result<VxScalarPtr> MakeLiteralScalar(const Literal& literal,
+ const std::shared_ptr<arrow::DataType>&
arrow_type) {
+ VxScalarPtr scalar(nullptr, vx_scalar_free);
+ if (literal.IsNull()) {
+ return scalar;
+ }
+ vx_error* error = nullptr;
+ switch (literal.GetType()) {
+ case FieldType::BOOLEAN:
+ scalar.reset(vx_scalar_new_bool(literal.GetValue<bool>(), false));
+ break;
+ case FieldType::TINYINT:
+ scalar.reset(vx_scalar_new_i8(literal.GetValue<int8_t>(), false));
+ break;
+ case FieldType::SMALLINT:
+ scalar.reset(vx_scalar_new_i16(literal.GetValue<int16_t>(),
false));
+ break;
+ case FieldType::INT:
+ scalar.reset(vx_scalar_new_i32(literal.GetValue<int32_t>(),
false));
+ break;
+ // Vortex stores a date column as a date[days] extension and rejects
comparing it against a
+ // plain int scalar, so build a date-typed scalar.
+ case FieldType::DATE:
+ scalar.reset(vx_scalar_new_date(literal.GetValue<int32_t>(),
false, &error));
+ if (scalar == nullptr && error == nullptr) {
+ // A valid Paimon constant can exceed the native civil-date
range.
+ return scalar;
+ }
+ break;
+ case FieldType::BIGINT:
+ scalar.reset(vx_scalar_new_i64(literal.GetValue<int64_t>(),
false));
+ break;
+ case FieldType::FLOAT:
+ case FieldType::DOUBLE:
+ // Native totalOrder distinguishes NaN signs and payloads; Paimon
treats all NaNs alike.
+ return scalar;
+ case FieldType::STRING: {
+ const auto value = literal.GetValue<std::string>();
+ scalar.reset(vx_scalar_new_utf8(value.data(), value.size(), false,
&error));
+ break;
+ }
+ case FieldType::DECIMAL: {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Decimal> normalized,
+ NormalizeDecimalLiteral(literal,
arrow_type));
+ if (!normalized.has_value()) {
+ return scalar;
+ }
+ // Cast validated the full source value before selecting a
narrower representation.
+ const Decimal& decimal = normalized.value();
+ const auto precision = static_cast<uint8_t>(decimal.Precision());
+ const auto scale = static_cast<int8_t>(decimal.Scale());
+ if (decimal.IsCompact()) {
+
scalar.reset(vx_scalar_new_decimal_i64(decimal.ToUnscaledLong(), precision,
scale,
+ false, &error));
+ } else {
+ // Little-endian 16-byte two's complement, low word first.
+ uint8_t bytes16[16];
+ const uint64_t low = decimal.LowBits();
+ const uint64_t high = decimal.HighBits();
+ std::memcpy(bytes16, &low, sizeof(low));
+ std::memcpy(bytes16 + sizeof(low), &high, sizeof(high));
+ scalar.reset(
+ vx_scalar_new_decimal_i128_le(bytes16, precision, scale,
false, &error));
+ }
+ break;
+ }
+ case FieldType::TIMESTAMP: {
+ int64_t value = 0;
+ vx_time_unit unit = VX_TIME_UNIT_MILLISECOND;
+ if (!GetTimestampValue(literal, arrow_type, &value, &unit)) {
+ return scalar;
+ }
+ const auto& timestamp_type = checked_cast<const
arrow::TimestampType&>(*arrow_type);
+ const char* zone = timestamp_type.timezone().empty() ? nullptr :
"UTC";
+ scalar.reset(vx_scalar_new_timestamp(value, unit, zone, false,
&error));
+ if (scalar == nullptr && error == nullptr) {
+ // Preserve residual comparison when native temporal limits
are narrower.
+ return scalar;
+ }
+ break;
+ }
+ default:
+ return scalar;
+ }
+ PAIMON_RETURN_NOT_OK(CheckNativeResult("create Vortex predicate scalar",
scalar.get(), error));
+ return scalar;
+}
+
+Result<VxExpressionPtr> MakeLiteralExpr(const Literal& literal,
+ const
std::shared_ptr<arrow::DataType>& arrow_type) {
+ PAIMON_ASSIGN_OR_RAISE(VxScalarPtr scalar, MakeLiteralScalar(literal,
arrow_type));
+ if (scalar == nullptr) {
+ return NullExpr();
+ }
+ vx_error* error = nullptr;
+ VxExpressionPtr expression(vx_expression_literal(scalar.get(), &error),
vx_expression_free);
+ PAIMON_RETURN_NOT_OK(
+ CheckNativeResult("create Vortex literal expression",
expression.get(), error));
+ return expression;
+}
+
+// Type mismatches violate the predicate contract; valid but unsafe
conversions fall back.
+Result<bool> LiteralMatchesColumn(const Literal& literal,
Review Comment:
**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.
--
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]
