feat: implement merge multiple DVs

Author: HuaHuaY

src/iceberg/CMakeLists.txt

@@ -24,6 +24,7 @@ set(ICEBERG_SOURCES
     catalog/session_catalog.cc
     catalog/session_context.cc
     delete_file_index.cc
+    deletes/roaring_position_bitmap.cc
     expression/aggregate.cc
     expression/binder.cc
     expression/evaluator.cc
@@ -75,6 +76,11 @@ set(ICEBERG_SOURCES
     partition_field.cc
     partition_spec.cc
     partition_summary.cc
+    puffin/file_metadata.cc
+    puffin/json_serde.cc
+    puffin/puffin_format.cc
+    puffin/puffin_reader.cc
+    puffin/puffin_writer.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     row/partition_values.cc
@@ -146,23 +152,27 @@ list(APPEND
      ICEBERG_STATIC_BUILD_INTERFACE_LIBS
      "$<IF:$<BOOL:${NANOARROW_VENDORED}>,nanoarrow::nanoarrow_static,$<IF:$<TARGET_EXISTS:nanoarrow::nanoarrow_static>,nanoarrow::nanoarrow_static,nanoarrow::nanoarrow_shared>>"
      nlohmann_json::nlohmann_json
+     roaring::roaring
      spdlog::spdlog
      ZLIB::ZLIB)
 list(APPEND
      ICEBERG_SHARED_BUILD_INTERFACE_LIBS
      "$<IF:$<BOOL:${NANOARROW_VENDORED}>,nanoarrow::nanoarrow_static,$<IF:$<TARGET_EXISTS:nanoarrow::nanoarrow_shared>,nanoarrow::nanoarrow_shared,nanoarrow::nanoarrow_static>>"
      nlohmann_json::nlohmann_json
+     roaring::roaring
      spdlog::spdlog
      ZLIB::ZLIB)
 list(APPEND
      ICEBERG_STATIC_INSTALL_INTERFACE_LIBS
      "$<IF:$<BOOL:${NANOARROW_VENDORED}>,iceberg::nanoarrow_static,$<IF:$<TARGET_EXISTS:nanoarrow::nanoarrow_static>,nanoarrow::nanoarrow_static,nanoarrow::nanoarrow_shared>>"
      "$<IF:$<BOOL:${NLOHMANN_JSON_VENDORED}>,iceberg::nlohmann_json,$<IF:$<TARGET_EXISTS:nlohmann_json::nlohmann_json>,nlohmann_json::nlohmann_json,nlohmann_json::nlohmann_json>>"
+     "$<IF:$<BOOL:${CROARING_VENDORED}>,iceberg::roaring,roaring::roaring>"
      "$<IF:$<BOOL:${SPDLOG_VENDORED}>,iceberg::spdlog,spdlog::spdlog>")
 list(APPEND
      ICEBERG_SHARED_INSTALL_INTERFACE_LIBS
      "$<IF:$<BOOL:${NANOARROW_VENDORED}>,iceberg::nanoarrow_static,$<IF:$<TARGET_EXISTS:nanoarrow::nanoarrow_shared>,nanoarrow::nanoarrow_shared,nanoarrow::nanoarrow_static>>"
      "$<IF:$<BOOL:${NLOHMANN_JSON_VENDORED}>,iceberg::nlohmann_json,$<IF:$<TARGET_EXISTS:nlohmann_json::nlohmann_json>,nlohmann_json::nlohmann_json,nlohmann_json::nlohmann_json>>"
+     "$<IF:$<BOOL:${CROARING_VENDORED}>,iceberg::roaring,roaring::roaring>"
      "$<IF:$<BOOL:${SPDLOG_VENDORED}>,iceberg::spdlog,spdlog::spdlog>")
 
 add_iceberg_lib(iceberg
@@ -190,33 +200,25 @@ set(ICEBERG_DATA_SOURCES
     data/position_delete_writer.cc
     data/writer.cc
     deletes/position_delete_index.cc
-    deletes/position_delete_range_consumer.cc
-    deletes/roaring_position_bitmap.cc
-    puffin/file_metadata.cc
-    puffin/json_serde.cc
-    puffin/puffin_format.cc
-    puffin/puffin_reader.cc
-    puffin/puffin_writer.cc)
+    deletes/position_delete_range_consumer.cc)
 
 set(ICEBERG_DATA_STATIC_BUILD_INTERFACE_LIBS)
 set(ICEBERG_DATA_SHARED_BUILD_INTERFACE_LIBS)
 set(ICEBERG_DATA_STATIC_INSTALL_INTERFACE_LIBS)
 set(ICEBERG_DATA_SHARED_INSTALL_INTERFACE_LIBS)
 
 list(APPEND ICEBERG_DATA_STATIC_BUILD_INTERFACE_LIBS
-     "$<IF:$<TARGET_EXISTS:iceberg_static>,iceberg_static,iceberg_shared>"
-     roaring::roaring)
+     "$<IF:$<TARGET_EXISTS:iceberg_static>,iceberg_static,iceberg_shared>")
 list(APPEND ICEBERG_DATA_SHARED_BUILD_INTERFACE_LIBS
-     "$<IF:$<TARGET_EXISTS:iceberg_shared>,iceberg_shared,iceberg_static>"
-     roaring::roaring)
+     "$<IF:$<TARGET_EXISTS:iceberg_shared>,iceberg_shared,iceberg_static>")
 list(APPEND
      ICEBERG_DATA_STATIC_INSTALL_INTERFACE_LIBS
      "$<IF:$<TARGET_EXISTS:iceberg::iceberg_static>,iceberg::iceberg_static,iceberg::iceberg_shared>"
-     "$<IF:$<BOOL:${CROARING_VENDORED}>,iceberg::roaring,roaring::roaring>")
+)
 list(APPEND
      ICEBERG_DATA_SHARED_INSTALL_INTERFACE_LIBS
      "$<IF:$<TARGET_EXISTS:iceberg::iceberg_shared>,iceberg::iceberg_shared,iceberg::iceberg_static>"
-     "$<IF:$<BOOL:${CROARING_VENDORED}>,iceberg::roaring,roaring::roaring>")
+)
 
 add_iceberg_lib(iceberg_data
                 SOURCES

src/iceberg/deletes/roaring_position_bitmap.h

@@ -29,7 +29,7 @@
 #include <string>
 #include <string_view>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 
 namespace iceberg {
@@ -48,7 +48,7 @@ class PositionDeleteIndex;
 /// \note This class is used to represent deletion vectors. The Puffin reader/writer
 /// handle adding the additional required framing (length prefix, magic bytes, CRC-32)
 /// for `deletion-vector-v1` persistence.
-class ICEBERG_DATA_EXPORT RoaringPositionBitmap {
+class ICEBERG_EXPORT RoaringPositionBitmap {
  public:
   /// \brief Maximum supported position (aligned with the Java implementation).
   static constexpr int64_t kMaxPosition = 0x7FFFFFFE80000000LL;

src/iceberg/meson.build

@@ -49,6 +49,7 @@ iceberg_sources = files(
     'catalog/session_catalog.cc',
     'catalog/session_context.cc',
     'delete_file_index.cc',
+    'deletes/roaring_position_bitmap.cc',
     'expression/aggregate.cc',
     'expression/binder.cc',
     'expression/evaluator.cc',
@@ -100,6 +101,11 @@ iceberg_sources = files(
     'partition_field.cc',
     'partition_spec.cc',
     'partition_summary.cc',
+    'puffin/file_metadata.cc',
+    'puffin/json_serde.cc',
+    'puffin/puffin_format.cc',
+    'puffin/puffin_reader.cc',
+    'puffin/puffin_writer.cc',
     'row/arrow_array_wrapper.cc',
     'row/manifest_wrapper.cc',
     'row/partition_values.cc',
@@ -173,12 +179,6 @@ iceberg_data_sources = files(
     'data/writer.cc',
     'deletes/position_delete_index.cc',
     'deletes/position_delete_range_consumer.cc',
-    'deletes/roaring_position_bitmap.cc',
-    'puffin/file_metadata.cc',
-    'puffin/json_serde.cc',
-    'puffin/puffin_format.cc',
-    'puffin/puffin_reader.cc',
-    'puffin/puffin_writer.cc',
 )
 
 # CRoaring does not export symbols, so on Windows it must
@@ -193,7 +193,13 @@ nlohmann_json_dep = dependency('nlohmann_json')
 spdlog_dep = dependency('spdlog')
 zlib_dep = dependency('zlib')
 
-iceberg_deps = [nanoarrow_dep, nlohmann_json_dep, spdlog_dep, zlib_dep]
+iceberg_deps = [
+    nanoarrow_dep,
+    nlohmann_json_dep,
+    croaring_dep,
+    spdlog_dep,
+    zlib_dep,
+]
 
 iceberg_lib = library(
     'iceberg',
@@ -219,7 +225,7 @@ iceberg_dep = declare_dependency(
 )
 meson.override_dependency('iceberg', iceberg_dep)
 
-iceberg_data_deps = [iceberg_dep, croaring_dep]
+iceberg_data_deps = [iceberg_dep]
 iceberg_data_lib = library(
     'iceberg_data',
     sources: iceberg_data_sources,

src/iceberg/puffin/file_metadata.h

@@ -29,7 +29,7 @@
 #include <unordered_map>
 #include <vector>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 
 namespace iceberg::puffin {
@@ -41,12 +41,12 @@ enum class PuffinCompressionCodec {
   kZstd,
 };
 
-ICEBERG_DATA_EXPORT std::string_view CodecName(PuffinCompressionCodec codec);
+ICEBERG_EXPORT std::string_view CodecName(PuffinCompressionCodec codec);
 
-ICEBERG_DATA_EXPORT Result<PuffinCompressionCodec> PuffinCompressionCodecFromName(
+ICEBERG_EXPORT Result<PuffinCompressionCodec> PuffinCompressionCodecFromName(
     std::string_view codec_name);
 
-ICEBERG_DATA_EXPORT std::string ToString(PuffinCompressionCodec codec);
+ICEBERG_EXPORT std::string ToString(PuffinCompressionCodec codec);
 
 /// \brief Standard blob types defined by the Iceberg specification.
 struct StandardBlobTypes {
@@ -67,7 +67,7 @@ struct StandardPuffinProperties {
 };
 
 /// \brief A blob in a Puffin file.
-struct ICEBERG_DATA_EXPORT Blob {
+struct ICEBERG_EXPORT Blob {
   /// See StandardBlobTypes for known types.
   std::string type;
   /// Ordered list of field IDs the blob was computed from.
@@ -84,10 +84,10 @@ struct ICEBERG_DATA_EXPORT Blob {
   friend bool operator==(const Blob& lhs, const Blob& rhs) = default;
 };
 
-ICEBERG_DATA_EXPORT std::string ToString(const Blob& blob);
+ICEBERG_EXPORT std::string ToString(const Blob& blob);
 
 /// \brief Metadata about a blob stored in a Puffin file footer.
-struct ICEBERG_DATA_EXPORT BlobMetadata {
+struct ICEBERG_EXPORT BlobMetadata {
   /// See StandardBlobTypes for known types.
   std::string type;
   /// Ordered list of field IDs the blob was computed from.
@@ -105,16 +105,16 @@ struct ICEBERG_DATA_EXPORT BlobMetadata {
   friend bool operator==(const BlobMetadata& lhs, const BlobMetadata& rhs) = default;
 };
 
-ICEBERG_DATA_EXPORT std::string ToString(const BlobMetadata& blob_metadata);
+ICEBERG_EXPORT std::string ToString(const BlobMetadata& blob_metadata);
 
 /// \brief Metadata about a Puffin file.
-struct ICEBERG_DATA_EXPORT FileMetadata {
+struct ICEBERG_EXPORT FileMetadata {
   std::vector<BlobMetadata> blobs;
   std::unordered_map<std::string, std::string> properties;
 
   friend bool operator==(const FileMetadata& lhs, const FileMetadata& rhs) = default;
 };
 
-ICEBERG_DATA_EXPORT std::string ToString(const FileMetadata& file_metadata);
+ICEBERG_EXPORT std::string ToString(const FileMetadata& file_metadata);
 
 }  // namespace iceberg::puffin

src/iceberg/puffin/json_serde_internal.h

@@ -27,30 +27,30 @@
 
 #include <nlohmann/json_fwd.hpp>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/puffin/type_fwd.h"
 #include "iceberg/result.h"
 
 namespace iceberg::puffin {
 
 /// \brief Serialize a BlobMetadata to JSON.
-ICEBERG_DATA_EXPORT nlohmann::json ToJson(const BlobMetadata& blob_metadata);
+ICEBERG_EXPORT nlohmann::json ToJson(const BlobMetadata& blob_metadata);
 
 /// \brief Deserialize a BlobMetadata from JSON.
-ICEBERG_DATA_EXPORT Result<BlobMetadata> BlobMetadataFromJson(const nlohmann::json& json);
+ICEBERG_EXPORT Result<BlobMetadata> BlobMetadataFromJson(const nlohmann::json& json);
 
 /// \brief Serialize a FileMetadata to JSON.
-ICEBERG_DATA_EXPORT nlohmann::json ToJson(const FileMetadata& file_metadata);
+ICEBERG_EXPORT nlohmann::json ToJson(const FileMetadata& file_metadata);
 
 /// \brief Deserialize a FileMetadata from JSON.
-ICEBERG_DATA_EXPORT Result<FileMetadata> FileMetadataFromJson(const nlohmann::json& json);
+ICEBERG_EXPORT Result<FileMetadata> FileMetadataFromJson(const nlohmann::json& json);
 
 /// \brief Serialize a FileMetadata to a JSON string.
-ICEBERG_DATA_EXPORT std::string ToJsonString(const FileMetadata& file_metadata,
-                                             bool pretty = false);
+ICEBERG_EXPORT std::string ToJsonString(const FileMetadata& file_metadata,
+                                        bool pretty = false);
 
 /// \brief Deserialize a FileMetadata from a JSON string.
-ICEBERG_DATA_EXPORT Result<FileMetadata> FileMetadataFromJsonString(
+ICEBERG_EXPORT Result<FileMetadata> FileMetadataFromJsonString(
     std::string_view json_string);
 
 }  // namespace iceberg::puffin

src/iceberg/puffin/puffin_format.h

@@ -28,14 +28,14 @@
 #include <span>
 #include <vector>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/puffin/file_metadata.h"
 #include "iceberg/result.h"
 
 namespace iceberg::puffin {
 
 /// \brief Puffin file format constants.
-struct ICEBERG_DATA_EXPORT PuffinFormat {
+struct ICEBERG_EXPORT PuffinFormat {
   /// Magic bytes: "PFA1" (Puffin Fratercula arctica, version 1)
   static constexpr std::array<uint8_t, 4> kMagicV1 = {0x50, 0x46, 0x41, 0x31};
 
@@ -63,17 +63,17 @@ enum class PuffinFlag : uint8_t {
 };
 
 /// \brief Check if a flag is set in the flags bytes.
-ICEBERG_DATA_EXPORT bool IsFlagSet(std::span<const uint8_t, 4> flags, PuffinFlag flag);
+ICEBERG_EXPORT bool IsFlagSet(std::span<const uint8_t, 4> flags, PuffinFlag flag);
 
 /// \brief Set a flag in the flags bytes.
-ICEBERG_DATA_EXPORT void SetFlag(std::span<uint8_t, 4> flags, PuffinFlag flag);
+ICEBERG_EXPORT void SetFlag(std::span<uint8_t, 4> flags, PuffinFlag flag);
 
 /// \brief Compress data using the specified codec.
-ICEBERG_DATA_EXPORT Result<std::vector<std::byte>> Compress(
-    PuffinCompressionCodec codec, std::span<const std::byte> input);
+ICEBERG_EXPORT Result<std::vector<std::byte>> Compress(PuffinCompressionCodec codec,
+                                                       std::span<const std::byte> input);
 
 /// \brief Decompress data using the specified codec.
-ICEBERG_DATA_EXPORT Result<std::vector<std::byte>> Decompress(
+ICEBERG_EXPORT Result<std::vector<std::byte>> Decompress(
     PuffinCompressionCodec codec, std::span<const std::byte> input);
 
 }  // namespace iceberg::puffin

src/iceberg/puffin/puffin_reader.h

@@ -29,7 +29,7 @@
 #include <utility>
 #include <vector>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/puffin/file_metadata.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -39,7 +39,7 @@ namespace iceberg::puffin {
 /// \brief Reader for Puffin files.
 ///
 /// Reads from an InputFile with seek support for efficient blob access.
-class ICEBERG_DATA_EXPORT PuffinReader {
+class ICEBERG_EXPORT PuffinReader {
  public:
   /// \brief Create a PuffinReader for the given input file.
   /// \param input_file The input file to read from.

src/iceberg/puffin/puffin_writer.cc

@@ -152,10 +152,10 @@ Status PuffinWriter::Finish() {
 
   ICEBERG_ASSIGN_OR_RAISE(auto end_pos, stream_->Position());
   footer_size_ = end_pos - footer_start;
+  file_size_ = end_pos;
   footer_written_ = true;
   ICEBERG_RETURN_UNEXPECTED(stream_->Flush());
   ICEBERG_RETURN_UNEXPECTED(stream_->Close());
-  ICEBERG_ASSIGN_OR_RAISE(file_size_, stream_->Position());
   finished_ = true;
   return {};
 }

src/iceberg/puffin/puffin_writer.h

@@ -30,7 +30,7 @@
 #include <unordered_map>
 #include <vector>
 
-#include "iceberg/iceberg_data_export.h"
+#include "iceberg/iceberg_export.h"
 #include "iceberg/puffin/file_metadata.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -40,7 +40,7 @@ namespace iceberg::puffin {
 /// \brief Writer for Puffin files.
 ///
 /// Writes blobs and footer to an OutputFile stream.
-class ICEBERG_DATA_EXPORT PuffinWriter {
+class ICEBERG_EXPORT PuffinWriter {
  public:
   /// \brief Create a PuffinWriter for the given output file.
   /// \param output_file The output file to write to.