From 3b1191161102819ad2896f2b87efb098bcf0e0a9 Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Thu, 2 Jul 2026 15:09:32 +0800 Subject: [PATCH 1/8] feat(inspect): implement SnapshotsTable scanning - Add Scan() virtual method and Scan() convenience overload to MetadataTable - Add SnapshotSelection struct for time-travel snapshot resolution - Add supports_time_travel() concrete method driven by kind() - Implement SnapshotsTable::Scan() to materialize snapshot rows via ArrowRowBuilder --- src/iceberg/inspect/metadata_table.cc | 17 ++ src/iceberg/inspect/metadata_table.h | 36 +++++ src/iceberg/inspect/snapshots_table.cc | 45 ++++++ src/iceberg/inspect/snapshots_table.h | 7 + src/iceberg/test/CMakeLists.txt | 7 +- src/iceberg/test/history_table_test.cc | 54 +++++++ src/iceberg/test/metadata_table_test.cc | 80 ++-------- src/iceberg/test/metadata_table_test_base.h | 161 +++++++++++++++++++ src/iceberg/test/snapshots_table_test.cc | 162 ++++++++++++++++++++ 9 files changed, 504 insertions(+), 65 deletions(-) create mode 100644 src/iceberg/test/history_table_test.cc create mode 100644 src/iceberg/test/metadata_table_test_base.h create mode 100644 src/iceberg/test/snapshots_table_test.cc diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 5e9504003..4298e3653 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -35,6 +35,23 @@ MetadataTable::MetadataTable(std::shared_ptr source_table, MetadataTable::~MetadataTable() = default; +bool MetadataTable::supports_time_travel() const noexcept { + // Time travel is supported for tables that read from a single snapshot's + // manifests. Tables that scan all snapshots or return in-memory history do + // not. + switch (kind()) { + case Kind::kSnapshots: + case Kind::kHistory: + return false; + } + return false; +} + +Result MetadataTable::Scan( + std::optional /*snapshot_selection*/) { + return NotSupported("Scan is not supported for this metadata table type"); +} + Result> MetadataTable::Make(std::shared_ptr
table, Kind kind) { if (table == nullptr) [[unlikely]] { diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 51c5f7920..320a95b17 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -23,14 +23,28 @@ /// \brief Define base APIs for metadata tables. #include +#include +#include +#include "iceberg/arrow_c_data.h" #include "iceberg/iceberg_export.h" #include "iceberg/result.h" #include "iceberg/table_identifier.h" #include "iceberg/type_fwd.h" +#include "iceberg/util/timepoint.h" namespace iceberg { +/// \brief Parameters for snapshot selection (time travel). +struct SnapshotSelection { + /// \brief The snapshot ID to read. + std::optional snapshot_id; + /// \brief Read the snapshot that was current at this timestamp. + std::optional as_of_timestamp; + /// \brief Read the snapshot referenced by this named ref (branch or tag). + std::optional ref_name; +}; + /// \brief Base class for Iceberg metadata tables. class ICEBERG_EXPORT MetadataTable { public: @@ -46,6 +60,28 @@ class ICEBERG_EXPORT MetadataTable { virtual Kind kind() const noexcept = 0; + /// \brief Whether this metadata table supports time-travel queries. + /// + /// Time travel is supported for tables that read from a single snapshot's + /// manifests (e.g., Entries, Files, Manifests, Partitions). Tables that + /// scan all snapshots (All*) or return in-memory history (Snapshots, + /// History, Refs) do not support time travel. + bool supports_time_travel() const noexcept; + + /// \brief Scan the metadata table using the current snapshot. + /// + /// Convenience overload — delegates to Scan(std::nullopt). + Result Scan() { return Scan(std::nullopt); } + + /// \brief Scan the metadata table and return all rows as an Arrow struct array. + /// + /// The returned ArrowArray is a struct array where each element is one row. + /// The caller takes ownership and must call ArrowArrayRelease when done. + /// + /// The default implementation returns NotSupported. Subclasses override this + /// to materialize their data. + virtual Result Scan(std::optional snapshot_selection); + const TableIdentifier& name() const { return identifier_; } const std::shared_ptr& schema() const { return schema_; } diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 4b0c3ce9f..717d953d4 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -19,15 +19,18 @@ #include "iceberg/inspect/snapshots_table.h" +#include #include #include #include +#include "iceberg/arrow_row_builder_internal.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" #include "iceberg/table_identifier.h" #include "iceberg/type.h" +#include "iceberg/util/macros.h" namespace iceberg { namespace { @@ -65,4 +68,46 @@ Result> SnapshotsTable::Make( return std::unique_ptr(new SnapshotsTable(std::move(table))); } +Result SnapshotsTable::Scan( + std::optional /*snapshot_selection*/) { + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema())); + + for (const auto& snapshot : source_table()->snapshots()) { + // column 0: committed_at (timestamptz -> int64 micros) + ICEBERG_RETURN_UNEXPECTED(AppendInt( + builder.column(0), std::chrono::duration_cast( + snapshot->timestamp_ms.time_since_epoch()) + .count())); + + // column 1: snapshot_id (long) + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot->snapshot_id)); + + // column 2: parent_id (long, optional) + if (snapshot->parent_snapshot_id.has_value()) { + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(2), *snapshot->parent_snapshot_id)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + } + + // column 3: operation (string, optional) + auto op = snapshot->Operation(); + if (op.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *op)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + } + + // column 4: manifest_list (string, optional) + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot->manifest_list)); + + // column 5: summary (map) + ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot->summary)); + + ICEBERG_RETURN_UNEXPECTED(builder.FinishRow()); + } + + return std::move(builder).Finish(); +} + } // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h index 9af1bcacb..ef5c0dfc4 100644 --- a/src/iceberg/inspect/snapshots_table.h +++ b/src/iceberg/inspect/snapshots_table.h @@ -40,6 +40,13 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable { Kind kind() const noexcept override { return Kind::kSnapshots; } + /// \brief Scan all snapshots as rows. + /// + /// The snapshots table always returns every known snapshot, so the + /// snapshot_selection parameter is ignored. + Result Scan( + std::optional /*snapshot_selection*/) override; + private: explicit SnapshotsTable(std::shared_ptr
table); }; diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 98129a6d2..6a02bd691 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -186,7 +186,12 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(catalog_test USE_BUNDLE SOURCES in_memory_catalog_test.cc) - add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc) + add_iceberg_test(metadata_table_test + USE_BUNDLE + SOURCES + history_table_test.cc + metadata_table_test.cc + snapshots_table_test.cc) add_iceberg_test(eval_expr_test USE_BUNDLE diff --git a/src/iceberg/test/history_table_test.cc b/src/iceberg/test/history_table_test.cc new file mode 100644 index 000000000..b27bdef30 --- /dev/null +++ b/src/iceberg/test/history_table_test.cc @@ -0,0 +1,54 @@ +/* + * 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. + */ + +/// \file history_table_test.cc +/// Unit tests for HistoryTable. + +#include +#include + +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +std::shared_ptr MakeHistorySchema() { + return std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "parent_id", int64()), + SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); +} + +} // namespace + +class HistoryTableTest : public MetadataTableTestBase {}; + +TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) { + ICEBERG_UNWRAP_OR_FAIL(auto history_table, + MetadataTable::Make(table_, MetadataTable::Kind::kHistory)); + EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema()); +} + +} // namespace iceberg diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc index 1e0a664c3..fd59af3b3 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -33,88 +33,40 @@ #include "iceberg/type.h" namespace iceberg { -namespace { - -std::shared_ptr MakeSnapshotsSchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeOptional(4, "operation", string()), - SchemaField::MakeOptional(5, "manifest_list", string()), - SchemaField::MakeOptional( - 6, "summary", - std::make_shared(SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); -} - -std::shared_ptr MakeHistorySchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); -} - -} // namespace class MetadataTableTest : public ::testing::Test { protected: void SetUp() override { - io_ = std::make_shared(); - catalog_ = std::make_shared(); - auto schema = std::make_shared( std::vector{SchemaField::MakeRequired(1, "id", int64()), SchemaField::MakeOptional(2, "name", string())}, 1); - metadata_ = std::make_shared( + auto metadata = std::make_shared( TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1}); - TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}}, - .name = "source_table"}; - auto source_table_result = - Table::Make(source_ident, metadata_, "s3://bucket/meta.json", io_, catalog_); - EXPECT_THAT(source_table_result, IsOk()); - source_table_ = *source_table_result; - - auto snapshots_table_result = - MetadataTable::Make(source_table_, MetadataTable::Kind::kSnapshots); - EXPECT_THAT(snapshots_table_result, IsOk()); - snapshots_table_ = std::move(*snapshots_table_result); + TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"}; + ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata, "s3://bucket/meta.json", + std::make_shared(), + std::make_shared())); } - std::shared_ptr io_; - std::shared_ptr catalog_; - std::shared_ptr metadata_; - std::shared_ptr
source_table_; - std::unique_ptr snapshots_table_; + std::shared_ptr
table_; }; -TEST_F(MetadataTableTest, Constructor) { - EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); - EXPECT_EQ(snapshots_table_->source_table(), source_table_); - EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots"); - EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector{"db"})); - EXPECT_NE(snapshots_table_->schema(), nullptr); -} - -TEST_F(MetadataTableTest, SnapshotsSchemaMatchesIcebergSchema) { - EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema()); -} - -TEST_F(MetadataTableTest, HistorySchemaMatchesIcebergSchema) { - auto history_table_result = - MetadataTable::Make(source_table_, MetadataTable::Kind::kHistory); - ASSERT_THAT(history_table_result, IsOk()); - - EXPECT_TRUE(*(*history_table_result)->schema() == *MakeHistorySchema()); -} - TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) { auto result = MetadataTable::Make(nullptr, MetadataTable::Kind::kSnapshots); EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); EXPECT_THAT(result, HasErrorMessage("Table cannot be null")); } +TEST_F(MetadataTableTest, SupportsTimeTravel) { + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots)); + EXPECT_FALSE(snapshots_table->supports_time_travel()); + + ICEBERG_UNWRAP_OR_FAIL(auto history_table, + MetadataTable::Make(table_, MetadataTable::Kind::kHistory)); + EXPECT_FALSE(history_table->supports_time_travel()); +} + } // namespace iceberg diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h new file mode 100644 index 000000000..bd75cf083 --- /dev/null +++ b/src/iceberg/test/metadata_table_test_base.h @@ -0,0 +1,161 @@ +/* + * 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. + */ + +/// \file metadata_table_test_base.h +/// Shared test base for all metadata table tests. +/// +/// Provides common helpers (FinishAndImport, MakeTestSnapshots, +/// MakeTableWithSnapshots) and the MockFileIO + MockCatalog fixture that +/// every metadata table test needs. + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/schema_internal.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/type.h" +#include "iceberg/util/timepoint.h" + +namespace iceberg { + +/// \brief Base class for all metadata table tests. +/// +/// Provides MockFileIO and MockCatalog instances plus helpers shared across +/// metadata table tests (SnapshotsTable, HistoryTable, RefsTable, ...). +class MetadataTableTestBase : public ::testing::Test { + protected: + void SetUp() override { + io_ = std::make_shared(); + catalog_ = std::make_shared(); + + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeOptional(2, "name", string())}, + 1); + metadata_ = std::make_shared( + TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1}); + + TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}}, + .name = "source_table"}; + ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(source_ident, metadata_, + "s3://bucket/meta.json", io_, catalog_)); + } + + /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. + static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray array, + const Schema& schema) { + ArrowSchema c_schema; + EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); + auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + + // ImportRecordBatch takes ownership of the array and releases it. + return ::arrow::ImportRecordBatch(&array, arrow_schema).ValueOrDie(); + } + + /// \brief Create two snapshots matching the Java TestDataTaskParser test data. + /// + /// Snapshot 1: id=1, no parent, timestamp=1234567890000, operation="append" + /// Snapshot 2: id=2, parent=1, timestamp=9876543210000, operation="append" + static std::pair, std::shared_ptr> + MakeTestSnapshots() { + std::unordered_map summary1{ + {"added-data-files", "1"}, {"added-records", "1"}, + {"added-files-size", "10"}, {"changed-partition-count", "1"}, + {"total-records", "1"}, {"total-files-size", "10"}, + {"total-data-files", "1"}, {"total-delete-files", "0"}, + {"total-position-deletes", "0"}, {"total-equality-deletes", "0"}, + {"operation", "append"}, + }; + + std::unordered_map summary2{ + {"added-data-files", "1"}, {"added-records", "1"}, + {"added-files-size", "10"}, {"changed-partition-count", "1"}, + {"total-records", "2"}, {"total-files-size", "20"}, + {"total-data-files", "2"}, {"total-delete-files", "0"}, + {"total-position-deletes", "0"}, {"total-equality-deletes", "0"}, + {"operation", "append"}, + }; + + auto snap1 = std::make_shared(Snapshot{ + .snapshot_id = 1, + .parent_snapshot_id = std::nullopt, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1234567890000), + .manifest_list = "file:/tmp/manifest1.avro", + .summary = std::move(summary1), + .schema_id = 1, + }); + + auto snap2 = std::make_shared(Snapshot{ + .snapshot_id = 2, + .parent_snapshot_id = 1, + .sequence_number = 2, + .timestamp_ms = TimePointMsFromUnixMs(9876543210000), + .manifest_list = "file:/tmp/manifest2.avro", + .summary = std::move(summary2), + .schema_id = 1, + }); + + return {snap1, snap2}; + } + + /// \brief Create a Table with the given snapshots. + Result> MakeTableWithSnapshots( + std::vector> snapshots, int64_t current_snapshot_id) { + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeOptional(2, "name", string())}, + 1); + auto metadata = std::make_shared(TableMetadata{ + .format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = current_snapshot_id, + .snapshots = std::move(snapshots), + }); + + TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "test_table"}; + return Table::Make(ident, metadata, "s3://bucket/meta.json", io_, catalog_); + } + + std::shared_ptr io_; + std::shared_ptr catalog_; + std::shared_ptr metadata_; + std::shared_ptr
table_; +}; + +} // namespace iceberg diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc new file mode 100644 index 000000000..86b8af3ec --- /dev/null +++ b/src/iceberg/test/snapshots_table_test.cc @@ -0,0 +1,162 @@ +/* + * 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 +#include + +#include +#include +#include +#include +#include + +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +std::shared_ptr MakeSnapshotsSchema() { + return std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "parent_id", int64()), + SchemaField::MakeOptional(4, "operation", string()), + SchemaField::MakeOptional(5, "manifest_list", string()), + SchemaField::MakeOptional( + 6, "summary", + std::make_shared(SchemaField::MakeRequired(7, "key", string()), + SchemaField::MakeRequired(8, "value", string())))}); +} + +} // namespace + +class SnapshotsTableTest : public MetadataTableTestBase { + protected: + void SetUp() override { + MetadataTableTestBase::SetUp(); + + auto [snap1, snap2] = MakeTestSnapshots(); + snap1_ = snap1; + snap2_ = snap2; + + ICEBERG_UNWRAP_OR_FAIL( + table_, MakeTableWithSnapshots({snap1, snap2}, /*current_snapshot_id=*/2)); + + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, + MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots)); + } + + std::shared_ptr
table_; + std::shared_ptr snap1_; + std::shared_ptr snap2_; + std::unique_ptr snapshots_table_; +}; + +TEST_F(SnapshotsTableTest, Construct) { + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, + MetadataTable::Make(MetadataTableTestBase::table_, + MetadataTable::Kind::kSnapshots)); + EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); + EXPECT_EQ(snapshots_table_->source_table(), MetadataTableTestBase::table_); + EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots"); + EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector{"db"})); + EXPECT_NE(snapshots_table_->schema(), nullptr); +} + +TEST_F(SnapshotsTableTest, SchemaMatchesIcebergSchema) { + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, + MetadataTable::Make(MetadataTableTestBase::table_, + MetadataTable::Kind::kSnapshots)); + EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema()); +} + +TEST_F(SnapshotsTableTest, Scan) { + // Scan the snapshots table once and verify all columns of the result. + ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan()); + auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + + // Row and column counts. + EXPECT_EQ(batch->num_rows(), 2); + EXPECT_EQ(batch->num_columns(), 6); + + // Column 0: committed_at (timestamptz) — microseconds since epoch. + auto committed_at = std::static_pointer_cast<::arrow::TimestampArray>(batch->column(0)); + EXPECT_EQ(committed_at->Value(0), 1234567890000 * 1000); + EXPECT_EQ(committed_at->Value(1), 9876543210000 * 1000); + + // Column 1: snapshot_id (long) — returned in storage order. + auto snapshot_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + EXPECT_EQ(snapshot_ids->Value(0), 1); + EXPECT_EQ(snapshot_ids->Value(1), 2); + + // Column 2: parent_id (long) — first snapshot has no parent. + auto parent_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(2)); + EXPECT_TRUE(parent_ids->IsNull(0)); + EXPECT_FALSE(parent_ids->IsNull(1)); + EXPECT_EQ(parent_ids->Value(1), 1); + + // Column 3: operation (string). + auto operations = std::static_pointer_cast<::arrow::StringArray>(batch->column(3)); + EXPECT_EQ(operations->GetString(0), "append"); + EXPECT_EQ(operations->GetString(1), "append"); + + // Column 4: manifest_list (string). + auto manifest_lists = std::static_pointer_cast<::arrow::StringArray>(batch->column(4)); + EXPECT_EQ(manifest_lists->GetString(0), "file:/tmp/manifest1.avro"); + EXPECT_EQ(manifest_lists->GetString(1), "file:/tmp/manifest2.avro"); + + // Column 5: summary (map) — each summary has 11 entries + // (10 data + 1 operation). + auto summaries = std::static_pointer_cast<::arrow::MapArray>(batch->column(5)); + EXPECT_FALSE(summaries->IsNull(0)); + EXPECT_FALSE(summaries->IsNull(1)); + EXPECT_EQ(summaries->value_length(0), 11); + EXPECT_EQ(summaries->value_length(1), 11); +} + +TEST_F(SnapshotsTableTest, ScanSnapshotSelectionIgnored) { + // SnapshotsTable always returns all snapshots regardless of selection. + SnapshotSelection sel{.snapshot_id = 999}; + ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(sel)); + auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + // Should still return all 2 snapshots, not filtered to snapshot 999. + EXPECT_EQ(batch->num_rows(), 2); +} + +TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) { + // A table with zero snapshots should return zero rows. + ICEBERG_UNWRAP_OR_FAIL(auto empty_table, + MakeTableWithSnapshots({}, /*current_snapshot_id=*/-1)); + + ICEBERG_UNWRAP_OR_FAIL( + snapshots_table_, + MetadataTable::Make(empty_table, MetadataTable::Kind::kSnapshots)); + + ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(std::nullopt)); + auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + EXPECT_EQ(batch->num_rows(), 0); + EXPECT_EQ(batch->num_columns(), 6); +} + +} // namespace iceberg From 6c112cc0eeefd9870b6d5359c150fc959efb4452 Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Wed, 22 Jul 2026 13:32:30 +0800 Subject: [PATCH 2/8] fix(inspect): address metadata table review feedback --- src/iceberg/inspect/metadata_table.cc | 14 ++------------ src/iceberg/inspect/metadata_table.h | 9 ++++----- src/iceberg/inspect/snapshots_table.cc | 2 +- src/iceberg/inspect/snapshots_table.h | 2 +- src/iceberg/test/metadata_table_test_base.h | 2 +- 5 files changed, 9 insertions(+), 20 deletions(-) diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 4298e3653..e638a6226 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -35,20 +35,10 @@ MetadataTable::MetadataTable(std::shared_ptr
source_table, MetadataTable::~MetadataTable() = default; -bool MetadataTable::supports_time_travel() const noexcept { - // Time travel is supported for tables that read from a single snapshot's - // manifests. Tables that scan all snapshots or return in-memory history do - // not. - switch (kind()) { - case Kind::kSnapshots: - case Kind::kHistory: - return false; - } - return false; -} +bool MetadataTable::supports_time_travel() const noexcept { return false; } Result MetadataTable::Scan( - std::optional /*snapshot_selection*/) { + const std::optional& /*snapshot_selection*/) { return NotSupported("Scan is not supported for this metadata table type"); } diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 320a95b17..03e66c3f4 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -62,10 +62,8 @@ class ICEBERG_EXPORT MetadataTable { /// \brief Whether this metadata table supports time-travel queries. /// - /// Time travel is supported for tables that read from a single snapshot's - /// manifests (e.g., Entries, Files, Manifests, Partitions). Tables that - /// scan all snapshots (All*) or return in-memory history (Snapshots, - /// History, Refs) do not support time travel. + /// The currently supported snapshots and history metadata tables do not + /// support time travel. bool supports_time_travel() const noexcept; /// \brief Scan the metadata table using the current snapshot. @@ -80,7 +78,8 @@ class ICEBERG_EXPORT MetadataTable { /// /// The default implementation returns NotSupported. Subclasses override this /// to materialize their data. - virtual Result Scan(std::optional snapshot_selection); + virtual Result Scan( + const std::optional& snapshot_selection); const TableIdentifier& name() const { return identifier_; } diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 717d953d4..34cb289a1 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -69,7 +69,7 @@ Result> SnapshotsTable::Make( } Result SnapshotsTable::Scan( - std::optional /*snapshot_selection*/) { + const std::optional& /*snapshot_selection*/) { ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema())); for (const auto& snapshot : source_table()->snapshots()) { diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h index ef5c0dfc4..67430026c 100644 --- a/src/iceberg/inspect/snapshots_table.h +++ b/src/iceberg/inspect/snapshots_table.h @@ -45,7 +45,7 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable { /// The snapshots table always returns every known snapshot, so the /// snapshot_selection parameter is ignored. Result Scan( - std::optional /*snapshot_selection*/) override; + const std::optional& /*snapshot_selection*/) override; private: explicit SnapshotsTable(std::shared_ptr
table); diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h index bd75cf083..c3480c829 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/metadata_table_test_base.h @@ -78,7 +78,7 @@ class MetadataTableTestBase : public ::testing::Test { /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray array, const Schema& schema) { - ArrowSchema c_schema; + ArrowSchema c_schema{}; EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); From 791474e991dcdbf8234ed82f0b7a535e1a886144 Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Thu, 23 Jul 2026 09:44:36 +0800 Subject: [PATCH 3/8] test(inspect): address snapshots table review feedback --- src/iceberg/test/snapshots_table_test.cc | 41 ++++++++++++++++-------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc index 86b8af3ec..5fa7bae53 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/snapshots_table_test.cc @@ -17,8 +17,11 @@ * under the License. */ +#include #include +#include #include +#include #include #include @@ -49,6 +52,19 @@ std::shared_ptr MakeSnapshotsSchema() { SchemaField::MakeRequired(8, "value", string())))}); } +std::vector> GetMapEntries( + const std::shared_ptr<::arrow::MapArray>& map_array, int64_t row) { + auto keys = std::static_pointer_cast<::arrow::StringArray>(map_array->keys()); + auto values = std::static_pointer_cast<::arrow::StringArray>(map_array->items()); + std::vector> entries; + entries.reserve(map_array->value_length(row)); + const auto offset = map_array->value_offset(row); + for (int64_t index = offset; index < offset + map_array->value_length(row); ++index) { + entries.emplace_back(keys->GetString(index), values->GetString(index)); + } + return entries; +} + } // namespace class SnapshotsTableTest : public MetadataTableTestBase { @@ -57,9 +73,6 @@ class SnapshotsTableTest : public MetadataTableTestBase { MetadataTableTestBase::SetUp(); auto [snap1, snap2] = MakeTestSnapshots(); - snap1_ = snap1; - snap2_ = snap2; - ICEBERG_UNWRAP_OR_FAIL( table_, MakeTableWithSnapshots({snap1, snap2}, /*current_snapshot_id=*/2)); @@ -67,27 +80,18 @@ class SnapshotsTableTest : public MetadataTableTestBase { MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots)); } - std::shared_ptr
table_; - std::shared_ptr snap1_; - std::shared_ptr snap2_; std::unique_ptr snapshots_table_; }; TEST_F(SnapshotsTableTest, Construct) { - ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, - MetadataTable::Make(MetadataTableTestBase::table_, - MetadataTable::Kind::kSnapshots)); EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); - EXPECT_EQ(snapshots_table_->source_table(), MetadataTableTestBase::table_); - EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots"); + EXPECT_EQ(snapshots_table_->source_table(), table_); + EXPECT_EQ(snapshots_table_->name().name, "test_table.snapshots"); EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector{"db"})); EXPECT_NE(snapshots_table_->schema(), nullptr); } TEST_F(SnapshotsTableTest, SchemaMatchesIcebergSchema) { - ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, - MetadataTable::Make(MetadataTableTestBase::table_, - MetadataTable::Kind::kSnapshots)); EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema()); } @@ -133,6 +137,15 @@ TEST_F(SnapshotsTableTest, Scan) { EXPECT_FALSE(summaries->IsNull(1)); EXPECT_EQ(summaries->value_length(0), 11); EXPECT_EQ(summaries->value_length(1), 11); + + auto first_summary = GetMapEntries(summaries, 0); + EXPECT_THAT(first_summary, ::testing::Contains(::testing::Pair("operation", "append"))); + EXPECT_THAT(first_summary, ::testing::Contains(::testing::Pair("total-records", "1"))); + + auto second_summary = GetMapEntries(summaries, 1); + EXPECT_THAT(second_summary, + ::testing::Contains(::testing::Pair("operation", "append"))); + EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2"))); } TEST_F(SnapshotsTableTest, ScanSnapshotSelectionIgnored) { From a5c20c20adbd5e010039fff6d6a427e4ae7a89e8 Mon Sep 17 00:00:00 2001 From: wzhuo Date: Fri, 24 Jul 2026 09:38:18 +0800 Subject: [PATCH 4/8] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/iceberg/test/metadata_table_test_base.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h index c3480c829..bbc748627 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/metadata_table_test_base.h @@ -76,7 +76,7 @@ class MetadataTableTestBase : public ::testing::Test { } /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. - static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray array, +static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray&& array, const Schema& schema) { ArrowSchema c_schema{}; EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); From ab62d36b65189a6957b92f66696018058f1e6ad4 Mon Sep 17 00:00:00 2001 From: wzhuo Date: Fri, 24 Jul 2026 09:39:13 +0800 Subject: [PATCH 5/8] Fix formatting issues in metadata_table_test_base.h --- src/iceberg/test/metadata_table_test_base.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h index bbc748627..8f8bfb5ae 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/metadata_table_test_base.h @@ -76,7 +76,7 @@ class MetadataTableTestBase : public ::testing::Test { } /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. -static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray&& array, + static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray&& array, const Schema& schema) { ArrowSchema c_schema{}; EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); From 6680bf9efa2bb0fa1e9370fdcaccc14945e0ef9a Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Mon, 27 Jul 2026 10:05:22 +0800 Subject: [PATCH 6/8] fix(inspect): address remaining review feedback --- src/iceberg/inspect/snapshots_table.cc | 4 +++ src/iceberg/test/metadata_table_test.cc | 6 ++++- src/iceberg/test/metadata_table_test_base.h | 25 +++++++++++++----- src/iceberg/test/snapshots_table_test.cc | 28 +++++++++++++++++---- 4 files changed, 51 insertions(+), 12 deletions(-) diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 34cb289a1..6e6fbee2e 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -73,6 +73,10 @@ Result SnapshotsTable::Scan( ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema())); for (const auto& snapshot : source_table()->snapshots()) { + if (snapshot == nullptr) [[unlikely]] { + continue; + } + // column 0: committed_at (timestamptz -> int64 micros) ICEBERG_RETURN_UNEXPECTED(AppendInt( builder.column(0), std::chrono::duration_cast( diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc index fd59af3b3..98f3ef002 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -22,6 +22,7 @@ #include #include +#include "iceberg/constants.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" @@ -42,7 +43,10 @@ class MetadataTableTest : public ::testing::Test { SchemaField::MakeOptional(2, "name", string())}, 1); auto metadata = std::make_shared( - TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1}); + TableMetadata{.format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = kInvalidSnapshotId}); TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"}; ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata, "s3://bucket/meta.json", diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h index 8f8bfb5ae..bde70042d 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/metadata_table_test_base.h @@ -37,6 +37,7 @@ #include #include +#include "iceberg/constants.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/schema_internal.h" @@ -67,7 +68,10 @@ class MetadataTableTestBase : public ::testing::Test { SchemaField::MakeOptional(2, "name", string())}, 1); metadata_ = std::make_shared( - TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1}); + TableMetadata{.format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = kInvalidSnapshotId}); TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"}; @@ -76,14 +80,23 @@ class MetadataTableTestBase : public ::testing::Test { } /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. - static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray&& array, - const Schema& schema) { + static Result> FinishAndImport( + ArrowArray&& array, const Schema& schema) { ArrowSchema c_schema{}; - EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); - auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &c_schema)); + + auto arrow_schema_result = ::arrow::ImportSchema(&c_schema); + if (!arrow_schema_result.ok()) { + return InvalidArrowData(arrow_schema_result.status().ToString()); + } // ImportRecordBatch takes ownership of the array and releases it. - return ::arrow::ImportRecordBatch(&array, arrow_schema).ValueOrDie(); + auto batch_result = ::arrow::ImportRecordBatch( + &array, std::move(arrow_schema_result).MoveValueUnsafe()); + if (!batch_result.ok()) { + return InvalidArrowData(batch_result.status().ToString()); + } + return std::move(batch_result).MoveValueUnsafe(); } /// \brief Create two snapshots matching the Java TestDataTaskParser test data. diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc index 5fa7bae53..f96ec0443 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/snapshots_table_test.cc @@ -29,6 +29,7 @@ #include #include +#include "iceberg/constants.h" #include "iceberg/inspect/metadata_table.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" @@ -98,7 +99,8 @@ TEST_F(SnapshotsTableTest, SchemaMatchesIcebergSchema) { TEST_F(SnapshotsTableTest, Scan) { // Scan the snapshots table once and verify all columns of the result. ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan()); - auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + ICEBERG_UNWRAP_OR_FAIL(auto batch, + FinishAndImport(std::move(array), *snapshots_table_->schema())); // Row and column counts. EXPECT_EQ(batch->num_rows(), 2); @@ -152,24 +154,40 @@ TEST_F(SnapshotsTableTest, ScanSnapshotSelectionIgnored) { // SnapshotsTable always returns all snapshots regardless of selection. SnapshotSelection sel{.snapshot_id = 999}; ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(sel)); - auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + ICEBERG_UNWRAP_OR_FAIL(auto batch, + FinishAndImport(std::move(array), *snapshots_table_->schema())); // Should still return all 2 snapshots, not filtered to snapshot 999. EXPECT_EQ(batch->num_rows(), 2); } TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) { // A table with zero snapshots should return zero rows. - ICEBERG_UNWRAP_OR_FAIL(auto empty_table, - MakeTableWithSnapshots({}, /*current_snapshot_id=*/-1)); + ICEBERG_UNWRAP_OR_FAIL( + auto empty_table, + MakeTableWithSnapshots({}, /*current_snapshot_id=*/kInvalidSnapshotId)); ICEBERG_UNWRAP_OR_FAIL( snapshots_table_, MetadataTable::Make(empty_table, MetadataTable::Kind::kSnapshots)); ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(std::nullopt)); - auto batch = FinishAndImport(std::move(array), *snapshots_table_->schema()); + ICEBERG_UNWRAP_OR_FAIL(auto batch, + FinishAndImport(std::move(array), *snapshots_table_->schema())); EXPECT_EQ(batch->num_rows(), 0); EXPECT_EQ(batch->num_columns(), 6); } +TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) { + auto [snap1, snap2] = MakeTestSnapshots(); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots({snap1, nullptr, snap2}, + /*current_snapshot_id=*/2)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table, MetadataTable::Kind::kSnapshots)); + + ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batch, + FinishAndImport(std::move(array), *snapshots_table->schema())); + EXPECT_EQ(batch->num_rows(), 2); +} + } // namespace iceberg From 69cfbe9945d0256586cbcb18a75596b611863657 Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Fri, 31 Jul 2026 11:23:42 +0800 Subject: [PATCH 7/8] feat(arrow): expose row count from ArrowRowBuilder --- src/iceberg/arrow_row_builder.cc | 2 ++ src/iceberg/arrow_row_builder_internal.h | 3 +++ src/iceberg/test/arrow_row_builder_test.cc | 3 +++ 3 files changed, 8 insertions(+) diff --git a/src/iceberg/arrow_row_builder.cc b/src/iceberg/arrow_row_builder.cc index 26e7cb4a2..9b07b4f4f 100644 --- a/src/iceberg/arrow_row_builder.cc +++ b/src/iceberg/arrow_row_builder.cc @@ -75,6 +75,8 @@ ArrowRowBuilder::~ArrowRowBuilder() { int64_t ArrowRowBuilder::num_columns() const { return array_.n_children; } +int64_t ArrowRowBuilder::num_rows() const { return array_.length; } + ArrowArray* ArrowRowBuilder::column(int64_t index) { if (index < 0 || index >= array_.n_children) { return nullptr; diff --git a/src/iceberg/arrow_row_builder_internal.h b/src/iceberg/arrow_row_builder_internal.h index db1b66f63..e9c55f07a 100644 --- a/src/iceberg/arrow_row_builder_internal.h +++ b/src/iceberg/arrow_row_builder_internal.h @@ -85,6 +85,9 @@ class ICEBERG_EXPORT ArrowRowBuilder { /// \brief The number of top-level columns in the batch. int64_t num_columns() const; + /// \brief The number of completed rows in the batch. + int64_t num_rows() const; + /// \brief Access the nanoarrow child builder for a top-level column. /// /// \param index Zero-based column index. Returns nullptr if out of range. diff --git a/src/iceberg/test/arrow_row_builder_test.cc b/src/iceberg/test/arrow_row_builder_test.cc index 45fb3b787..d37fe3458 100644 --- a/src/iceberg/test/arrow_row_builder_test.cc +++ b/src/iceberg/test/arrow_row_builder_test.cc @@ -73,6 +73,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ICEBERG_UNWRAP_OR_FAIL(auto builder, ArrowRowBuilder::Make(*schema)); ASSERT_EQ(builder.num_columns(), 5); + ASSERT_EQ(builder.num_rows(), 0); // Row 0 ASSERT_THAT(AppendInt(builder.column(0), 1), IsOk()); @@ -81,6 +82,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ASSERT_THAT(AppendBoolean(builder.column(3), true), IsOk()); ASSERT_THAT(AppendStringMap(builder.column(4), {{"k", "v"}}), IsOk()); ASSERT_THAT(builder.FinishRow(), IsOk()); + ASSERT_EQ(builder.num_rows(), 1); // Row 1 ASSERT_THAT(AppendInt(builder.column(0), 2), IsOk()); @@ -89,6 +91,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ASSERT_THAT(AppendBoolean(builder.column(3), false), IsOk()); ASSERT_THAT(AppendStringMap(builder.column(4), {}), IsOk()); ASSERT_THAT(builder.FinishRow(), IsOk()); + ASSERT_EQ(builder.num_rows(), 2); auto batch = FinishAndImport(std::move(builder), *schema); ASSERT_EQ(batch->num_rows(), 2); From 894539e94008964824f852bcebbc1e6fdcce6aed Mon Sep 17 00:00:00 2001 From: Zhuo Wang Date: Fri, 31 Jul 2026 16:10:32 +0800 Subject: [PATCH 8/8] refactor(inspect): refine metadata table scan APIs --- src/iceberg/inspect/history_table.cc | 33 ++-- src/iceberg/inspect/history_table.h | 4 + src/iceberg/inspect/metadata_table.cc | 39 ++--- src/iceberg/inspect/metadata_table.h | 107 +++++++----- src/iceberg/inspect/snapshots_table.cc | 178 +++++++++++++------- src/iceberg/inspect/snapshots_table.h | 8 +- src/iceberg/test/history_table_test.cc | 5 +- src/iceberg/test/metadata_table_test.cc | 9 +- src/iceberg/test/metadata_table_test_base.h | 27 ++- src/iceberg/test/snapshots_table_test.cc | 118 ++++++------- 10 files changed, 302 insertions(+), 226 deletions(-) diff --git a/src/iceberg/inspect/history_table.cc b/src/iceberg/inspect/history_table.cc index 7fa840043..03bbcca97 100644 --- a/src/iceberg/inspect/history_table.cc +++ b/src/iceberg/inspect/history_table.cc @@ -26,37 +26,32 @@ #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" -#include "iceberg/table_identifier.h" #include "iceberg/type.h" +#include "iceberg/util/macros.h" namespace iceberg { -namespace { -std::shared_ptr MakeHistoryTableSchema() { - return std::make_shared(std::vector{ +HistoryTable::HistoryTable(std::shared_ptr
table) + : MetadataTable(std::move(table)) {} + +HistoryTable::~HistoryTable() = default; + +const std::shared_ptr& HistoryTable::schema() const { + static const auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), SchemaField::MakeRequired(2, "snapshot_id", int64()), SchemaField::MakeOptional(3, "parent_id", int64()), SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); + return schema; } -TableIdentifier MakeHistoryTableName(const TableIdentifier& source_name) { - return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".history"}; -} - -} // namespace - -HistoryTable::HistoryTable(std::shared_ptr
table) - : MetadataTable(table, MakeHistoryTableName(table->name()), - MakeHistoryTableSchema()) {} - -HistoryTable::~HistoryTable() = default; - Result> HistoryTable::Make(std::shared_ptr
table) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); return std::unique_ptr(new HistoryTable(std::move(table))); } +Result HistoryTable::Scan() { + return NotSupported("Scan is not supported for the history table"); +} + } // namespace iceberg diff --git a/src/iceberg/inspect/history_table.h b/src/iceberg/inspect/history_table.h index 21f1f8002..7d863fc2c 100644 --- a/src/iceberg/inspect/history_table.h +++ b/src/iceberg/inspect/history_table.h @@ -40,6 +40,10 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable { Kind kind() const noexcept override { return Kind::kHistory; } + const std::shared_ptr& schema() const override; + + Result Scan() override; + private: explicit HistoryTable(std::shared_ptr
table); }; diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index e638a6226..7bc94c511 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -22,40 +22,33 @@ #include #include -#include "iceberg/inspect/history_table.h" -#include "iceberg/inspect/snapshots_table.h" - namespace iceberg { -MetadataTable::MetadataTable(std::shared_ptr
source_table, - TableIdentifier identifier, std::shared_ptr schema) - : identifier_(std::move(identifier)), - schema_(std::move(schema)), - source_table_(std::move(source_table)) {} +MetadataTable::MetadataTable(std::shared_ptr
source_table) + : source_table_(std::move(source_table)) {} MetadataTable::~MetadataTable() = default; bool MetadataTable::supports_time_travel() const noexcept { return false; } -Result MetadataTable::Scan( - const std::optional& /*snapshot_selection*/) { - return NotSupported("Scan is not supported for this metadata table type"); +const std::shared_ptr
& MetadataTable::source_table() const { + return source_table_; } -Result> MetadataTable::Make(std::shared_ptr
table, - Kind kind) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } +TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr
source_table) + : MetadataTable(std::move(source_table)) {} + +TimeTravelMetadataTable::~TimeTravelMetadataTable() = default; - switch (kind) { - case Kind::kSnapshots: - return SnapshotsTable::Make(table); - case Kind::kHistory: - return HistoryTable::Make(table); - } +bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; } + +Result TimeTravelMetadataTable::Scan() { + return ScanSnapshot(SnapshotSelection{}); +} - return NotSupported("Unsupported metadata table type"); +Result TimeTravelMetadataTable::Scan( + const SnapshotSelection& snapshot_selection) { + return ScanSnapshot(snapshot_selection); } } // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 03e66c3f4..6ca55a5a3 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -20,81 +20,108 @@ #pragma once /// \file iceberg/inspect/metadata_table.h -/// \brief Define base APIs for metadata tables. +/// \brief Base APIs for inspecting Iceberg metadata tables. +#include #include -#include #include +#include +#include #include "iceberg/arrow_c_data.h" #include "iceberg/iceberg_export.h" #include "iceberg/result.h" -#include "iceberg/table_identifier.h" #include "iceberg/type_fwd.h" #include "iceberg/util/timepoint.h" namespace iceberg { -/// \brief Parameters for snapshot selection (time travel). -struct SnapshotSelection { - /// \brief The snapshot ID to read. - std::optional snapshot_id; - /// \brief Read the snapshot that was current at this timestamp. - std::optional as_of_timestamp; - /// \brief Read the snapshot referenced by this named ref (branch or tag). - std::optional ref_name; -}; - -/// \brief Base class for Iceberg metadata tables. +/// \brief Base interface for an Iceberg metadata table. class ICEBERG_EXPORT MetadataTable { public: + /// \brief Supported metadata table kinds. enum class Kind { kSnapshots, kHistory, }; - static Result> Make(std::shared_ptr
table, - Kind kind); + /// \brief Maximum number of rows emitted in each Arrow batch. + static constexpr int64_t kBatchSize = 1024; + + /// \brief Create a metadata table of the requested concrete type. + /// + /// \tparam MetadataTableType Concrete class derived from MetadataTable. + /// \param table Source table whose metadata will be exposed. + /// \return The constructed metadata table, or an error. + template + requires std::derived_from + static Result> Make(std::shared_ptr
table) { + return MetadataTableType::Make(std::move(table)); + } virtual ~MetadataTable(); + /// \brief Return this metadata table's kind. virtual Kind kind() const noexcept = 0; - /// \brief Whether this metadata table supports time-travel queries. - /// - /// The currently supported snapshots and history metadata tables do not - /// support time travel. - bool supports_time_travel() const noexcept; + /// \brief Return the schema of rows emitted by scans. + virtual const std::shared_ptr& schema() const = 0; - /// \brief Scan the metadata table using the current snapshot. + /// \brief Return the source table whose metadata is exposed. + const std::shared_ptr
& source_table() const; + + /// \brief Return whether this metadata table supports time travel. + virtual bool supports_time_travel() const noexcept; + + /// \brief Scan the metadata table without time travel. /// - /// Convenience overload — delegates to Scan(std::nullopt). - Result Scan() { return Scan(std::nullopt); } + /// The caller owns the returned stream and must release it with + /// ArrowArrayStreamRelease. + virtual Result Scan() = 0; + + protected: + explicit MetadataTable(std::shared_ptr
source_table); + + private: + std::shared_ptr
source_table_; +}; - /// \brief Scan the metadata table and return all rows as an Arrow struct array. +/// \brief Snapshot selection parameters for a time-travel scan. +struct SnapshotSelection { + /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp. /// - /// The returned ArrowArray is a struct array where each element is one row. - /// The caller takes ownership and must call ArrowArrayRelease when done. + /// std::monostate selects the current snapshot. + std::variant snapshot; + + /// \brief Resolve the snapshot relative to this branch or tag. /// - /// The default implementation returns NotSupported. Subclasses override this - /// to materialize their data. - virtual Result Scan( - const std::optional& snapshot_selection); + /// An empty string uses the main branch. + std::string ref_name; +}; - const TableIdentifier& name() const { return identifier_; } +/// \brief Base interface for metadata tables that support time travel. +class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable { + public: + ~TimeTravelMetadataTable() override; + + /// \brief Return true because this interface supports time travel. + bool supports_time_travel() const noexcept final; - const std::shared_ptr& schema() const { return schema_; } + /// \brief Scan using the current snapshot on the main branch. + Result Scan() final; - const std::shared_ptr
& source_table() const { return source_table_; } + /// \brief Scan using the requested snapshot selection. + /// + /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection. + /// \return An Arrow stream containing the metadata table rows, or an error. + Result Scan(const SnapshotSelection& snapshot_selection); protected: - explicit MetadataTable(std::shared_ptr
source_table, TableIdentifier identifier, - std::shared_ptr schema); + explicit TimeTravelMetadataTable(std::shared_ptr
source_table); - private: - TableIdentifier identifier_; - std::shared_ptr schema_; - std::shared_ptr
source_table_; + /// \brief Implement a scan for the requested snapshot selection. + virtual Result ScanSnapshot( + const SnapshotSelection& snapshot_selection) = 0; }; } // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 6e6fbee2e..f0a37843b 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -20,98 +20,154 @@ #include "iceberg/inspect/snapshots_table.h" #include +#include #include +#include #include #include +#include + +#include "iceberg/arrow/nanoarrow_status_internal.h" +#include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/arrow_row_builder_internal.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" +#include "iceberg/schema_internal.h" +#include "iceberg/snapshot.h" #include "iceberg/table.h" -#include "iceberg/table_identifier.h" #include "iceberg/type.h" #include "iceberg/util/macros.h" namespace iceberg { namespace { -std::shared_ptr MakeSnapshotsTableSchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeOptional(4, "operation", string()), - SchemaField::MakeOptional(5, "manifest_list", string()), - SchemaField::MakeOptional(6, "summary", - std::make_shared( - SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); -} +Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) { + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(0), std::chrono::duration_cast( + snapshot.timestamp_ms.time_since_epoch()) + .count())); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot.snapshot_id)); + + if (snapshot.parent_snapshot_id.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), *snapshot.parent_snapshot_id)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + } -TableIdentifier MakeSnapshotsTableName(const TableIdentifier& source_name) { - return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".snapshots"}; -} + auto operation = snapshot.Operation(); + if (operation.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + } -} // namespace + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list)); -SnapshotsTable::SnapshotsTable(std::shared_ptr
table) - : MetadataTable(table, MakeSnapshotsTableName(table->name()), - MakeSnapshotsTableSchema()) {} + auto summary = snapshot.summary; + summary.erase(SnapshotSummaryFields::kOperation); + if (summary.empty()) { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5))); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary)); + } -SnapshotsTable::~SnapshotsTable() = default; + return builder.FinishRow(); +} -Result> SnapshotsTable::Make( - std::shared_ptr
table) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); +class SnapshotsTableStream { + public: + static Result> Make( + std::shared_ptr
table, const iceberg::Schema& schema) { + ArrowSchema arrow_schema{}; + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); + return std::unique_ptr( + new SnapshotsTableStream(std::move(table), std::move(arrow_schema))); } - return std::unique_ptr(new SnapshotsTable(std::move(table))); -} -Result SnapshotsTable::Scan( - const std::optional& /*snapshot_selection*/) { - ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema())); + ~SnapshotsTableStream() { std::ignore = Close(); } - for (const auto& snapshot : source_table()->snapshots()) { - if (snapshot == nullptr) [[unlikely]] { - continue; + Status Close() { + table_.reset(); + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); } + return {}; + } - // column 0: committed_at (timestamptz -> int64 micros) - ICEBERG_RETURN_UNEXPECTED(AppendInt( - builder.column(0), std::chrono::duration_cast( - snapshot->timestamp_ms.time_since_epoch()) - .count())); - - // column 1: snapshot_id (long) - ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot->snapshot_id)); - - // column 2: parent_id (long, optional) - if (snapshot->parent_snapshot_id.has_value()) { - ICEBERG_RETURN_UNEXPECTED( - AppendInt(builder.column(2), *snapshot->parent_snapshot_id)); - } else { - ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + Result> Next() { + const auto& snapshots = table_->snapshots(); + if (next_snapshot_ == snapshots.size()) { + return std::nullopt; } - // column 3: operation (string, optional) - auto op = snapshot->Operation(); - if (op.has_value()) { - ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *op)); - } else { - ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_)); + while (next_snapshot_ < snapshots.size() && + builder.num_rows() < MetadataTable::kBatchSize) { + const auto& snapshot = snapshots[next_snapshot_++]; + if (snapshot == nullptr) [[unlikely]] { + continue; + } + ICEBERG_RETURN_UNEXPECTED(AppendSnapshot(builder, *snapshot)); + } + if (builder.num_rows() == 0) { + return std::nullopt; } - // column 4: manifest_list (string, optional) - ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot->manifest_list)); - - // column 5: summary (map) - ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot->summary)); + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } - ICEBERG_RETURN_UNEXPECTED(builder.FinishRow()); + Result Schema() { + if (arrow_schema_.release == nullptr) [[unlikely]] { + return InvalidArgument("Cannot read schema from a closed snapshots table stream"); + } + ArrowSchema schema_copy{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); + return schema_copy; } - return std::move(builder).Finish(); + private: + SnapshotsTableStream(std::shared_ptr
table, ArrowSchema arrow_schema) + : table_(std::move(table)), arrow_schema_(std::move(arrow_schema)) {} + + std::shared_ptr
table_; + ArrowSchema arrow_schema_{}; + size_t next_snapshot_ = 0; +}; + +} // namespace + +SnapshotsTable::SnapshotsTable(std::shared_ptr
table) + : MetadataTable(std::move(table)) {} + +SnapshotsTable::~SnapshotsTable() = default; + +const std::shared_ptr& SnapshotsTable::schema() const { + static const auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "parent_id", int64()), + SchemaField::MakeOptional(4, "operation", string()), + SchemaField::MakeOptional(5, "manifest_list", string()), + SchemaField::MakeOptional(6, "summary", + std::make_shared( + SchemaField::MakeRequired(7, "key", string()), + SchemaField::MakeRequired(8, "value", string())))}); + return schema; +} + +Result> SnapshotsTable::Make( + std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + return std::unique_ptr(new SnapshotsTable(std::move(table))); +} + +Result SnapshotsTable::Scan() { + ICEBERG_ASSIGN_OR_RAISE(auto stream, + SnapshotsTableStream::Make(source_table(), *schema())); + return MakeArrowArrayStream(std::move(stream)); } } // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h index 67430026c..d2f0ddf90 100644 --- a/src/iceberg/inspect/snapshots_table.h +++ b/src/iceberg/inspect/snapshots_table.h @@ -40,12 +40,12 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable { Kind kind() const noexcept override { return Kind::kSnapshots; } + const std::shared_ptr& schema() const override; + /// \brief Scan all snapshots as rows. /// - /// The snapshots table always returns every known snapshot, so the - /// snapshot_selection parameter is ignored. - Result Scan( - const std::optional& /*snapshot_selection*/) override; + /// The snapshots table always returns every known snapshot. + Result Scan() override; private: explicit SnapshotsTable(std::shared_ptr
table); diff --git a/src/iceberg/test/history_table_test.cc b/src/iceberg/test/history_table_test.cc index b27bdef30..8da311c02 100644 --- a/src/iceberg/test/history_table_test.cc +++ b/src/iceberg/test/history_table_test.cc @@ -20,6 +20,8 @@ /// \file history_table_test.cc /// Unit tests for HistoryTable. +#include "iceberg/inspect/history_table.h" + #include #include @@ -46,8 +48,7 @@ std::shared_ptr MakeHistorySchema() { class HistoryTableTest : public MetadataTableTestBase {}; TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) { - ICEBERG_UNWRAP_OR_FAIL(auto history_table, - MetadataTable::Make(table_, MetadataTable::Kind::kHistory)); + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema()); } diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc index 98f3ef002..b014ef962 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -23,6 +23,8 @@ #include #include "iceberg/constants.h" +#include "iceberg/inspect/history_table.h" +#include "iceberg/inspect/snapshots_table.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" @@ -58,18 +60,17 @@ class MetadataTableTest : public ::testing::Test { }; TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) { - auto result = MetadataTable::Make(nullptr, MetadataTable::Kind::kSnapshots); + auto result = MetadataTable::Make(nullptr); EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); EXPECT_THAT(result, HasErrorMessage("Table cannot be null")); } TEST_F(MetadataTableTest, SupportsTimeTravel) { ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, - MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots)); + MetadataTable::Make(table_)); EXPECT_FALSE(snapshots_table->supports_time_travel()); - ICEBERG_UNWRAP_OR_FAIL(auto history_table, - MetadataTable::Make(table_, MetadataTable::Kind::kHistory)); + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); EXPECT_FALSE(history_table->supports_time_travel()); } diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h index bde70042d..d15163ba8 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/metadata_table_test_base.h @@ -20,7 +20,7 @@ /// \file metadata_table_test_base.h /// Shared test base for all metadata table tests. /// -/// Provides common helpers (FinishAndImport, MakeTestSnapshots, +/// Provides common helpers (ReadAllBatches, MakeTestSnapshots, /// MakeTableWithSnapshots) and the MockFileIO + MockCatalog fixture that /// every metadata table test needs. @@ -79,24 +79,19 @@ class MetadataTableTestBase : public ::testing::Test { "s3://bucket/meta.json", io_, catalog_)); } - /// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch. - static Result> FinishAndImport( - ArrowArray&& array, const Schema& schema) { - ArrowSchema c_schema{}; - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &c_schema)); - - auto arrow_schema_result = ::arrow::ImportSchema(&c_schema); - if (!arrow_schema_result.ok()) { - return InvalidArrowData(arrow_schema_result.status().ToString()); + /// \brief Import and consume a Scan()-produced ArrowArrayStream. + static Result>> ReadAllBatches( + ArrowArrayStream&& stream) { + auto reader_result = ::arrow::ImportRecordBatchReader(&stream); + if (!reader_result.ok()) { + return InvalidArrowData(reader_result.status().ToString()); } - // ImportRecordBatch takes ownership of the array and releases it. - auto batch_result = ::arrow::ImportRecordBatch( - &array, std::move(arrow_schema_result).MoveValueUnsafe()); - if (!batch_result.ok()) { - return InvalidArrowData(batch_result.status().ToString()); + auto batches_result = reader_result.ValueUnsafe()->ToRecordBatches(); + if (!batches_result.ok()) { + return InvalidArrowData(batches_result.status().ToString()); } - return std::move(batch_result).MoveValueUnsafe(); + return std::move(batches_result).MoveValueUnsafe(); } /// \brief Create two snapshots matching the Java TestDataTaskParser test data. diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc index f96ec0443..5cda84052 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/snapshots_table_test.cc @@ -17,6 +17,8 @@ * under the License. */ +#include "iceberg/inspect/snapshots_table.h" + #include #include #include @@ -31,28 +33,12 @@ #include "iceberg/constants.h" #include "iceberg/inspect/metadata_table.h" -#include "iceberg/schema.h" -#include "iceberg/schema_field.h" #include "iceberg/test/matchers.h" #include "iceberg/test/metadata_table_test_base.h" -#include "iceberg/type.h" namespace iceberg { namespace { -std::shared_ptr MakeSnapshotsSchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeOptional(4, "operation", string()), - SchemaField::MakeOptional(5, "manifest_list", string()), - SchemaField::MakeOptional( - 6, "summary", - std::make_shared(SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); -} - std::vector> GetMapEntries( const std::shared_ptr<::arrow::MapArray>& map_array, int64_t row) { auto keys = std::static_pointer_cast<::arrow::StringArray>(map_array->keys()); @@ -77,8 +63,7 @@ class SnapshotsTableTest : public MetadataTableTestBase { ICEBERG_UNWRAP_OR_FAIL( table_, MakeTableWithSnapshots({snap1, snap2}, /*current_snapshot_id=*/2)); - ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, - MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots)); + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, MetadataTable::Make(table_)); } std::unique_ptr snapshots_table_; @@ -87,20 +72,15 @@ class SnapshotsTableTest : public MetadataTableTestBase { TEST_F(SnapshotsTableTest, Construct) { EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); EXPECT_EQ(snapshots_table_->source_table(), table_); - EXPECT_EQ(snapshots_table_->name().name, "test_table.snapshots"); - EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector{"db"})); EXPECT_NE(snapshots_table_->schema(), nullptr); } -TEST_F(SnapshotsTableTest, SchemaMatchesIcebergSchema) { - EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema()); -} - TEST_F(SnapshotsTableTest, Scan) { // Scan the snapshots table once and verify all columns of the result. - ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan()); - ICEBERG_UNWRAP_OR_FAIL(auto batch, - FinishAndImport(std::move(array), *snapshots_table_->schema())); + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + const auto& batch = batches.front(); // Row and column counts. EXPECT_EQ(batch->num_rows(), 2); @@ -132,49 +112,38 @@ TEST_F(SnapshotsTableTest, Scan) { EXPECT_EQ(manifest_lists->GetString(0), "file:/tmp/manifest1.avro"); EXPECT_EQ(manifest_lists->GetString(1), "file:/tmp/manifest2.avro"); - // Column 5: summary (map) — each summary has 11 entries - // (10 data + 1 operation). + // Column 5: summary (map) excludes the separate operation field. auto summaries = std::static_pointer_cast<::arrow::MapArray>(batch->column(5)); EXPECT_FALSE(summaries->IsNull(0)); EXPECT_FALSE(summaries->IsNull(1)); - EXPECT_EQ(summaries->value_length(0), 11); - EXPECT_EQ(summaries->value_length(1), 11); + EXPECT_EQ(summaries->value_length(0), 10); + EXPECT_EQ(summaries->value_length(1), 10); auto first_summary = GetMapEntries(summaries, 0); - EXPECT_THAT(first_summary, ::testing::Contains(::testing::Pair("operation", "append"))); + EXPECT_THAT( + first_summary, + ::testing::Not(::testing::Contains(::testing::Pair("operation", "append")))); EXPECT_THAT(first_summary, ::testing::Contains(::testing::Pair("total-records", "1"))); auto second_summary = GetMapEntries(summaries, 1); - EXPECT_THAT(second_summary, - ::testing::Contains(::testing::Pair("operation", "append"))); + EXPECT_THAT( + second_summary, + ::testing::Not(::testing::Contains(::testing::Pair("operation", "append")))); EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2"))); } -TEST_F(SnapshotsTableTest, ScanSnapshotSelectionIgnored) { - // SnapshotsTable always returns all snapshots regardless of selection. - SnapshotSelection sel{.snapshot_id = 999}; - ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(sel)); - ICEBERG_UNWRAP_OR_FAIL(auto batch, - FinishAndImport(std::move(array), *snapshots_table_->schema())); - // Should still return all 2 snapshots, not filtered to snapshot 999. - EXPECT_EQ(batch->num_rows(), 2); -} - TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) { // A table with zero snapshots should return zero rows. ICEBERG_UNWRAP_OR_FAIL( auto empty_table, MakeTableWithSnapshots({}, /*current_snapshot_id=*/kInvalidSnapshotId)); - ICEBERG_UNWRAP_OR_FAIL( - snapshots_table_, - MetadataTable::Make(empty_table, MetadataTable::Kind::kSnapshots)); + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, + MetadataTable::Make(empty_table)); - ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table_->Scan(std::nullopt)); - ICEBERG_UNWRAP_OR_FAIL(auto batch, - FinishAndImport(std::move(array), *snapshots_table_->schema())); - EXPECT_EQ(batch->num_rows(), 0); - EXPECT_EQ(batch->num_columns(), 6); + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); } TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) { @@ -182,12 +151,47 @@ TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) { ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots({snap1, nullptr, snap2}, /*current_snapshot_id=*/2)); ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, - MetadataTable::Make(table, MetadataTable::Kind::kSnapshots)); + MetadataTable::Make(table)); - ICEBERG_UNWRAP_OR_FAIL(auto array, snapshots_table->Scan()); - ICEBERG_UNWRAP_OR_FAIL(auto batch, - FinishAndImport(std::move(array), *snapshots_table->schema())); - EXPECT_EQ(batch->num_rows(), 2); + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + EXPECT_EQ(batches.front()->num_rows(), 2); +} + +TEST_F(SnapshotsTableTest, ScanTreatsEmptySummaryAsNull) { + auto [missing_summary, operation_only_summary] = MakeTestSnapshots(); + missing_summary->summary.clear(); + operation_only_summary->summary = { + {SnapshotSummaryFields::kOperation, DataOperation::kAppend}}; + ICEBERG_UNWRAP_OR_FAIL(auto table, + MakeTableWithSnapshots({missing_summary, operation_only_summary}, + /*current_snapshot_id=*/2)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto summaries = + std::static_pointer_cast<::arrow::MapArray>(batches.front()->column(5)); + EXPECT_TRUE(summaries->IsNull(0)); + EXPECT_TRUE(summaries->IsNull(1)); +} + +TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) { + auto snapshot = MakeTestSnapshots().first; + std::vector> snapshots(1025, snapshot); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots(std::move(snapshots), + /*current_snapshot_id=*/1)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 2); + EXPECT_EQ(batches[0]->num_rows(), 1024); + EXPECT_EQ(batches[1]->num_rows(), 1); } } // namespace iceberg