This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a899511c fix(arrow): make full batch validation configurable (#1043)
a899511c is described below
commit a899511ce9eda52e2b79cdb2abfaefccd4841ea3
Author: Guotao Yu <[email protected]>
AuthorDate: Fri Oct 30 17:33:57 2026 +0800
fix(arrow): make full batch validation configurable (#1043)
---
src/iceberg/arrow_c_data_util.cc | 45 ++++++----
src/iceberg/arrow_c_data_util_internal.h | 17 ++--
src/iceberg/data/file_scan_task_reader.cc | 8 +-
src/iceberg/file_reader.h | 3 +
src/iceberg/test/arrow_c_data_util_test.cc | 113 ++++++++++++++++++++-----
src/iceberg/test/file_scan_task_reader_test.cc | 43 ++++++++++
6 files changed, 185 insertions(+), 44 deletions(-)
diff --git a/src/iceberg/arrow_c_data_util.cc b/src/iceberg/arrow_c_data_util.cc
index eb7e90b7..915eeabc 100644
--- a/src/iceberg/arrow_c_data_util.cc
+++ b/src/iceberg/arrow_c_data_util.cc
@@ -238,7 +238,8 @@ ProjectionContext::ProjectionContext(ProjectionContext&&
other) noexcept
input_arrow_schema_(other.input_arrow_schema_),
output_arrow_schema_(other.output_arrow_schema_),
project_batch_function_(std::exchange(other.project_batch_function_,
nullptr)),
- project_batch_state_(std::move(other.project_batch_state_)) {
+ project_batch_state_(std::move(other.project_batch_state_)),
+ validate_batch_(other.validate_batch_) {
other.input_arrow_schema_.release = nullptr;
other.output_arrow_schema_.release = nullptr;
}
@@ -264,6 +265,7 @@ ProjectionContext&
ProjectionContext::operator=(ProjectionContext&& other) noexc
other.output_arrow_schema_.release = nullptr;
project_batch_function_ = std::exchange(other.project_batch_function_,
nullptr);
project_batch_state_ = std::move(other.project_batch_state_);
+ validate_batch_ = other.validate_batch_;
return *this;
}
@@ -278,7 +280,7 @@ ProjectionContext::~ProjectionContext() {
Result<ProjectionContext> ProjectionContext::Make(
const Schema& input_schema, const Schema& output_schema,
- ProjectionContext::ProjectBatchFunction project_batch_function) {
+ ProjectionContext::ProjectBatchFunction project_batch_function, bool
validate_batch) {
ICEBERG_ASSIGN_OR_RAISE(
auto selected_field_indices,
BuildSelectedFieldIndices(input_schema.fields(),
output_schema.fields()));
@@ -290,6 +292,7 @@ Result<ProjectionContext> ProjectionContext::Make(
ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(input_schema,
&context.input_arrow_schema_));
ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(output_schema,
&context.output_arrow_schema_));
context.project_batch_function_ = project_batch_function;
+ context.validate_batch_ = validate_batch;
return context;
}
@@ -319,6 +322,8 @@ ProjectionContext::ProjectBatchState&
ProjectionContext::project_batch_state() {
return project_batch_state_;
}
+bool ProjectionContext::validate_batch() const { return validate_batch_; }
+
void ProjectionContext::RegisterProjectBatchFunction(
ProjectionContext::ProjectBatchFunction project_batch_function) {
ICEBERG_DCHECK(project_batch_function != nullptr,
@@ -341,21 +346,13 @@ auto ProjectionContext::ResolveProjectBatchFunction()
namespace {
-Result<ArrowArray> ProjectBatchNanoarrow(
- const ArrowSchema& input_arrow_schema, const ArrowArray& input_batch,
- std::span<const int32_t> row_indices, const ArrowSchema&
output_arrow_schema,
- std::span<const int32_t> selected_field_indices) {
- ArrowArrayView input_view;
+Result<ArrowArray> ProjectBatchNanoarrow(const ArrowSchema& input_arrow_schema,
+ const ArrowArray& input_batch,
+ std::span<const int32_t> row_indices,
+ const ArrowSchema&
output_arrow_schema,
+ std::span<const int32_t>
selected_field_indices,
+ const ArrowArrayView& input_view) {
ArrowError error;
- ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
- ArrowArrayViewInitFromSchema(&input_view, &input_arrow_schema, &error),
error);
- internal::ArrowArrayViewGuard input_view_guard(&input_view);
- ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
- ArrowArrayViewSetArray(&input_view, &input_batch, &error), error);
- ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
- ArrowArrayViewValidate(&input_view, NANOARROW_VALIDATION_LEVEL_DEFAULT,
&error),
- error);
-
ArrowArray output_array;
ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
ArrowArrayInitFromSchema(&output_array, &output_arrow_schema, &error),
error);
@@ -393,6 +390,20 @@ Result<ArrowArray> ProjectBatch(ArrowArray* input_batch,
ICEBERG_PRECHECK(input_batch != nullptr, "input_batch must not be null");
internal::ArrowArrayGuard input_batch_guard(input_batch);
+ ArrowArrayView input_view;
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayViewInitFromSchema(&input_view,
&projection.input_arrow_schema(), &error),
+ error);
+ internal::ArrowArrayViewGuard input_view_guard(&input_view);
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayViewSetArray(&input_view, input_batch, &error), error);
+ if (projection.validate_batch()) {
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayViewValidate(&input_view, NANOARROW_VALIDATION_LEVEL_FULL,
&error),
+ error);
+ }
+
auto project_batch_function = projection.project_batch_function();
if (project_batch_function != nullptr) {
// ProjectBatch owns input_batch. Arrow-backed implementations import it
and clear
@@ -402,7 +413,7 @@ Result<ArrowArray> ProjectBatch(ArrowArray* input_batch,
return ProjectBatchNanoarrow(projection.input_arrow_schema(), *input_batch,
row_indices,
projection.output_arrow_schema(),
- projection.selected_field_indices());
+ projection.selected_field_indices(),
input_view);
}
namespace {
diff --git a/src/iceberg/arrow_c_data_util_internal.h
b/src/iceberg/arrow_c_data_util_internal.h
index 899695b6..c5d20397 100644
--- a/src/iceberg/arrow_c_data_util_internal.h
+++ b/src/iceberg/arrow_c_data_util_internal.h
@@ -68,13 +68,16 @@ class ICEBERG_EXPORT ProjectionContext {
/// \param project_batch_function Optional implementation returned by
/// ProjectionContext::ResolveProjectBatchFunction, or nullptr to use the
nanoarrow
/// path.
+ /// \param validate_batch Whether to perform full content validation.
Disable only for
+ /// trusted input.
/// \note It validates that output_schema selects or reorders complete
top-level fields
/// by field id. Nested pruning and type changes are rejected. The
input_schema and
/// output_schema passed to Make must outlive the context. ProjectBatch may
lazily
/// initialize backend cache; do not share one context across concurrent
calls.
static Result<ProjectionContext> Make(const Schema& input_schema,
const Schema& output_schema,
- ProjectBatchFunction
project_batch_function);
+ ProjectBatchFunction
project_batch_function,
+ bool validate_batch = true);
ProjectionContext(ProjectionContext&&) noexcept;
ProjectionContext& operator=(ProjectionContext&&) noexcept;
@@ -97,6 +100,8 @@ class ICEBERG_EXPORT ProjectionContext {
ProjectBatchState& project_batch_state();
+ bool validate_batch() const;
+
private:
ProjectionContext() = default;
@@ -109,6 +114,7 @@ class ICEBERG_EXPORT ProjectionContext {
ArrowSchema output_arrow_schema_{};
ProjectBatchFunction project_batch_function_ = nullptr;
ProjectBatchState project_batch_state_;
+ bool validate_batch_ = true;
};
/// \brief Concept for sources that can be wrapped as ArrowArrayStreams.
@@ -228,10 +234,11 @@ ICEBERG_EXPORT Result<ArrowArrayStream>
MakeArrowArrayStream(
/// \brief Project selected rows from a batch into complete top-level fields.
///
/// `input_batch` is consumed by this function. If the projection carries a
registered
-/// implementation, the call is delegated to it; otherwise the built-in
nanoarrow
-/// implementation is used. The projection must have been created for the
stable schema
-/// pair that describes the input batch and requested output. This function
does not
-/// revalidate schema compatibility on each batch.
+/// implementation, the call is delegated to it after validation; otherwise
the built-in
+/// nanoarrow implementation is used. The projection must have been created
for the stable
+/// schema pair that describes the input batch and requested output. This
function does
+/// not revalidate schema compatibility on each batch. Input arrays are
validated
+/// according to the context policy before either backend runs.
///
/// \param input_batch Owned Arrow C Data batch to project.
/// \param row_indices Zero-based row positions to copy from `input_batch`.
diff --git a/src/iceberg/data/file_scan_task_reader.cc
b/src/iceberg/data/file_scan_task_reader.cc
index bf23c5c3..8d3c7328 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/src/iceberg/data/file_scan_task_reader.cc
@@ -213,9 +213,11 @@ class FileScanTaskReader::Impl {
auto required_schema = delete_filter->RequiredSchema();
auto project_batch_function =
ProjectionContext::ResolveProjectBatchFunction();
// ProjectionContext borrows schemas that are kept in
MergeOnReadStreamSource.
- ICEBERG_ASSIGN_OR_RAISE(auto projection_context,
- ProjectionContext::Make(*required_schema,
*projected_schema_,
- project_batch_function));
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto projection_context,
+ ProjectionContext::Make(*required_schema, *projected_schema_,
+ project_batch_function,
+
properties_.Get(ReaderProperties::kArrowValidateBatch)));
auto options = MakeReaderOptions(task, io_, required_schema, filter,
case_sensitive_,
name_mapping_, properties_,
data_file->first_row_id,
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
index 13e8b3ad..eded9191 100644
--- a/src/iceberg/file_reader.h
+++ b/src/iceberg/file_reader.h
@@ -80,6 +80,9 @@ class ICEBERG_EXPORT ReaderProperties : public
ConfigBase<ReaderProperties> {
/// Only the Parquet reader honors this option; other readers ignore it.
/// Default: false (use 32-bit offset list).
inline static Entry<bool> kArrowUseLargeList{"read.arrow.use-large-list",
false};
+ /// \brief Validate Arrow batches before merge-on-read projection.
+ /// Default: true. Disable only when the reader is trusted to produce valid
arrays.
+ inline static Entry<bool> kArrowValidateBatch{"read.arrow.validate-batch",
true};
/// \brief Use footer statistics to prune Parquet row groups.
inline static Entry<bool> kParquetRowGroupFilter{
"read.parquet.row-group-filter.enabled", true};
diff --git a/src/iceberg/test/arrow_c_data_util_test.cc
b/src/iceberg/test/arrow_c_data_util_test.cc
index 35bb30fd..a21f88c4 100644
--- a/src/iceberg/test/arrow_c_data_util_test.cc
+++ b/src/iceberg/test/arrow_c_data_util_test.cc
@@ -20,6 +20,7 @@
/// \file arrow_c_data_util_test.cc
/// Verifies ProjectBatch behavior across registered implementations.
+#include <array>
#include <cstdint>
#include <memory>
#include <string>
@@ -76,15 +77,15 @@ ProjectionContext::ProjectBatchFunction
ArrowComputeFunction() {
std::shared_ptr<::arrow::RecordBatch> RunProjectBatch(
const ::arrow::RecordBatch& batch, const std::vector<int32_t>&
alive_indices,
const Schema& required_schema, const Schema& projected_schema,
- ProjectionContext::ProjectBatchFunction project_batch_function) {
+ ProjectionContext::ProjectBatchFunction project_batch_function, bool
validate_batch) {
ArrowSchema c_schema;
ArrowArray c_array;
EXPECT_TRUE(::arrow::ExportRecordBatch(batch, &c_array, &c_schema).ok());
ArrowSchemaGuard schema_guard(&c_schema);
ArrowArrayGuard array_guard(&c_array);
- auto projection =
- ProjectionContext::Make(required_schema, projected_schema,
project_batch_function);
+ auto projection = ProjectionContext::Make(required_schema, projected_schema,
+ project_batch_function,
validate_batch);
EXPECT_THAT(projection, IsOk());
auto result = ProjectBatch(&c_array, alive_indices, projection.value());
@@ -103,22 +104,15 @@ void ExpectProjectBatch(const ::arrow::RecordBatch& batch,
const Schema& required_schema, const Schema&
projected_schema,
std::string_view expected_json) {
auto expected = MakeBatch(projected_schema, expected_json);
- auto nanoarrow =
- RunProjectBatch(batch, alive_indices, required_schema, projected_schema,
nullptr);
- auto arrow_compute = RunProjectBatch(batch, alive_indices, required_schema,
- projected_schema,
ArrowComputeFunction());
-
- EXPECT_TRUE(nanoarrow->Equals(*expected)) << "nanoarrow:\n"
- << nanoarrow->ToString() <<
"expected:\n"
- << expected->ToString();
- EXPECT_TRUE(arrow_compute->Equals(*expected))
- << "arrow_compute:\n"
- << arrow_compute->ToString() << "expected:\n"
- << expected->ToString();
- EXPECT_TRUE(nanoarrow->Equals(*arrow_compute))
- << "nanoarrow:\n"
- << nanoarrow->ToString() << "arrow_compute:\n"
- << arrow_compute->ToString();
+ for (bool validate_batch : {true, false}) {
+ auto nanoarrow = RunProjectBatch(batch, alive_indices, required_schema,
+ projected_schema, nullptr,
validate_batch);
+ auto arrow_compute =
+ RunProjectBatch(batch, alive_indices, required_schema,
projected_schema,
+ ArrowComputeFunction(), validate_batch);
+ EXPECT_TRUE(nanoarrow->Equals(*expected)) << nanoarrow->ToString();
+ EXPECT_TRUE(arrow_compute->Equals(*expected)) << arrow_compute->ToString();
+ }
}
std::shared_ptr<Schema> MakeFullSchema() {
@@ -231,6 +225,87 @@ TEST(ProjectBatchTest, ProjectionRejectsNestedPruning) {
EXPECT_THAT(projection, IsError(ErrorKind::kInvalidArgument));
}
+TEST(ProjectBatchTest, ValidateBatchControlsNonMonotonicStringOffsets) {
+ for (bool validate_batch : {true, false}) {
+ for (auto function :
+ {ProjectionContext::ProjectBatchFunction{nullptr},
ArrowComputeFunction()}) {
+ auto schema = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(1, "value",
string())});
+
+ // DEFAULT accepts the first/last offsets; FULL rejects the intermediate
decrease.
+ ArrowArray root;
+ ASSERT_EQ(ArrowArrayInitFromType(&root, NANOARROW_TYPE_STRUCT),
NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayAllocateChildren(&root, 1), NANOARROW_OK);
+
+ ArrowArray* value_array = root.children[0];
+ ASSERT_EQ(ArrowArrayInitFromType(value_array, NANOARROW_TYPE_STRING),
NANOARROW_OK);
+ const std::array<int32_t, 3> raw_offsets = {0, 5, 2};
+ ASSERT_EQ(ArrowBufferAppend(ArrowArrayBuffer(value_array, 1),
raw_offsets.data(),
+ raw_offsets.size() * sizeof(int32_t)),
+ NANOARROW_OK);
+ ASSERT_EQ(ArrowBufferAppend(ArrowArrayBuffer(value_array, 2), "ab", 2),
+ NANOARROW_OK);
+ value_array->length = 2;
+ value_array->null_count = 0;
+ ASSERT_EQ(ArrowArrayFinishBuildingDefault(value_array, nullptr),
NANOARROW_OK);
+
+ root.length = 2;
+ root.null_count = 0;
+ ASSERT_EQ(ArrowArrayFinishBuildingDefault(&root, nullptr), NANOARROW_OK);
+
+ auto projection =
+ ProjectionContext::Make(*schema, *schema, function, validate_batch);
+ ASSERT_THAT(projection, IsOk());
+
+ // Empty selection skips the content check without reading corrupt data.
+ auto result = ProjectBatch(&root, /*row_indices=*/{},
projection.value());
+ if (validate_batch) {
+ EXPECT_THAT(result, IsError(ErrorKind::kInvalidArrowData));
+ EXPECT_THAT(result, HasErrorMessage("element size >= 0"));
+ } else {
+ EXPECT_THAT(result, IsOk());
+ }
+ if (result.has_value()) {
+ ArrowArrayGuard guard(&result.value());
+ EXPECT_EQ(result->length, 0);
+ }
+ EXPECT_EQ(root.release, nullptr);
+ }
+ }
+}
+
+TEST(ProjectBatchTest, SkippedContentValidationStillRejectsInvalidStructure) {
+ auto schema = MakeFullSchema();
+ auto batch = MakeBatch(*schema, R"([[1,"a",1.0]])");
+ for (auto function :
+ {ProjectionContext::ProjectBatchFunction{nullptr},
ArrowComputeFunction()}) {
+ ArrowArray array;
+ ASSERT_TRUE(::arrow::ExportRecordBatch(*batch, &array).ok());
+ array.length = -1;
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto projection,
+ ProjectionContext::Make(*schema, *schema, function,
/*validate_batch=*/false));
+ auto result = ProjectBatch(&array, {}, projection);
+ EXPECT_THAT(result, IsError(ErrorKind::kInvalidArrowData));
+ EXPECT_THAT(result, HasErrorMessage("length"));
+ EXPECT_EQ(array.release, nullptr);
+ }
+}
+
+TEST(ProjectBatchTest, MovePreservesValidateBatch) {
+ auto schema = MakeFullSchema();
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto without_validation,
+ ProjectionContext::Make(*schema, *schema, nullptr,
/*validate_batch=*/false));
+ auto moved = std::move(without_validation);
+ EXPECT_FALSE(moved.validate_batch());
+ ICEBERG_UNWRAP_OR_FAIL(auto validation,
+ ProjectionContext::Make(*schema, *schema, nullptr));
+ EXPECT_TRUE(validation.validate_batch());
+ validation = std::move(moved);
+ EXPECT_FALSE(validation.validate_batch());
+}
+
TEST(WriteProjectionTest, IdenticalSchemasTransferBatchWithoutConversion) {
auto input_schema = MakeFullSchema();
auto write_schema = MakeFullSchema();
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc
b/src/iceberg/test/file_scan_task_reader_test.cc
index 8b0d79d1..596ec488 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -671,6 +671,49 @@ TEST_F(FileScanTaskReaderTest,
OpenWithDeletionVectorFiltersRows) {
ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3,
"Baz"]])"));
}
+TEST_F(FileScanTaskReaderTest, ReadsArrowValidateBatchProperty) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto data_file,
+ MakeDataFile(table_schema_,
+ R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz",
"green"]])"));
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto pos_delete,
MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+ {1}, data_file->file_path));
+ FileScanTask task(data_file, {pos_delete});
+
+ struct RestoreProjection {
+ ProjectionContext::ProjectBatchFunction function =
+ ProjectionContext::ResolveProjectBatchFunction();
+ ~RestoreProjection() {
ProjectionContext::RegisterProjectBatchFunction(function); }
+ } restore;
+ ProjectionContext::RegisterProjectBatchFunction(
+ [](ArrowArray*, std::span<const int32_t>,
+ ProjectionContext& projection) -> Result<ArrowArray> {
+ return InvalidArgument("projection mode: {}",
+ projection.validate_batch() ? "full" : "basic");
+ });
+
+ for (bool validate_batch : {true, false}) {
+ FileScanTaskReader::Options options{
+ .io = file_io_,
+ .table_schema = table_schema_,
+ .projected_schema = projected_schema_,
+ };
+ if (!validate_batch) {
+ options.properties[ReaderProperties::kArrowValidateBatch.key()] =
"false";
+ }
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
FileScanTaskReader::Make(std::move(options)));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+
+ ArrowArray array{};
+ EXPECT_NE(stream.get_next(&stream, &array), 0);
+ EXPECT_THAT(stream.get_last_error(&stream),
+ ::testing::HasSubstr(validate_batch ? "projection mode: full"
+ : "projection mode:
basic"));
+ stream.release(&stream);
+ }
+}
+
TEST_F(FileScanTaskReaderTest,
OpenWithEqualityDeletesAddsAndPrunesDeleteOnlyColumns) {
ICEBERG_UNWRAP_OR_FAIL(
auto data_file,
