Skip to content

fix(row): make Arrow wrapper validation configurable #1077

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gty404 wants to merge 1 commit into from
+231 −40
Closed

fix(row): make Arrow wrapper validation configurable #1077

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
30 changes: 17 additions & 13 deletions src/iceberg/data/delete_filter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -460,64 +460,66 @@ Result<std::unique_ptr<DeleteFilter>> DeleteFilter::Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> table_schema, std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, bool need_row_pos_col,
std::shared_ptr<DeleteCounter> counter) {
std::shared_ptr<DeleteCounter> counter, bool validate_batch) {
ICEBERG_ASSIGN_OR_RAISE(auto field_lookup, MakeFieldLookup(table_schema));
return Make(std::move(file_path), delete_files, std::move(requested_schema),
std::move(io), std::move(field_lookup), need_row_pos_col,
std::move(counter));
std::move(counter), validate_batch);
}

Result<std::unique_ptr<DeleteFilter>> DeleteFilter::Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<TableMetadata> table_metadata,
std::shared_ptr<Schema> requested_schema, std::shared_ptr<FileIO> io,
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter) {
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter, bool validate_batch) {
ICEBERG_PRECHECK(table_metadata != nullptr, "Table metadata must not be null");

ICEBERG_ASSIGN_OR_RAISE(auto field_lookup, MakeFieldLookup(std::move(table_metadata)));
return Make(std::move(file_path), delete_files, std::move(requested_schema),
std::move(io), std::move(field_lookup), need_row_pos_col,
std::move(counter));
std::move(counter), validate_batch);
}

Result<std::unique_ptr<DeleteFilter>> DeleteFilter::Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> table_schema, std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, std::span<const std::shared_ptr<Schema>> schemas,
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter) {
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter, bool validate_batch) {
ICEBERG_ASSIGN_OR_RAISE(auto field_lookup, MakeFieldLookup(table_schema, schemas));
return Make(std::move(file_path), delete_files, std::move(requested_schema),
std::move(io), std::move(field_lookup), need_row_pos_col,
std::move(counter));
std::move(counter), validate_batch);
}

Result<std::unique_ptr<DeleteFilter>> DeleteFilter::Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> requested_schema, std::shared_ptr<FileIO> io,
FieldLookup field_lookup, bool need_row_pos_col,
std::shared_ptr<DeleteCounter> counter) {
std::shared_ptr<DeleteCounter> counter, bool validate_batch) {
ICEBERG_PRECHECK(requested_schema != nullptr, "Requested schema must not be null");
ICEBERG_PRECHECK(field_lookup != nullptr, "Field lookup must not be null");
ICEBERG_PRECHECK(delete_files.empty() || io != nullptr,
"FileIO must not be null when delete files are present");

auto filter = std::unique_ptr<DeleteFilter>(
new DeleteFilter(std::move(file_path), std::move(requested_schema), std::move(io),
std::move(field_lookup), need_row_pos_col, std::move(counter)));
auto filter = std::unique_ptr<DeleteFilter>(new DeleteFilter(
std::move(file_path), std::move(requested_schema), std::move(io),
std::move(field_lookup), need_row_pos_col, std::move(counter), validate_batch));
ICEBERG_RETURN_UNEXPECTED(filter->Init(delete_files));
return filter;
}

DeleteFilter::DeleteFilter(std::string file_path,
std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, FieldLookup field_lookup,
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter)
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter,
bool validate_batch)
: file_path_(std::move(file_path)),
requested_schema_(std::move(requested_schema)),
field_lookup_(std::move(field_lookup)),
need_row_pos_col_(need_row_pos_col),
validate_batch_(validate_batch),
counter_(std::move(counter)),
delete_loader_(std::move(io)) {}
delete_loader_(std::move(io), validate_batch) {}

DeleteFilter::~DeleteFilter() = default;

