fix(arrow): make full batch validation configurable

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 {

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

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,

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};

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

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,