diff --git a/dev/dv-fixtures/generate_go_fixtures.go b/dev/dv-fixtures/generate_go_fixtures.go new file mode 100644 index 000000000..67d26d6c0 --- /dev/null +++ b/dev/dv-fixtures/generate_go_fixtures.go @@ -0,0 +1,152 @@ +// 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. + +//go:build ignore + +package main + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "strconv" + + "github.com/apache/iceberg-go/puffin" + "github.com/apache/iceberg-go/table/dv" +) + +type fixtureBlob struct { + referencedDataFile string + positions []uint64 + ranges []positionRange +} + +type positionRange struct { + start uint64 + end uint64 +} + +func writeFixture(outputDir, fileName, createdBy string, blobs []fixtureBlob) error { + var output bytes.Buffer + writer, err := puffin.NewWriter(&output) + if err != nil { + return err + } + if err := writer.SetCreatedBy(createdBy); err != nil { + return err + } + + for _, blob := range blobs { + bitmap := dv.NewRoaringPositionBitmap() + for _, position := range blob.positions { + bitmap.Set(position) + } + for _, positionRange := range blob.ranges { + bitmap.SetRange(positionRange.start, positionRange.end) + } + payload, err := dv.SerializeDV(bitmap) + if err != nil { + return err + } + _, err = writer.AddBlob(puffin.BlobMetadataInput{ + Type: puffin.BlobTypeDeletionVector, + SnapshotID: -1, + SequenceNumber: -1, + Fields: []int32{}, + Properties: map[string]string{ + "referenced-data-file": blob.referencedDataFile, + "cardinality": strconv.FormatInt(bitmap.Cardinality(), 10), + }, + }, payload) + if err != nil { + return err + } + } + + if err := writer.Finish(); err != nil { + return err + } + return os.WriteFile(filepath.Join(outputDir, fileName), output.Bytes(), 0o644) +} + +func main() { + if len(os.Args) != 2 { + fmt.Fprintln(os.Stderr, "usage: go run generate_go_fixtures.go OUTPUT_DIR") + os.Exit(2) + } + outputDir := os.Args[1] + if err := os.MkdirAll(outputDir, 0o755); err != nil { + panic(err) + } + + err := writeFixture(outputDir, "single-blob-dv.puffin", + "iceberg-go test fixture", []fixtureBlob{{ + referencedDataFile: "data/test.parquet", + positions: []uint64{1, 3, 5, 7, 9}, + }}) + if err != nil { + panic(err) + } + + err = writeFixture(outputDir, "multi-blob-dv.puffin", + "iceberg-go cross-language fixture", []fixtureBlob{ + { + referencedDataFile: "s3://warehouse/db/table/data/go-file-001.parquet", + positions: []uint64{ + 0, 100, 200, (uint64(1) << 32) + 7, + }, + }, + { + referencedDataFile: "s3://warehouse/db/table/data/go-file-002.parquet", + positions: []uint64{ + 50, 150, (uint64(2) << 32) + 9, + }, + }, + }) + if err != nil { + panic(err) + } + + position := func(bucket, container, value uint64) uint64 { + return (bucket << 32) + (container << 16) + value + } + allContainerPositions := []uint64{ + position(0, 0, 5), + position(0, 0, 7), + position(1, 0, 10), + position(1, 0, 20), + } + for bucket := uint64(0); bucket < 2; bucket++ { + for value := uint64(0); value < 10000; value += 2 { + allContainerPositions = + append(allContainerPositions, position(bucket, 2, value)) + } + } + err = writeFixture(outputDir, "all-container-types-dv.puffin", + "iceberg-go cross-language fixture", []fixtureBlob{{ + referencedDataFile: "s3://warehouse/db/table/data/all-containers.parquet", + positions: allContainerPositions, + ranges: []positionRange{ + {start: position(0, 1, 1), end: position(0, 1, 1000)}, + {start: position(1, 1, 10), end: position(1, 1, 500)}, + }, + }}) + if err != nil { + panic(err) + } +} diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 8a98274ff..2457f8eed 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -55,6 +55,7 @@ set(ICEBERG_SOURCES geospatial.cc inspect/history_table.cc inspect/metadata_table.cc + inspect/position_deletes_table.cc inspect/snapshots_table.cc inheritable_metadata.cc json_serde.cc diff --git a/src/iceberg/arrow_c_data_util.cc b/src/iceberg/arrow_c_data_util.cc index a77012a69..8139ad0da 100644 --- a/src/iceberg/arrow_c_data_util.cc +++ b/src/iceberg/arrow_c_data_util.cc @@ -220,6 +220,12 @@ Status AppendValue(const ArrowSchema& input_schema, const ArrowArray& input_arra } // namespace +Status AppendArrayValue(const ArrowSchema& input_schema, const ArrowArray& input_array, + const ArrowArrayView& input_view, int64_t row_index, + ArrowArray* output_array) { + return AppendValue(input_schema, input_array, input_view, row_index, output_array); +} + ProjectionContext::ProjectionContext(ProjectionContext&& other) noexcept : input_schema_(std::exchange(other.input_schema_, nullptr)), output_schema_(std::exchange(other.output_schema_, nullptr)), diff --git a/src/iceberg/arrow_c_data_util_internal.h b/src/iceberg/arrow_c_data_util_internal.h index e02db29a6..709d72ea0 100644 --- a/src/iceberg/arrow_c_data_util_internal.h +++ b/src/iceberg/arrow_c_data_util_internal.h @@ -35,6 +35,8 @@ #include "iceberg/result.h" #include "iceberg/type_fwd.h" +struct ArrowArrayView; + namespace iceberg { /// \brief Cached state for ProjectBatch over one input/output schema pair. @@ -242,4 +244,10 @@ ICEBERG_EXPORT Result ProjectBatch(ArrowArray* input_batch, std::span row_indices, ProjectionContext& projection); +/// \brief Append one value from an Arrow array into a compatible nanoarrow builder. +ICEBERG_EXPORT Status AppendArrayValue(const ArrowSchema& input_schema, + const ArrowArray& input_array, + const ArrowArrayView& input_view, + int64_t row_index, ArrowArray* output_array); + } // namespace iceberg diff --git a/src/iceberg/deletes/dv_util.cc b/src/iceberg/deletes/dv_util.cc index 670ccfedb..e9fb6449d 100644 --- a/src/iceberg/deletes/dv_util.cc +++ b/src/iceberg/deletes/dv_util.cc @@ -17,6 +17,7 @@ * under the License. */ +#include #include #include #include @@ -36,13 +37,63 @@ #include "iceberg/metadata_columns.h" #include "iceberg/partition_spec.h" #include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/puffin_reader.h" #include "iceberg/result.h" #include "iceberg/util/content_file_util.h" #include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" #include "iceberg/version.h" namespace iceberg { +namespace { + +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +Status ValidateDVBlobMetadata(const puffin::BlobMetadata& blob, + const DataFile& delete_file) { + ICEBERG_PRECHECK(blob.type == puffin::StandardBlobTypes::kDeletionVectorV1, + "Invalid deletion vector blob type '{}', expected '{}'", blob.type, + puffin::StandardBlobTypes::kDeletionVectorV1); + ICEBERG_PRECHECK(blob.snapshot_id == -1, + "Deletion vector requires snapshot-id -1, got {}", blob.snapshot_id); + ICEBERG_PRECHECK(blob.sequence_number == -1, + "Deletion vector requires sequence-number -1, got {}", + blob.sequence_number); + ICEBERG_PRECHECK(blob.compression_codec.empty(), + "Deletion vector must not be compressed, got '{}'", + blob.compression_codec); + + auto referenced_data_file = + blob.properties.find(std::string(kReferencedDataFileProperty)); + ICEBERG_PRECHECK(referenced_data_file != blob.properties.end() && + !referenced_data_file->second.empty(), + "Deletion vector blob requires non-empty '{}' property", + kReferencedDataFileProperty); + ICEBERG_PRECHECK( + referenced_data_file->second == *delete_file.referenced_data_file, + "Manifest referenced_data_file '{}' does not match Puffin '{}' property '{}'", + *delete_file.referenced_data_file, kReferencedDataFileProperty, + referenced_data_file->second); + + auto cardinality = blob.properties.find(std::string(kCardinalityProperty)); + ICEBERG_PRECHECK(cardinality != blob.properties.end(), + "Deletion vector blob requires '{}' property", kCardinalityProperty); + ICEBERG_ASSIGN_OR_RAISE(auto parsed_cardinality, + StringUtils::ParseNumber(cardinality->second)); + ICEBERG_PRECHECK(parsed_cardinality >= 0, + "Deletion vector cardinality must be non-negative, got {}", + parsed_cardinality); + ICEBERG_PRECHECK( + parsed_cardinality == delete_file.record_count, + "Manifest record_count {} does not match Puffin cardinality property {}", + delete_file.record_count, parsed_cardinality); + return {}; +} + +} // namespace + Result>> DVUtil::MergeAndWriteDVs( std::span groups, std::string_view output_path, const std::shared_ptr& io) { @@ -84,6 +135,10 @@ Result DVUtil::ReadDV(const std::shared_ptr& dele delete_file->content_size_in_bytes.has_value(), "Deletion vector requires content_offset and content_size_in_bytes: {}", delete_file->file_path); + ICEBERG_PRECHECK(delete_file->referenced_data_file.has_value() && + !delete_file->referenced_data_file->empty(), + "Deletion vector requires referenced_data_file: {}", + delete_file->file_path); const int64_t offset = delete_file->content_offset.value(); const int64_t length = delete_file->content_size_in_bytes.value(); @@ -94,14 +149,21 @@ Result DVUtil::ReadDV(const std::shared_ptr& dele "Cannot read deletion vector larger than 2GB: {}", length); ICEBERG_ASSIGN_OR_RAISE(auto input_file, io->NewInputFile(delete_file->file_path)); - ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open()); - - std::vector bytes(static_cast(length)); - ICEBERG_RETURN_UNEXPECTED(stream->ReadFully(offset, bytes)); - ICEBERG_RETURN_UNEXPECTED(stream->Close()); - - std::span blob(reinterpret_cast(bytes.data()), - bytes.size()); + ICEBERG_ASSIGN_OR_RAISE(auto reader, puffin::PuffinReader::Make(std::move(input_file))); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->ReadFileMetadata()); + auto blob_metadata = std::ranges::find_if( + metadata.blobs, [offset](const auto& blob) { return blob.offset == offset; }); + ICEBERG_PRECHECK(blob_metadata != metadata.blobs.end(), + "No Puffin blob starts at manifest content_offset {}", offset); + ICEBERG_PRECHECK( + blob_metadata->length == length, + "Puffin blob at offset {} has length {}, manifest content_size_in_bytes is {}", + offset, blob_metadata->length, length); + ICEBERG_RETURN_UNEXPECTED(ValidateDVBlobMetadata(*blob_metadata, *delete_file)); + + ICEBERG_ASSIGN_OR_RAISE(auto blob_data, reader->ReadBlob(*blob_metadata)); + std::span blob(reinterpret_cast(blob_data.second.data()), + blob_data.second.size()); return PositionDeleteIndex::Deserialize(blob, delete_file); } @@ -118,8 +180,9 @@ Result DVUtil::WriteDVBlob(puffin::PuffinWriter& writer, .data = std::move(data), .requested_compression = puffin::PuffinCompressionCodec::kNone, }; - blob.properties.emplace("referenced-data-file", std::string(referenced_data_file)); - blob.properties.emplace("cardinality", std::format("{}", positions.Cardinality())); + blob.properties.emplace(kReferencedDataFileProperty, std::string(referenced_data_file)); + blob.properties.emplace(kCardinalityProperty, + std::format("{}", positions.Cardinality())); return writer.Write(blob); } diff --git a/src/iceberg/deletes/position_delete_index.cc b/src/iceberg/deletes/position_delete_index.cc index 53e33e635..dd59cb4da 100644 --- a/src/iceberg/deletes/position_delete_index.cc +++ b/src/iceberg/deletes/position_delete_index.cc @@ -136,6 +136,10 @@ int64_t PositionDeleteIndex::Cardinality() const { return static_cast(bitmap_.Cardinality()); } +void PositionDeleteIndex::ForEach(const std::function& fn) const { + bitmap_.ForEach(fn); +} + void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) { bitmap_.Or(other.bitmap_); delete_files_.insert(delete_files_.end(), other.delete_files_.begin(), diff --git a/src/iceberg/deletes/position_delete_index.h b/src/iceberg/deletes/position_delete_index.h index 6f301210e..828e43a2c 100644 --- a/src/iceberg/deletes/position_delete_index.h +++ b/src/iceberg/deletes/position_delete_index.h @@ -23,6 +23,7 @@ /// Index of deleted row positions for a data file. #include +#include #include #include #include @@ -66,6 +67,9 @@ class ICEBERG_EXPORT PositionDeleteIndex { /// \brief Get the number of deleted positions. int64_t Cardinality() const; + /// \brief Iterate over deleted positions in ascending order. + void ForEach(const std::function& fn) const; + /// \brief Merge another index into this one. /// \param other The index to merge (union operation) void Merge(const PositionDeleteIndex& other); diff --git a/src/iceberg/deletes/roaring_position_bitmap.cc b/src/iceberg/deletes/roaring_position_bitmap.cc index a2827d4bb..2020160f1 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.cc +++ b/src/iceberg/deletes/roaring_position_bitmap.cc @@ -271,6 +271,7 @@ Result RoaringPositionBitmap::Deserialize(std::string_vie --remaining_count; } + ICEBERG_PRECHECK(remaining == 0, "Trailing data after bitmaps: {} bytes", remaining); return RoaringPositionBitmap(std::move(impl)); } diff --git a/src/iceberg/inspect/meson.build b/src/iceberg/inspect/meson.build index 5c738008a..e67e95660 100644 --- a/src/iceberg/inspect/meson.build +++ b/src/iceberg/inspect/meson.build @@ -16,6 +16,11 @@ # under the License. install_headers( - ['history_table.h', 'metadata_table.h', 'snapshots_table.h'], + [ + 'history_table.h', + 'metadata_table.h', + 'position_deletes_table.h', + 'snapshots_table.h', + ], subdir: 'iceberg/inspect', ) diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 5e9504003..1638df5d3 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -23,6 +23,7 @@ #include #include "iceberg/inspect/history_table.h" +#include "iceberg/inspect/position_deletes_table.h" #include "iceberg/inspect/snapshots_table.h" namespace iceberg { @@ -35,6 +36,10 @@ MetadataTable::MetadataTable(std::shared_ptr source_table, MetadataTable::~MetadataTable() = default; +Result MetadataTable::Scan(const Schema& /*projected_schema*/) { + return NotSupported("Scan is not supported for this metadata table type"); +} + Result> MetadataTable::Make(std::shared_ptr
table, Kind kind) { if (table == nullptr) [[unlikely]] { @@ -46,6 +51,8 @@ Result> MetadataTable::Make(std::shared_ptr
+#include "iceberg/arrow_c_data.h" #include "iceberg/iceberg_export.h" #include "iceberg/result.h" #include "iceberg/table_identifier.h" @@ -37,6 +38,7 @@ class ICEBERG_EXPORT MetadataTable { enum class Kind { kSnapshots, kHistory, + kPositionDeletes, }; static Result> Make(std::shared_ptr
table, @@ -46,6 +48,14 @@ class ICEBERG_EXPORT MetadataTable { virtual Kind kind() const noexcept = 0; + /// \brief Scan all rows using the metadata table's full schema. + Result Scan() { return Scan(*schema_); } + + /// \brief Scan all rows projected to the requested top-level fields. + /// + /// The default implementation returns NotSupported. + virtual Result Scan(const Schema& projected_schema); + const TableIdentifier& name() const { return identifier_; } const std::shared_ptr& schema() const { return schema_; } diff --git a/src/iceberg/inspect/position_deletes_table.cc b/src/iceberg/inspect/position_deletes_table.cc new file mode 100644 index 000000000..7a7a8c45e --- /dev/null +++ b/src/iceberg/inspect/position_deletes_table.cc @@ -0,0 +1,740 @@ +/* + * 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 "iceberg/inspect/position_deletes_table.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow_c_data_guard_internal.h" +#include "iceberg/arrow_c_data_util_internal.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/deletes/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/expression/literal.h" +#include "iceberg/file_reader.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/nanoarrow_status_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/transform.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/type_util.h" + +namespace iceberg { +namespace { + +using PartitionMapping = std::vector>; + +struct PositionDeletesSchema { + std::shared_ptr schema; + std::shared_ptr partition_type; + std::unordered_map partition_mappings; +}; + +struct UnifiedPartitionField { + int32_t partition_field_id; + int32_t source_id; + std::shared_ptr transform; + SchemaField field; +}; + +TableIdentifier MakePositionDeletesTableName(const TableIdentifier& source_name) { + return TableIdentifier{.ns = source_name.ns, + .name = source_name.name + ".position_deletes"}; +} + +Status CollectFieldIds(const Type& type, std::unordered_set& field_ids) { + if (!type.is_nested()) { + return {}; + } + for (const auto& field : static_cast(type).fields()) { + if (!field_ids.insert(field.field_id()).second) { + return InvalidSchema("Duplicate field ID {} in position_deletes schema", + field.field_id()); + } + ICEBERG_RETURN_UNEXPECTED(CollectFieldIds(*field.type(), field_ids)); + } + return {}; +} + +Result TakeFreshFieldId(int64_t& candidate, + std::unordered_set& used_ids) { + constexpr int64_t kMaxFieldId = std::numeric_limits::max(); + for (int pass = 0; pass < 2; ++pass) { + while (candidate <= kMaxFieldId) { + const auto field_id = static_cast(candidate++); + if (used_ids.insert(field_id).second) { + return field_id; + } + } + candidate = 1; + } + return InvalidSchema("No field ID is available for position_deletes partition fields"); +} + +Result> ResolvePartitionField( + const PartitionField& partition_field, const Schema& current_schema) { + ICEBERG_ASSIGN_OR_RAISE(auto source_field, + current_schema.FindFieldById(partition_field.source_id())); + if (!source_field.has_value()) { + return std::nullopt; + } + return SchemaField::MakeOptional( + partition_field.field_id(), std::string(partition_field.name()), + partition_field.transform()->ResultType(source_field->get().type())); +} + +Result MakePositionDeletesSchema(const Table& table) { + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, table.schema()); + + std::vector> specs = table.metadata()->partition_specs; + std::ranges::sort(specs, {}, [](const auto& spec) { + return spec == nullptr ? std::numeric_limits::min() : spec->spec_id(); + }); + std::ranges::reverse(specs); + + std::unordered_map fields_by_partition_id; + for (const auto& spec : specs) { + ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null"); + for (const auto& partition_field : spec->fields()) { + ICEBERG_PRECHECK( + partition_field.transform()->transform_type() != TransformType::kUnknown, + "Cannot build position_deletes partition type for unknown " + "transform on field {}", + partition_field.field_id()); + ICEBERG_ASSIGN_OR_RAISE(auto field, + ResolvePartitionField(partition_field, *table_schema)); + if (!field.has_value()) { + continue; + } + auto state = UnifiedPartitionField{ + .partition_field_id = partition_field.field_id(), + .source_id = partition_field.source_id(), + .transform = partition_field.transform(), + .field = std::move(*field), + }; + auto [existing_it, inserted] = + fields_by_partition_id.try_emplace(partition_field.field_id(), state); + if (inserted) { + continue; + } + + auto& existing = existing_it->second; + if (existing.source_id != partition_field.source_id()) { + return InvalidSchema("Partition field ID {} has conflicting source IDs {} and {}", + partition_field.field_id(), existing.source_id, + partition_field.source_id()); + } + const bool existing_void = + existing.transform->transform_type() == TransformType::kVoid; + const bool current_void = + partition_field.transform()->transform_type() == TransformType::kVoid; + if (!existing_void && !current_void && + *existing.transform != *partition_field.transform()) { + return InvalidSchema( + "Partition field ID {} has incompatible transforms {} and {}", + partition_field.field_id(), existing.transform->ToString(), + partition_field.transform()->ToString()); + } + if (!existing_void && !current_void && + *existing.field.type() != *state.field.type()) { + return InvalidSchema("Partition field ID {} has incompatible types {} and {}", + partition_field.field_id(), + existing.field.type()->ToString(), + state.field.type()->ToString()); + } + if (existing_void && !current_void) { + existing.transform = std::move(state.transform); + existing.field = existing.field.WithType(state.field.type()); + } + } + } + + std::vector unified_partition_fields; + unified_partition_fields.reserve(fields_by_partition_id.size()); + for (auto& [_, field] : fields_by_partition_id) { + unified_partition_fields.push_back(std::move(field)); + } + std::ranges::sort(unified_partition_fields, {}, + &UnifiedPartitionField::partition_field_id); + + constexpr std::array kMetadataFieldIds{ + MetadataColumns::kDeleteFilePathColumnId, + MetadataColumns::kDeleteFilePosColumnId, + MetadataColumns::kDeleteFileRowColumnId, + MetadataColumns::kPartitionColumnId, + MetadataColumns::kSpecIdColumnId, + MetadataColumns::kFilePathColumnId, + MetadataColumns::kContentOffsetColumnId, + MetadataColumns::kContentSizeInBytesColumnId, + }; + std::unordered_set used_ids(kMetadataFieldIds.begin(), + kMetadataFieldIds.end()); + std::unordered_set current_schema_ids; + ICEBERG_RETURN_UNEXPECTED(CollectFieldIds(*table_schema, current_schema_ids)); + for (int32_t field_id : current_schema_ids) { + if (!used_ids.insert(field_id).second) { + return InvalidSchema("Table field ID {} conflicts with a metadata field ID", + field_id); + } + } + for (const auto& historical_schema : table.metadata()->schemas) { + ICEBERG_PRECHECK(historical_schema != nullptr, "Table schema cannot be null"); + std::unordered_set historical_ids; + ICEBERG_RETURN_UNEXPECTED(CollectFieldIds(*historical_schema, historical_ids)); + used_ids.insert(historical_ids.begin(), historical_ids.end()); + } + + int64_t next_field_id = + std::max(1, static_cast(table.metadata()->last_column_id) + 1); + std::vector partition_fields; + partition_fields.reserve(unified_partition_fields.size()); + for (const auto& field : unified_partition_fields) { + ICEBERG_ASSIGN_OR_RAISE(auto field_id, TakeFreshFieldId(next_field_id, used_ids)); + partition_fields.push_back(SchemaField::MakeOptional( + field_id, field.field.name(), field.field.type(), field.field.doc())); + } + + auto partition_type = std::make_shared(partition_fields); + std::unordered_map partition_mappings; + for (const auto& spec : specs) { + PartitionMapping mapping(partition_fields.size()); + for (size_t output_idx = 0; output_idx < partition_fields.size(); ++output_idx) { + for (size_t input_idx = 0; input_idx < spec->fields().size(); ++input_idx) { + if (spec->fields()[input_idx].field_id() == + unified_partition_fields[output_idx].partition_field_id && + spec->fields()[input_idx].source_id() == + unified_partition_fields[output_idx].source_id) { + mapping[output_idx] = input_idx; + break; + } + } + } + partition_mappings.emplace(spec->spec_id(), std::move(mapping)); + } + + std::vector fields{ + MetadataColumns::kDeleteFilePath, + MetadataColumns::kDeleteFilePos, + SchemaField::MakeOptional( + MetadataColumns::kDeleteFileRowColumnId, + MetadataColumns::kDeleteFileRowFieldName, + std::make_shared(std::vector( + table_schema->fields().begin(), table_schema->fields().end())), + MetadataColumns::kDeleteFileRowDoc), + }; + if (!partition_fields.empty()) { + fields.push_back(SchemaField::MakeRequired( + MetadataColumns::kPartitionColumnId, "partition", partition_type, + "Partition that position delete row belongs to")); + } + fields.push_back( + SchemaField::MakeRequired(MetadataColumns::kSpecIdColumnId, "spec_id", int32(), + "Spec ID used to track the file containing a row")); + fields.push_back( + SchemaField::MakeRequired(MetadataColumns::kFilePathColumnId, "delete_file_path", + string(), "Path of the file in which a row is stored")); + + if (table.metadata()->format_version >= 3) { + fields.push_back(SchemaField::MakeOptional( + MetadataColumns::kContentOffsetColumnId, "content_offset", int64(), + "The offset in the DV where the content starts")); + fields.push_back(SchemaField::MakeOptional( + MetadataColumns::kContentSizeInBytesColumnId, "content_size_in_bytes", int64(), + "The length in bytes of the DV blob")); + } + + auto schema = std::make_shared(std::move(fields)); + ICEBERG_RETURN_UNEXPECTED(schema->HighestFieldId()); + return PositionDeletesSchema{ + .schema = std::move(schema), + .partition_type = std::move(partition_type), + .partition_mappings = std::move(partition_mappings), + }; +} + +template +Result GetLiteralValue(const Literal& literal, const Type& expected_type) { + if (const auto* value = std::get_if(&literal.value())) { + return value; + } + return InvalidArrowData("Partition value has type {} but metadata schema expects {}", + literal.type()->ToString(), expected_type.ToString()); +} + +Status AppendLiteralValue(ArrowArray* array, const Literal& literal, + const std::shared_ptr& type) { + if (literal.IsNull()) { + return AppendNull(array); + } + + const Literal* value = &literal; + std::optional promoted; + if (*literal.type() != *type) { + if (!IsPromotionAllowed(literal.type(), type)) { + return InvalidArrowData( + "Partition value has type {} but metadata schema expects {}", + literal.type()->ToString(), type->ToString()); + } + ICEBERG_ASSIGN_OR_RAISE( + auto casted, literal.CastTo(std::static_pointer_cast(type))); + promoted.emplace(std::move(casted)); + value = &*promoted; + } + + switch (type->type_id()) { + case TypeId::kBoolean: { + ICEBERG_ASSIGN_OR_RAISE(auto bool_value, GetLiteralValue(*value, *type)); + return AppendBoolean(array, *bool_value); + } + case TypeId::kInt: + case TypeId::kDate: { + ICEBERG_ASSIGN_OR_RAISE(auto int_value, GetLiteralValue(*value, *type)); + return AppendInt(array, *int_value); + } + case TypeId::kLong: + case TypeId::kTime: + case TypeId::kTimestamp: + case TypeId::kTimestampTz: + case TypeId::kTimestampNs: + case TypeId::kTimestampTzNs: { + ICEBERG_ASSIGN_OR_RAISE(auto long_value, GetLiteralValue(*value, *type)); + return AppendInt(array, *long_value); + } + case TypeId::kFloat: { + ICEBERG_ASSIGN_OR_RAISE(auto float_value, GetLiteralValue(*value, *type)); + return AppendDouble(array, *float_value); + } + case TypeId::kDouble: { + ICEBERG_ASSIGN_OR_RAISE(auto double_value, GetLiteralValue(*value, *type)); + return AppendDouble(array, *double_value); + } + case TypeId::kString: { + ICEBERG_ASSIGN_OR_RAISE(auto string_value, + GetLiteralValue(*value, *type)); + return AppendString(array, *string_value); + } + case TypeId::kFixed: + case TypeId::kBinary: { + ICEBERG_ASSIGN_OR_RAISE(auto bytes_value, + GetLiteralValue>(*value, *type)); + return AppendBytes(array, *bytes_value); + } + case TypeId::kDecimal: { + ICEBERG_ASSIGN_OR_RAISE(auto decimal_value, + GetLiteralValue(*value, *type)); + return AppendBytes(array, decimal_value->ToBytes()); + } + case TypeId::kUuid: { + ICEBERG_ASSIGN_OR_RAISE(auto uuid_value, GetLiteralValue(*value, *type)); + return AppendBytes(array, uuid_value->bytes()); + } + default: + return InvalidArrowData("Unsupported partition type: {}", type->ToString()); + } +} + +Status AppendPartition(ArrowArray* array, const StructType& partition_type, + const PartitionValues& partition, + const PartitionMapping& mapping) { + ICEBERG_PRECHECK(std::cmp_equal(array->n_children, mapping.size()), + "Partition builder does not match metadata table schema"); + for (size_t output_idx = 0; output_idx < mapping.size(); ++output_idx) { + if (!mapping[output_idx].has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendNull(array->children[output_idx])); + continue; + } + const size_t input_idx = *mapping[output_idx]; + ICEBERG_PRECHECK(input_idx < partition.num_fields(), + "Partition values do not match partition spec"); + ICEBERG_ASSIGN_OR_RAISE(auto literal, partition.ValueAt(input_idx)); + ICEBERG_RETURN_UNEXPECTED( + AppendLiteralValue(array->children[output_idx], literal.get(), + partition_type.fields()[output_idx].type())); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array)); + return {}; +} + +Status AppendOptionalInt(ArrowArray* array, const std::optional& value) { + return value.has_value() ? AppendInt(array, *value) : AppendNull(array); +} + +struct OutputLayout { + explicit OutputLayout(bool has_partition, bool is_v3) + : partition(has_partition ? std::optional(3) : std::nullopt), + spec_id(has_partition ? 4 : 3), + delete_file_path(spec_id + 1), + content_offset(is_v3 ? std::optional(delete_file_path + 1) + : std::nullopt), + content_size(is_v3 ? std::optional(delete_file_path + 2) : std::nullopt) { + } + + std::optional partition; + size_t spec_id; + size_t delete_file_path; + std::optional content_offset; + std::optional content_size; +}; + +struct DeletedRowValue { + const ArrowSchema* schema; + const ArrowArray* array; + const ArrowArrayView* view; + int64_t row_index; +}; + +Status AppendPositionDeleteRow( + ArrowRowBuilder& builder, const OutputLayout& layout, const DataFile& delete_file, + std::string_view data_file_path, int64_t pos, int32_t spec_id, + const StructType& partition_type, + const std::unordered_map& partition_mappings, + const DeletedRowValue* deleted_row) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(0), data_file_path)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), pos)); + if (deleted_row == nullptr) { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendArrayValue(*deleted_row->schema, *deleted_row->array, + *deleted_row->view, deleted_row->row_index, + builder.column(2))); + } + + if (layout.partition.has_value()) { + auto mapping = partition_mappings.find(spec_id); + ICEBERG_PRECHECK(mapping != partition_mappings.end(), + "Partition spec ID {} not found", spec_id); + ICEBERG_RETURN_UNEXPECTED(AppendPartition(builder.column(*layout.partition), + partition_type, delete_file.partition, + mapping->second)); + } + + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(layout.spec_id), spec_id)); + ICEBERG_RETURN_UNEXPECTED( + AppendString(builder.column(layout.delete_file_path), delete_file.file_path)); + if (layout.content_offset.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(builder.column(*layout.content_offset), + delete_file.content_offset)); + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(builder.column(*layout.content_size), + delete_file.content_size_in_bytes)); + } + return builder.FinishRow(); +} + +SchemaField ClearPositionDeleteReadDefaults(SchemaField field) { + return field.WithInitialDefault(nullptr).WithWriteDefault(nullptr); +} + +std::shared_ptr MakePositionDeleteReadType(const std::shared_ptr& type) { + switch (type->type_id()) { + case TypeId::kStruct: { + const auto& struct_type = static_cast(*type); + std::vector fields; + fields.reserve(struct_type.fields().size()); + for (const auto& field : struct_type.fields()) { + fields.push_back(ClearPositionDeleteReadDefaults( + field.WithType(MakePositionDeleteReadType(field.type()))) + .AsOptional()); + } + return std::make_shared(std::move(fields)); + } + case TypeId::kList: { + const auto& list_type = static_cast(*type); + return std::make_shared( + ClearPositionDeleteReadDefaults(list_type.element().WithType( + MakePositionDeleteReadType(list_type.element().type())))); + } + case TypeId::kMap: { + const auto& map_type = static_cast(*type); + return std::make_shared( + ClearPositionDeleteReadDefaults(map_type.key()), + ClearPositionDeleteReadDefaults(map_type.value().WithType( + MakePositionDeleteReadType(map_type.value().type())))); + } + default: + return type; + } +} + +std::shared_ptr PositionDeleteFileSchema(const Schema& table_schema, + bool include_row) { + std::vector fields{ + MetadataColumns::kDeleteFilePath, + MetadataColumns::kDeleteFilePos, + }; + if (include_row) { + std::vector row_fields; + row_fields.reserve(table_schema.fields().size()); + for (const auto& field : table_schema.fields()) { + row_fields.push_back(ClearPositionDeleteReadDefaults( + field.WithType(MakePositionDeleteReadType(field.type()))) + .AsOptional()); + } + fields.push_back(SchemaField::MakeRequired( + MetadataColumns::kDeleteFileRowColumnId, MetadataColumns::kDeleteFileRowFieldName, + std::make_shared(std::move(row_fields)), + MetadataColumns::kDeleteFileRowDoc)); + } + return std::make_shared(std::move(fields)); +} + +struct PositionDeleteReader { + std::unique_ptr reader; + bool has_row; +}; + +Result OpenPositionDeleteFile(const DataFile& file, + const Schema& table_schema, + const std::shared_ptr& io) { + ICEBERG_PRECHECK(file.file_format == FileFormatType::kParquet, + "Unsupported position delete format: {}", ToString(file.file_format)); + auto reader = ReaderFactoryRegistry::Open( + file.file_format, ReaderOptions{ + .path = file.file_path, + .length = static_cast(file.file_size_in_bytes), + .io = io, + .projection = PositionDeleteFileSchema(table_schema, true), + }); + if (reader.has_value()) { + return PositionDeleteReader{.reader = std::move(*reader), .has_row = true}; + } + + const auto missing_row_message = std::format("Missing required field with id: {}", + MetadataColumns::kDeleteFileRowColumnId); + if (reader.error().kind != ErrorKind::kInvalidSchema || + reader.error().message != missing_row_message) { + return std::unexpected(reader.error()); + } + + ICEBERG_ASSIGN_OR_RAISE( + auto without_row, + ReaderFactoryRegistry::Open( + file.file_format, + ReaderOptions{ + .path = file.file_path, + .length = static_cast(file.file_size_in_bytes), + .io = io, + .projection = PositionDeleteFileSchema(table_schema, false), + })); + return PositionDeleteReader{.reader = std::move(without_row), .has_row = false}; +} + +Status AppendParquetDeletes( + ArrowRowBuilder& builder, const OutputLayout& layout, + const std::shared_ptr& delete_file, int32_t spec_id, + const StructType& partition_type, + const std::unordered_map& partition_mappings, + const Schema& table_schema, const std::shared_ptr& io) { + ICEBERG_ASSIGN_OR_RAISE(auto delete_reader, + OpenPositionDeleteFile(*delete_file, table_schema, io)); + auto& reader = delete_reader.reader; + ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader->Schema()); + internal::ArrowSchemaGuard schema_guard(&arrow_schema); + const int64_t expected_columns = delete_reader.has_row ? 3 : 2; + ICEBERG_PRECHECK(arrow_schema.n_children == expected_columns, + "Position delete reader returned {} columns, expected {}", + arrow_schema.n_children, expected_columns); + + ArrowArrayView array_view; + internal::ArrowArrayViewGuard view_guard(&array_view); + ArrowError error; + ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR( + ArrowArrayViewInitFromSchema(&array_view, &arrow_schema, &error), error); + + while (true) { + ICEBERG_ASSIGN_OR_RAISE(auto batch_opt, reader->Next()); + if (!batch_opt.has_value()) { + break; + } + + auto& batch = *batch_opt; + internal::ArrowArrayGuard batch_guard(&batch); + ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR( + ArrowArrayViewSetArray(&array_view, &batch, &error), error); + ICEBERG_PRECHECK(batch.n_children == expected_columns, + "Position delete batch has {} columns, expected {}", + batch.n_children, expected_columns); + + const auto* path_view = array_view.children[0]; + const auto* pos_view = array_view.children[1]; + if (ArrowArrayViewComputeNullCount(path_view) != 0 || + ArrowArrayViewComputeNullCount(pos_view) != 0) { + return InvalidArrowData( + "position delete file has null values in required pos/file_path columns"); + } + if (delete_reader.has_row && + ArrowArrayViewComputeNullCount(array_view.children[2]) != 0) { + return InvalidArrowData("position delete file has null row values"); + } + + const int64_t* positions = pos_view->buffer_views[1].data.as_int64 + pos_view->offset; + for (int64_t row = 0; row < batch.length; ++row) { + ICEBERG_PRECHECK(positions[row] >= 0, "Invalid negative delete position: {}", + positions[row]); + const ArrowStringView path = ArrowArrayViewGetStringUnsafe(path_view, row); + std::optional deleted_row; + if (delete_reader.has_row) { + deleted_row.emplace(DeletedRowValue{ + .schema = arrow_schema.children[2], + .array = batch.children[2], + .view = array_view.children[2], + .row_index = row, + }); + } + ICEBERG_RETURN_UNEXPECTED(AppendPositionDeleteRow( + builder, layout, *delete_file, + std::string_view(path.data, static_cast(path.size_bytes)), + positions[row], spec_id, partition_type, partition_mappings, + deleted_row ? &*deleted_row : nullptr)); + } + } + + return reader->Close(); +} + +Status AppendDeletionVector( + ArrowRowBuilder& builder, const OutputLayout& layout, + const std::shared_ptr& delete_file, int32_t spec_id, + const StructType& partition_type, + const std::unordered_map& partition_mappings, + const std::shared_ptr& io) { + ICEBERG_PRECHECK(delete_file->referenced_data_file.has_value(), + "Deletion vector requires a referenced data file"); + ICEBERG_ASSIGN_OR_RAISE(auto positions, DVUtil::ReadDV(delete_file, io)); + + Status status = {}; + positions.ForEach([&](int64_t pos) { + if (status.has_value()) { + status = AppendPositionDeleteRow(builder, layout, *delete_file, + *delete_file->referenced_data_file, pos, spec_id, + partition_type, partition_mappings, nullptr); + } + }); + return status; +} + +Result ProjectResult(ArrowArray array, const Schema& input_schema, + const Schema& projected_schema) { + if (input_schema == projected_schema) { + return array; + } + + internal::ArrowArrayGuard array_guard(&array); + ICEBERG_ASSIGN_OR_RAISE( + auto projection, + ProjectionContext::Make(input_schema, projected_schema, + ProjectionContext::ResolveProjectBatchFunction())); + std::vector rows(static_cast(array.length)); + std::iota(rows.begin(), rows.end(), 0); + array_guard.Release(); + return ProjectBatch(&array, rows, projection); +} + +} // namespace + +PositionDeletesTable::PositionDeletesTable( + std::shared_ptr
table, std::shared_ptr schema, + std::shared_ptr partition_type, + std::unordered_map partition_mappings) + : MetadataTable(table, MakePositionDeletesTableName(table->name()), + std::move(schema)), + partition_type_(std::move(partition_type)), + partition_mappings_(std::move(partition_mappings)) {} + +PositionDeletesTable::~PositionDeletesTable() = default; + +Result> PositionDeletesTable::Make( + std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + ICEBERG_ASSIGN_OR_RAISE(auto schema, MakePositionDeletesSchema(*table)); + return std::unique_ptr(new PositionDeletesTable( + std::move(table), std::move(schema.schema), std::move(schema.partition_type), + std::move(schema.partition_mappings))); +} + +Result PositionDeletesTable::Scan(const Schema& projected_schema) { + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema())); + const bool has_partition = !partition_type_->fields().empty(); + const bool is_v3 = source_table()->metadata()->format_version >= 3; + const OutputLayout layout(has_partition, is_v3); + + if (source_table()->metadata()->current_snapshot_id != kInvalidSnapshotId) { + ICEBERG_ASSIGN_OR_RAISE(auto current_snapshot, source_table()->current_snapshot()); + SnapshotCache cache(current_snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DeleteManifests(source_table()->io())); + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, source_table()->specs()); + const auto& specs = specs_ref.get(); + + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, source_table()->io(), table_schema, specs)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries()); + for (const auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Manifest entry must have a data file"); + const auto& file = entry.data_file; + if (file->content != DataFile::Content::kPositionDeletes) { + continue; + } + const int32_t spec_id = + file->partition_spec_id.value_or(manifest.partition_spec_id); + if (file->IsDeletionVector()) { + ICEBERG_RETURN_UNEXPECTED( + AppendDeletionVector(builder, layout, file, spec_id, *partition_type_, + partition_mappings_, source_table()->io())); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendParquetDeletes( + builder, layout, file, spec_id, *partition_type_, partition_mappings_, + *table_schema, source_table()->io())); + } + } + } + } + + ICEBERG_ASSIGN_OR_RAISE(auto result, std::move(builder).Finish()); + return ProjectResult(std::move(result), *schema(), projected_schema); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/position_deletes_table.h b/src/iceberg/inspect/position_deletes_table.h new file mode 100644 index 000000000..ae0886616 --- /dev/null +++ b/src/iceberg/inspect/position_deletes_table.h @@ -0,0 +1,63 @@ +/* + * 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 + +/// \file iceberg/inspect/position_deletes_table.h +/// \brief Define the position_deletes metadata table. + +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Metadata table that expands physical position-delete storage into rows. +class ICEBERG_EXPORT PositionDeletesTable : public MetadataTable { + public: + /// \brief Create a position_deletes metadata table for a source table. + /// \param table Source table whose current snapshot will be inspected. + /// \return A metadata table or an error if its schema cannot be constructed. + static Result> Make(std::shared_ptr
table); + + ~PositionDeletesTable() override; + + Kind kind() const noexcept override { return Kind::kPositionDeletes; } + + using MetadataTable::Scan; + Result Scan(const Schema& projected_schema) override; + + private: + PositionDeletesTable( + std::shared_ptr
table, std::shared_ptr schema, + std::shared_ptr partition_type, + std::unordered_map>> partition_mappings); + + std::shared_ptr partition_type_; + std::unordered_map>> partition_mappings_; +}; + +} // namespace iceberg diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 989f4ae03..020dbda35 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -109,6 +109,7 @@ iceberg_sources = files( 'inheritable_metadata.cc', 'inspect/history_table.cc', 'inspect/metadata_table.cc', + 'inspect/position_deletes_table.cc', 'inspect/snapshots_table.cc', 'json_serde.cc', 'location_provider.cc', diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 5ca9fd915..3e316c3c1 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -158,6 +158,8 @@ add_iceberg_test(util_test add_iceberg_test(puffin_test USE_DATA SOURCES + dv_util_test.cc + puffin_dv_interop_test.cc puffin_format_test.cc puffin_json_test.cc puffin_reader_writer_test.cc) @@ -188,6 +190,9 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc) + add_iceberg_test(position_deletes_table_test USE_BUNDLE SOURCES + position_deletes_table_test.cc) + add_iceberg_test(eval_expr_test USE_BUNDLE SOURCES diff --git a/src/iceberg/test/dv_util_test.cc b/src/iceberg/test/dv_util_test.cc new file mode 100644 index 000000000..d84eb0784 --- /dev/null +++ b/src/iceberg/test/dv_util_test.cc @@ -0,0 +1,303 @@ +/* + * 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 +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/deletes/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/json_serde_internal.h" +#include "iceberg/puffin/puffin_format.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/endian.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +namespace { + +constexpr std::string_view kReferencedDataFile = "data.parquet"; +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +struct MetadataTestCase { + std::string name; + std::function&)> mutate_data; + std::function mutate_blob; + std::function mutate_manifest; + std::string expected_error; +}; + +void AddTrailingBitmapData(std::vector& data) { + constexpr size_t kLengthPrefixBytes = 4; + constexpr size_t kCrcBytes = 4; + + data.insert(data.end() - kCrcBytes, 0); + const auto length = static_cast(data.size() - kLengthPrefixBytes - kCrcBytes); + WriteBigEndian(length, data.data()); + + uLong crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, reinterpret_cast(data.data() + kLengthPrefixBytes), + static_cast(length)); + WriteBigEndian(static_cast(crc), data.data() + data.size() - kCrcBytes); +} + +Result> WriteDVFixture(const std::shared_ptr& io, + const MetadataTestCase& test_case) { + PositionDeleteIndex positions; + positions.Delete(1); + positions.Delete(3); + positions.Delete(5); + ICEBERG_ASSIGN_OR_RAISE(auto data, positions.Serialize()); + if (test_case.mutate_data) { + test_case.mutate_data(data); + } + + puffin::Blob blob{ + .type = std::string(puffin::StandardBlobTypes::kDeletionVectorV1), + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .snapshot_id = -1, + .sequence_number = -1, + .data = std::move(data), + .requested_compression = puffin::PuffinCompressionCodec::kNone, + .properties = + { + {std::string(kReferencedDataFileProperty), + std::string(kReferencedDataFile)}, + {std::string(kCardinalityProperty), "3"}, + }, + }; + if (test_case.mutate_blob) { + test_case.mutate_blob(blob); + } + + const std::string path = "memory://" + test_case.name + ".puffin"; + const auto codec = + blob.requested_compression.value_or(puffin::PuffinCompressionCodec::kNone); + puffin::BlobMetadata blob_metadata{ + .type = blob.type, + .input_fields = blob.input_fields, + .snapshot_id = blob.snapshot_id, + .sequence_number = blob.sequence_number, + .offset = puffin::PuffinFormat::kMagicLength, + .length = static_cast(blob.data.size()), + .compression_codec = std::string(puffin::CodecName(codec)), + .properties = blob.properties, + }; + const std::string footer = + puffin::ToJsonString(puffin::FileMetadata{.blobs = {blob_metadata}}); + + std::vector file_data; + auto append = [&file_data](const void* data, size_t size) { + auto bytes = std::span(reinterpret_cast(data), size); + file_data.insert(file_data.end(), bytes.begin(), bytes.end()); + }; + append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size()); + append(blob.data.data(), blob.data.size()); + append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size()); + append(footer.data(), footer.size()); + + std::array footer_struct{}; + WriteLittleEndian(static_cast(footer.size()), footer_struct.data()); + std::memcpy(footer_struct.data() + puffin::PuffinFormat::kFooterStructMagicOffset, + puffin::PuffinFormat::kMagicV1.data(), + puffin::PuffinFormat::kMagicV1.size()); + append(footer_struct.data(), footer_struct.size()); + io->AddFile(path, file_data); + + auto delete_file = std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = path, + .file_format = FileFormatType::kPuffin, + .record_count = 3, + .file_size_in_bytes = static_cast(file_data.size()), + .referenced_data_file = std::string(kReferencedDataFile), + .content_offset = blob_metadata.offset, + .content_size_in_bytes = blob_metadata.length, + }); + if (test_case.mutate_manifest) { + test_case.mutate_manifest(*delete_file); + } + return delete_file; +} + +} // namespace + +TEST(DVUtilTest, RejectsMalformedPuffinMetadata) { + const std::vector cases = { + { + .name = "wrong-offset", + .mutate_manifest = [](DataFile& file) { ++*file.content_offset; }, + .expected_error = "No Puffin blob starts", + }, + { + .name = "wrong-length", + .mutate_manifest = [](DataFile& file) { --*file.content_size_in_bytes; }, + .expected_error = "manifest content_size_in_bytes", + }, + { + .name = "missing-manifest-reference", + .mutate_manifest = [](DataFile& file) { file.referenced_data_file.reset(); }, + .expected_error = "requires referenced_data_file", + }, + { + .name = "empty-manifest-reference", + .mutate_manifest = [](DataFile& file) { file.referenced_data_file = ""; }, + .expected_error = "requires referenced_data_file", + }, + { + .name = "wrong-type", + .mutate_blob = [](puffin::Blob& blob) { blob.type = "test-blob"; }, + .expected_error = "Invalid deletion vector blob type", + }, + { + .name = "wrong-snapshot", + .mutate_blob = [](puffin::Blob& blob) { blob.snapshot_id = 0; }, + .expected_error = "snapshot-id -1", + }, + { + .name = "wrong-sequence", + .mutate_blob = [](puffin::Blob& blob) { blob.sequence_number = 0; }, + .expected_error = "sequence-number -1", + }, + { + .name = "compressed", + .mutate_blob = + [](puffin::Blob& blob) { + blob.requested_compression = puffin::PuffinCompressionCodec::kZstd; + }, + .expected_error = "must not be compressed", + }, + { + .name = "missing-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties.erase(std::string(kReferencedDataFileProperty)); + }, + .expected_error = "requires non-empty 'referenced-data-file'", + }, + { + .name = "empty-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kReferencedDataFileProperty)] = ""; + }, + .expected_error = "requires non-empty 'referenced-data-file'", + }, + { + .name = "mismatched-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kReferencedDataFileProperty)] = + "other.parquet"; + }, + .expected_error = "does not match Puffin", + }, + { + .name = "missing-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties.erase(std::string(kCardinalityProperty)); + }, + .expected_error = "requires 'cardinality'", + }, + { + .name = "invalid-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "three"; + }, + .expected_error = "Failed to parse", + }, + { + .name = "negative-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "-1"; + }, + .expected_error = "must be non-negative", + }, + { + .name = "mismatched-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "4"; + }, + .expected_error = "Manifest record_count 3", + }, + { + .name = "trailing-bitmap-data", + .mutate_data = AddTrailingBitmapData, + .expected_error = "Trailing data after bitmaps", + }, + }; + + for (const auto& test_case : cases) { + SCOPED_TRACE(test_case.name); + auto io = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + EXPECT_THAT(DVUtil::ReadDV(delete_file, io), + HasErrorMessage(test_case.expected_error)); + } +} + +TEST(DVUtilTest, ReadsValidPuffinMetadata) { + auto io = std::make_shared(); + MetadataTestCase test_case{.name = "valid"}; + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); + + EXPECT_EQ(positions.Cardinality(), 3); + EXPECT_TRUE(positions.IsDeleted(1)); + EXPECT_TRUE(positions.IsDeleted(3)); + EXPECT_TRUE(positions.IsDeleted(5)); +} + +TEST(DVUtilTest, ReadsUsingActualFileSize) { + auto io = std::make_shared(); + MetadataTestCase test_case{ + .name = "stale-manifest-file-size", + .mutate_manifest = [](DataFile& file) { file.file_size_in_bytes = 0; }, + }; + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); + + EXPECT_EQ(positions.Cardinality(), 3); +} + +} // namespace iceberg diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build index 6844df5e9..aeb909cca 100644 --- a/src/iceberg/test/meson.build +++ b/src/iceberg/test/meson.build @@ -63,6 +63,10 @@ iceberg_tests = { 'update_schema_test.cc', ), }, + 'position_deletes_table_test': { + 'sources': files('position_deletes_table_test.cc'), + 'use_data': true, + }, 'logging_test': { 'sources': files( 'cerr_logger_test.cc', diff --git a/src/iceberg/test/position_deletes_table_test.cc b/src/iceberg/test/position_deletes_table_test.cc new file mode 100644 index 000000000..fd44ba43b --- /dev/null +++ b/src/iceberg/test/position_deletes_table_test.cc @@ -0,0 +1,616 @@ +/* + * 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 "iceberg/inspect/position_deletes_table.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow/arrow_register.h" +#include "iceberg/arrow/arrow_status_internal.h" +#include "iceberg/data/position_delete_writer.h" +#include "iceberg/deletes/dv_writer.h" +#include "iceberg/file_format.h" +#include "iceberg/file_writer.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/parquet/parquet_register.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/schema.h" +#include "iceberg/schema_internal.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/scan_test_base.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +using ::testing::ElementsAre; + +class PositionDeletesTableTest : public ScanTestBase { + protected: + void SetUp() override { + ScanTestBase::SetUp(); + parquet::RegisterAll(); + arrow::RegisterAll(); + } + + std::shared_ptr WritePositionDeletes( + std::string path, const std::vector>& deletes, + const std::shared_ptr& spec, const PartitionValues& partition) { + PositionDeleteWriterOptions options{ + .path = std::move(path), + .schema = schema_, + .spec = spec, + .partition = partition, + .format = FileFormatType::kParquet, + .io = file_io_, + .flush_threshold = 10000, + .properties = {{"write.parquet.compression-codec", "uncompressed"}}, + }; + auto writer = PositionDeleteWriter::Make(options).value(); + for (const auto& [file_path, pos] : deletes) { + ICEBERG_THROW_NOT_OK(writer->WriteDelete(file_path, pos)); + } + ICEBERG_THROW_NOT_OK(writer->Close()); + return writer->Metadata().value().data_files.front(); + } + + Result> WritePositionDeletesWithRows( + std::string path, std::string_view json, std::vector row_fields, + bool row_required, const std::shared_ptr& spec, + const PartitionValues& partition) { + auto row_type = std::make_shared(std::move(row_fields)); + auto row_field = + row_required ? SchemaField::MakeRequired(MetadataColumns::kDeleteFileRowColumnId, + MetadataColumns::kDeleteFileRowFieldName, + std::move(row_type), + MetadataColumns::kDeleteFileRowDoc) + : SchemaField::MakeOptional(MetadataColumns::kDeleteFileRowColumnId, + MetadataColumns::kDeleteFileRowFieldName, + std::move(row_type), + MetadataColumns::kDeleteFileRowDoc); + auto delete_schema = std::make_shared(std::vector{ + MetadataColumns::kDeleteFilePath, + MetadataColumns::kDeleteFilePos, + std::move(row_field), + }); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, WriterFactoryRegistry::Open( + FileFormatType::kParquet, + WriterOptions{ + .path = path, + .schema = delete_schema, + .io = file_io_, + .properties = WriterProperties::FromMap( + {{"write.parquet.compression-codec", "uncompressed"}}), + })); + + ArrowSchema arrow_c_schema; + ICEBERG_THROW_NOT_OK(ToArrowSchema(*delete_schema, &arrow_c_schema)); + auto arrow_type = ::arrow::ImportType(&arrow_c_schema).ValueOrDie(); + auto rows = ::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_type->fields()), + std::string(json)) + .ValueOrDie(); + ArrowArray arrow_array; + ICEBERG_ARROW_RETURN_NOT_OK(::arrow::ExportArray(*rows, &arrow_array)); + ICEBERG_RETURN_UNEXPECTED(writer->Write(&arrow_array)); + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto length, writer->length()); + + return std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = std::move(path), + .file_format = FileFormatType::kParquet, + .partition = partition, + .record_count = rows->length(), + .file_size_in_bytes = length, + .partition_spec_id = spec->spec_id(), + }); + } + + std::vector> WriteDeletionVectors( + std::string path, + const std::vector>>& deletes, + const std::shared_ptr& spec, const PartitionValues& partition) { + auto writer = DVWriter::Make(DVWriterOptions{ + .path = std::move(path), + .io = file_io_, + .load_previous_deletes = [](std::string_view) + -> Result> { + return std::nullopt; + }, + }) + .value(); + for (const auto& [file_path, positions] : deletes) { + for (int64_t pos : positions) { + ICEBERG_THROW_NOT_OK(writer->Delete(file_path, pos, spec, partition)); + } + } + ICEBERG_THROW_NOT_OK(writer->Close()); + return writer->Metadata().value().data_files; + } + + std::shared_ptr MakeSnapshot(int8_t table_format_version, int64_t snapshot_id, + int64_t sequence_number, + const std::vector& manifests) { + auto manifest_list = WriteManifestList(table_format_version, snapshot_id, 0, + sequence_number, manifests); + return std::make_shared(Snapshot{ + .snapshot_id = snapshot_id, + .parent_snapshot_id = std::nullopt, + .sequence_number = sequence_number, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = std::move(manifest_list), + .summary = {{"operation", "delete"}}, + .schema_id = schema_->schema_id(), + }); + } + + std::shared_ptr
MakeTable(int8_t format_version, + const std::shared_ptr& spec, + std::shared_ptr snapshot = nullptr) { + return MakeTableWithSpecs(format_version, {spec}, spec, std::move(snapshot)); + } + + std::shared_ptr
MakeTableWithSpecs( + int8_t format_version, std::vector> specs, + const std::shared_ptr& default_spec, + std::shared_ptr snapshot = nullptr) { + std::vector> snapshots; + if (snapshot != nullptr) { + snapshots.push_back(snapshot); + } + int32_t last_partition_id = PartitionSpec::kInvalidPartitionFieldId; + for (const auto& spec : specs) { + last_partition_id = std::max(last_partition_id, spec->last_assigned_field_id()); + } + auto metadata = std::make_shared(TableMetadata{ + .format_version = format_version, + .table_uuid = "test-table-uuid", + .location = "/tmp/table", + .last_sequence_number = snapshot == nullptr ? 0 : snapshot->sequence_number, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = schema_->HighestFieldId().value(), + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = std::move(specs), + .default_spec_id = default_spec->spec_id(), + .last_partition_id = last_partition_id, + .current_snapshot_id = + snapshot == nullptr ? kInvalidSnapshotId : snapshot->snapshot_id, + .snapshots = std::move(snapshots), + .default_sort_order_id = 0, + }); + return Table::Make(TableIdentifier{.name = "table"}, std::move(metadata), + "/tmp/table/metadata.json", file_io_, + std::make_shared<::testing::NiceMock>()) + .value(); + } + + std::unique_ptr MakePositionDeletesTable( + const std::shared_ptr
& table) { + return MetadataTable::Make(table, MetadataTable::Kind::kPositionDeletes).value(); + } + + static std::shared_ptr<::arrow::RecordBatch> Import(ArrowArray array, + const Schema& schema) { + ArrowSchema c_schema; + ICEBERG_THROW_NOT_OK(ToArrowSchema(schema, &c_schema)); + auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + return ::arrow::ImportRecordBatch(&array, arrow_schema).ValueOrDie(); + } +}; + +TEST_P(PositionDeletesTableTest, SchemaAndEmptyScan) { + auto table = MakePositionDeletesTable(MakeTable(GetParam(), unpartitioned_spec_)); + + EXPECT_EQ(table->kind(), MetadataTable::Kind::kPositionDeletes); + EXPECT_EQ(table->name().name, "table.position_deletes"); + + std::vector names; + for (const auto& field : table->schema()->fields()) { + names.emplace_back(field.name()); + } + std::vector expected{"file_path", "pos", "row", "spec_id", + "delete_file_path"}; + if (GetParam() >= 3) { + expected.push_back("content_offset"); + expected.push_back("content_size_in_bytes"); + } + EXPECT_EQ(names, expected); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + EXPECT_EQ(batch->num_rows(), 0); +} + +TEST_P(PositionDeletesTableTest, ExpandsParquetAndProjectsColumns) { + auto delete_file = WritePositionDeletes( + "position-deletes.parquet", {{"data-a.parquet", 2}, {"data-b.parquet", 7}}, + partitioned_spec_, PartitionValues(Literal::Int(11))); + constexpr int64_t kSnapshotId = 10; + auto manifest = WriteDeleteManifest( + GetParam(), kSnapshotId, + {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + partitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(GetParam(), partitioned_spec_, + MakeSnapshot(GetParam(), kSnapshotId, 1, {manifest}))); + + const auto& fields = table->schema()->fields(); + auto projected = + std::make_unique(std::vector{fields[5], fields[1], fields[3]}); + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan(*projected)); + auto batch = Import(std::move(array), *projected); + + ASSERT_EQ(batch->num_rows(), 2); + EXPECT_EQ(batch->schema()->field(0)->name(), "delete_file_path"); + EXPECT_EQ(batch->schema()->field(1)->name(), "pos"); + EXPECT_EQ(batch->schema()->field(2)->name(), "partition"); + auto delete_paths = std::static_pointer_cast<::arrow::StringArray>(batch->column(0)); + auto positions = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + auto partitions = std::static_pointer_cast<::arrow::StructArray>(batch->column(2)); + auto buckets = std::static_pointer_cast<::arrow::Int32Array>(partitions->field(0)); + std::vector actual_delete_paths{delete_paths->GetString(0), + delete_paths->GetString(1)}; + std::vector actual_positions{positions->Value(0), positions->Value(1)}; + std::vector actual_buckets{buckets->Value(0), buckets->Value(1)}; + EXPECT_THAT(actual_delete_paths, + ElementsAre("position-deletes.parquet", "position-deletes.parquet")); + EXPECT_THAT(actual_positions, ElementsAre(2, 7)); + EXPECT_THAT(actual_buckets, ElementsAre(11, 11)); +} + +TEST_F(PositionDeletesTableTest, SurfacesDeletedRowsFromParquet) { + ICEBERG_UNWRAP_OR_FAIL( + auto delete_file, + WritePositionDeletesWithRows( + "position-deletes-with-rows.parquet", + R"([["data.parquet", 5, [42, "deleted"]]])", + std::vector(schema_->fields().begin(), schema_->fields().end()), + true, unpartitioned_spec_, PartitionValues{})); + constexpr int64_t kSnapshotId = 15; + auto manifest = WriteDeleteManifest( + 2, kSnapshotId, {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(2, unpartitioned_spec_, MakeSnapshot(2, kSnapshotId, 1, {manifest}))); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 1); + + auto rows = std::static_pointer_cast<::arrow::StructArray>(batch->column(2)); + ASSERT_FALSE(rows->IsNull(0)); + auto ids = std::static_pointer_cast<::arrow::Int32Array>(rows->field(0)); + auto data = std::static_pointer_cast<::arrow::StringArray>(rows->field(1)); + EXPECT_EQ(ids->Value(0), 42); + EXPECT_EQ(data->GetString(0), "deleted"); +} + +TEST_F(PositionDeletesTableTest, ProjectsSubsetDeletedRowsFromParquet) { + ICEBERG_UNWRAP_OR_FAIL( + auto delete_file, + WritePositionDeletesWithRows("position-deletes-with-subset-rows.parquet", + R"([["data.parquet", 5, ["deleted"]]])", + {schema_->fields()[1]}, true, unpartitioned_spec_, + PartitionValues{})); + constexpr int64_t kSnapshotId = 16; + auto manifest = WriteDeleteManifest( + 2, kSnapshotId, {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(2, unpartitioned_spec_, MakeSnapshot(2, kSnapshotId, 1, {manifest}))); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 1); + + auto rows = std::static_pointer_cast<::arrow::StructArray>(batch->column(2)); + ASSERT_FALSE(rows->IsNull(0)); + auto ids = std::static_pointer_cast<::arrow::Int32Array>(rows->field(0)); + auto data = std::static_pointer_cast<::arrow::StringArray>(rows->field(1)); + EXPECT_TRUE(ids->IsNull(0)); + EXPECT_EQ(data->GetString(0), "deleted"); +} + +TEST_F(PositionDeletesTableTest, DoesNotDefaultOmittedDeletedRowFields) { + schema_ = std::make_shared( + std::vector{ + SchemaField::MakeRequired(1, "id", int32()) + .WithInitialDefault(std::make_shared(Literal::Int(99))), + SchemaField::MakeRequired(2, "data", string()), + }, + 2); + ICEBERG_UNWRAP_OR_FAIL( + auto delete_file, + WritePositionDeletesWithRows("position-deletes-with-defaulted-subset-row.parquet", + R"([["data.parquet", 5, ["deleted"]]])", + {schema_->fields()[1]}, true, unpartitioned_spec_, + PartitionValues{})); + constexpr int64_t kSnapshotId = 17; + auto manifest = WriteDeleteManifest( + 3, kSnapshotId, {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(3, unpartitioned_spec_, MakeSnapshot(3, kSnapshotId, 1, {manifest}))); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 1); + + auto rows = std::static_pointer_cast<::arrow::StructArray>(batch->column(2)); + auto ids = std::static_pointer_cast<::arrow::Int32Array>(rows->field(0)); + auto data = std::static_pointer_cast<::arrow::StringArray>(rows->field(1)); + EXPECT_TRUE(ids->IsNull(0)); + EXPECT_EQ(data->GetString(0), "deleted"); +} + +TEST_F(PositionDeletesTableTest, RejectsNullDeletedRowsFromParquet) { + ICEBERG_UNWRAP_OR_FAIL( + auto delete_file, + WritePositionDeletesWithRows( + "position-deletes-with-null-row.parquet", R"([["data.parquet", 5, null]])", + std::vector(schema_->fields().begin(), schema_->fields().end()), + false, unpartitioned_spec_, PartitionValues{})); + constexpr int64_t kSnapshotId = 19; + auto manifest = WriteDeleteManifest( + 2, kSnapshotId, {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(2, unpartitioned_spec_, MakeSnapshot(2, kSnapshotId, 1, {manifest}))); + + auto result = table->Scan(); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArrowData)); + EXPECT_THAT(result, HasErrorMessage("null row values")); +} + +TEST_F(PositionDeletesTableTest, ReassignsPartitionIdsCollidingWithNestedRows) { + schema_ = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "payload", + std::make_shared(std::vector{ + SchemaField::MakeRequired(1000, "id", int32()), + })), + SchemaField::MakeRequired(2, "data", string()), + }); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, + PartitionSpec::Make(2, {PartitionField(1000, 1000, "id", Transform::Identity())})); + auto shared_spec = std::shared_ptr(std::move(spec)); + auto table = MakePositionDeletesTable(MakeTable(2, shared_spec)); + + EXPECT_THAT(table->schema()->HighestFieldId(), IsOk()); + const auto& fields = table->schema()->fields(); + auto partition_type = std::static_pointer_cast(fields[3].type()); + ASSERT_EQ(partition_type->fields().size(), 1); + EXPECT_NE(partition_type->fields()[0].field_id(), 1000); + EXPECT_NE(partition_type->fields()[0].field_id(), MetadataColumns::kPartitionColumnId); +} + +TEST_F(PositionDeletesTableTest, RejectsIncompatiblePartitionEvolution) { + ICEBERG_UNWRAP_OR_FAIL( + auto old_spec, + PartitionSpec::Make(1, {PartitionField(2, 1000, "data", Transform::Identity())})); + ICEBERG_UNWRAP_OR_FAIL( + auto new_spec, + PartitionSpec::Make(2, {PartitionField(2, 1000, "data", Transform::Bucket(16))})); + auto shared_old_spec = std::shared_ptr(std::move(old_spec)); + auto shared_new_spec = std::shared_ptr(std::move(new_spec)); + auto source = + MakeTableWithSpecs(2, {shared_old_spec, shared_new_spec}, shared_new_spec); + + auto table = MetadataTable::Make(source, MetadataTable::Kind::kPositionDeletes); + EXPECT_THAT(table, IsError(ErrorKind::kInvalidSchema)); + EXPECT_THAT(table, HasErrorMessage("incompatible transforms")); +} + +TEST_F(PositionDeletesTableTest, PromotesHistoricalPartitionValues) { + ICEBERG_UNWRAP_OR_FAIL( + auto spec, + PartitionSpec::Make(1, {PartitionField(1, 1000, "id", Transform::Identity())})); + auto shared_spec = std::shared_ptr(std::move(spec)); + auto historical_schema = schema_; + auto delete_file = WritePositionDeletes("position-deletes-int-partition.parquet", + {{"data.parquet", 5}}, shared_spec, + PartitionValues(Literal::Int(11))); + constexpr int64_t kSnapshotId = 18; + auto manifest = WriteDeleteManifest( + 2, kSnapshotId, {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, delete_file)}, + shared_spec); + + schema_ = std::make_shared( + std::vector{ + SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeRequired(2, "data", string()), + }, + 2); + auto source = MakeTable(2, shared_spec, MakeSnapshot(2, kSnapshotId, 1, {manifest})); + source->metadata()->schemas.insert(source->metadata()->schemas.begin(), + historical_schema); + + auto table = MakePositionDeletesTable(source); + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 1); + auto partitions = std::static_pointer_cast<::arrow::StructArray>(batch->column(3)); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(partitions->field(0)); + EXPECT_EQ(ids->Value(0), 11); +} + +TEST_F(PositionDeletesTableTest, OmitsDroppedHistoricalPartitionSources) { + auto historical_schema = std::make_shared( + std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string()), + SchemaField::MakeOptional(3, "dropped", string()), + }, + 1); + ICEBERG_UNWRAP_OR_FAIL( + auto old_spec, PartitionSpec::Make( + 1, {PartitionField(3, 1000, "dropped", Transform::Identity())})); + auto shared_old_spec = std::shared_ptr(std::move(old_spec)); + auto source = + MakeTableWithSpecs(2, {shared_old_spec, unpartitioned_spec_}, unpartitioned_spec_); + source->metadata()->schemas.insert(source->metadata()->schemas.begin(), + historical_schema); + + auto table = MakePositionDeletesTable(source); + std::vector names; + for (const auto& field : table->schema()->fields()) { + names.emplace_back(field.name()); + } + EXPECT_THAT(names, + ElementsAre("file_path", "pos", "row", "spec_id", "delete_file_path")); +} + +TEST_F(PositionDeletesTableTest, ProjectsEmptyScan) { + auto table = MakePositionDeletesTable(MakeTable(3, unpartitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto projected, table->schema()->Select(std::vector{ + "pos", "delete_file_path"})); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan(*projected)); + auto batch = Import(std::move(array), *projected); + EXPECT_EQ(batch->num_rows(), 0); + ASSERT_EQ(batch->num_columns(), 2); + EXPECT_EQ(batch->schema()->field(0)->name(), "pos"); + EXPECT_EQ(batch->schema()->field(1)->name(), "delete_file_path"); +} + +TEST_F(PositionDeletesTableTest, RejectsNestedRowProjection) { + auto table = MakePositionDeletesTable(MakeTable(3, unpartitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto projected, + table->schema()->Select(std::vector{"row.id"})); + + auto result = table->Scan(*projected); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("only supports complete top-level fields")); +} + +TEST_F(PositionDeletesTableTest, ExpandsUpgradedMixedTable) { + auto parquet_file = WritePositionDeletes( + "old-position-deletes.parquet", {{"old-data.parquet", 3}, {"old-data.parquet", 9}}, + unpartitioned_spec_, PartitionValues{}); + auto dv_files = + WriteDeletionVectors("new-deletes.puffin", {{"new-data.parquet", {4, 12}}}, + unpartitioned_spec_, PartitionValues{}); + + constexpr int64_t kSnapshotId = 20; + auto old_manifest = WriteDeleteManifest( + 2, kSnapshotId, + {MakeEntry(ManifestStatus::kExisting, kSnapshotId, 1, parquet_file)}, + unpartitioned_spec_); + auto new_manifest = WriteDeleteManifest( + 3, kSnapshotId, + {MakeEntry(ManifestStatus::kAdded, kSnapshotId, 2, dv_files.front())}, + unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(3, unpartitioned_spec_, + MakeSnapshot(3, kSnapshotId, 2, {old_manifest, new_manifest}))); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 4); + + auto data_paths = std::static_pointer_cast<::arrow::StringArray>(batch->column(0)); + auto positions = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + auto rows = std::static_pointer_cast<::arrow::StructArray>(batch->column(2)); + auto delete_paths = std::static_pointer_cast<::arrow::StringArray>(batch->column(4)); + auto offsets = std::static_pointer_cast<::arrow::Int64Array>(batch->column(5)); + auto sizes = std::static_pointer_cast<::arrow::Int64Array>(batch->column(6)); + + EXPECT_EQ(rows->null_count(), 4); + std::vector actual_data_paths{ + data_paths->GetString(0), data_paths->GetString(1), data_paths->GetString(2), + data_paths->GetString(3)}; + std::vector actual_positions{positions->Value(0), positions->Value(1), + positions->Value(2), positions->Value(3)}; + std::vector actual_delete_paths{delete_paths->GetString(0), + delete_paths->GetString(2)}; + EXPECT_THAT(actual_data_paths, ElementsAre("old-data.parquet", "old-data.parquet", + "new-data.parquet", "new-data.parquet")); + EXPECT_THAT(actual_positions, ElementsAre(3, 9, 4, 12)); + EXPECT_THAT(actual_delete_paths, + ElementsAre("old-position-deletes.parquet", "new-deletes.puffin")); + EXPECT_TRUE(offsets->IsNull(0)); + EXPECT_TRUE(sizes->IsNull(0)); + EXPECT_FALSE(offsets->IsNull(2)); + EXPECT_FALSE(sizes->IsNull(2)); + EXPECT_EQ(offsets->Value(2), *dv_files.front()->content_offset); + EXPECT_EQ(sizes->Value(2), *dv_files.front()->content_size_in_bytes); +} + +TEST_F(PositionDeletesTableTest, ExpandsMultiplePuffinBlobs) { + auto dv_files = + WriteDeletionVectors("multi-deletes.puffin", + {{"data-a.parquet", {1, 5}}, {"data-b.parquet", {2, 8, 13}}}, + unpartitioned_spec_, PartitionValues{}); + ASSERT_EQ(dv_files.size(), 2); + + constexpr int64_t kSnapshotId = 30; + std::vector entries; + for (const auto& file : dv_files) { + entries.push_back(MakeEntry(ManifestStatus::kAdded, kSnapshotId, 1, file)); + } + auto manifest = + WriteDeleteManifest(3, kSnapshotId, std::move(entries), unpartitioned_spec_); + auto table = MakePositionDeletesTable( + MakeTable(3, unpartitioned_spec_, MakeSnapshot(3, kSnapshotId, 1, {manifest}))); + + ICEBERG_UNWRAP_OR_FAIL(auto array, table->Scan()); + auto batch = Import(std::move(array), *table->schema()); + ASSERT_EQ(batch->num_rows(), 5); + + auto data_paths = std::static_pointer_cast<::arrow::StringArray>(batch->column(0)); + auto positions = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + std::vector> rows; + for (int64_t i = 0; i < batch->num_rows(); ++i) { + rows.emplace_back(data_paths->GetString(i), positions->Value(i)); + } + EXPECT_THAT(rows, ElementsAre(std::tuple{"data-a.parquet", int64_t{1}}, + std::tuple{"data-a.parquet", int64_t{5}}, + std::tuple{"data-b.parquet", int64_t{2}}, + std::tuple{"data-b.parquet", int64_t{8}}, + std::tuple{"data-b.parquet", int64_t{13}})); + EXPECT_NE(dv_files[0]->content_offset, dv_files[1]->content_offset); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, PositionDeletesTableTest, + ::testing::Values(2, 3)); + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/test/puffin_dv_interop_test.cc b/src/iceberg/test/puffin_dv_interop_test.cc new file mode 100644 index 000000000..b322fe0ae --- /dev/null +++ b/src/iceberg/test/puffin_dv_interop_test.cc @@ -0,0 +1,247 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include + +#include "iceberg/deletes/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/puffin_reader.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/std_io.h" +#include "iceberg/test/test_resource.h" + +namespace iceberg::puffin { + +namespace { + +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +int64_t Position(int64_t bucket, int64_t container, int64_t value) { + return (bucket << 32) + (container << 16) + value; +} + +std::vector AllContainerPositions() { + std::vector positions = { + Position(0, 0, 5), + Position(0, 0, 7), + Position(1, 0, 10), + Position(1, 0, 20), + }; + positions.reserve(10004); + for (int64_t bucket = 0; bucket < 2; ++bucket) { + for (int64_t value = 0; value < 10000; value += 2) { + positions.push_back(Position(bucket, 2, value)); + } + } + return positions; +} + +struct PositionRange { + int64_t begin; + int64_t end; +}; + +struct ExpectedBlob { + std::string referenced_data_file; + std::vector input_fields; + int64_t offset; + int64_t length; + int64_t cardinality; + std::vector positions; + std::vector ranges; +}; + +std::shared_ptr MakeDeleteFile(const ExpectedBlob& expected, + const std::string& fixture_path, + int64_t file_size) { + return std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = fixture_path, + .file_format = FileFormatType::kPuffin, + .record_count = expected.cardinality, + .file_size_in_bytes = file_size, + .referenced_data_file = expected.referenced_data_file, + .content_offset = expected.offset, + .content_size_in_bytes = expected.length, + }); +} + +void AssertPositions(const ExpectedBlob& expected, + const std::shared_ptr& delete_file, + const std::shared_ptr& io) { + ASSERT_TRUE(delete_file->content_offset.has_value()); + ASSERT_TRUE(delete_file->content_size_in_bytes.has_value()); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); + EXPECT_EQ(positions.Cardinality(), expected.cardinality); + int64_t expected_cardinality = static_cast(expected.positions.size()); + for (const auto& range : expected.ranges) { + expected_cardinality += range.end - range.begin; + } + ASSERT_EQ(expected_cardinality, expected.cardinality); + for (int64_t position : expected.positions) { + EXPECT_TRUE(positions.IsDeleted(position)) << "Missing position " << position; + } + for (const auto& range : expected.ranges) { + for (int64_t position = range.begin; position < range.end; ++position) { + ASSERT_TRUE(positions.IsDeleted(position)) << "Missing position " << position; + } + } +} + +void AssertFixture(const std::string& resource_name, + const std::vector& expected_blobs) { + const std::string fixture_path = GetResourcePath(resource_name); + const auto file_size = static_cast(std::filesystem::file_size(fixture_path)); + auto io = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(fixture_path)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, PuffinReader::Make(std::move(input_file))); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, reader->ReadFileMetadata()); + + ASSERT_EQ(metadata.blobs.size(), expected_blobs.size()); + std::unordered_set seen; + for (const auto& blob : metadata.blobs) { + const auto& referenced_data_file = + blob.properties.at(std::string(kReferencedDataFileProperty)); + auto expected = std::ranges::find_if(expected_blobs, [&](const auto& candidate) { + return candidate.referenced_data_file == referenced_data_file; + }); + ASSERT_NE(expected, expected_blobs.end()) + << "Unexpected referenced data file " << referenced_data_file; + ASSERT_TRUE(seen.insert(referenced_data_file).second) + << "Duplicate referenced data file " << referenced_data_file; + + EXPECT_EQ(blob.type, StandardBlobTypes::kDeletionVectorV1); + EXPECT_EQ(blob.input_fields, expected->input_fields); + EXPECT_EQ(blob.snapshot_id, -1); + EXPECT_EQ(blob.sequence_number, -1); + EXPECT_EQ(blob.offset, expected->offset); + EXPECT_EQ(blob.length, expected->length); + EXPECT_TRUE(blob.compression_codec.empty()); + EXPECT_EQ(blob.properties.at(std::string(kCardinalityProperty)), + std::to_string(expected->cardinality)); + + AssertPositions(*expected, MakeDeleteFile(*expected, fixture_path, file_size), io); + } + EXPECT_EQ(seen.size(), expected_blobs.size()); +} + +} // namespace + +TEST(PuffinDVInteropTest, ReadsJavaSingleBlobFixture) { + AssertFixture( + "deletion_vectors/java/single-blob-dv.puffin", + {{ + .referenced_data_file = "s3://warehouse/db/table/data/00000-0-abc.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 4, + .length = 50, + .cardinality = 5, + .positions = {1, 3, 5, 7, 9}, + }}); +} + +TEST(PuffinDVInteropTest, ReadsJavaMultiBlobFixture) { + AssertFixture( + "deletion_vectors/java/multi-blob-dv.puffin", + { + { + .referenced_data_file = "s3://warehouse/db/table/data/file-001.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 4, + .length = 46, + .cardinality = 3, + .positions = {0, 100, 200}, + }, + { + .referenced_data_file = "s3://warehouse/db/table/data/file-002.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 50, + .length = 44, + .cardinality = 2, + .positions = {50, 150}, + }, + }); +} + +TEST(PuffinDVInteropTest, ReadsGoSingleBlobFixture) { + AssertFixture("deletion_vectors/go/single-blob-dv.puffin", + {{ + .referenced_data_file = "data/test.parquet", + .input_fields = {}, + .offset = 4, + .length = 50, + .cardinality = 5, + .positions = {1, 3, 5, 7, 9}, + }}); +} + +TEST(PuffinDVInteropTest, ReadsGoMultiBlobFixture) { + AssertFixture( + "deletion_vectors/go/multi-blob-dv.puffin", + { + { + .referenced_data_file = "s3://warehouse/db/table/data/go-file-001.parquet", + .input_fields = {}, + .offset = 4, + .length = 68, + .cardinality = 4, + .positions = {0, 100, 200, (int64_t{1} << 32) + 7}, + }, + { + .referenced_data_file = "s3://warehouse/db/table/data/go-file-002.parquet", + .input_fields = {}, + .offset = 72, + .length = 66, + .cardinality = 3, + .positions = {50, 150, (int64_t{2} << 32) + 9}, + }, + }); +} + +TEST(PuffinDVInteropTest, ReadsGoAllContainerTypesFixture) { + AssertFixture( + "deletion_vectors/go/all-container-types-dv.puffin", + {{ + .referenced_data_file = "s3://warehouse/db/table/data/all-containers.parquet", + .input_fields = {}, + .offset = 4, + .length = 16466, + .cardinality = 11493, + .positions = AllContainerPositions(), + .ranges = + { + {Position(0, 1, 1), Position(0, 1, 1000)}, + {Position(1, 1, 10), Position(1, 1, 500)}, + }, + }}); +} + +} // namespace iceberg::puffin diff --git a/src/iceberg/test/resources/deletion_vectors/README.md b/src/iceberg/test/resources/deletion_vectors/README.md new file mode 100644 index 000000000..983cbfe3e --- /dev/null +++ b/src/iceberg/test/resources/deletion_vectors/README.md @@ -0,0 +1,44 @@ +# Deletion vector interoperability fixtures + +These fixtures verify deletion-vector compatibility against bytes produced by +implementations other than Iceberg C++. + +## Java fixtures + +The files under `java/` were copied byte-for-byte from Apache Iceberg Go commit +`3020adbbc3faff047da6f483f739f1b5e1de611b`, where they are consumed by +`table/dv/dv_cross_client_test.go`. + +- `single-blob-dv.puffin` and `multi-blob-dv.puffin` were produced by Apache + Iceberg Java using + `iceberg-go/dev/dv-fixtures/GenerateDVFixtures.java`. + +The committed Puffin bytes were reproduced against Apache Iceberg Java commit +`76d35b1e40f77edcad19646bb6afdd9f05249964`. To regenerate them, follow +`iceberg-go/dev/dv-fixtures/README.md` at the Iceberg Go commit above using +that Java revision, then copy the generated files into `java/`. + +## Go fixtures + +The files under `go/` are produced with the Iceberg Go Puffin and deletion +vector writers. From an Iceberg Go checkout at commit +`3020adbbc3faff047da6f483f739f1b5e1de611b`, run: + +```bash +go run /path/to/iceberg-cpp/dev/dv-fixtures/generate_go_fixtures.go \ + /path/to/iceberg-cpp/src/iceberg/test/resources/deletion_vectors/go +``` + +The multi-blob fixture includes positions in three distinct high-32-bit +buckets. `all-container-types-dv.puffin` covers array, run, and bitmap +containers inside a complete Puffin file. + +## Checksums + +```text +ab8309671c0c5ef1956f4f1d7b907f4a69ffeb154ab77d5a4855d7dd4779108c java/single-blob-dv.puffin +fed7edbb5a343a6c6fc4706ff4c213ab3f0a50916baeb228619f1f2c956f3f27 java/multi-blob-dv.puffin +dd293e827439cd22053a9356ae6d56f0d5d69e648b1850d39663cdc5c7ec5a77 go/single-blob-dv.puffin +000a77697d5ddce01e0242786b91fc15482a8b8f4fa3c8e214a936d7c2fc978b go/multi-blob-dv.puffin +d9f374706891b4780f8a59f155a7cc4cae4161b6415578abb4790f6eed793a52 go/all-container-types-dv.puffin +``` diff --git a/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin new file mode 100644 index 000000000..5582a008e Binary files /dev/null and b/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin differ diff --git a/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin new file mode 100644 index 000000000..4404137af Binary files /dev/null and b/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin differ diff --git a/src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin new file mode 100644 index 000000000..84c6baf79 Binary files /dev/null and b/src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin differ diff --git a/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin new file mode 100644 index 000000000..b9037900c Binary files /dev/null and b/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin differ diff --git a/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin new file mode 100644 index 000000000..5c0007e21 Binary files /dev/null and b/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin differ diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 0b19adaf5..cef0de8ef 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -198,6 +198,7 @@ class ManifestListWriter; class ManifestReader; class ManifestWriter; class PartitionSummary; +class PositionDeletesTable; /// \brief File I/O. struct ReaderOptions;