feat(inspect): add files metadata table

src/iceberg/CMakeLists.txt

@@ -58,6 +58,8 @@ set(ICEBERG_SOURCES
     geospatial.cc
     incremental_append_scan.cc
     incremental_changelog_scan.cc
+    inspect/files_table.cc
+    inspect/files_table_util_internal.cc
     inspect/history_table.cc
     inspect/metadata_scan.cc
     inspect/metadata_table.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -80,6 +80,7 @@
 #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/metadata_scan.h"
 #include "iceberg/inspect/metadata_scan_task.h"

src/iceberg/inspect/files_table.cc

@@ -0,0 +1,94 @@
+/*
+ * 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 <mutex>
+#include <utility>
+
+#include "iceberg/inspect/files_table_util_internal.h"
+#include "iceberg/inspect/metadata_scan_util_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+FilesTable::FilesTable(std::shared_ptr<const Table> table)
+    : MetadataTable(std::move(table)) {}
+
+FilesTable::~FilesTable() = default;
+
+std::shared_ptr<Schema> FilesTable::schema() const {
+  std::lock_guard lock(schema_mutex_);
+  return schema_;
+}
+
+Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<const 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->SchemaForScan(*files->source_table()->metadata()));
+  return files;
+}
+
+Result<std::shared_ptr<Schema>> FilesTable::SchemaForScan(
+    const TableMetadata& metadata) const {
+  ICEBERG_ASSIGN_OR_RAISE(auto table_schema, metadata.Schema());
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(metadata));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema,
+                          internal::FilesTableSchema(*table_schema, partition_type));
+  {
+    std::lock_guard lock(schema_mutex_);
+    schema_ = schema;
+  }
+  return schema;
+}
+
+Result<std::unique_ptr<MetadataScan>> FilesTable::MakeScan(
+    internal::MetadataScanContext context) const {
+  return internal::MakeSnapshotMetadataScan(std::move(context), ReadRows);
+}
+
+Result<ArrowArrayStream> FilesTable::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 table_schema, context.table_metadata->Schema());
+  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));
+  auto schema = context.projected_schema;
+  const int32_t readable_metrics_id = context.schema->fields().back().field_id();
+  return internal::MakeMetadataBatchStream(
+      schema, std::move(files), batch_size,
+      [schema, readable_metrics_id, 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,
+                                        readable_metrics_id, file);
+      });
+}
+
+}  // namespace iceberg

src/iceberg/inspect/files_table.h

@@ -0,0 +1,67 @@
+/*
+ * 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 <mutex>
+
+#include "iceberg/arrow_c_data.h"
+#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 MetadataTable {
+ public:
+  static Result<std::unique_ptr<FilesTable>> Make(std::shared_ptr<const Table> table);
+
+  ~FilesTable() override;
+
+  Kind kind() const noexcept override { return Kind::kFiles; }
+
+  /// \brief Schema from construction or the most recent scan builder.
+  /// Existing scans and tasks retain their captured schema.
+  std::shared_ptr<Schema> schema() const override;
+
+ protected:
+  Result<std::shared_ptr<Schema>> SchemaForScan(
+      const TableMetadata& metadata) const override;
+
+  Result<std::unique_ptr<MetadataScan>> MakeScan(
+      internal::MetadataScanContext context) const override;
+
+ private:
+  static Result<ArrowArrayStream> ReadRows(const internal::MetadataScanContext& context,
+                                           const std::shared_ptr<FileIO>& io,
+                                           int64_t batch_size);
+
+  explicit FilesTable(std::shared_ptr<const Table> table);
+
+  mutable std::mutex schema_mutex_;
+  mutable std::shared_ptr<Schema> schema_;
+};
+
+}  // namespace iceberg