feat(inspect): add partitions metadata table

src/iceberg/CMakeLists.txt

@@ -58,6 +58,7 @@ set(ICEBERG_SOURCES
     geospatial.cc
     incremental_append_scan.cc
     incremental_changelog_scan.cc
+    inspect/partitions_table.cc
     inspect/history_table.cc
     inspect/metadata_scan.cc
     inspect/metadata_table.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -85,6 +85,7 @@
 #include "iceberg/inspect/metadata_scan_task.h"
 #include "iceberg/inspect/metadata_scan_task_reader.h"
 #include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/partitions_table.h"
 #include "iceberg/inspect/refs_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/labels.h"

src/iceberg/inspect/metadata_table.cc

@@ -24,6 +24,7 @@
 
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/partitions_table.h"
 #include "iceberg/inspect/refs_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 
@@ -43,6 +44,8 @@ Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(
       return RefsTable::Make(std::move(table));
     case MetadataTableKind::kHistory:
       return HistoryTable::Make(std::move(table));
+    case MetadataTableKind::kPartitions:
+      return PartitionsTable::Make(std::move(table));
     default:
       break;
   }

src/iceberg/inspect/partitions_table.cc

@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_manifest_util_internal.h"
+#include "iceberg/inspect/metadata_partition_util_internal.h"
+#include "iceberg/inspect/metadata_scan_util_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+  PartitionValues values;
+  size_t projected_fields;
+
+  bool operator==(const PartitionKey& other) const {
+    if (projected_fields != other.projected_fields ||
+        values.num_fields() != other.values.num_fields()) {
+      return false;
+    }
+    for (size_t index = 0; index < values.num_fields(); ++index) {
+      const auto& lhs = values.values()[index];
+      const auto& rhs = other.values.values()[index];
+      if (lhs.IsNull() || rhs.IsNull()) {
+        if (lhs.IsNull() != rhs.IsNull()) {
+          return false;
+        }
+      } else if (lhs.IsNaN() && rhs.IsNaN()) {
+        // Java partition equality canonicalizes all NaN signs and payloads.
+        continue;
+      } else if (lhs != rhs) {
+        return false;
+      }
+    }
+    return true;
+  }
+};
+
+struct PartitionKeyHash {
+  size_t operator()(const PartitionKey& key) const noexcept {
+    size_t result = 17;
+    for (const auto& value : key.values.values()) {
+      const size_t value_hash =
+          value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+      result = result * 37 + value_hash;
+    }
+    return result * 37 + key.projected_fields;
+  }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+  size_t count = 0;
+  for (const auto& field : partition_type.fields()) {
+    count += std::ranges::any_of(
+        spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+          return spec_field.field_id() == field_id;
+        });
+  }
+  return count;
+}
+
+struct PartitionStats {
+  explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+  PartitionValues partition;
+  int32_t spec_id = PartitionSpec::kInitialSpecId;
+  int64_t data_record_count = 0;
+  int32_t data_file_count = 0;
+  int64_t data_file_size = 0;
+  int64_t position_delete_record_count = 0;
+  int32_t position_delete_file_count = 0;
+  int64_t equality_delete_record_count = 0;
+  int32_t equality_delete_file_count = 0;
+  std::optional<TimePointMs> last_updated_at;
+  std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+                       const StructType& partition_type,
+                       const PartitionStats& partition) {
+  for (size_t index = 0; index < schema.fields().size(); ++index) {
+    auto* array = builder.column(index);
+    switch (schema.fields()[index].field_id()) {
+      case kPartitionFieldId:
+        ICEBERG_RETURN_UNEXPECTED(
+            internal::AppendPartitionValues(array, partition_type, partition.partition));
+        break;
+      case kSpecIdFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+        break;
+      case kRecordCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+        break;
+      case kFileCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+        break;
+      case kTotalDataFileSizeFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+        break;
+      case kPositionDeleteRecordCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(
+            AppendInt(array, partition.position_delete_record_count));
+        break;
+      case kPositionDeleteFileCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+        break;
+      case kEqualityDeleteRecordCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(
+            AppendInt(array, partition.equality_delete_record_count));
+        break;
+      case kEqualityDeleteFileCountFieldId:
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+        break;
+      case kLastUpdatedAtFieldId:
+        if (partition.last_updated_at.has_value()) {
+          ICEBERG_RETURN_UNEXPECTED(
+              AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+                                   partition.last_updated_at->time_since_epoch())
+                                   .count()));
+        } else {
+          ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+        }
+        break;
+      case kLastUpdatedSnapshotIdFieldId:
+        if (partition.last_updated_snapshot_id.has_value()) {
+          ICEBERG_RETURN_UNEXPECTED(
+              AppendInt(array, *partition.last_updated_snapshot_id));
+        } else {
+          ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+        }
+        break;
+      default:
+        return InvalidSchema("Unsupported partitions metadata field {}",
+                             schema.fields()[index].field_id());
+    }
+  }
+  return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+  switch (file.content) {
+    case DataFile::Content::kData:
+      partition.data_record_count += file.record_count;
+      ++partition.data_file_count;
+      partition.data_file_size += file.file_size_in_bytes;
+      break;
+    case DataFile::Content::kPositionDeletes:
+      partition.position_delete_record_count += file.record_count;
+      ++partition.position_delete_file_count;
+      break;
+    case DataFile::Content::kEqualityDeletes:
+      partition.equality_delete_record_count += file.record_count;
+      ++partition.equality_delete_file_count;
+      break;
+  }
+}
+
+}  // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<const Table> table)
+    : MetadataTable(std::move(table)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+std::shared_ptr<Schema> PartitionsTable::schema() const {
+  std::lock_guard lock(schema_mutex_);
+  return schema_;
+}
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+    std::shared_ptr<const Table> table) {
+  ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+  auto partitions =
+      std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
+  ICEBERG_RETURN_UNEXPECTED(
+      partitions->SchemaForScan(*partitions->source_table()->metadata()));
+  return partitions;
+}
+
+Result<std::shared_ptr<Schema>> PartitionsTable::SchemaForScan(
+    const TableMetadata& metadata) const {
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(metadata));
+
+  std::vector<SchemaField> fields;
+  if (!partition_type->fields().empty()) {
+    fields.push_back(
+        SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
+    fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32()));
+  }
+  fields.push_back(
+      SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64()));
+  fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32()));
+  fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId,
+                                             "total_data_file_size_in_bytes", int64()));
+  fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId,
+                                             "position_delete_record_count", int64()));
+  fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId,
+                                             "position_delete_file_count", int32()));
+  fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId,
+                                             "equality_delete_record_count", int64()));
+  fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId,
+                                             "equality_delete_file_count", int32()));
+  fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at",
+                                             timestamp_tz()));
+  fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
+                                             "last_updated_snapshot_id", int64()));
+
+  auto schema = std::make_shared<Schema>(std::move(fields));
+  {
+    std::lock_guard lock(schema_mutex_);
+    schema_ = schema;
+  }
+  return schema;
+}
+
+Result<std::unique_ptr<MetadataScan>> PartitionsTable::MakeScan(
+    internal::MetadataScanContext context) const {
+  return internal::MakeSnapshotMetadataScan(std::move(context), ReadRows);
+}
+
+Result<ArrowArrayStream> PartitionsTable::ReadRows(
+    const internal::MetadataScanContext& context, const std::shared_ptr<FileIO>& io,
+    int64_t batch_size) {
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+                          internal::UnifiedPartitionType(*context.table_metadata));
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+                                             *context.table_metadata, context));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto files, internal::LiveFiles(
+                      *context.table_metadata, io, snapshot,
+                      {"content", "partition", "record_count", "file_size_in_bytes"}));
+
+  std::vector<PartitionStats> partitions;
+  std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;
+  std::unordered_map<int64_t, std::shared_ptr<Snapshot>> snapshots;
+  while (true) {
+    ICEBERG_ASSIGN_OR_RAISE(auto next_file, files->Next());
+    if (!next_file.has_value()) {
+      break;
+    }
+    const auto& live_file = *next_file;
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_values, internal::ProjectPartitionValues(
+                                                       *partition_type, *live_file.spec,
+                                                       live_file.file->partition));
+    PartitionKey key{
+        .values = std::move(partition_values),
+        .projected_fields = ProjectedFieldCount(*partition_type, *live_file.spec)};
+    auto [position, inserted] = positions.try_emplace(key, partitions.size());
+    if (inserted) {
+      partitions.emplace_back(std::move(key.values));
+    }
+    auto& partition = partitions[position->second];
+    UpdateCounts(partition, *live_file.file);
+
+    if (live_file.snapshot_id.has_value()) {
+      auto snapshot_iter = snapshots.find(*live_file.snapshot_id);
+      if (snapshot_iter == snapshots.end()) {
+        auto file_snapshot = context.table_metadata->SnapshotById(*live_file.snapshot_id);
+        if (!file_snapshot.has_value() &&
+            file_snapshot.error().kind != ErrorKind::kNotFound) {
+          return unexpected<Error>(file_snapshot.error());
+        }
+        snapshot_iter =
+            snapshots.emplace(*live_file.snapshot_id, file_snapshot.value_or(nullptr))
+                .first;
+      }
+      const auto& file_snapshot = snapshot_iter->second;
+      if (file_snapshot != nullptr &&
+          (!partition.last_updated_at.has_value() ||
+           file_snapshot->timestamp_ms > *partition.last_updated_at)) {
+        partition.spec_id = live_file.spec->spec_id();
+        partition.last_updated_at = file_snapshot->timestamp_ms;
+        partition.last_updated_snapshot_id = file_snapshot->snapshot_id;
+      }
+    }
+  }
+
+  auto schema = context.projected_schema;
+  return internal::MakeMetadataBatchStream(
+      schema, std::move(partitions), batch_size,
+      [schema, partition_type = std::move(partition_type)](
+          ArrowRowBuilder& builder, const PartitionStats& partition) {
+        return AppendPartition(builder, *schema, *partition_type, partition);
+      });
+}
+
+}  // namespace iceberg