diff --git a/.clang-tidy b/.clang-tidy index 5b13a0551..229bb0ec1 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -43,6 +43,6 @@ CheckOptions: - key: modernize-use-scoped-lock.WarnOnSingleLocks value: 'false' - key: misc-include-cleaner.IgnoreHeaders - value: 'arrow/.*;avro/.*;aws/.*;cpr/.*;gmock/.*;gtest/.*;nanoarrow/.*;nlohmann/.*;parquet/.*;roaring/.*;spdlog/.*;sqlpp23/.*;sqlite3\.h;thrift/.*;utf8proc\.h;zlib\.h' + value: 'arrow/.*;avro/.*;aws/.*;cpr/.*;gmock/.*;gtest/.*;nanoarrow/.*;nlohmann/.*;parquet/.*;roaring/.*;spdlog/.*;sqlpp23/.*;thrift/.*' HeaderFilterRegex: 'src/iceberg|example' diff --git a/CMakeLists.txt b/CMakeLists.txt index db54b24ed..7cbd5e55e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -28,10 +28,9 @@ project(Iceberg DESCRIPTION "Iceberg C++ Project" LANGUAGES CXX) -set(ICEBERG_VERSION_SUFFIX "-SNAPSHOT") -set(ICEBERG_VERSION_STRING "${PROJECT_VERSION}${ICEBERG_VERSION_SUFFIX}") -configure_file("${CMAKE_CURRENT_SOURCE_DIR}/src/iceberg/version.h.in" - "${CMAKE_BINARY_DIR}/src/iceberg/version.h") +include(IcebergBuildUtils) +iceberg_configure_version_header("${CMAKE_CURRENT_SOURCE_DIR}/src/iceberg/version.h.in" + "${CMAKE_BINARY_DIR}/src/iceberg/version.h") set(CMAKE_CXX_STANDARD 23) set(CMAKE_CXX_STANDARD_REQUIRED ON) @@ -79,7 +78,6 @@ if(ICEBERG_BUILD_REST_INTEGRATION_TESTS AND WIN32) endif() include(CMakeParseArguments) -include(IcebergBuildUtils) include(IcebergSanitizer) include(IcebergSccache) include(IcebergThirdpartyToolchain) diff --git a/cmake_modules/IcebergBuildUtils.cmake b/cmake_modules/IcebergBuildUtils.cmake index da23f37a2..689ec44b7 100644 --- a/cmake_modules/IcebergBuildUtils.cmake +++ b/cmake_modules/IcebergBuildUtils.cmake @@ -20,6 +20,28 @@ include(CMakePackageConfigHelpers) +function(iceberg_configure_version_header INPUT_FILE OUTPUT_FILE) + set(ICEBERG_VERSION_SUFFIX "-SNAPSHOT") + set(ICEBERG_VERSION_STRING "${PROJECT_VERSION}${ICEBERG_VERSION_SUFFIX}") + set(ICEBERG_GIT_COMMIT_ID "unknown") + + find_package(Git QUIET) + if(GIT_FOUND) + execute_process(COMMAND "${GIT_EXECUTABLE}" rev-parse HEAD + WORKING_DIRECTORY "${PROJECT_SOURCE_DIR}" + OUTPUT_VARIABLE ICEBERG_GIT_COMMIT_ID + RESULT_VARIABLE ICEBERG_GIT_RESULT + OUTPUT_STRIP_TRAILING_WHITESPACE ERROR_QUIET) + if(NOT ICEBERG_GIT_RESULT EQUAL 0 OR NOT ICEBERG_GIT_COMMIT_ID) + set(ICEBERG_GIT_COMMIT_ID "unknown") + endif() + endif() + + set(ICEBERG_FULL_VERSION_STRING + "Apache Iceberg-CPP ${ICEBERG_VERSION_STRING} (commit ${ICEBERG_GIT_COMMIT_ID})") + configure_file("${INPUT_FILE}" "${OUTPUT_FILE}" @ONLY) +endfunction() + function(iceberg_install_cmake_package PACKAGE_NAME EXPORT_NAME) set(CONFIG_CMAKE "${PACKAGE_NAME}-config.cmake") set(BUILT_CONFIG_CMAKE "${CMAKE_CURRENT_BINARY_DIR}/${CONFIG_CMAKE}") diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 049a61256..35b7b91b5 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -196,6 +196,8 @@ set(ICEBERG_DATA_SOURCES data/data_writer.cc data/delete_filter.cc data/delete_loader.cc + data/deletion_vector_writer.cc + data/dv_util.cc data/equality_delete_writer.cc data/file_scan_task_reader.cc data/position_delete_writer.cc diff --git a/src/iceberg/data/data_writer.cc b/src/iceberg/data/data_writer.cc index 420471280..512f77cc9 100644 --- a/src/iceberg/data/data_writer.cc +++ b/src/iceberg/data/data_writer.cc @@ -20,7 +20,9 @@ #include "iceberg/data/data_writer.h" #include +#include +#include "iceberg/data/writer.h" // IWYU pragma: keep #include "iceberg/file_writer.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/partition_spec.h" @@ -58,7 +60,7 @@ class DataWriter::Impl { return {}; } - Result Metadata() { + Result Metadata() { ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer"); ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics()); @@ -98,7 +100,7 @@ class DataWriter::Impl { options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt, }); - FileWriter::WriteResult result; + WriteResult result; result.data_files.push_back(std::move(data_file)); return result; } @@ -127,6 +129,6 @@ Result DataWriter::Length() const { return impl_->Length(); } Status DataWriter::Close() { return impl_->Close(); } -Result DataWriter::Metadata() { return impl_->Metadata(); } +Result DataWriter::Metadata() { return impl_->Metadata(); } } // namespace iceberg diff --git a/src/iceberg/data/delete_loader.cc b/src/iceberg/data/delete_loader.cc index 922173401..a95fab7e6 100644 --- a/src/iceberg/data/delete_loader.cc +++ b/src/iceberg/data/delete_loader.cc @@ -19,23 +19,29 @@ #include "iceberg/data/delete_loader.h" +#include #include +#include #include #include +#include #include #include #include "iceberg/arrow/nanoarrow_status_internal.h" #include "iceberg/arrow_c_data_guard_internal.h" +#include "iceberg/data/dv_util_internal.h" #include "iceberg/deletes/position_delete_index.h" #include "iceberg/deletes/position_delete_range_consumer.h" +#include "iceberg/file_io.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/metadata_columns.h" #include "iceberg/result.h" #include "iceberg/row/arrow_array_wrapper.h" #include "iceberg/schema.h" +#include "iceberg/util/content_file_util.h" #include "iceberg/util/macros.h" #include "iceberg/util/struct_like_set.h" @@ -83,6 +89,21 @@ bool StringEquals(const ArrowArrayView* view, int64_t row_idx, std::string_view return sv.data != nullptr && std::memcmp(sv.data, target.data(), target.size()) == 0; } +Status ValidateDV(const DataFile& dv, std::string_view data_file_path) { + ICEBERG_PRECHECK(dv.content_offset.has_value(), "Invalid DV, offset cannot be null: {}", + ContentFileUtil::DVDesc(dv)); + ICEBERG_PRECHECK(dv.content_size_in_bytes.has_value(), "Invalid DV, length is null: {}", + ContentFileUtil::DVDesc(dv)); + ICEBERG_PRECHECK( + dv.content_size_in_bytes.value() <= std::numeric_limits::max(), + "Can't read DV larger than 2GB: {}", dv.content_size_in_bytes.value()); + ICEBERG_PRECHECK(dv.referenced_data_file.has_value() && + dv.referenced_data_file.value() == data_file_path, + "DV is expected to reference {}, not {}", data_file_path, + dv.referenced_data_file.value_or("")); + return {}; +} + } // namespace DeleteLoader::DeleteLoader(std::shared_ptr io) : io_(std::move(io)) {} @@ -91,7 +112,7 @@ DeleteLoader::~DeleteLoader() = default; Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteIndex& index, std::string_view data_file_path) const { - // TODO(gangwu): push down path filter to open the file. + // TODO(gangwu): Add cache hooks, worker pool, and filter pushdown. ICEBERG_ASSIGN_OR_RAISE(auto reader, OpenDeleteFile(file, PosDeleteSchema(), io_)); ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader->Schema()); @@ -170,30 +191,40 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde return reader->Close(); } -Status DeleteLoader::LoadDV(const DataFile& file, PositionDeleteIndex& index) const { - return NotSupported("Loading deletion vectors is not yet supported"); -} - Result DeleteLoader::LoadPositionDeletes( std::span> delete_files, std::string_view data_file_path) const { - PositionDeleteIndex index; + for (const auto& file : delete_files) { + ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null"); + } + + if (ContentFileUtil::ContainsSingleDV(delete_files)) { + const auto& dv = delete_files.front(); + ICEBERG_RETURN_UNEXPECTED(ValidateDV(*dv, data_file_path)); + return DVUtil::ReadDV(dv, io_); + } + + std::vector> position_delete_files; for (const auto& file : delete_files) { + ICEBERG_PRECHECK(!file->IsDeletionVector(), + "Deletion vector cannot be mixed with position delete files: {}", + file->file_path); + ICEBERG_PRECHECK(file->content == DataFile::Content::kPositionDeletes, + "Expected position delete file but got content type {}", + ToString(file->content)); + if (file->referenced_data_file.has_value() && file->referenced_data_file.value() != data_file_path) { continue; } - if (file->IsDeletionVector()) { - ICEBERG_RETURN_UNEXPECTED(LoadDV(*file, index)); - continue; - } + position_delete_files.push_back(file); + } - ICEBERG_PRECHECK(file->content == DataFile::Content::kPositionDeletes, - "Expected position delete file but got content type {}", - ToString(file->content)); + PositionDeleteIndex index(position_delete_files); + for (const auto& file : position_delete_files) { ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(*file, index, data_file_path)); } diff --git a/src/iceberg/data/delete_loader.h b/src/iceberg/data/delete_loader.h index a422f0683..360914e85 100644 --- a/src/iceberg/data/delete_loader.h +++ b/src/iceberg/data/delete_loader.h @@ -70,9 +70,6 @@ class ICEBERG_DATA_EXPORT DeleteLoader { Status LoadPositionDelete(const DataFile& file, PositionDeleteIndex& index, std::string_view data_file_path) const; - /// \brief Load a single deletion vector file into the index. - Status LoadDV(const DataFile& file, PositionDeleteIndex& index) const; - std::shared_ptr io_; }; diff --git a/src/iceberg/data/deletion_vector_writer.cc b/src/iceberg/data/deletion_vector_writer.cc new file mode 100644 index 000000000..8ff890627 --- /dev/null +++ b/src/iceberg/data/deletion_vector_writer.cc @@ -0,0 +1,256 @@ +/* + * 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/data/deletion_vector_writer.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/data/writer.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/deletes/roaring_position_bitmap.h" +#include "iceberg/file_format.h" +#include "iceberg/file_io.h" // IWYU pragma: keep +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/partition_spec.h" +#include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/puffin_writer.h" +#include "iceberg/result.h" +#include "iceberg/row/partition_values.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 kReferencedDataFile = "referenced-data-file"; +constexpr std::string_view kCardinality = "cardinality"; +} // namespace + +class DeletionVectorWriter::Impl { + public: + explicit Impl(DeletionVectorWriterOptions options) : options_(std::move(options)) {} + + // Accumulated positions and metadata for a single referenced data file. + struct Deletes { + PositionDeleteIndex positions; + std::shared_ptr spec; + PartitionValues partition; + }; + + Deletes& DeletesFor(std::string_view referenced_data_file, + const std::shared_ptr& spec, + const PartitionValues& partition) { + auto it = deletes_by_path_.lower_bound(referenced_data_file); + if (it == deletes_by_path_.end() || + deletes_by_path_.key_comp()(referenced_data_file, it->first)) { + it = + deletes_by_path_.emplace_hint(it, std::string(referenced_data_file), Deletes{}); + it->second.spec = spec; + it->second.partition = partition; + } + return it->second; + } + + Status Delete(std::string_view referenced_data_file, int64_t pos, + const std::shared_ptr& spec, + const PartitionValues& partition) { + ICEBERG_CHECK(!closed_, "Cannot delete after writer is closed"); + ICEBERG_PRECHECK(!referenced_data_file.empty(), + "Deletion vector requires a non-empty referenced data file"); + ICEBERG_PRECHECK(spec != nullptr, "Deletion vector requires a partition spec"); + ICEBERG_PRECHECK(pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition, + "Deletion vector position out of range [0, {}]: {}", + RoaringPositionBitmap::kMaxPosition, pos); + DeletesFor(referenced_data_file, spec, partition).positions.Delete(pos); + return {}; + } + + Status Delete(std::string_view referenced_data_file, + const PositionDeleteIndex& positions, + const std::shared_ptr& spec, + const PartitionValues& partition) { + ICEBERG_CHECK(!closed_, "Cannot delete after writer is closed"); + ICEBERG_PRECHECK(!referenced_data_file.empty(), + "Deletion vector requires a non-empty referenced data file"); + ICEBERG_PRECHECK(spec != nullptr, "Deletion vector requires a partition spec"); + DeletesFor(referenced_data_file, spec, partition).positions.Merge(positions); + return {}; + } + + Status Close() { + if (closed_) { + return {}; + } + + if (deletes_by_path_.empty()) { + closed_ = true; + return {}; + } + + for (auto& [path, deletes] : deletes_by_path_) { + ICEBERG_RETURN_UNEXPECTED(LoadPreviousDeletes(path, deletes)); + } + + ICEBERG_ASSIGN_OR_RAISE(auto output_file, options_.io->NewOutputFile(options_.path)); + const std::string output_path(options_.path); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, puffin::PuffinWriter::Make( + std::move(output_file), + {{std::string(puffin::StandardPuffinProperties::kCreatedBy), + ICEBERG_FULL_VERSION_STRING}})); + + for (auto& [path, deletes] : deletes_by_path_) { + ICEBERG_RETURN_UNEXPECTED(Write(*writer, path, deletes)); + } + + ICEBERG_RETURN_UNEXPECTED(writer->Finish()); + ICEBERG_ASSIGN_OR_RAISE(const int64_t file_size, writer->FileSize()); + + for (const auto& [path, _] : deletes_by_path_) { + result_.referenced_data_files.push_back(path); + ICEBERG_ASSIGN_OR_RAISE(auto data_file, CreateDV(output_path, file_size, path)); + result_.data_files.push_back(std::move(data_file)); + } + + closed_ = true; + return {}; + } + + Result Metadata() { + ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer"); + return result_; + } + + private: + Status LoadPreviousDeletes(std::string_view path, Deletes& deletes) { + ICEBERG_ASSIGN_OR_RAISE(auto previous, options_.load_previous_deletes(path)); + if (!previous.has_value()) { + return {}; + } + + deletes.positions.Merge(*previous); + for (const auto& delete_file : previous->delete_files()) { + ICEBERG_ASSIGN_OR_RAISE(bool file_scoped, + ContentFileUtil::IsFileScoped(*delete_file)); + if (file_scoped) { + result_.rewritten_delete_files.push_back(delete_file); + } + } + return {}; + } + + Status Write(puffin::PuffinWriter& writer, std::string_view path, Deletes& deletes) { + const int64_t cardinality = deletes.positions.Cardinality(); + ICEBERG_ASSIGN_OR_RAISE(auto data, deletes.positions.Serialize()); + + puffin::Blob blob{ + .type = std::string(puffin::StandardBlobTypes::kDeletionVectorV1), + .input_fields = {MetadataColumns::kFilePositionColumnId}, + // Snapshot ID and sequence number are inherited; the spec requires -1. + .snapshot_id = -1, + .sequence_number = -1, + .data = std::move(data), + .requested_compression = puffin::PuffinCompressionCodec::kNone, + }; + blob.properties.emplace(std::string(kReferencedDataFile), path); + blob.properties.emplace(std::string(kCardinality), std::format("{}", cardinality)); + + ICEBERG_ASSIGN_OR_RAISE(auto blob_metadata, writer.Write(blob)); + blobs_by_path_.insert_or_assign(std::string(path), std::move(blob_metadata)); + return {}; + } + + Result> CreateDV(const std::string& path, int64_t size, + std::string_view referenced_data_file) { + auto deletes = deletes_by_path_.find(referenced_data_file); + ICEBERG_CHECK(deletes != deletes_by_path_.end(), "Missing deletion vector for {}", + referenced_data_file); + auto blob_metadata = blobs_by_path_.find(referenced_data_file); + ICEBERG_CHECK(blob_metadata != blobs_by_path_.end(), + "Missing deletion vector blob for {}", referenced_data_file); + + return std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = path, + .file_format = FileFormatType::kPuffin, + .partition = deletes->second.partition, + .record_count = deletes->second.positions.Cardinality(), + .file_size_in_bytes = size, + // TODO(gangwu): support encryption key metadata + .referenced_data_file = std::string(referenced_data_file), + .content_offset = blob_metadata->second.offset, + .content_size_in_bytes = blob_metadata->second.length, + .partition_spec_id = deletes->second.spec + ? std::make_optional(deletes->second.spec->spec_id()) + : std::nullopt, + }); + } + + DeletionVectorWriterOptions options_; + std::map deletes_by_path_; + std::map blobs_by_path_; + WriteResult result_; + bool closed_ = false; +}; + +DeletionVectorWriter::DeletionVectorWriter(std::unique_ptr impl) + : impl_(std::move(impl)) {} + +DeletionVectorWriter::~DeletionVectorWriter() = default; + +Result> DeletionVectorWriter::Make( + DeletionVectorWriterOptions options) { + ICEBERG_PRECHECK(!options.path.empty(), "DeletionVectorWriter requires an output path"); + ICEBERG_PRECHECK(options.io != nullptr, "DeletionVectorWriter requires a FileIO"); + ICEBERG_PRECHECK(options.load_previous_deletes != nullptr, + "DeletionVectorWriter requires a load_previous_deletes callback"); + return std::unique_ptr( + new DeletionVectorWriter(std::make_unique(std::move(options)))); +} + +Status DeletionVectorWriter::Delete(std::string_view referenced_data_file, int64_t pos, + const std::shared_ptr& spec, + const PartitionValues& partition) { + return impl_->Delete(referenced_data_file, pos, spec, partition); +} + +Status DeletionVectorWriter::Delete(std::string_view referenced_data_file, + const PositionDeleteIndex& positions, + const std::shared_ptr& spec, + const PartitionValues& partition) { + return impl_->Delete(referenced_data_file, positions, spec, partition); +} + +Status DeletionVectorWriter::Close() { return impl_->Close(); } + +Result DeletionVectorWriter::Metadata() { return impl_->Metadata(); } + +} // namespace iceberg diff --git a/src/iceberg/data/deletion_vector_writer.h b/src/iceberg/data/deletion_vector_writer.h new file mode 100644 index 000000000..bd5bb66a5 --- /dev/null +++ b/src/iceberg/data/deletion_vector_writer.h @@ -0,0 +1,83 @@ +/* + * 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/data/deletion_vector_writer.h +/// Writer that emits deletion vectors as `deletion-vector-v1` blobs in a Puffin file. + +#include +#include +#include +#include +#include +#include + +#include "iceberg/data/writer.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/iceberg_data_export.h" +#include "iceberg/result.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Options for creating a DeletionVectorWriter. +struct ICEBERG_DATA_EXPORT DeletionVectorWriterOptions { + std::string path; + std::shared_ptr io; + /// Loads existing deletes for a data file to merge, if any. + std::function>(std::string_view)> + load_previous_deletes; +}; + +/// \brief A deletion vector file writer. +class ICEBERG_DATA_EXPORT DeletionVectorWriter { + public: + ~DeletionVectorWriter(); + + /// \brief Create a new DeletionVectorWriter. + static Result> Make( + DeletionVectorWriterOptions options); + + /// \brief Mark a row position as deleted for the given data file. + Status Delete(std::string_view referenced_data_file, int64_t pos, + const std::shared_ptr& spec, + const PartitionValues& partition); + + /// \brief Mark all positions in the given index as deleted for a data file. + Status Delete(std::string_view referenced_data_file, + const PositionDeleteIndex& positions, + const std::shared_ptr& spec, + const PartitionValues& partition); + + /// \brief Write all accumulated deletion vectors to the Puffin file and close. + Status Close(); + + /// \brief The result of writing; valid only after Close(). + Result Metadata(); + + private: + class Impl; + std::unique_ptr impl_; + + explicit DeletionVectorWriter(std::unique_ptr impl); +}; + +} // namespace iceberg diff --git a/src/iceberg/data/dv_util.cc b/src/iceberg/data/dv_util.cc new file mode 100644 index 000000000..175644841 --- /dev/null +++ b/src/iceberg/data/dv_util.cc @@ -0,0 +1,69 @@ +/* + * 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 "iceberg/data/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/file_io.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/result.h" +#include "iceberg/util/content_file_util.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result DVUtil::ReadDV(const std::shared_ptr& delete_file, + const std::shared_ptr& io) { + ICEBERG_PRECHECK(delete_file != nullptr, "Delete file must not be null"); + ICEBERG_PRECHECK(io != nullptr, "DV read requires a FileIO"); + ICEBERG_PRECHECK(ContentFileUtil::IsDV(*delete_file), + "Cannot read, not a deletion vector: {}", delete_file->file_path); + ICEBERG_PRECHECK( + delete_file->content_offset.has_value() && + delete_file->content_size_in_bytes.has_value(), + "Deletion vector requires content_offset and content_size_in_bytes: {}", + delete_file->file_path); + + const int64_t offset = delete_file->content_offset.value(); + const int64_t length = delete_file->content_size_in_bytes.value(); + ICEBERG_PRECHECK(offset >= 0 && length >= 0, + "Invalid deletion vector offset/length: offset={}, length={}", offset, + length); + ICEBERG_PRECHECK(length <= std::numeric_limits::max(), + "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()); + return PositionDeleteIndex::Deserialize(blob, delete_file); +} + +} // namespace iceberg diff --git a/src/iceberg/data/dv_util_internal.h b/src/iceberg/data/dv_util_internal.h new file mode 100644 index 000000000..4738b55bb --- /dev/null +++ b/src/iceberg/data/dv_util_internal.h @@ -0,0 +1,40 @@ +/* + * 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/data/dv_util_internal.h +/// Internal deletion vector helpers. + +#include + +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/iceberg_data_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +class ICEBERG_DATA_EXPORT DVUtil { + public: + static Result ReadDV(const std::shared_ptr& delete_file, + const std::shared_ptr& io); +}; + +} // namespace iceberg diff --git a/src/iceberg/data/equality_delete_writer.cc b/src/iceberg/data/equality_delete_writer.cc index 6604c03e4..86293addb 100644 --- a/src/iceberg/data/equality_delete_writer.cc +++ b/src/iceberg/data/equality_delete_writer.cc @@ -20,7 +20,9 @@ #include "iceberg/data/equality_delete_writer.h" #include +#include +#include "iceberg/data/writer.h" // IWYU pragma: keep #include "iceberg/file_writer.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/partition_spec.h" @@ -57,7 +59,7 @@ class EqualityDeleteWriter::Impl { return {}; } - Result Metadata() { + Result Metadata() { ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer"); ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics()); @@ -99,7 +101,7 @@ class EqualityDeleteWriter::Impl { options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt, }); - FileWriter::WriteResult result; + WriteResult result; result.data_files.push_back(std::move(data_file)); return result; } @@ -134,9 +136,7 @@ Result EqualityDeleteWriter::Length() const { return impl_->Length(); } Status EqualityDeleteWriter::Close() { return impl_->Close(); } -Result EqualityDeleteWriter::Metadata() { - return impl_->Metadata(); -} +Result EqualityDeleteWriter::Metadata() { return impl_->Metadata(); } std::span EqualityDeleteWriter::equality_field_ids() const { return impl_->equality_field_ids(); diff --git a/src/iceberg/data/meson.build b/src/iceberg/data/meson.build index bbb26db27..eaae8a2b4 100644 --- a/src/iceberg/data/meson.build +++ b/src/iceberg/data/meson.build @@ -20,6 +20,7 @@ install_headers( 'data_writer.h', 'delete_filter.h', 'delete_loader.h', + 'deletion_vector_writer.h', 'equality_delete_writer.h', 'file_scan_task_reader.h', 'position_delete_writer.h', diff --git a/src/iceberg/data/position_delete_writer.cc b/src/iceberg/data/position_delete_writer.cc index 3238dc506..e487e0e97 100644 --- a/src/iceberg/data/position_delete_writer.cc +++ b/src/iceberg/data/position_delete_writer.cc @@ -21,12 +21,14 @@ #include #include +#include #include #include #include "iceberg/arrow/nanoarrow_status_internal.h" #include "iceberg/arrow_c_data_guard_internal.h" +#include "iceberg/data/writer.h" // IWYU pragma: keep #include "iceberg/file_writer.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/metadata_columns.h" @@ -92,7 +94,7 @@ class PositionDeleteWriter::Impl { return {}; } - Result Metadata() { + Result Metadata() { ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer"); ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics()); @@ -159,7 +161,7 @@ class PositionDeleteWriter::Impl { options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt, }); - FileWriter::WriteResult result; + WriteResult result; result.data_files.push_back(std::move(data_file)); return result; } @@ -233,8 +235,6 @@ Result PositionDeleteWriter::Length() const { return impl_->Length(); } Status PositionDeleteWriter::Close() { return impl_->Close(); } -Result PositionDeleteWriter::Metadata() { - return impl_->Metadata(); -} +Result PositionDeleteWriter::Metadata() { return impl_->Metadata(); } } // namespace iceberg diff --git a/src/iceberg/data/writer.h b/src/iceberg/data/writer.h index a6f5f063d..0efcdc274 100644 --- a/src/iceberg/data/writer.h +++ b/src/iceberg/data/writer.h @@ -24,6 +24,7 @@ #include #include +#include #include #include "iceberg/arrow_c_data.h" @@ -33,6 +34,18 @@ namespace iceberg { +/// \brief File metadata for files produced by a writer. +struct ICEBERG_DATA_EXPORT WriteResult { + /// Usually a writer produces a single data or delete file. + /// Position delete writer may produce multiple file-scoped delete files. + /// In the future, multiple files can be produced if file rolling is supported. + std::vector> data_files; + /// Data files referenced by the produced delete files. + std::vector referenced_data_files; + /// Previously written delete files that were merged and should be removed. + std::vector> rewritten_delete_files; +}; + /// \brief Base interface for data file writers. class ICEBERG_DATA_EXPORT FileWriter { public: @@ -48,14 +61,6 @@ class ICEBERG_DATA_EXPORT FileWriter { /// \brief Close the writer and finalize the file. virtual Status Close() = 0; - /// \brief File metadata for all files produced by this writer. - struct ICEBERG_DATA_EXPORT WriteResult { - /// Usually a writer produces a single data or delete file. - /// Position delete writer may produce multiple file-scoped delete files. - /// In the future, multiple files can be produced if file rolling is supported. - std::vector> data_files; - }; - /// \brief Get file metadata for all files produced by this writer. /// \note This method should be called after Close(). virtual Result Metadata() = 0; diff --git a/src/iceberg/deletes/position_delete_index.cc b/src/iceberg/deletes/position_delete_index.cc index f09d0b480..53e33e635 100644 --- a/src/iceberg/deletes/position_delete_index.cc +++ b/src/iceberg/deletes/position_delete_index.cc @@ -19,8 +19,109 @@ #include "iceberg/deletes/position_delete_index.h" +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/deletes/roaring_position_bitmap.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/result.h" +#include "iceberg/util/endian.h" +#include "iceberg/util/macros.h" + namespace iceberg { +namespace { + +// `deletion-vector-v1` blob framing. See: +// https://iceberg.apache.org/puffin-spec/#deletion-vector-v1-blob-type +constexpr std::array kMagic = {0xD1, 0xD3, 0x39, 0x64}; +constexpr int32_t kLengthPrefixBytes = 4; +constexpr int32_t kMagicBytes = 4; +constexpr int32_t kCrcBytes = 4; + +uint32_t ComputeCrc32(std::span bytes) { + uLong crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, reinterpret_cast(bytes.data()), + static_cast(bytes.size())); + return static_cast(crc); +} + +Result ReadBitmapDataLength(std::span blob, + const DataFile& delete_file) { + ICEBERG_PRECHECK(delete_file.content_size_in_bytes.has_value(), + "Deletion vector requires content_size_in_bytes: {}", + delete_file.file_path); + ICEBERG_PRECHECK( + std::cmp_equal(blob.size(), *delete_file.content_size_in_bytes), + "Deletion vector blob size {} does not match content_size_in_bytes {}: {}", + blob.size(), *delete_file.content_size_in_bytes, delete_file.file_path); + ICEBERG_PRECHECK(blob.size() >= kLengthPrefixBytes + kMagicBytes + kCrcBytes, + "Deletion vector blob too small: {} bytes", blob.size()); + + const auto length = ReadBigEndian(blob.data()); + const size_t expected_size = + kLengthPrefixBytes + static_cast(length) + kCrcBytes; + ICEBERG_PRECHECK(length >= kMagicBytes, "Invalid deletion vector length prefix: {}", + length); + ICEBERG_PRECHECK(blob.size() == expected_size, + "Deletion vector blob size mismatch: {} bytes, expected {}", + blob.size(), expected_size); + return length; +} + +Result DeserializeBitmap(std::span blob, + int32_t length, + const DataFile& delete_file) { + const uint8_t* bitmap_data = blob.data() + kLengthPrefixBytes; + ICEBERG_PRECHECK(std::memcmp(bitmap_data, kMagic.data(), kMagic.size()) == 0, + "Invalid deletion vector magic"); + + std::string_view vector_bytes(reinterpret_cast(bitmap_data + kMagicBytes), + static_cast(length) - kMagicBytes); + ICEBERG_ASSIGN_OR_RAISE(auto bitmap, RoaringPositionBitmap::Deserialize(vector_bytes)); + ICEBERG_PRECHECK(std::cmp_equal(bitmap.Cardinality(), delete_file.record_count), + "Deletion vector cardinality {} does not match record count {}: {}", + bitmap.Cardinality(), delete_file.record_count, delete_file.file_path); + return bitmap; +} + +Status ValidateChecksum(std::span blob, int32_t length) { + const uint8_t* bitmap_data = blob.data() + kLengthPrefixBytes; + const auto stored_crc = ReadBigEndian(bitmap_data + length); + const auto actual_crc = + ComputeCrc32(std::span(bitmap_data, static_cast(length))); + ICEBERG_PRECHECK(stored_crc == actual_crc, + "Deletion vector CRC mismatch: stored {:#010x}, computed {:#010x}", + stored_crc, actual_crc); + return {}; +} + +} // namespace + +PositionDeleteIndex::PositionDeleteIndex(RoaringPositionBitmap bitmap) + : bitmap_(std::move(bitmap)) {} + +PositionDeleteIndex::PositionDeleteIndex(std::shared_ptr delete_file) { + if (delete_file != nullptr) { + delete_files_.push_back(std::move(delete_file)); + } +} + +PositionDeleteIndex::PositionDeleteIndex( + std::vector> delete_files) + : delete_files_(std::move(delete_files)) {} + void PositionDeleteIndex::Delete(int64_t pos) { bitmap_.Add(pos); } void PositionDeleteIndex::Delete(int64_t pos_start, int64_t pos_end) { @@ -37,6 +138,40 @@ int64_t PositionDeleteIndex::Cardinality() const { void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) { bitmap_.Or(other.bitmap_); + delete_files_.insert(delete_files_.end(), other.delete_files_.begin(), + other.delete_files_.end()); +} + +Result> PositionDeleteIndex::Serialize() { + bitmap_.Optimize(); // run-length encode before serializing + std::vector blob(kLengthPrefixBytes); + blob.insert(blob.end(), kMagic.begin(), kMagic.end()); + ICEBERG_ASSIGN_OR_RAISE(const auto vector_size, bitmap_.SerializeTo(blob)); + + const size_t magic_and_vector_size = kMagicBytes + vector_size; + ICEBERG_PRECHECK(magic_and_vector_size <= std::numeric_limits::max(), + "Deletion vector is too large to serialize: {} bytes", + magic_and_vector_size); + + WriteBigEndian(static_cast(magic_and_vector_size), blob.data()); + const auto crc_offset = blob.size(); + blob.resize(crc_offset + kCrcBytes); + WriteBigEndian(ComputeCrc32(std::span(blob.data() + kLengthPrefixBytes, + magic_and_vector_size)), + blob.data() + crc_offset); + return blob; +} + +Result PositionDeleteIndex::Deserialize( + std::span blob, std::shared_ptr delete_file) { + ICEBERG_PRECHECK(delete_file != nullptr, + "Deletion vector requires a source delete file"); + ICEBERG_ASSIGN_OR_RAISE(const auto length, ReadBitmapDataLength(blob, *delete_file)); + ICEBERG_ASSIGN_OR_RAISE(auto bitmap, DeserializeBitmap(blob, length, *delete_file)); + ICEBERG_RETURN_UNEXPECTED(ValidateChecksum(blob, length)); + PositionDeleteIndex index(std::move(bitmap)); + index.delete_files_.push_back(std::move(delete_file)); + return index; } void PositionDeleteIndex::BulkAddForKey(int32_t key, diff --git a/src/iceberg/deletes/position_delete_index.h b/src/iceberg/deletes/position_delete_index.h index 592216ed6..cbf2db478 100644 --- a/src/iceberg/deletes/position_delete_index.h +++ b/src/iceberg/deletes/position_delete_index.h @@ -29,6 +29,8 @@ #include "iceberg/deletes/roaring_position_bitmap.h" #include "iceberg/iceberg_data_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" namespace iceberg { @@ -40,6 +42,8 @@ namespace iceberg { class ICEBERG_DATA_EXPORT PositionDeleteIndex { public: PositionDeleteIndex() = default; + explicit PositionDeleteIndex(std::shared_ptr delete_file); + explicit PositionDeleteIndex(std::vector> delete_files); ~PositionDeleteIndex() = default; /// \brief Mark a position as deleted. @@ -66,7 +70,33 @@ class ICEBERG_DATA_EXPORT PositionDeleteIndex { /// \param other The index to merge (union operation) void Merge(const PositionDeleteIndex& other); + /// \brief The delete files whose positions were merged into this index. + /// + /// Populated by constructors and Deserialize, and preserved across Merge. + /// Callers use these to report the delete files that were rewritten when + /// replacing them with a new deletion vector. + const std::vector>& delete_files() const { + return delete_files_; + } + + /// \brief Serialize the index into a `deletion-vector-v1` blob. + /// + /// The positions are run-length encoded, then framed per the Puffin spec: + /// https://iceberg.apache.org/puffin-spec/#deletion-vector-v1-blob-type + Result> Serialize(); + + /// \brief Deserialize a `deletion-vector-v1` blob into an index. + /// + /// Validates the blob framing (length prefix, magic sequence, CRC-32) and, + /// against the source delete file, that the blob length matches + /// `content_size_in_bytes` and the bitmap cardinality matches `record_count`. + /// The source delete file is retained and exposed via delete_files(). + static Result Deserialize(std::span blob, + std::shared_ptr delete_file); + private: + explicit PositionDeleteIndex(RoaringPositionBitmap bitmap); + // Bulk-add positions sharing high-32-bit `key`. Private hook for // `ForEachPositionDelete`'s bulk path; keeps `Delete` the sole public // mutation surface. @@ -77,6 +107,7 @@ class ICEBERG_DATA_EXPORT PositionDeleteIndex { std::vector& scratch); RoaringPositionBitmap bitmap_; + std::vector> delete_files_; }; } // namespace iceberg diff --git a/src/iceberg/deletes/roaring_position_bitmap.cc b/src/iceberg/deletes/roaring_position_bitmap.cc index e31efeec1..a2827d4bb 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.cc +++ b/src/iceberg/deletes/roaring_position_bitmap.cc @@ -19,9 +19,11 @@ #include "iceberg/deletes/roaring_position_bitmap.h" +#include #include #include #include +#include #include #include @@ -57,6 +59,16 @@ Status ValidatePosition(int64_t pos) { return {}; } +void WriteBitmaps(const std::vector& bitmaps, uint8_t* buf) { + WriteLittleEndian(static_cast(bitmaps.size()), buf); + buf += kBitmapCountSizeBytes; + for (int32_t key = 0; std::cmp_less(key, bitmaps.size()); ++key) { + WriteLittleEndian(key, buf); + buf += kBitmapKeySizeBytes; + buf += bitmaps[key].write(reinterpret_cast(buf), /*portable=*/true); + } +} + } // namespace struct RoaringPositionBitmap::Impl { @@ -186,23 +198,18 @@ size_t RoaringPositionBitmap::SerializedSizeInBytes() const { Result RoaringPositionBitmap::Serialize() const { size_t size = SerializedSizeInBytes(); std::string result(size, '\0'); - char* buf = result.data(); - - // Write bitmap count (array length including empties) - WriteLittleEndian(static_cast(impl_->bitmaps.size()), buf); - buf += kBitmapCountSizeBytes; - - // Write each bitmap with its key - for (int32_t key = 0; std::cmp_less(key, impl_->bitmaps.size()); ++key) { - WriteLittleEndian(key, buf); - buf += kBitmapKeySizeBytes; - size_t written = impl_->bitmaps[key].write(buf, /*portable=*/true); - buf += written; - } - + WriteBitmaps(impl_->bitmaps, reinterpret_cast(result.data())); return result; } +Result RoaringPositionBitmap::SerializeTo(std::vector& output) const { + const size_t offset = output.size(); + const size_t size = SerializedSizeInBytes(); + output.resize(offset + size); + WriteBitmaps(impl_->bitmaps, output.data() + offset); + return size; +} + Result RoaringPositionBitmap::Deserialize(std::string_view bytes) { const char* buf = bytes.data(); size_t remaining = bytes.size(); diff --git a/src/iceberg/deletes/roaring_position_bitmap.h b/src/iceberg/deletes/roaring_position_bitmap.h index bfb7d7c9e..50807dc6a 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.h +++ b/src/iceberg/deletes/roaring_position_bitmap.h @@ -28,6 +28,7 @@ #include #include #include +#include #include "iceberg/iceberg_data_export.h" #include "iceberg/result.h" @@ -118,6 +119,8 @@ class ICEBERG_DATA_EXPORT RoaringPositionBitmap { // `PositionDeleteIndex::BulkAddForKey`; per-key grouping is the caller's // job, keeping this a thin wrapper around CRoaring's `addMany`. void AddManyForKey(int32_t key, std::span positions); + // Appends the portable serialized bitmap and returns bytes appended. + Result SerializeTo(std::vector& output) const; friend class PositionDeleteIndex; }; diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 7340bfa28..e3bef01af 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -18,6 +18,7 @@ conf_data = configuration_data() version = meson.project_version() version_suffix = '-SNAPSHOT' +version_string = version + version_suffix components = version.split('.') assert( components.length() >= 3, @@ -29,7 +30,30 @@ ver_patch = components[2] conf_data.set('PROJECT_VERSION_MAJOR', ver_major) conf_data.set('PROJECT_VERSION_MINOR', ver_minor) conf_data.set('PROJECT_VERSION_PATCH', ver_patch) -conf_data.set('ICEBERG_VERSION_STRING', version + version_suffix) +conf_data.set('ICEBERG_VERSION_STRING', version_string) +git = find_program('git', required: false) +git_commit_id = 'unknown' +if git.found() + git_result = run_command( + git, + '-C', + meson.project_source_root(), + 'rev-parse', + 'HEAD', + check: false, + ) + if git_result.returncode() == 0 + git_commit_id_candidate = git_result.stdout().strip() + if git_commit_id_candidate != '' + git_commit_id = git_commit_id_candidate + endif + endif +endif +conf_data.set('ICEBERG_GIT_COMMIT_ID', git_commit_id) +conf_data.set( + 'ICEBERG_FULL_VERSION_STRING', + 'Apache Iceberg-CPP ' + version_string + ' (commit ' + git_commit_id + ')', +) conf_data.set('PROJECT_NAME', meson.project_name()) configure_file( input: 'version.h.in', @@ -174,6 +198,8 @@ iceberg_data_sources = files( 'data/data_writer.cc', 'data/delete_filter.cc', 'data/delete_loader.cc', + 'data/deletion_vector_writer.cc', + 'data/dv_util.cc', 'data/equality_delete_writer.cc', 'data/file_scan_task_reader.cc', 'data/position_delete_writer.cc', diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 7a1b45b70..98129a6d2 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -254,6 +254,7 @@ if(ICEBERG_BUILD_BUNDLE) data_writer_test.cc delete_filter_test.cc delete_loader_test.cc + deletion_vector_writer_test.cc file_scan_task_reader_test.cc literal_util_test.cc) diff --git a/src/iceberg/test/delete_filter_test.cc b/src/iceberg/test/delete_filter_test.cc index 89d1b6b85..4b5e888e5 100644 --- a/src/iceberg/test/delete_filter_test.cc +++ b/src/iceberg/test/delete_filter_test.cc @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -34,9 +35,12 @@ #include #include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/data/deletion_vector_writer.h" #include "iceberg/data/equality_delete_writer.h" #include "iceberg/data/position_delete_writer.h" +#include "iceberg/deletes/position_delete_index.h" #include "iceberg/file_format.h" +#include "iceberg/file_io.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/metadata_columns.h" @@ -249,6 +253,25 @@ class DeleteFilterTest : public ::testing::Test { static constexpr std::string_view kDataPath = "data.parquet"; + Result> DeletionVectorFile( + const std::string& path, const std::vector& positions, + const std::string& data_path = std::string(kDataPath)) { + DeletionVectorWriterOptions options{ + .path = path, + .io = file_io_, + .load_previous_deletes = [](std::string_view) + -> Result> { return std::nullopt; }, + }; + ICEBERG_ASSIGN_OR_RAISE(auto writer, DeletionVectorWriter::Make(std::move(options))); + for (int64_t pos : positions) { + ICEBERG_RETURN_UNEXPECTED( + writer->Delete(data_path, pos, partition_spec_, PartitionValues{})); + } + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata()); + return metadata.data_files[0]; + } + std::shared_ptr file_io_; std::shared_ptr table_schema_; std::shared_ptr partition_spec_; @@ -1156,7 +1179,7 @@ TEST_F(DeleteFilterTest, DeletionVectorErrorPropagatesFromCompute) { ICEBERG_UNWRAP_OR_FAIL(auto batch, MakeBatch(*filter.value()->RequiredSchema(), R"([[1, 0]])")); auto alive = filter.value()->ComputeAliveRows(batch.schema, batch.array); - ASSERT_THAT(alive, IsError(ErrorKind::kNotSupported)); + ASSERT_THAT(alive, IsError(ErrorKind::kInvalidArgument)); } TEST_F(DeleteFilterTest, EmptyBatchPropagatesDeleteLoadErrors) { @@ -1175,7 +1198,27 @@ TEST_F(DeleteFilterTest, EmptyBatchPropagatesDeleteLoadErrors) { auto alive = filter.value()->ComputeAliveRows(batch.schema, batch.array); - ASSERT_THAT(alive, IsError(ErrorKind::kNotSupported)); + ASSERT_THAT(alive, IsError(ErrorKind::kInvalidArgument)); +} + +TEST_F(DeleteFilterTest, DeletionVectorComputeAliveRows) { + // Write a real deletion vector with DeletionVectorWriter, then load it through + // DeleteFilter (DeleteLoader::LoadDV) and verify deleted positions are filtered. + ICEBERG_UNWRAP_OR_FAIL(auto dv, DeletionVectorFile("dv-alive.puffin", {1, 3})); + std::vector> delete_files = {dv}; + auto requested_schema = Project({1}); + auto filter = DeleteFilter::Make(std::string(kDataPath), delete_files, table_schema_, + requested_schema, file_io_, + /*need_row_pos_col=*/true); + ASSERT_THAT(filter, IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto batch, + MakeBatch(*filter.value()->RequiredSchema(), + R"([[10, 0], [20, 1], [30, 2], [40, 3], [50, 4]])")); + auto alive = filter.value()->ComputeAliveRows(batch.schema, batch.array); + + ASSERT_THAT(alive, IsOk()); + ExpectAliveRows(alive.value(), {0, 2, 4}); } TEST_F(DeleteFilterTest, CounterAccumulatesAcrossBatches) { diff --git a/src/iceberg/test/delete_loader_test.cc b/src/iceberg/test/delete_loader_test.cc index b392065cf..a5dda22e9 100644 --- a/src/iceberg/test/delete_loader_test.cc +++ b/src/iceberg/test/delete_loader_test.cc @@ -19,6 +19,9 @@ #include "iceberg/data/delete_loader.h" +#include +#include + #include #include #include @@ -26,19 +29,23 @@ #include #include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/data/deletion_vector_writer.h" #include "iceberg/data/equality_delete_writer.h" #include "iceberg/data/position_delete_writer.h" #include "iceberg/deletes/position_delete_index.h" #include "iceberg/file_format.h" +#include "iceberg/file_io.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/parquet/parquet_register.h" #include "iceberg/partition_spec.h" +#include "iceberg/result.h" #include "iceberg/row/partition_values.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/schema_internal.h" #include "iceberg/test/matchers.h" #include "iceberg/type.h" +#include "iceberg/util/macros.h" #include "iceberg/util/struct_like_set.h" namespace iceberg { @@ -110,6 +117,27 @@ class DeleteLoaderTest : public ::testing::Test { return writer->Metadata().value().data_files[0]; } + /// Write a deletion vector to a Puffin file and return the DataFile metadata. + std::shared_ptr WriteDeletionVector(const std::string& path, + const std::string& referenced_data_file, + const std::vector& positions) { + auto writer = + DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .path = path, + .io = file_io_, + .load_previous_deletes = [](std::string_view) + -> Result> { + return std::nullopt; + }}) + .value(); + for (int64_t pos : positions) { + ICEBERG_THROW_NOT_OK( + writer->Delete(referenced_data_file, pos, partition_spec_, PartitionValues{})); + } + ICEBERG_THROW_NOT_OK(writer->Close()); + return writer->Metadata().value().data_files[0]; + } + std::shared_ptr file_io_; std::shared_ptr schema_; std::shared_ptr partition_spec_; @@ -169,8 +197,7 @@ TEST_F(DeleteLoaderTest, LoadPositionDeletesFiltersByDataFilePath) { {"data_b.parquet", 10}, {"data_b.parquet", 20}}); - // Mixed paths -> writer must NOT set the hint, forcing the loader's - // per-row filter path. Locks the routing in case the writer behavior changes. + // Mixed paths keep the loader on the per-row filter path. ASSERT_FALSE(delete_file->referenced_data_file.has_value()); std::vector> files = {delete_file}; @@ -197,6 +224,46 @@ TEST_F(DeleteLoaderTest, LoadPositionDeletesFiltersByDataFilePath) { ASSERT_TRUE(result_none.value().IsEmpty()); } +// The loaded index keeps the source Parquet position delete file, so a DV +// writer can report file-scoped deletes as rewritten. +TEST_F(DeleteLoaderTest, LoadPositionDeletesTracksSourceDeleteFile) { + // Single referenced data file -> writer sets the hint -> fast path. + auto file_scoped = WritePositionDeletes("pos_src_scoped.parquet", + {{"data.parquet", 0}, {"data.parquet", 5}}); + ASSERT_TRUE(file_scoped->referenced_data_file.has_value()); + + std::vector> files = {file_scoped}; + auto result = loader_->LoadPositionDeletes(files, "data.parquet"); + ASSERT_THAT(result, IsOk()); + ASSERT_EQ(result.value().delete_files().size(), 1u); + EXPECT_EQ(result.value().delete_files()[0], file_scoped); +} + +TEST_F(DeleteLoaderTest, LoadPositionDeletesTracksSourceForFilterPath) { + auto mixed = WritePositionDeletes("pos_src_mixed.parquet", + {{"data_a.parquet", 0}, {"data_b.parquet", 10}}); + ASSERT_FALSE(mixed->referenced_data_file.has_value()); + + std::vector> files = {mixed}; + + auto matched = loader_->LoadPositionDeletes(files, "data_a.parquet"); + ASSERT_THAT(matched, IsOk()); + ASSERT_EQ(matched.value().delete_files().size(), 1u); + EXPECT_EQ(matched.value().delete_files()[0], mixed); + + auto unmatched = loader_->LoadPositionDeletes(files, "data_c.parquet"); + ASSERT_THAT(unmatched, IsOk()); + ASSERT_TRUE(unmatched.value().IsEmpty()); + ASSERT_EQ(unmatched.value().delete_files().size(), 1u); + EXPECT_EQ(unmatched.value().delete_files()[0], mixed); +} + +TEST_F(DeleteLoaderTest, LoadPositionDeletesRejectsNullFile) { + std::vector> files = {nullptr}; + auto result = loader_->LoadPositionDeletes(files, "data.parquet"); + ASSERT_THAT(result, IsError(ErrorKind::kInvalidArgument)); +} + TEST_F(DeleteLoaderTest, LoadPositionDeletesSkipsMismatchedReferencedDataFile) { auto delete_file = std::make_shared(DataFile{ .content = DataFile::Content::kPositionDeletes, @@ -212,10 +279,7 @@ TEST_F(DeleteLoaderTest, LoadPositionDeletesSkipsMismatchedReferencedDataFile) { } TEST_F(DeleteLoaderTest, LoadPositionDeletesFastPathHonorsReferencedDataFile) { - // Single-file writes -> writer sets referenced_data_file -> loader takes - // the zero-copy fast path. Sized above the consumer's 64-element sniff - // threshold so the dispatcher's real coalesce/bulk logic runs end-to-end, - // not just the small-input shortcut covered by LoadPositionDeletesSingleFile. + // Large enough to exercise the bulk path. constexpr int64_t kRowCount = 128; std::vector> deletes; deletes.reserve(kRowCount); @@ -239,15 +303,91 @@ TEST_F(DeleteLoaderTest, LoadPositionDeletesFastPathHonorsReferencedDataFile) { ASSERT_FALSE(index.IsDeleted(kRowCount)); } -TEST_F(DeleteLoaderTest, LoadPositionDeletesRejectsDV) { +TEST_F(DeleteLoaderTest, LoadDeletionVector) { + auto dv_file = + WriteDeletionVector("dv-a.puffin", "data.parquet", {0, 5, 10, 4'000'000'000LL}); + + std::vector> files = {dv_file}; + auto result = loader_->LoadPositionDeletes(files, "data.parquet"); + ASSERT_THAT(result, IsOk()); + + auto& index = result.value(); + ASSERT_EQ(index.Cardinality(), 4); + ASSERT_TRUE(index.IsDeleted(0)); + ASSERT_TRUE(index.IsDeleted(5)); + ASSERT_TRUE(index.IsDeleted(10)); + ASSERT_TRUE(index.IsDeleted(4'000'000'000LL)); + ASSERT_FALSE(index.IsDeleted(1)); +} + +TEST_F(DeleteLoaderTest, LoadDeletionVectorRejectsMismatchedReferencedDataFile) { + // A DV referencing another data file must not be silently skipped: the single + // DV applies to its referenced file, so a mismatch is bad grouping/metadata. + auto dv_file = WriteDeletionVector("dv-b.puffin", "other-data.parquet", {1, 2, 3}); + + std::vector> files = {dv_file}; + auto result = loader_->LoadPositionDeletes(files, "data.parquet"); + ASSERT_THAT(result, IsError(ErrorKind::kInvalidArgument)); +} + +TEST_F(DeleteLoaderTest, LoadDeletionVectorRequiresContentOffsetAndSize) { auto dv_file = std::make_shared(DataFile{ .content = DataFile::Content::kPositionDeletes, .file_path = "dv.puffin", .file_format = FileFormatType::kPuffin, + .referenced_data_file = "data.parquet", }); std::vector> files = {dv_file}; auto result = loader_->LoadPositionDeletes(files, "data.parquet"); - ASSERT_THAT(result, IsError(ErrorKind::kNotSupported)); + ASSERT_THAT(result, IsError(ErrorKind::kInvalidArgument)); +} + +TEST_F(DeleteLoaderTest, LoadDeletionVectorRejectsCardinalityMismatch) { + auto dv_file = WriteDeletionVector("dv-card.puffin", "data.parquet", {0, 1, 2}); + // Corrupt the recorded cardinality so it no longer matches the bitmap. + dv_file->record_count = 99; + + std::vector> files = {dv_file}; + auto result = loader_->LoadPositionDeletes(files, "data.parquet"); + ASSERT_THAT(result, IsError(ErrorKind::kInvalidArgument)); +} + +// End to end: a previously written Parquet position delete is loaded, merged +// into a new DV, and reported in rewritten_delete_files when it is file-scoped. +TEST_F(DeleteLoaderTest, DVWriterReportsFileScopedPositionDeleteAsRewritten) { + // A file-scoped position delete (single referenced data file). + auto previous = WritePositionDeletes("prev_pos.parquet", + {{"data.parquet", 100}, {"data.parquet", 200}}); + ASSERT_TRUE(previous->referenced_data_file.has_value()); + + auto* loader = loader_.get(); + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .path = "dv-replace.puffin", + .io = file_io_, + .load_previous_deletes = + [&](std::string_view path) -> Result> { + std::vector> files = {previous}; + ICEBERG_ASSIGN_OR_RAISE(auto index, loader->LoadPositionDeletes(files, path)); + if (index.IsEmpty()) { + return std::nullopt; + } + return std::optional(std::move(index)); + }, + })); + + ICEBERG_THROW_NOT_OK( + writer->Delete("data.parquet", 0, partition_spec_, PartitionValues{})); + ICEBERG_THROW_NOT_OK(writer->Close()); + + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + ASSERT_EQ(result.data_files.size(), 1u); + // New position plus the two previous positions. + EXPECT_EQ(result.data_files[0]->record_count, 3); + // The previous Parquet position delete is file-scoped and must be reported. + ASSERT_EQ(result.rewritten_delete_files.size(), 1u); + EXPECT_EQ(result.rewritten_delete_files[0], previous); } TEST_F(DeleteLoaderTest, LoadPositionDeletesRejectsWrongContent) { diff --git a/src/iceberg/test/deletion_vector_writer_test.cc b/src/iceberg/test/deletion_vector_writer_test.cc new file mode 100644 index 000000000..605b2d4e1 --- /dev/null +++ b/src/iceberg/test/deletion_vector_writer_test.cc @@ -0,0 +1,396 @@ +/* + * 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/data/deletion_vector_writer.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/data/delete_loader.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/deletes/roaring_position_bitmap.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#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/row/partition_values.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/macros.h" +#include "iceberg/version.h" + +namespace iceberg { + +namespace { + +std::shared_ptr FindByReferencedFile( + const std::vector>& files, const std::string& ref) { + for (const auto& file : files) { + if (file->referenced_data_file == ref) { + return file; + } + } + return nullptr; +} + +std::shared_ptr UnpartitionedSpec() { + return PartitionSpec::Unpartitioned(); +} + +const puffin::BlobMetadata* FindBlobByReferencedFile( + const std::vector& blobs, const std::string& ref) { + for (const auto& blob : blobs) { + auto it = blob.properties.find("referenced-data-file"); + if (it != blob.properties.end() && it->second == ref) { + return &blob; + } + } + return nullptr; +} + +// A load_previous_deletes callback for data files that have no existing deletes. +Result> NoPreviousDeletes(std::string_view) { + return std::nullopt; +} + +DeletionVectorWriterOptions MakeDVWriterOptions( + std::shared_ptr io, std::string path, + std::function>(std::string_view)> + load_previous_deletes = NoPreviousDeletes) { + return DeletionVectorWriterOptions{ + .path = std::move(path), + .io = std::move(io), + .load_previous_deletes = std::move(load_previous_deletes), + }; +} + +} // namespace + +TEST(DeletionVectorWriterTest, WriteThenLoadEndToEnd) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + + std::vector> delete_files; + { + ICEBERG_UNWRAP_OR_FAIL(auto writer, DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://deletes.puffin"))); + + ASSERT_THAT(writer->Delete("data-a.parquet", 0, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Delete("data-a.parquet", 5, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Delete("data-a.parquet", 10, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Delete("data-b.parquet", 1, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Delete("data-b.parquet", 2, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Close(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + delete_files = result.data_files; + // Each referenced data file is reported once. + EXPECT_EQ(result.referenced_data_files.size(), 2u); + // No previous deletes were loaded, so nothing was rewritten. + EXPECT_TRUE(result.rewritten_delete_files.empty()); + } + + // One DataFile per referenced data file. + ASSERT_EQ(delete_files.size(), 2u); + + auto dv_a = FindByReferencedFile(delete_files, "data-a.parquet"); + auto dv_b = FindByReferencedFile(delete_files, "data-b.parquet"); + ASSERT_NE(dv_a, nullptr); + ASSERT_NE(dv_b, nullptr); + + EXPECT_EQ(dv_a->content, DataFile::Content::kPositionDeletes); + EXPECT_EQ(dv_a->file_format, FileFormatType::kPuffin); + EXPECT_TRUE(dv_a->IsDeletionVector()); + EXPECT_EQ(dv_a->file_path, "memory://deletes.puffin"); + EXPECT_EQ(dv_a->record_count, 3); + EXPECT_TRUE(dv_a->content_offset.has_value()); + EXPECT_TRUE(dv_a->content_size_in_bytes.has_value()); + EXPECT_GT(dv_a->file_size_in_bytes, 0); + EXPECT_EQ(dv_a->partition_spec_id, spec->spec_id()); + EXPECT_EQ(dv_b->record_count, 2); + + // Both blobs live in the same Puffin file but at different offsets. + EXPECT_EQ(dv_a->file_path, dv_b->file_path); + EXPECT_NE(dv_a->content_offset.value(), dv_b->content_offset.value()); + + ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile("memory://deletes.puffin")); + ICEBERG_UNWRAP_OR_FAIL(auto puffin_reader, + puffin::PuffinReader::Make(std::move(input_file))); + ICEBERG_UNWRAP_OR_FAIL(auto puffin_metadata, puffin_reader->ReadFileMetadata()); + ASSERT_EQ(puffin_metadata.blobs.size(), 2u); + EXPECT_EQ(puffin_metadata.properties.at("created-by"), ICEBERG_FULL_VERSION_STRING); + + auto assert_blob = [](const puffin::BlobMetadata& blob, const DataFile& file, + const std::string& ref, const std::string& cardinality) { + EXPECT_EQ(blob.type, puffin::StandardBlobTypes::kDeletionVectorV1); + EXPECT_EQ(blob.input_fields, + std::vector{MetadataColumns::kFilePositionColumnId}); + EXPECT_EQ(blob.snapshot_id, -1); + EXPECT_EQ(blob.sequence_number, -1); + EXPECT_TRUE(blob.compression_codec.empty()); + EXPECT_EQ(blob.offset, file.content_offset.value()); + EXPECT_EQ(blob.length, file.content_size_in_bytes.value()); + EXPECT_EQ(blob.properties.at("referenced-data-file"), ref); + EXPECT_EQ(blob.properties.at("cardinality"), cardinality); + }; + ASSERT_NE(FindBlobByReferencedFile(puffin_metadata.blobs, "data-a.parquet"), nullptr); + ASSERT_NE(FindBlobByReferencedFile(puffin_metadata.blobs, "data-b.parquet"), nullptr); + assert_blob(*FindBlobByReferencedFile(puffin_metadata.blobs, "data-a.parquet"), *dv_a, + "data-a.parquet", "3"); + assert_blob(*FindBlobByReferencedFile(puffin_metadata.blobs, "data-b.parquet"), *dv_b, + "data-b.parquet", "2"); + + // Load each data file's DV separately. + DeleteLoader loader(io); + { + auto result = loader.LoadPositionDeletes({&dv_a, 1}, "data-a.parquet"); + ASSERT_THAT(result, IsOk()); + auto& index = result.value(); + EXPECT_EQ(index.Cardinality(), 3); + EXPECT_TRUE(index.IsDeleted(0)); + EXPECT_TRUE(index.IsDeleted(5)); + EXPECT_TRUE(index.IsDeleted(10)); + EXPECT_FALSE(index.IsDeleted(1)); + } + + { + auto result = loader.LoadPositionDeletes({&dv_b, 1}, "data-b.parquet"); + ASSERT_THAT(result, IsOk()); + auto& index = result.value(); + EXPECT_EQ(index.Cardinality(), 2); + EXPECT_TRUE(index.IsDeleted(1)); + EXPECT_TRUE(index.IsDeleted(2)); + EXPECT_FALSE(index.IsDeleted(0)); + } +} + +// The PositionDeleteIndex overload bulk-adds positions for a data file. +TEST(DeletionVectorWriterTest, DeleteFromIndex) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + + PositionDeleteIndex positions; + positions.Delete(0); + positions.Delete(3, 6); // [3, 6) -> 3, 4, 5 + + ICEBERG_UNWRAP_OR_FAIL(auto writer, DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://from-index.puffin"))); + ASSERT_THAT(writer->Delete("data.parquet", positions, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Close(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + ASSERT_EQ(result.data_files.size(), 1u); + EXPECT_EQ(result.data_files[0]->record_count, 4); + + DeleteLoader loader(io); + auto loaded = loader.LoadPositionDeletes(result.data_files, "data.parquet"); + ASSERT_THAT(loaded, IsOk()); + EXPECT_EQ(loaded.value().Cardinality(), 4); + EXPECT_TRUE(loaded.value().IsDeleted(0)); + EXPECT_TRUE(loaded.value().IsDeleted(5)); + EXPECT_FALSE(loaded.value().IsDeleted(6)); +} + +// Previously written deletes are merged into the new vector, and the file-scoped +// delete files they came from are reported as rewritten. +TEST(DeletionVectorWriterTest, LoadPreviousDeletesMergesAndReportsRewritten) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + + // Build a previous DV index carrying its source (file-scoped) delete file, + // the same way DeleteLoader would produce it. + PositionDeleteIndex previous_positions; + previous_positions.Delete(100); + previous_positions.Delete(200); + ICEBERG_UNWRAP_OR_FAIL(auto previous_blob, previous_positions.Serialize()); + auto previous_dv = std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = "memory://old.puffin", + .file_format = FileFormatType::kPuffin, + .record_count = 2, + .referenced_data_file = "data.parquet", + .content_offset = 0, + .content_size_in_bytes = static_cast(previous_blob.size()), + }); + + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://merged.puffin", + [&](std::string_view path) -> Result> { + if (path != "data.parquet") { + return std::nullopt; + } + ICEBERG_ASSIGN_OR_RAISE( + auto index, PositionDeleteIndex::Deserialize(previous_blob, previous_dv)); + return std::optional(std::move(index)); + }))); + + ASSERT_THAT(writer->Delete("data.parquet", 0, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Close(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + ASSERT_EQ(result.data_files.size(), 1u); + // New position plus the two previous positions. + EXPECT_EQ(result.data_files[0]->record_count, 3); + // The previous DV is file-scoped, so it is reported for removal. + ASSERT_EQ(result.rewritten_delete_files.size(), 1u); + EXPECT_EQ(result.rewritten_delete_files[0]->file_path, "memory://old.puffin"); + + DeleteLoader loader(io); + auto loaded = loader.LoadPositionDeletes(result.data_files, "data.parquet"); + ASSERT_THAT(loaded, IsOk()); + EXPECT_EQ(loaded.value().Cardinality(), 3); + EXPECT_TRUE(loaded.value().IsDeleted(0)); + EXPECT_TRUE(loaded.value().IsDeleted(100)); + EXPECT_TRUE(loaded.value().IsDeleted(200)); +} + +TEST(DeletionVectorWriterTest, PreviousDeletesWithoutSourceFilesAreNotRewritten) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://merged-partition.puffin", + [&](std::string_view path) -> Result> { + PositionDeleteIndex index; + index.Delete(50); + return std::optional(std::move(index)); + }))); + + ASSERT_THAT(writer->Delete("data.parquet", 0, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Close(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + ASSERT_EQ(result.data_files.size(), 1u); + // The previous position was merged in. + EXPECT_EQ(result.data_files[0]->record_count, 2); + // The previous delete is partition-scoped, so it is not rewritten. + EXPECT_TRUE(result.rewritten_delete_files.empty()); + + DeleteLoader loader(io); + auto loaded = loader.LoadPositionDeletes(result.data_files, "data.parquet"); + ASSERT_THAT(loaded, IsOk()); + EXPECT_EQ(loaded.value().Cardinality(), 2); + EXPECT_TRUE(loaded.value().IsDeleted(0)); + EXPECT_TRUE(loaded.value().IsDeleted(50)); +} + +TEST(DeletionVectorWriterTest, EmptyWriterProducesNoDataFiles) { + auto io = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL(auto writer, + DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .path = "memory://empty.puffin", + .io = io, + .load_previous_deletes = NoPreviousDeletes})); + ASSERT_THAT(writer->Close(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + EXPECT_TRUE(result.data_files.empty()); + EXPECT_THAT(io->NewInputFile("memory://empty.puffin"), IsError(ErrorKind::kNotFound)); +} + +TEST(DeletionVectorWriterTest, DeleteRejectsEmptyReferencedFile) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + ICEBERG_UNWRAP_OR_FAIL(auto writer, DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://invalid.puffin"))); + EXPECT_THAT(writer->Delete("", 0, spec, PartitionValues{}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(writer->Delete("data.parquet", 0, nullptr, PartitionValues{}), + IsError(ErrorKind::kInvalidArgument)); + PositionDeleteIndex positions; + positions.Delete(0); + EXPECT_THAT(writer->Delete("data.parquet", positions, nullptr, PartitionValues{}), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(DeletionVectorWriterTest, DeleteRejectsOutOfRangePosition) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + ICEBERG_UNWRAP_OR_FAIL(auto writer, DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://invalid.puffin"))); + // Negative and out-of-range positions are rejected rather than silently + // dropped by the underlying bitmap. + EXPECT_THAT(writer->Delete("data.parquet", -1, spec, PartitionValues{}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(writer->Delete("data.parquet", RoaringPositionBitmap::kMaxPosition + 1, + spec, PartitionValues{}), + IsError(ErrorKind::kInvalidArgument)); +} + +// Close propagates a load_previous_deletes failure and returns no metadata. +TEST(DeletionVectorWriterTest, ClosePropagatesLoadPreviousDeletesError) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://err.puffin", + [](std::string_view) -> Result> { + return IOError("boom"); + }))); + + ASSERT_THAT(writer->Delete("data.parquet", 0, spec, PartitionValues{}), IsOk()); + EXPECT_THAT(writer->Close(), IsError(ErrorKind::kIOError)); + // No partial delete metadata is exposed after a failed Close. + EXPECT_THAT(writer->Metadata(), IsError(ErrorKind::kValidationFailed)); + // No orphan Puffin file was created. + EXPECT_THAT(io->NewInputFile("memory://err.puffin"), IsError(ErrorKind::kNotFound)); +} + +TEST(DeletionVectorWriterTest, MakeRejectsMissingOutputPath) { + EXPECT_THAT(DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .io = std::make_shared(), + .load_previous_deletes = NoPreviousDeletes}), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(DeletionVectorWriterTest, MakeRejectsMissingLoadPreviousDeletes) { + auto io = std::make_shared(); + EXPECT_THAT(DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .path = "x.puffin", + .io = io, + }), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(DeletionVectorWriterTest, DeleteAfterCloseFails) { + auto io = std::make_shared(); + auto spec = UnpartitionedSpec(); + ICEBERG_UNWRAP_OR_FAIL(auto writer, DeletionVectorWriter::Make(MakeDVWriterOptions( + io, "memory://closed.puffin"))); + ASSERT_THAT(writer->Close(), IsOk()); + EXPECT_THAT(writer->Delete("data-a.parquet", 0, spec, PartitionValues{}), + IsError(ErrorKind::kValidationFailed)); +} + +} // namespace iceberg diff --git a/src/iceberg/test/endian_test.cc b/src/iceberg/test/endian_test.cc index 717baeafe..93aadde4e 100644 --- a/src/iceberg/test/endian_test.cc +++ b/src/iceberg/test/endian_test.cc @@ -94,6 +94,9 @@ TEST(EndianTest, BufferReadWriteRoundTrip) { WriteLittleEndian(int32_t{-1}, buf.data()); EXPECT_EQ(ReadLittleEndian(buf.data()), -1); + + WriteBigEndian(int32_t{0x12345678}, buf.data()); + EXPECT_EQ(ReadBigEndian(buf.data()), 0x12345678); } } // namespace iceberg diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 1630a108a..10e9c8e5f 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -36,11 +36,15 @@ #include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/arrow/arrow_register.h" +#include "iceberg/arrow_c_data.h" #include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/arrow_c_data_util_internal.h" +#include "iceberg/data/deletion_vector_writer.h" #include "iceberg/data/equality_delete_writer.h" #include "iceberg/data/position_delete_writer.h" +#include "iceberg/deletes/position_delete_index.h" #include "iceberg/file_format.h" +#include "iceberg/file_io.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/parquet/parquet_register.h" @@ -298,6 +302,25 @@ class FileScanTaskReaderTest : public TempFileTestBase { return metadata.data_files[0]; } + Result> MakeDeletionVectorFile( + const std::string& path, const std::vector& positions, + const std::string& data_path) { + DeletionVectorWriterOptions options{ + .path = path, + .io = file_io_, + .load_previous_deletes = [](std::string_view) + -> Result> { return std::nullopt; }, + }; + ICEBERG_ASSIGN_OR_RAISE(auto writer, DeletionVectorWriter::Make(std::move(options))); + for (int64_t pos : positions) { + ICEBERG_RETURN_UNEXPECTED( + writer->Delete(data_path, pos, partition_spec_, PartitionValues{})); + } + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata()); + return metadata.data_files[0]; + } + void VerifyStream(struct ArrowArrayStream* stream, std::string_view expected_json) { auto record_batch_reader = ::arrow::ImportRecordBatchReader(stream).ValueOrDie(); @@ -401,6 +424,30 @@ TEST_F(FileScanTaskReaderTest, OpenWithPositionDeletesFiltersRowsAndPrunesPos) { ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3, "Baz"]])")); } +TEST_F(FileScanTaskReaderTest, OpenWithDeletionVectorFiltersRows) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, + MakeDataFile(table_schema_, + R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz", "green"]])")); + ICEBERG_UNWRAP_OR_FAIL( + auto dv, MakeDeletionVectorFile(CreateNewTempFilePathWithSuffix(".puffin"), {1}, + data_file->file_path)); + FileScanTask task(data_file, {dv}); + + FileScanTaskReader::Options options{ + .io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projected_schema_, + }; + ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options))); + auto stream_result = reader->Open(task); + ASSERT_THAT(stream_result, IsOk()); + auto stream = std::move(stream_result.value()); + + ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3, "Baz"]])")); +} + TEST_F(FileScanTaskReaderTest, OpenWithEqualityDeletesAddsAndPrunesDeleteOnlyColumns) { ICEBERG_UNWRAP_OR_FAIL( auto data_file, diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build index 12460237d..0e5399347 100644 --- a/src/iceberg/test/meson.build +++ b/src/iceberg/test/meson.build @@ -124,6 +124,7 @@ iceberg_tests = { }, 'puffin_test': { 'sources': files( + 'deletion_vector_writer_test.cc', 'puffin_format_test.cc', 'puffin_json_test.cc', 'puffin_reader_writer_test.cc', diff --git a/src/iceberg/test/position_delete_index_test.cc b/src/iceberg/test/position_delete_index_test.cc index 0c24a587b..1075ee6fa 100644 --- a/src/iceberg/test/position_delete_index_test.cc +++ b/src/iceberg/test/position_delete_index_test.cc @@ -19,10 +19,38 @@ #include "iceberg/deletes/position_delete_index.h" +#include +#include +#include + #include +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/result.h" +#include "iceberg/test/matchers.h" + namespace iceberg { +namespace { + +// A source delete file matching a serialized blob: content_size_in_bytes and +// record_count are what Deserialize validates against. +std::shared_ptr DeleteFileFor(const std::vector& blob, + int64_t record_count) { + return std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = "memory://dv.puffin", + .file_format = FileFormatType::kPuffin, + .record_count = record_count, + .referenced_data_file = "data.parquet", + .content_offset = 0, + .content_size_in_bytes = static_cast(blob.size()), + }); +} + +} // namespace + TEST(PositionDeleteIndexTest, TestEmptyIndex) { PositionDeleteIndex index; ASSERT_TRUE(index.IsEmpty()); @@ -200,4 +228,141 @@ TEST(PositionDeleteIndexTest, TestMergeIdempotence) { ASSERT_TRUE(index1.IsDeleted(20)); } +// ==================== deletion-vector-v1 serialization ==================== + +TEST(PositionDeleteIndexTest, SerializeRoundTrip) { + PositionDeleteIndex index; + for (int64_t pos : + {int64_t{0}, int64_t{1}, int64_t{5}, int64_t{100}, int64_t{4'000'000'000}}) { + index.Delete(pos); + } + + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + ICEBERG_UNWRAP_OR_FAIL(auto restored, + PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 5))); + + EXPECT_EQ(restored.Cardinality(), 5); + for (int64_t pos : + {int64_t{0}, int64_t{1}, int64_t{5}, int64_t{100}, int64_t{4'000'000'000}}) { + EXPECT_TRUE(restored.IsDeleted(pos)); + } +} + +TEST(PositionDeleteIndexTest, SerializeEmptyRoundTrip) { + PositionDeleteIndex index; + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + ICEBERG_UNWRAP_OR_FAIL(auto restored, + PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 0))); + EXPECT_TRUE(restored.IsEmpty()); +} + +// Spans two high-32-bit keys and exercises all Roaring container types +// (sparse "array", dense "bitset", and run containers after optimization). +TEST(PositionDeleteIndexTest, SerializeAllContainerTypesAcrossKeys) { + constexpr int64_t kKeyStride = 0x100000000LL; // 2^32: high-32-bit key + constexpr int64_t kContainerStride = 1 << 16; // 2^16: Roaring container + auto pos = [](int64_t key, int64_t container, int64_t value) { + return key * kKeyStride + container * kContainerStride + value; + }; + + PositionDeleteIndex index; + int64_t expected = 0; + for (int64_t key : {int64_t{0}, int64_t{1}}) { + index.Delete(pos(key, 0, 5)); + index.Delete(pos(key, 0, 7)); + expected += 2; + index.Delete(pos(key, 1, 1), pos(key, 1, 1000)); + expected += 999; + index.Delete(pos(key, 2, 1), pos(key, 2, kContainerStride)); + expected += kContainerStride - 1; + } + + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + ICEBERG_UNWRAP_OR_FAIL(auto restored, PositionDeleteIndex::Deserialize( + blob, DeleteFileFor(blob, expected))); + + EXPECT_EQ(restored.Cardinality(), expected); + EXPECT_TRUE(restored.IsDeleted(pos(0, 0, 5))); + EXPECT_TRUE(restored.IsDeleted(pos(1, 2, kContainerStride - 1))); + EXPECT_TRUE(restored.IsDeleted(pos(0, 1, 999))); + EXPECT_FALSE(restored.IsDeleted(pos(0, 0, 6))); + EXPECT_FALSE(restored.IsDeleted(pos(1, 1, 1000))); // range end is exclusive +} + +TEST(PositionDeleteIndexTest, DeserializeRejectsCorruptedCrc) { + PositionDeleteIndex index; + index.Delete(1); + index.Delete(2); + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + + blob.back() ^= 0xFF; + EXPECT_THAT(PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 2)), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(PositionDeleteIndexTest, DeserializeRejectsBadMagic) { + PositionDeleteIndex index; + index.Delete(1); + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + + blob[4] = 0x00; + EXPECT_THAT(PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 1)), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(PositionDeleteIndexTest, DeserializeRejectsTruncatedBlob) { + std::vector blob = {0x00, 0x00}; + EXPECT_THAT(PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 0)), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(PositionDeleteIndexTest, DeserializeRejectsCardinalityMismatch) { + PositionDeleteIndex index; + index.Delete(1); + index.Delete(2); + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + + // record_count in metadata disagrees with the bitmap cardinality. + EXPECT_THAT(PositionDeleteIndex::Deserialize(blob, DeleteFileFor(blob, 99)), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(PositionDeleteIndexTest, DeserializeRejectsContentSizeMismatch) { + PositionDeleteIndex index; + index.Delete(1); + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + + auto delete_file = DeleteFileFor(blob, 1); + delete_file->content_size_in_bytes = static_cast(blob.size()) + 1; + EXPECT_THAT(PositionDeleteIndex::Deserialize(blob, delete_file), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(PositionDeleteIndexTest, DeserializePreservesSourceDeleteFile) { + PositionDeleteIndex index; + index.Delete(1); + index.Delete(2); + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + + auto delete_file = DeleteFileFor(blob, 2); + ICEBERG_UNWRAP_OR_FAIL(auto restored, + PositionDeleteIndex::Deserialize(blob, delete_file)); + ASSERT_EQ(restored.delete_files().size(), 1u); + EXPECT_EQ(restored.delete_files()[0], delete_file); +} + +TEST(PositionDeleteIndexTest, ConstructorsPreserveSourceDeleteFiles) { + auto file_a = std::make_shared(DataFile{.file_path = "a.parquet"}); + auto file_b = std::make_shared(DataFile{.file_path = "b.parquet"}); + + PositionDeleteIndex single(file_a); + ASSERT_EQ(single.delete_files().size(), 1u); + EXPECT_EQ(single.delete_files()[0], file_a); + + PositionDeleteIndex multiple(std::vector>{file_a, file_b}); + ASSERT_EQ(multiple.delete_files().size(), 2u); + EXPECT_EQ(multiple.delete_files()[0], file_a); + EXPECT_EQ(multiple.delete_files()[1], file_b); +} + } // namespace iceberg diff --git a/src/iceberg/util/endian.h b/src/iceberg/util/endian.h index 638273a1c..8aa559f40 100644 --- a/src/iceberg/util/endian.h +++ b/src/iceberg/util/endian.h @@ -112,4 +112,21 @@ T ReadLittleEndian(const void* input) { return FromLittleEndian(value); } +/// \brief Write a value in big-endian format to a buffer. +/// \note Caller must ensure output has at least sizeof(T) bytes available. +template +void WriteBigEndian(T value, void* output) { + auto be = ToBigEndian(value); + std::memcpy(output, &be, sizeof(be)); +} + +/// \brief Read a value in big-endian format from a buffer. +/// \note Caller must ensure input has at least sizeof(T) bytes available. +template +T ReadBigEndian(const void* input) { + T value; + std::memcpy(&value, input, sizeof(value)); + return FromBigEndian(value); +} + } // namespace iceberg diff --git a/src/iceberg/util/gzip_internal.cc b/src/iceberg/util/gzip_internal.cc index 680e2bc33..63e5acb2a 100644 --- a/src/iceberg/util/gzip_internal.cc +++ b/src/iceberg/util/gzip_internal.cc @@ -19,11 +19,15 @@ #include "iceberg/util/gzip_internal.h" +#include #include #include +#include +#include #include +#include "iceberg/result.h" #include "iceberg/util/macros.h" namespace iceberg { diff --git a/src/iceberg/util/string_util.cc b/src/iceberg/util/string_util.cc index 8b98397db..b4c717861 100644 --- a/src/iceberg/util/string_util.cc +++ b/src/iceberg/util/string_util.cc @@ -22,7 +22,13 @@ #include #include +#include +#include +#include +#include +#include +#include "iceberg/result.h" #include "iceberg/util/macros.h" namespace iceberg { diff --git a/src/iceberg/util/string_util.h b/src/iceberg/util/string_util.h index 123167324..8202e60a3 100644 --- a/src/iceberg/util/string_util.h +++ b/src/iceberg/util/string_util.h @@ -226,4 +226,11 @@ struct ICEBERG_EXPORT StringEqual { } }; +/// \brief Transparent less function that supports std::string_view as lookup key +struct ICEBERG_EXPORT StringLess { + using is_transparent = void; + + bool operator()(std::string_view lhs, std::string_view rhs) const { return lhs < rhs; } +}; + } // namespace iceberg diff --git a/src/iceberg/version.h.in b/src/iceberg/version.h.in index af7fe01e6..c7f8149fa 100644 --- a/src/iceberg/version.h.in +++ b/src/iceberg/version.h.in @@ -23,6 +23,8 @@ #define ICEBERG_VERSION_MINOR @PROJECT_VERSION_MINOR@ #define ICEBERG_VERSION_PATCH @PROJECT_VERSION_PATCH@ +#define ICEBERG_GIT_COMMIT_ID "@ICEBERG_GIT_COMMIT_ID@" #define ICEBERG_VERSION_STRING "@ICEBERG_VERSION_STRING@" +#define ICEBERG_FULL_VERSION_STRING "@ICEBERG_FULL_VERSION_STRING@" #define ICEBERG_PROJECT_NAME "@PROJECT_NAME@"