feat(inspect): add files, partitions, and manifests metadata tables

src/iceberg/CMakeLists.txt

@@ -56,8 +56,12 @@ set(ICEBERG_SOURCES
     geospatial.cc
     incremental_append_scan.cc
     incremental_changelog_scan.cc
+    inspect/files_table.cc
     inspect/history_table.cc
+    inspect/manifests_table.cc
     inspect/metadata_table.cc
+    inspect/metadata_table_util_internal.cc
+    inspect/partitions_table.cc
     inspect/snapshots_table.cc
     inheritable_metadata.cc
     json_serde.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -78,8 +78,11 @@
 #include "iceberg/incremental_append_scan.h"
 #include "iceberg/incremental_changelog_scan.h"
 #include "iceberg/inheritable_metadata.h"
+#include "iceberg/inspect/files_table.h"
 #include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/manifests_table.h"
 #include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/partitions_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/logging/cerr_logger.h"

src/iceberg/inspect/files_table.cc

@@ -0,0 +1,79 @@
+/*
+ * 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/files_table.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+FilesTable::FilesTable(std::shared_ptr<Table> table)
+    : TimeTravelMetadataTable(std::move(table)) {}
+
+FilesTable::~FilesTable() = default;
+
+const std::shared_ptr<Schema>& FilesTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<Table> table) {
+  ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+  auto files = std::unique_ptr<FilesTable>(new FilesTable(std::move(table)));
+  ICEBERG_RETURN_UNEXPECTED(files->RefreshSchema());
+  return files;
+}
+
+Status FilesTable::RefreshSchema() {
+  ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema());
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+                          internal::UnifiedPartitionType(*source_table()));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema,
+                          internal::FilesTableSchema(*table_schema, partition_type));
+  schema_ = std::move(schema);
+  table_schema_ = std::move(table_schema);
+  partition_type_ = std::move(partition_type);
+  return {};
+}
+
+Result<ArrowArrayStream> FilesTable::ScanSnapshot(
+    const SnapshotSelection& snapshot_selection) {
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+                                             *source_table(), snapshot_selection));
+  ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+  ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LiveFiles(*source_table(), snapshot));
+  auto schema = schema_;
+  auto table_schema = table_schema_;
+  auto partition_type = partition_type_;
+  return internal::MakeMetadataTableStream(
+      *schema_, std::move(files),
+      [schema = std::move(schema), table_schema = std::move(table_schema),
+       partition_type = std::move(partition_type)](ArrowRowBuilder& builder,
+                                                   const internal::LiveFile& file) {
+        return internal::AppendDataFile(builder, *schema, *table_schema, *partition_type,
+                                        file);
+      });
+}
+
+}  // namespace iceberg

src/iceberg/inspect/files_table.h

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/inspect/files_table.h
+/// \brief Define the files metadata table.
+
+#include <memory>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Metadata table containing live data and delete files in a snapshot.
+class ICEBERG_EXPORT FilesTable : public TimeTravelMetadataTable {
+ public:
+  static Result<std::unique_ptr<FilesTable>> Make(std::shared_ptr<Table> table);
+
+  ~FilesTable() override;
+
+  Kind kind() const noexcept override { return Kind::kFiles; }
+
+  /// \brief Schema from construction or the most recent scan.
+  /// A new scan rebuilds this schema from the source table's current metadata.
+  /// Previously returned streams retain their own schemas.
+  const std::shared_ptr<Schema>& schema() const override;
+
+ protected:
+  Result<ArrowArrayStream> ScanSnapshot(
+      const SnapshotSelection& snapshot_selection) override;
+
+ private:
+  explicit FilesTable(std::shared_ptr<Table> table);
+
+  Status RefreshSchema();
+
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<Schema> table_schema_;
+  std::shared_ptr<StructType> partition_type_;
+};
+
+}  // namespace iceberg

src/iceberg/inspect/manifests_table.cc

@@ -0,0 +1,219 @@
+/*
+ * 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/manifests_table.h"
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+struct ManifestRow {
+  ManifestFile manifest;
+  std::shared_ptr<PartitionSpec> spec;
+};
+
+Result<std::string> HumanReadableBound(const PartitionSpec& spec,
+                                       const StructType& partition_type, size_t index,
+                                       const std::vector<uint8_t>& bytes) {
+  ICEBERG_PRECHECK(index < spec.fields().size() && index < partition_type.fields().size(),
+                   "Partition summary index {} is out of range", index);
+  auto primitive = internal::checked_pointer_cast<PrimitiveType>(
+      partition_type.fields()[index].type());
+  ICEBERG_ASSIGN_OR_RAISE(auto literal,
+                          Conversions::FromBytes(std::move(primitive), bytes));
+  return spec.fields()[index].transform()->ToHumanString(literal);
+}
+
+Status AppendPartitionSummaries(ArrowArray* array, const ManifestRow& row,
+                                const std::shared_ptr<StructType>& partition_type) {
+  ICEBERG_PRECHECK(row.manifest.partitions.size() <= row.spec->fields().size(),
+                   "Manifest '{}' has more partition summaries than spec {} fields",
+                   row.manifest.manifest_path, row.spec->spec_id());
+  auto* entries = array->children[0];
+  ICEBERG_PRECHECK(entries != nullptr && entries->n_children == 4,
+                   "Partition summaries must contain four fields");
+
+  for (size_t index = 0; index < row.manifest.partitions.size(); ++index) {
+    const auto& summary = row.manifest.partitions[index];
+    ICEBERG_RETURN_UNEXPECTED(AppendBoolean(entries->children[0], summary.contains_null));
+    if (summary.contains_nan.has_value()) {
+      ICEBERG_RETURN_UNEXPECTED(
+          AppendBoolean(entries->children[1], *summary.contains_nan));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(AppendNull(entries->children[1]));
+    }
+
+    if (summary.lower_bound.has_value()) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto lower,
+          HumanReadableBound(*row.spec, *partition_type, index, *summary.lower_bound));
+      ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], lower));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], "null"));
+    }
+    if (summary.upper_bound.has_value()) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto upper,
+          HumanReadableBound(*row.spec, *partition_type, index, *summary.upper_bound));
+      ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], upper));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], "null"));
+    }
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(entries));
+  }
+  ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array));
+  return {};
+}
+
+Status AppendManifestCount(ArrowArray* array, const std::optional<int32_t>& count,
+                           bool applicable) {
+  if (!applicable) {
+    return AppendInt(array, 0);
+  }
+  return count.has_value() ? AppendInt(array, *count) : AppendNull(array);
+}
+
+Status AppendManifest(ArrowRowBuilder& builder, const ManifestRow& row,
+                      const std::shared_ptr<Schema>& table_schema) {
+  const auto& manifest = row.manifest;
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendInt(builder.column(0), static_cast<int64_t>(manifest.content)));
+  ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), manifest.manifest_path));
+  ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), manifest.manifest_length));
+  ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(3), manifest.partition_spec_id));
+  ICEBERG_PRECHECK(manifest.added_snapshot_id.has_value(),
+                   "Manifest '{}' has no added snapshot ID", manifest.manifest_path);
+  ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(4), *manifest.added_snapshot_id));
+
+  const bool data = manifest.content == ManifestContent::kData;
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(5), manifest.added_files_count, data));
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(6), manifest.existing_files_count, data));
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(7), manifest.deleted_files_count, data));
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(8), manifest.added_files_count, !data));
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(9), manifest.existing_files_count, !data));
+  ICEBERG_RETURN_UNEXPECTED(
+      AppendManifestCount(builder.column(10), manifest.deleted_files_count, !data));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_type, row.spec->PartitionType(*table_schema));
+  ICEBERG_RETURN_UNEXPECTED(AppendPartitionSummaries(
+      builder.column(11), row, std::shared_ptr<StructType>(std::move(partition_type))));
+  return builder.FinishRow();
+}
+
+}  // namespace
+
+ManifestsTable::ManifestsTable(std::shared_ptr<Table> table)
+    : TimeTravelMetadataTable(std::move(table)) {}
+
+ManifestsTable::~ManifestsTable() = default;
+
+const std::shared_ptr<Schema>& ManifestsTable::schema() const {
+  static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(14, "content", int32()),
+      SchemaField::MakeRequired(1, "path", string()),
+      SchemaField::MakeRequired(2, "length", int64()),
+      SchemaField::MakeRequired(3, "partition_spec_id", int32()),
+      SchemaField::MakeRequired(4, "added_snapshot_id", int64()),
+      // Legacy manifests can omit counts; unknown counts must stay null.
+      SchemaField::MakeOptional(5, "added_data_files_count", int32()),
+      SchemaField::MakeOptional(6, "existing_data_files_count", int32()),
+      SchemaField::MakeOptional(7, "deleted_data_files_count", int32()),
+      SchemaField::MakeOptional(15, "added_delete_files_count", int32()),
+      SchemaField::MakeOptional(16, "existing_delete_files_count", int32()),
+      SchemaField::MakeOptional(17, "deleted_delete_files_count", int32()),
+      SchemaField::MakeRequired(
+          8, "partition_summaries",
+          list(SchemaField::MakeRequired(
+              9, std::string(ListType::kElementName),
+              struct_({SchemaField::MakeRequired(10, "contains_null", boolean()),
+                       SchemaField::MakeOptional(11, "contains_nan", boolean()),
+                       SchemaField::MakeOptional(12, "lower_bound", string()),
+                       SchemaField::MakeOptional(13, "upper_bound", string())})))),
+  });
+  return schema;
+}
+
+Result<std::unique_ptr<ManifestsTable>> ManifestsTable::Make(
+    std::shared_ptr<Table> table) {
+  ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+  return std::unique_ptr<ManifestsTable>(new ManifestsTable(std::move(table)));
+}
+
+Result<ArrowArrayStream> ManifestsTable::ScanSnapshot(
+    const SnapshotSelection& snapshot_selection) {
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+                                             *source_table(), snapshot_selection));
+  std::vector<ManifestRow> rows;
+  if (snapshot != nullptr) {
+    ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, source_table()->specs());
+    SnapshotReader snapshot_reader(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests,
+                            snapshot_reader.Manifests(source_table()->io()));
+    rows.reserve(manifests.size());
+    for (const auto& manifest : manifests) {
+      auto spec = specs_ref.get().find(manifest.partition_spec_id);
+      ICEBERG_CHECK(spec != specs_ref.get().end(),
+                    "Cannot find partition spec {} for manifest '{}'",
+                    manifest.partition_spec_id, manifest.manifest_path);
+      ICEBERG_PRECHECK(spec->second != nullptr, "Partition spec {} is null",
+                       manifest.partition_spec_id);
+      rows.push_back(ManifestRow{.manifest = manifest, .spec = spec->second});
+    }
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema());
+  return internal::MakeMetadataTableStream(
+      *schema(), std::move(rows),
+      [table_schema = std::move(table_schema)](ArrowRowBuilder& builder,
+                                               const ManifestRow& row) {
+        return AppendManifest(builder, row, table_schema);
+      });
+}
+
+}  // namespace iceberg