feat: add roaring-based position bitmap

Author: wgtmac

src/iceberg/CMakeLists.txt

@@ -25,6 +25,7 @@ set(ICEBERG_SOURCES
     data/position_delete_writer.cc
     data/writer.cc
     delete_file_index.cc
+    deletes/roaring_position_bitmap.cc
     expression/aggregate.cc
     expression/binder.cc
     expression/evaluator.cc
@@ -165,6 +166,7 @@ iceberg_install_all_headers(iceberg)
 
 add_subdirectory(catalog)
 add_subdirectory(data)
+add_subdirectory(deletes)
 add_subdirectory(expression)
 add_subdirectory(manifest)
 add_subdirectory(row)

src/iceberg/deletes/CMakeLists.txt

@@ -0,0 +1,18 @@
+# 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.
+
+iceberg_install_all_headers(iceberg/deletes)

src/iceberg/deletes/meson.build

@@ -0,0 +1,18 @@
+# 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.
+
+install_headers(['roaring_position_bitmap.h'], subdir: 'iceberg/deletes')

src/iceberg/deletes/roaring_position_bitmap.cc

@@ -0,0 +1,276 @@
+/*
+ * 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/deletes/roaring_position_bitmap.h"
+
+#include <cstdint>
+#include <exception>
+#include <limits>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "roaring/roaring.hh"
+
+namespace iceberg {
+
+namespace {
+
+constexpr size_t kBitmapCountSizeBytes = 8;
+constexpr size_t kBitmapKeySizeBytes = 4;
+
+// Extracts high 32 bits from a 64-bit position (the key).
+int32_t Key(int64_t pos) { return static_cast<int32_t>(pos >> 32); }
+
+// Extracts low 32 bits from a 64-bit position.
+uint32_t Pos32Bits(int64_t pos) { return static_cast<uint32_t>(pos); }
+
+// Combines key (high 32 bits) and pos32 (low 32 bits) into a 64-bit
+// position. The low 32 bits are zero-extended to avoid sign extension.
+int64_t ToPosition(int32_t key, uint32_t pos32) {
+  return (static_cast<int64_t>(key) << 32) | static_cast<int64_t>(pos32);
+}
+
+void WriteLE64(char* buf, int64_t value) {
+  auto v = static_cast<uint64_t>(value);
+  for (int i = 0; i < 8; ++i) {
+    buf[i] = static_cast<char>((v >> (i * 8)) & 0xFF);
+  }
+}
+
+void WriteLE32(char* buf, int32_t value) {
+  auto v = static_cast<uint32_t>(value);
+  for (int i = 0; i < 4; ++i) {
+    buf[i] = static_cast<char>((v >> (i * 8)) & 0xFF);
+  }
+}
+
+int64_t ReadLE64(const char* buf) {
+  auto b = reinterpret_cast<const uint8_t*>(buf);
+  uint64_t v = 0;
+  for (int i = 0; i < 8; ++i) {
+    v |= static_cast<uint64_t>(b[i]) << (i * 8);
+  }
+  return static_cast<int64_t>(v);
+}
+
+int32_t ReadLE32(const char* buf) {
+  auto b = reinterpret_cast<const uint8_t*>(buf);
+  uint32_t v = 0;
+  for (int i = 0; i < 4; ++i) {
+    v |= static_cast<uint32_t>(b[i]) << (i * 8);
+  }
+  return static_cast<int32_t>(v);
+}
+
+Status ValidatePosition(int64_t pos) {
+  if (pos < 0 || pos > RoaringPositionBitmap::kMaxPosition) {
+    return InvalidArgument("Bitmap supports positions that are >= 0 and <= {}: {}",
+                           RoaringPositionBitmap::kMaxPosition, pos);
+  }
+  return {};
+}
+
+}  // namespace
+
+struct RoaringPositionBitmap::Impl {
+  std::vector<roaring::Roaring> bitmaps;
+
+  void AllocateBitmapsIfNeeded(int32_t required_length) {
+    if (static_cast<int32_t>(bitmaps.size()) < required_length) {
+      bitmaps.resize(static_cast<size_t>(required_length));
+    }
+  }
+};
+
+RoaringPositionBitmap::RoaringPositionBitmap() : impl_(std::make_unique<Impl>()) {}
+
+RoaringPositionBitmap::~RoaringPositionBitmap() = default;
+
+RoaringPositionBitmap::RoaringPositionBitmap(RoaringPositionBitmap&&) noexcept = default;
+
+RoaringPositionBitmap& RoaringPositionBitmap::operator=(
+    RoaringPositionBitmap&&) noexcept = default;
+
+RoaringPositionBitmap::RoaringPositionBitmap(std::unique_ptr<Impl> impl)
+    : impl_(std::move(impl)) {}
+
+Status RoaringPositionBitmap::Add(int64_t pos) {
+  if (auto status = ValidatePosition(pos); !status) {
+    return status;
+  }
+  int32_t key = Key(pos);
+  uint32_t pos32 = Pos32Bits(pos);
+  impl_->AllocateBitmapsIfNeeded(key + 1);
+  impl_->bitmaps[key].add(pos32);
+  return {};
+}
+
+Status RoaringPositionBitmap::AddRange(int64_t pos_start, int64_t pos_end) {
+  for (int64_t pos = pos_start; pos < pos_end; ++pos) {
+    if (auto status = Add(pos); !status) {
+      return status;
+    }
+  }
+  return {};
+}
+
+Result<bool> RoaringPositionBitmap::Contains(int64_t pos) const {
+  if (auto status = ValidatePosition(pos); !status) {
+    return std::unexpected(status.error());
+  }
+  int32_t key = Key(pos);
+  uint32_t pos32 = Pos32Bits(pos);
+  return key < static_cast<int32_t>(impl_->bitmaps.size()) &&
+         impl_->bitmaps[key].contains(pos32);
+}
+
+bool RoaringPositionBitmap::IsEmpty() const { return Cardinality() == 0; }
+
+int64_t RoaringPositionBitmap::Cardinality() const {
+  int64_t total = 0;
+  for (const auto& bitmap : impl_->bitmaps) {
+    total += static_cast<int64_t>(bitmap.cardinality());
+  }
+  return total;
+}
+
+void RoaringPositionBitmap::Or(const RoaringPositionBitmap& other) {
+  impl_->AllocateBitmapsIfNeeded(static_cast<int32_t>(other.impl_->bitmaps.size()));
+  for (size_t key = 0; key < other.impl_->bitmaps.size(); ++key) {
+    impl_->bitmaps[key] |= other.impl_->bitmaps[key];
+  }
+}
+
+bool RoaringPositionBitmap::RunLengthEncode() {
+  bool changed = false;
+  for (auto& bitmap : impl_->bitmaps) {
+    changed |= bitmap.runOptimize();
+  }
+  return changed;
+}
+
+void RoaringPositionBitmap::ForEach(const std::function<void(int64_t)>& fn) const {
+  for (size_t key = 0; key < impl_->bitmaps.size(); ++key) {
+    for (uint32_t pos32 : impl_->bitmaps[key]) {
+      fn(ToPosition(static_cast<int32_t>(key), pos32));
+    }
+  }
+}
+
+int64_t RoaringPositionBitmap::SerializedSizeInBytes() const {
+  int64_t size = static_cast<int64_t>(kBitmapCountSizeBytes);
+  for (const auto& bitmap : impl_->bitmaps) {
+    size += static_cast<int64_t>(kBitmapKeySizeBytes) +
+            static_cast<int64_t>(bitmap.getSizeInBytes(/*portable=*/true));
+  }
+  return size;
+}
+
+Result<std::string> RoaringPositionBitmap::Serialize() const {
+  int64_t size = SerializedSizeInBytes();
+  std::string result(static_cast<size_t>(size), '\0');
+  char* buf = result.data();
+
+  // Write bitmap count (array length including empties)
+  WriteLE64(buf, static_cast<int64_t>(impl_->bitmaps.size()));
+  buf += kBitmapCountSizeBytes;
+
+  // Write each bitmap with its key
+  for (int32_t key = 0; key < static_cast<int32_t>(impl_->bitmaps.size()); ++key) {
+    WriteLE32(buf, key);
+    buf += kBitmapKeySizeBytes;
+    size_t written = impl_->bitmaps[key].write(buf, /*portable=*/true);
+    buf += written;
+  }
+
+  return result;
+}
+
+Result<RoaringPositionBitmap> RoaringPositionBitmap::Deserialize(std::string_view bytes) {
+  const char* buf = bytes.data();
+  size_t remaining = bytes.size();
+
+  if (remaining < kBitmapCountSizeBytes) {
+    return InvalidArgument("Buffer too small for bitmap count");
+  }
+
+  int64_t bitmap_count = ReadLE64(buf);
+  buf += kBitmapCountSizeBytes;
+  remaining -= kBitmapCountSizeBytes;
+
+  if (bitmap_count < 0 || bitmap_count > std::numeric_limits<int32_t>::max()) {
+    return InvalidArgument("Invalid bitmap count: {}", bitmap_count);
+  }
+
+  auto impl = std::make_unique<Impl>();
+  int32_t last_key = -1;
+  int32_t remaining_count = static_cast<int32_t>(bitmap_count);
+
+  while (remaining_count > 0) {
+    if (remaining < kBitmapKeySizeBytes) {
+      return InvalidArgument("Buffer too small for bitmap key");
+    }
+
+    int32_t key = ReadLE32(buf);
+    buf += kBitmapKeySizeBytes;
+    remaining -= kBitmapKeySizeBytes;
+
+    // Validate key (matches Java's readKey)
+    if (key < 0) {
+      return InvalidArgument("Invalid unsigned key: {}", key);
+    }
+    if (key > std::numeric_limits<int32_t>::max() - 1) {
+      return InvalidArgument("Key is too large: {}", key);
+    }
+    if (key <= last_key) {
+      return InvalidArgument("Keys must be sorted in ascending order");
+    }
+
+    // Fill gaps with empty bitmaps
+    while (last_key < key - 1) {
+      impl->bitmaps.emplace_back();
+      ++last_key;
+    }
+
+    // Read bitmap using portable safe deserialization.
+    // CRoaring's readSafe may throw on corrupted data.
+    roaring::Roaring bitmap;
+    try {
+      bitmap = roaring::Roaring::readSafe(buf, remaining);
+    } catch (const std::exception& e) {
+      return InvalidArgument("Failed to deserialize bitmap at key {}: {}", key, e.what());
+    }
+    size_t bitmap_size = bitmap.getSizeInBytes(/*portable=*/true);
+    if (bitmap_size > remaining) {
+      return InvalidArgument("Buffer too small for bitmap data at key {}", key);
+    }
+    buf += bitmap_size;
+    remaining -= bitmap_size;
+
+    impl->bitmaps.push_back(std::move(bitmap));
+    last_key = key;
+    --remaining_count;
+  }
+
+  return RoaringPositionBitmap(std::move(impl));
+}
+
+}  // namespace iceberg