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/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 5e9504003..7bc94c511 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -22,33 +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; -Result> MetadataTable::Make(std::shared_ptr
table, - Kind kind) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } +bool MetadataTable::supports_time_travel() const noexcept { return false; } + +const std::shared_ptr
& MetadataTable::source_table() const { + return source_table_; +} + +TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr
source_table) + : MetadataTable(std::move(source_table)) {} - switch (kind) { - case Kind::kSnapshots: - return SnapshotsTable::Make(table); - case Kind::kHistory: - return HistoryTable::Make(table); - } +TimeTravelMetadataTable::~TimeTravelMetadataTable() = default; + +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 51c5f7920..6ca55a5a3 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -20,46 +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 "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 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; - const TableIdentifier& name() const { return identifier_; } + /// \brief Return the schema of rows emitted by scans. + virtual const std::shared_ptr& schema() const = 0; + + /// \brief Return the source table whose metadata is exposed. + const std::shared_ptr
& source_table() const; - const std::shared_ptr& schema() const { return schema_; } + /// \brief Return whether this metadata table supports time travel. + virtual bool supports_time_travel() const noexcept; - const std::shared_ptr
& source_table() const { return source_table_; } + /// \brief Scan the metadata table without time travel. + /// + /// The caller owns the returned stream and must release it with + /// ArrowArrayStreamRelease. + virtual Result Scan() = 0; protected: - explicit MetadataTable(std::shared_ptr
source_table, TableIdentifier identifier, - std::shared_ptr schema); + explicit MetadataTable(std::shared_ptr
source_table); private: - TableIdentifier identifier_; - std::shared_ptr schema_; std::shared_ptr
source_table_; }; +/// \brief Snapshot selection parameters for a time-travel scan. +struct SnapshotSelection { + /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp. + /// + /// std::monostate selects the current snapshot. + std::variant snapshot; + + /// \brief Resolve the snapshot relative to this branch or tag. + /// + /// An empty string uses the main branch. + std::string ref_name; +}; + +/// \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; + + /// \brief Scan using the current snapshot on the main branch. + Result Scan() final; + + /// \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 TimeTravelMetadataTable(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 4b0c3ce9f..f0a37843b 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -19,21 +19,133 @@ #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{ +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))); + } + + auto operation = snapshot.Operation(); + if (operation.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + } + + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list)); + + 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)); + } + + return builder.FinishRow(); +} + +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))); + } + + ~SnapshotsTableStream() { std::ignore = Close(); } + + Status Close() { + table_.reset(); + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + return {}; + } + + Result> Next() { + const auto& snapshots = table_->snapshots(); + if (next_snapshot_ == snapshots.size()) { + return std::nullopt; + } + + 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; + } + + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } + + 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; + } + + 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()), @@ -43,26 +155,19 @@ std::shared_ptr MakeSnapshotsTableSchema() { std::make_shared( SchemaField::MakeRequired(7, "key", string()), SchemaField::MakeRequired(8, "value", string())))}); + return schema; } -TableIdentifier MakeSnapshotsTableName(const TableIdentifier& source_name) { - return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".snapshots"}; -} - -} // namespace - -SnapshotsTable::SnapshotsTable(std::shared_ptr
table) - : MetadataTable(table, MakeSnapshotsTableName(table->name()), - MakeSnapshotsTableSchema()) {} - -SnapshotsTable::~SnapshotsTable() = default; - Result> SnapshotsTable::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 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 9af1bcacb..d2f0ddf90 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; } + const std::shared_ptr& schema() const override; + + /// \brief Scan all snapshots as rows. + /// + /// The snapshots table always returns every known snapshot. + Result Scan() 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/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); diff --git a/src/iceberg/test/history_table_test.cc b/src/iceberg/test/history_table_test.cc new file mode 100644 index 000000000..8da311c02 --- /dev/null +++ b/src/iceberg/test/history_table_test.cc @@ -0,0 +1,55 @@ +/* + * 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 "iceberg/inspect/history_table.h" + +#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_)); + 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..b014ef962 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -22,6 +22,9 @@ #include #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" @@ -33,88 +36,42 @@ #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( - 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); + auto metadata = std::make_shared( + 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", + 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); + 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_)); + EXPECT_FALSE(snapshots_table->supports_time_travel()); + + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); + 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..d15163ba8 --- /dev/null +++ b/src/iceberg/test/metadata_table_test_base.h @@ -0,0 +1,169 @@ +/* + * 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 (ReadAllBatches, 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/constants.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/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, + .current_snapshot_id = kInvalidSnapshotId}); + + 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 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()); + } + + auto batches_result = reader_result.ValueUnsafe()->ToRecordBatches(); + if (!batches_result.ok()) { + return InvalidArrowData(batches_result.status().ToString()); + } + return std::move(batches_result).MoveValueUnsafe(); + } + + /// \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..5cda84052 --- /dev/null +++ b/src/iceberg/test/snapshots_table_test.cc @@ -0,0 +1,197 @@ +/* + * 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/snapshots_table.h" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" + +namespace iceberg { +namespace { + +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 { + protected: + void SetUp() override { + MetadataTableTestBase::SetUp(); + + auto [snap1, snap2] = MakeTestSnapshots(); + ICEBERG_UNWRAP_OR_FAIL( + table_, MakeTableWithSnapshots({snap1, snap2}, /*current_snapshot_id=*/2)); + + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, MetadataTable::Make(table_)); + } + + std::unique_ptr snapshots_table_; +}; + +TEST_F(SnapshotsTableTest, Construct) { + EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); + EXPECT_EQ(snapshots_table_->source_table(), table_); + EXPECT_NE(snapshots_table_->schema(), nullptr); +} + +TEST_F(SnapshotsTableTest, Scan) { + // Scan the snapshots table once and verify all columns of the result. + 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); + 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) 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), 10); + EXPECT_EQ(summaries->value_length(1), 10); + + auto first_summary = GetMapEntries(summaries, 0); + 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::Not(::testing::Contains(::testing::Pair("operation", "append")))); + EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "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)); + + 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) { + 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)); + + 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