zhouyc-ali commented on code in PR #471:
URL: https://github.com/apache/paimon-cpp/pull/471#discussion_r4477284526
##########
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:
Native error messages are now copied before stream release and included in
the status. Added tests for callback error precedence and null messages.
##########
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:
Renamed the helpers and flags to distinguish pushdown eligibility from type
validity. Type mismatches still return `Status::Invalid`.
##########
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:
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.
--
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]