Expand Down Expand Up @@ -750,7 +752,9 @@ Result<AliveRowSelection> DeleteFilter::ComputeAliveRows(const ArrowSchema& batc
batch.length <= static_cast<int64_t>(std::numeric_limits<int32_t>::max()),
"Batch length {} exceeds int32_t row index capacity", batch.length);
result.indices.reserve(static_cast<size_t>(batch.length));
ICEBERG_ASSIGN_OR_RAISE(auto row, ArrowArrayStructLike::Make(batch_schema, batch));
ICEBERG_ASSIGN_OR_RAISE(
auto row,
ArrowArrayStructLike::Make(batch_schema, batch, /*row_index=*/0, validate_batch_));

for (int64_t i = 0; i < batch.length; ++i) {
if (i > 0) {
Expand Down
14 changes: 9 additions & 5 deletions src/iceberg/data/delete_filter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,28 +110,30 @@ class ICEBERG_DATA_EXPORT DeleteFilter {
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> table_schema, std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, bool need_row_pos_col = true,
std::shared_ptr<DeleteCounter> counter = nullptr);
std::shared_ptr<DeleteCounter> counter = nullptr, bool validate_batch = true);

/// \brief Create a DeleteFilter using table metadata for schema-aware field lookup.
static Result<std::unique_ptr<DeleteFilter>> Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<TableMetadata> table_metadata,
std::shared_ptr<Schema> requested_schema, std::shared_ptr<FileIO> io,
bool need_row_pos_col = true, std::shared_ptr<DeleteCounter> counter = nullptr);
bool need_row_pos_col = true, std::shared_ptr<DeleteCounter> counter = nullptr,
bool validate_batch = true);

/// \brief Create a DeleteFilter with table schemas for dropped equality fields.
static Result<std::unique_ptr<DeleteFilter>> Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> table_schema, std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, std::span<const std::shared_ptr<Schema>> schemas,
bool need_row_pos_col = true, std::shared_ptr<DeleteCounter> counter = nullptr);
bool need_row_pos_col = true, std::shared_ptr<DeleteCounter> counter = nullptr,
bool validate_batch = true);

/// \brief Create a DeleteFilter with a custom field lookup.
static Result<std::unique_ptr<DeleteFilter>> Make(
std::string file_path, std::span<const std::shared_ptr<DataFile>> delete_files,
std::shared_ptr<Schema> requested_schema, std::shared_ptr<FileIO> io,
FieldLookup field_lookup, bool need_row_pos_col = true,
std::shared_ptr<DeleteCounter> counter = nullptr);
std::shared_ptr<DeleteCounter> counter = nullptr, bool validate_batch = true);

~DeleteFilter();

Expand Down Expand Up @@ -191,7 +193,8 @@ class ICEBERG_DATA_EXPORT DeleteFilter {

DeleteFilter(std::string file_path, std::shared_ptr<Schema> requested_schema,
std::shared_ptr<FileIO> io, FieldLookup field_lookup,
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter);
bool need_row_pos_col, std::shared_ptr<DeleteCounter> counter,
bool validate_batch);

Status Init(std::span<const std::shared_ptr<DataFile>> delete_files);
Result<std::shared_ptr<Schema>> ComputeRequiredSchema() const;
Expand All @@ -207,6 +210,7 @@ class ICEBERG_DATA_EXPORT DeleteFilter {
FieldLookup field_lookup_;

const bool need_row_pos_col_;
const bool validate_batch_;
// Position of `_pos` in required_schema_ when existent
std::optional<size_t> pos_field_position_;
std::shared_ptr<DeleteCounter> counter_;
Expand Down
6 changes: 4 additions & 2 deletions src/iceberg/data/delete_loader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,8 @@ Status ValidateDV(const DataFile& dv, std::string_view data_file_path) {

} // namespace

DeleteLoader::DeleteLoader(std::shared_ptr<FileIO> io) : io_(std::move(io)) {}
DeleteLoader::DeleteLoader(std::shared_ptr<FileIO> io, bool validate_batch)
: io_(std::move(io)), validate_batch_(validate_batch) {}

DeleteLoader::~DeleteLoader() = default;

Expand Down Expand Up @@ -255,7 +256,8 @@ Result<std::unique_ptr<UncheckedStructLikeSet>> DeleteLoader::LoadEqualityDelete
internal::ArrowArrayGuard batch_guard(&batch);

ICEBERG_ASSIGN_OR_RAISE(
auto row, ArrowArrayStructLike::Make(arrow_schema, batch, /*row_index=*/0));
auto row, ArrowArrayStructLike::Make(arrow_schema, batch, /*row_index=*/0,
validate_batch_));

