Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 34 additions & 17 deletions src/iceberg/arrow_c_data_util.cc
Original file line number Diff line number Diff line change
Expand Up @@ -227,7 +227,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_)),
validation_mode_(other.validation_mode_) {
other.input_arrow_schema_.release = nullptr;
other.output_arrow_schema_.release = nullptr;
}
Expand All @@ -253,6 +254,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_);
validation_mode_ = other.validation_mode_;
return *this;
}

Expand All @@ -267,7 +269,11 @@ ProjectionContext::~ProjectionContext() {

Result<ProjectionContext> ProjectionContext::Make(
const Schema& input_schema, const Schema& output_schema,
ProjectionContext::ProjectBatchFunction project_batch_function) {
ProjectionContext::ProjectBatchFunction project_batch_function,
ArrowValidationMode validation_mode) {
ICEBERG_PRECHECK(validation_mode == ArrowValidationMode::kFull ||
validation_mode == ArrowValidationMode::kBasic,
"Invalid Arrow validation mode");
ICEBERG_ASSIGN_OR_RAISE(
auto selected_field_indices,
BuildSelectedFieldIndices(input_schema.fields(), output_schema.fields()));
Expand All @@ -279,6 +285,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.validation_mode_ = validation_mode;

return context;
}
Expand Down Expand Up @@ -308,6 +315,10 @@ ProjectionContext::ProjectBatchState& ProjectionContext::project_batch_state() {
return project_batch_state_;
}

ArrowValidationMode ProjectionContext::validation_mode() const {
return validation_mode_;
}

void ProjectionContext::RegisterProjectBatchFunction(
ProjectionContext::ProjectBatchFunction project_batch_function) {
ICEBERG_DCHECK(project_batch_function != nullptr,
Expand All @@ -330,21 +341,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);
Expand Down Expand Up @@ -382,6 +385,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.validation_mode() == ArrowValidationMode::kFull) {
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
Expand All @@ -391,7 +408,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 iceberg
22 changes: 15 additions & 7 deletions src/iceberg/arrow_c_data_util_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include <vector>

#include "iceberg/arrow_c_data.h"
#include "iceberg/arrow_validation.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
Expand Down Expand Up @@ -68,13 +69,16 @@ class ICEBERG_EXPORT ProjectionContext {
/// \param project_batch_function Optional implementation returned by
/// ProjectionContext::ResolveProjectBatchFunction, or nullptr to use the nanoarrow
/// path.
/// \param validation_mode Full content validation by default; Basic requires 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);
static Result<ProjectionContext> Make(
const Schema& input_schema, const Schema& output_schema,
ProjectBatchFunction project_batch_function,
ArrowValidationMode validation_mode = ArrowValidationMode::kFull);

ProjectionContext(ProjectionContext&&) noexcept;
ProjectionContext& operator=(ProjectionContext&&) noexcept;
Expand All @@ -97,6 +101,8 @@ class ICEBERG_EXPORT ProjectionContext {

ProjectBatchState& project_batch_state();

ArrowValidationMode validation_mode() const;

private:
ProjectionContext() = default;

Expand All @@ -109,6 +115,7 @@ class ICEBERG_EXPORT ProjectionContext {
ArrowSchema output_arrow_schema_{};
ProjectBatchFunction project_batch_function_ = nullptr;
ProjectBatchState project_batch_state_;
ArrowValidationMode validation_mode_ = ArrowValidationMode::kFull;
};

/// \brief Concept for sources that can be wrapped as ArrowArrayStreams.
Expand Down Expand Up @@ -228,10 +235,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`.
Expand Down
35 changes: 35 additions & 0 deletions src/iceberg/arrow_validation.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
/*
* 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.
*/

#pragma once

#include <cstdint>

namespace iceberg {

/// Validation policy for Arrow arrays before accessing their contents.
enum class ArrowValidationMode : uint8_t {
/// Validate structure and contents, including intermediate offsets.
kFull,
/// Validate structure without scanning all contents. The caller must guarantee
/// that the array is valid and remains unchanged while it is being accessed.
kBasic,
};

} // namespace iceberg
1 change: 1 addition & 0 deletions src/iceberg/cpp20_compatibility_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
#pragma once

#include "iceberg/arrow_c_data.h"
#include "iceberg/arrow_validation.h"
#include "iceberg/catalog.h"
#include "iceberg/catalog/catalog_util.h"
#include "iceberg/catalog/memory/in_memory_catalog.h"
Expand Down
9 changes: 6 additions & 3 deletions src/iceberg/data/file_scan_task_reader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -211,9 +211,10 @@ 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, arrow_validation_mode_));

auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(),
name_mapping_, properties_, data_file->first_row_id,
Expand All @@ -235,6 +236,7 @@ class FileScanTaskReader::Impl {
projected_schema_(std::move(options.projected_schema)),
name_mapping_(std::move(options.name_mapping)),
properties_(ReaderProperties::FromMap(options.properties)),
arrow_validation_mode_(options.arrow_validation_mode),
field_lookup_(std::move(field_lookup)),
delete_counter_(std::move(delete_counter)) {}

Expand All @@ -243,6 +245,7 @@ class FileScanTaskReader::Impl {
std::shared_ptr<Schema> projected_schema_;
std::shared_ptr<NameMapping> name_mapping_;
ReaderProperties properties_;
ArrowValidationMode arrow_validation_mode_;
DeleteFilter::FieldLookup field_lookup_;
std::shared_ptr<DeleteCounter> delete_counter_;
};
Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/data/file_scan_task_reader.h
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include <vector>

#include "iceberg/arrow_c_data.h"
#include "iceberg/arrow_validation.h"
#include "iceberg/iceberg_data_export.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
Expand Down Expand Up @@ -59,6 +60,9 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
std::shared_ptr<NameMapping> name_mapping;
/// Format-specific or implementation-specific options for data readers.
std::unordered_map<std::string, std::string> properties;
/// Validation before merge-on-read batch projection. Basic requires trusted arrays.
/// This does not change validation in readers or delete filtering.
ArrowValidationMode arrow_validation_mode = ArrowValidationMode::kFull;
};

/// \brief Create a reusable task reader from shared read context.
Expand Down
Loading
Loading