for (int64_t i = 0; i < batch.length; ++i) {
if (i > 0) {
Expand Down
4 changes: 3 additions & 1 deletion src/iceberg/data/delete_loader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,8 @@ class ICEBERG_DATA_EXPORT DeleteLoader {
public:
/// \brief Create a DeleteLoader.
/// \param io FileIO instance for reading delete files
explicit DeleteLoader(std::shared_ptr<FileIO> io);
/// \param validate_batch Whether equality-delete Arrow batches are fully validated
explicit DeleteLoader(std::shared_ptr<FileIO> io, bool validate_batch = true);

~DeleteLoader();

Expand Down Expand Up @@ -71,6 +72,7 @@ class ICEBERG_DATA_EXPORT DeleteLoader {
std::string_view data_file_path) const;

std::shared_ptr<FileIO> io_;
const bool validate_batch_;
};

} // namespace iceberg
3 changes: 2 additions & 1 deletion src/iceberg/data/file_scan_task_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,8 @@ class FileScanTaskReader::Impl {
ICEBERG_ASSIGN_OR_RAISE(
auto delete_filter,
DeleteFilter::Make(data_file->file_path, task.delete_files(), projected_schema_,
io_, field_lookup_, has_position_deletes, delete_counter_));
io_, field_lookup_, has_position_deletes, delete_counter_,
properties_.Get(ReaderProperties::kArrowValidateBatch)));

auto required_schema = delete_filter->RequiredSchema();
auto project_batch_function = ProjectionContext::ResolveProjectBatchFunction();
Expand Down
76 changes: 61 additions & 15 deletions src/iceberg/row/arrow_array_wrapper.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,18 +85,22 @@ Result<Scalar> ExtractValue(const ArrowSchema* schema, const ArrowArray* array,
return Decimal(int_value);
}
case NANOARROW_TYPE_STRUCT: {
// The parent view validates nested children recursively; avoid rescanning them.
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<StructLike> struct_like,
ArrowArrayStructLike::Make(*schema, *array, index));
ArrowArrayStructLike::Make(*schema, *array, index,
/*validate_batch=*/false));
return struct_like;
}
case NANOARROW_TYPE_LIST: {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<ArrayLike> array_like,
ArrowArrayArrayLike::Make(*schema, *array, index));
ArrowArrayArrayLike::Make(*schema, *array, index,
/*validate_batch=*/false));
return array_like;
}
case NANOARROW_TYPE_MAP: {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<MapLike> map_like,
ArrowArrayMapLike::Make(*schema, *array, index));
ArrowArrayMapLike::Make(*schema, *array, index,
/*validate_batch=*/false));
return map_like;
}
case NANOARROW_TYPE_EXTENSION:
Expand All @@ -113,8 +117,12 @@ Result<Scalar> ExtractValue(const ArrowSchema* schema, const ArrowArray* array,

class ArrowArrayStructLike::Impl {
public:
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index)
: schema_(schema), array_(std::cref(array)), row_index_(row_index) {}
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch)
: schema_(schema),
array_(std::cref(array)),
row_index_(row_index),
validate_batch_(validate_batch) {}

~Impl() = default;

Expand Down Expand Up @@ -147,6 +155,10 @@ class ArrowArrayStructLike::Impl {
ArrowError error;
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return {};
}

Expand All @@ -161,6 +173,10 @@ class ArrowArrayStructLike::Impl {
ArrowArrayViewInitFromSchema(&array_view_, &schema_, &error));
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return {};
}

Expand All @@ -171,11 +187,13 @@ class ArrowArrayStructLike::Impl {
const ArrowSchema& schema_;
std::reference_wrapper<const ArrowArray> array_;
int64_t row_index_;
bool validate_batch_;
};

Result<std::unique_ptr<ArrowArrayStructLike>> ArrowArrayStructLike::Make(
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index) {
auto impl = std::make_unique<Impl>(schema, array, row_index);
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch) {
auto impl = std::make_unique<Impl>(schema, array, row_index, validate_batch);
ICEBERG_RETURN_UNEXPECTED(impl->Init());
return std::unique_ptr<ArrowArrayStructLike>(new ArrowArrayStructLike(std::move(impl)));
}
Expand All @@ -201,8 +219,12 @@ Status ArrowArrayStructLike::Reset(const ArrowArray& array, int64_t row_index) {

class ArrowArrayArrayLike::Impl {
public:
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index)
: schema_(schema), array_(std::cref(array)), row_index_(row_index) {}
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch)
: schema_(schema),
array_(std::cref(array)),
row_index_(row_index),
validate_batch_(validate_batch) {}

~Impl() = default;

Expand All @@ -229,6 +251,10 @@ class ArrowArrayArrayLike::Impl {
ArrowError error;
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return UpdateOffsets();
}

Expand All @@ -243,6 +269,10 @@ class ArrowArrayArrayLike::Impl {
ArrowArrayViewInitFromSchema(&array_view_, &schema_, &error));
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return UpdateOffsets();
}

Expand All @@ -268,11 +298,13 @@ class ArrowArrayArrayLike::Impl {

int64_t offset_ = 0;
int64_t length_ = 0;
bool validate_batch_;
};

Result<std::unique_ptr<ArrowArrayArrayLike>> ArrowArrayArrayLike::Make(
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index) {
auto impl = std::make_unique<Impl>(schema, array, row_index);
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch) {
auto impl = std::make_unique<Impl>(schema, array, row_index, validate_batch);
ICEBERG_RETURN_UNEXPECTED(impl->Init());
return std::unique_ptr<ArrowArrayArrayLike>(new ArrowArrayArrayLike(std::move(impl)));
}
Expand All @@ -298,8 +330,12 @@ Status ArrowArrayArrayLike::Reset(const ArrowArray& array, int64_t row_index) {

class ArrowArrayMapLike::Impl {
public:
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index)
: schema_(schema), array_(std::cref(array)), row_index_(row_index) {}
Impl(const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch)
: schema_(schema),
array_(std::cref(array)),
row_index_(row_index),
validate_batch_(validate_batch) {}

~Impl() = default;

Expand Down Expand Up @@ -340,6 +376,10 @@ class ArrowArrayMapLike::Impl {
ArrowError error;
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return UpdateOffsets();
}

Expand All @@ -354,6 +394,10 @@ class ArrowArrayMapLike::Impl {
ArrowArrayViewInitFromSchema(&array_view_, &schema_, &error));
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewSetArray(&array_view_, &array_.get(), &error));
if (validate_batch_) {
NANOARROW_RETURN_IF_NOT_OK(
ArrowArrayViewValidate(&array_view_, NANOARROW_VALIDATION_LEVEL_FULL, &error));
}
return UpdateOffsets();
}

Expand Down Expand Up @@ -387,11 +431,13 @@ class ArrowArrayMapLike::Impl {

int64_t offset_ = 0;
int64_t length_ = 0;
bool validate_batch_;
};

Result<std::unique_ptr<ArrowArrayMapLike>> ArrowArrayMapLike::Make(
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index) {
auto impl = std::make_unique<Impl>(schema, array, row_index);
const ArrowSchema& schema, const ArrowArray& array, int64_t row_index,
bool validate_batch) {
auto impl = std::make_unique<Impl>(schema, array, row_index, validate_batch);
ICEBERG_RETURN_UNEXPECTED(impl->Init());
return std::unique_ptr<ArrowArrayMapLike>(new ArrowArrayMapLike(std::move(impl)));
}
Expand Down
Loading
Loading