From f25565e91fe0c3595e11b5f2fcb020c7dd9e084c Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 13 Oct 2026 21:53:15 +0800 Subject: [PATCH 1/2] feat(inspect): add refs metadata table Co-authored-by: Codex --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/cpp20_compatibility_internal.h | 1 + src/iceberg/inspect/metadata_table.h | 1 + .../inspect/metadata_table_stream_internal.h | 149 ++++++++++++++++++ src/iceberg/inspect/refs_table.cc | 109 +++++++++++++ src/iceberg/inspect/refs_table.h | 51 ++++++ src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/inspect/refs_table_test.cc | 142 +++++++++++++++++ src/iceberg/type_fwd.h | 1 + 9 files changed, 456 insertions(+) create mode 100644 src/iceberg/inspect/metadata_table_stream_internal.h create mode 100644 src/iceberg/inspect/refs_table.cc create mode 100644 src/iceberg/inspect/refs_table.h create mode 100644 src/iceberg/test/inspect/refs_table_test.cc diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 657d487c2c..127d37dbaf 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -58,6 +58,7 @@ set(ICEBERG_SOURCES incremental_changelog_scan.cc inspect/history_table.cc inspect/metadata_table.cc + inspect/refs_table.cc inspect/snapshots_table.cc inheritable_metadata.cc json_serde.cc diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h index b5f46e488d..692e631277 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/src/iceberg/cpp20_compatibility_internal.h @@ -80,6 +80,7 @@ #include "iceberg/inheritable_metadata.h" #include "iceberg/inspect/history_table.h" #include "iceberg/inspect/metadata_table.h" +#include "iceberg/inspect/refs_table.h" #include "iceberg/inspect/snapshots_table.h" #include "iceberg/location_provider.h" #include "iceberg/logging/cerr_logger.h" diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 85000ee1e7..405bf17864 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -43,6 +43,7 @@ class ICEBERG_EXPORT MetadataTable { enum class Kind { kSnapshots, kHistory, + kRefs, }; /// \brief Maximum number of rows emitted in each Arrow batch. diff --git a/src/iceberg/inspect/metadata_table_stream_internal.h b/src/iceberg/inspect/metadata_table_stream_internal.h new file mode 100644 index 0000000000..079ea9c4da --- /dev/null +++ b/src/iceberg/inspect/metadata_table_stream_internal.h @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow_c_data_util_internal.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/nanoarrow_status_internal.h" +#include "iceberg/schema.h" +#include "iceberg/schema_internal.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/stream.h" + +namespace iceberg::internal { + +/// \brief Arrow stream backed by a metadata-table row iterator. +template +class MetadataTableRowsStream { + public: + using AppendRow = std::function; + + static Result> Make( + const Schema& schema, std::unique_ptr> rows, AppendRow append_row) { + ArrowSchema arrow_schema{}; + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); + return std::unique_ptr(new MetadataTableRowsStream( + std::move(rows), std::move(append_row), std::move(arrow_schema))); + } + + ~MetadataTableRowsStream() { + auto status = Close(); + static_cast(status); + } + + Status Close() { + rows_.reset(); + append_row_ = nullptr; + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + return {}; + } + + Result> Next() { + ICEBERG_PRECHECK(arrow_schema_.release != nullptr, + "Cannot read from a closed metadata table stream"); + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_)); + while (builder.num_rows() < MetadataTable::kBatchSize) { + ICEBERG_ASSIGN_OR_RAISE(auto row, rows_->Next()); + if (!row.has_value()) { + break; + } + ICEBERG_RETURN_UNEXPECTED(append_row_(builder, *row)); + } + if (builder.num_rows() == 0) { + return std::nullopt; + } + + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } + + Result Schema() { + ICEBERG_PRECHECK(arrow_schema_.release != nullptr, + "Cannot read schema from a closed metadata table stream"); + ArrowSchema schema_copy{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); + return schema_copy; + } + + private: + MetadataTableRowsStream(std::unique_ptr> rows, AppendRow append_row, + ArrowSchema arrow_schema) + : rows_(std::move(rows)), + append_row_(std::move(append_row)), + arrow_schema_(std::move(arrow_schema)) {} + + std::unique_ptr> rows_; + AppendRow append_row_; + ArrowSchema arrow_schema_{}; +}; + +template +Result MakeMetadataTableStream(const Schema& schema, + std::unique_ptr> rows, + AppendRow append_row) { + ICEBERG_ASSIGN_OR_RAISE( + auto stream, + MetadataTableRowsStream::Make( + schema, std::move(rows), + typename MetadataTableRowsStream::AppendRow(std::move(append_row)))); + return MakeArrowArrayStream(std::move(stream)); +} + +template +class MetadataTableVectorIterator : public Stream { + public: + explicit MetadataTableVectorIterator(std::vector rows) : rows_(std::move(rows)) {} + + protected: + Result> NextImpl() override { + if (next_row_ == rows_.size()) { + return std::nullopt; + } + return std::move(rows_[next_row_++]); + } + + private: + std::vector rows_; + size_t next_row_ = 0; +}; + +template +Result MakeMetadataTableStream(const Schema& schema, + std::vector rows, + AppendRow append_row) { + std::unique_ptr> iterator = + std::make_unique>(std::move(rows)); + return MakeMetadataTableStream(schema, std::move(iterator), std::move(append_row)); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/refs_table.cc b/src/iceberg/inspect/refs_table.cc new file mode 100644 index 0000000000..c6815a5233 --- /dev/null +++ b/src/iceberg/inspect/refs_table.cc @@ -0,0 +1,109 @@ +/* + * 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/refs_table.h" + +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table_stream_internal.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +struct RefRow { + std::string name; + std::string type; + int64_t snapshot_id; + std::optional max_ref_age_ms; + std::optional min_snapshots_to_keep; + std::optional max_snapshot_age_ms; +}; + +Status AppendOptional(ArrowArray* array, const auto& value) { + if (!value.has_value()) { + return AppendNull(array); + } + return AppendInt(array, static_cast(*value)); +} + +Status AppendRef(ArrowRowBuilder& builder, const RefRow& ref) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(0), ref.name)); + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), ref.type)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), ref.snapshot_id)); + ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(3), ref.max_ref_age_ms)); + ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(4), ref.min_snapshots_to_keep)); + ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(5), ref.max_snapshot_age_ms)); + return builder.FinishRow(); +} + +} // namespace + +RefsTable::RefsTable(std::shared_ptr table) : MetadataTable(std::move(table)) {} + +RefsTable::~RefsTable() = default; + +const std::shared_ptr& RefsTable::schema() const { + static const auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "name", string()), + SchemaField::MakeRequired(2, "type", string()), + SchemaField::MakeRequired(3, "snapshot_id", int64()), + SchemaField::MakeOptional(4, "max_reference_age_in_ms", int64()), + SchemaField::MakeOptional(5, "min_snapshots_to_keep", int32()), + SchemaField::MakeOptional(6, "max_snapshot_age_in_ms", int64()), + }); + return schema; +} + +Result> RefsTable::Make(std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + return std::unique_ptr(new RefsTable(std::move(table))); +} + +Result RefsTable::Scan() { + std::vector rows; + for (const auto& [name, ref] : source_table()->metadata()->refs) { + ICEBERG_PRECHECK(ref != nullptr, "Snapshot reference '{}' is null", name); + RefRow row{.name = name, + .type = ref->type() == SnapshotRefType::kBranch ? "BRANCH" : "TAG", + .snapshot_id = ref->snapshot_id, + .max_ref_age_ms = ref->max_ref_age_ms()}; + if (const auto* branch = std::get_if(&ref->retention)) { + row.min_snapshots_to_keep = branch->min_snapshots_to_keep; + row.max_snapshot_age_ms = branch->max_snapshot_age_ms; + } + rows.push_back(std::move(row)); + } + std::ranges::sort(rows, {}, &RefRow::name); + return internal::MakeMetadataTableStream(*schema(), std::move(rows), AppendRef); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/refs_table.h b/src/iceberg/inspect/refs_table.h new file mode 100644 index 0000000000..8c56e288a1 --- /dev/null +++ b/src/iceberg/inspect/refs_table.h @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/inspect/refs_table.h +/// \brief Define the snapshot references metadata table. + +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Metadata table containing the table's branch and tag snapshot references. +class ICEBERG_EXPORT RefsTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~RefsTable() override; + + Kind kind() const noexcept override { return Kind::kRefs; } + + const std::shared_ptr& schema() const override; + + Result Scan() override; + + private: + explicit RefsTable(std::shared_ptr
table); +}; + +} // namespace iceberg diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 058a9dd3cc..f215198285 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -191,6 +191,7 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES + inspect/refs_table_test.cc history_table_test.cc metadata_table_test.cc snapshots_table_test.cc) diff --git a/src/iceberg/test/inspect/refs_table_test.cc b/src/iceberg/test/inspect/refs_table_test.cc new file mode 100644 index 0000000000..ca7fc75313 --- /dev/null +++ b/src/iceberg/test/inspect/refs_table_test.cc @@ -0,0 +1,142 @@ +/* + * 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/refs_table.h" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +class RefsTableTest : public MetadataTableTestBase, + public ::testing::WithParamInterface { + protected: + Result> MakeTable( + std::vector> snapshots, int64_t current_snapshot_id, + std::unordered_map> refs = {}) { + auto metadata = std::make_shared(*metadata_); + metadata->format_version = GetParam(); + metadata->snapshots = std::move(snapshots); + metadata->current_snapshot_id = current_snapshot_id; + metadata->refs = std::move(refs); + return Table::Make( + TableIdentifier{.ns = Namespace{.levels = {"db"}}, .name = "table"}, + std::move(metadata), "s3://bucket/metadata.json", io_, catalog_); + } +}; + +TEST_P(RefsTableTest, ScansBranchesAndTagsAsRefs) { + ICEBERG_UNWRAP_OR_FAIL(auto main_ref, SnapshotRef::MakeBranch(2)); + ICEBERG_UNWRAP_OR_FAIL(auto dev_ref, SnapshotRef::MakeBranch(1, 3, 2000, 1000)); + ICEBERG_UNWRAP_OR_FAIL(auto release_ref, SnapshotRef::MakeTag(1, 5000)); + std::unordered_map> refs; + refs.emplace("main", std::move(main_ref)); + refs.emplace("dev", std::move(dev_ref)); + refs.emplace("release", std::move(release_ref)); + + auto first = std::make_shared(Snapshot{ + .snapshot_id = 1, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = "unused-1.avro", + }); + auto second = std::make_shared(Snapshot{ + .snapshot_id = 2, + .parent_snapshot_id = 1, + .sequence_number = 2, + .timestamp_ms = TimePointMsFromUnixMs(2000), + .manifest_list = "unused-2.avro", + }); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second}, 2, std::move(refs))); + + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, MetadataTable::Make(table)); + EXPECT_EQ(refs_table->kind(), MetadataTable::Kind::kRefs); + EXPECT_FALSE(refs_table->supports_time_travel()); + Schema expected_schema({ + SchemaField::MakeRequired(1, "name", string()), + SchemaField::MakeRequired(2, "type", string()), + SchemaField::MakeRequired(3, "snapshot_id", int64()), + SchemaField::MakeOptional(4, "max_reference_age_in_ms", int64()), + SchemaField::MakeOptional(5, "min_snapshots_to_keep", int32()), + SchemaField::MakeOptional(6, "max_snapshot_age_in_ms", int64()), + }); + EXPECT_EQ(*refs_table->schema(), expected_schema); + ICEBERG_UNWRAP_OR_FAIL(auto stream, refs_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 3); + auto names = std::static_pointer_cast<::arrow::StringArray>(batches[0]->column(0)); + auto types = std::static_pointer_cast<::arrow::StringArray>(batches[0]->column(1)); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(batches[0]->column(2)); + auto ref_age = std::static_pointer_cast<::arrow::Int64Array>(batches[0]->column(3)); + auto min_snapshots = + std::static_pointer_cast<::arrow::Int32Array>(batches[0]->column(4)); + auto snapshot_age = + std::static_pointer_cast<::arrow::Int64Array>(batches[0]->column(5)); + EXPECT_EQ(names->GetString(0), "dev"); + EXPECT_EQ(types->GetString(0), "BRANCH"); + EXPECT_EQ(ids->Value(0), 1); + EXPECT_EQ(ref_age->Value(0), 1000); + EXPECT_EQ(min_snapshots->Value(0), 3); + EXPECT_EQ(snapshot_age->Value(0), 2000); + EXPECT_EQ(names->GetString(1), "main"); + EXPECT_EQ(types->GetString(1), "BRANCH"); + EXPECT_EQ(ids->Value(1), 2); + EXPECT_TRUE(ref_age->IsNull(1)); + EXPECT_TRUE(min_snapshots->IsNull(1)); + EXPECT_TRUE(snapshot_age->IsNull(1)); + EXPECT_EQ(names->GetString(2), "release"); + EXPECT_EQ(types->GetString(2), "TAG"); + EXPECT_EQ(ids->Value(2), 1); + EXPECT_EQ(ref_age->Value(2), 5000); + EXPECT_TRUE(min_snapshots->IsNull(2)); + EXPECT_TRUE(snapshot_age->IsNull(2)); +} + +TEST_P(RefsTableTest, ScanEmpty) { + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({}, kInvalidSnapshotId)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, metadata_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, RefsTableTest, ::testing::Values(2, 3)); + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 84a9919097..8299d89234 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -279,6 +279,7 @@ class PositionDeleteIndex; /// \brief Metadata tables. class HistoryTable; class MetadataTable; +class RefsTable; class SnapshotsTable; /// \brief Table encryption From bd8e6c8dd359d4686d298f2a8f34b255ab969e28 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 13 Oct 2026 23:29:59 +0800 Subject: [PATCH 2/2] fix(inspect): harden and share metadata table streams Make row errors and exhaustion terminal, validate stream inputs, and store callbacks without type erasure. Reuse the helper for SnapshotsTable and cover batching, lazy reads, failures, cleanup, null metadata, and format version 1. Co-authored-by: Codex --- .../inspect/metadata_table_stream_internal.h | 70 +++--- src/iceberg/inspect/refs_table.cc | 25 +- src/iceberg/inspect/snapshots_table.cc | 78 ++---- src/iceberg/test/CMakeLists.txt | 7 +- .../test/{ => inspect}/history_table_test.cc | 2 +- .../inspect/metadata_table_stream_test.cc | 236 ++++++++++++++++++ .../test/{ => inspect}/metadata_table_test.cc | 0 .../{ => inspect}/metadata_table_test_base.h | 2 + src/iceberg/test/inspect/refs_table_test.cc | 93 ++++--- .../{ => inspect}/snapshots_table_test.cc | 2 +- 10 files changed, 377 insertions(+), 138 deletions(-) rename src/iceberg/test/{ => inspect}/history_table_test.cc (96%) create mode 100644 src/iceberg/test/inspect/metadata_table_stream_test.cc rename src/iceberg/test/{ => inspect}/metadata_table_test.cc (100%) rename src/iceberg/test/{ => inspect}/metadata_table_test_base.h (99%) rename src/iceberg/test/{ => inspect}/snapshots_table_test.cc (99%) diff --git a/src/iceberg/inspect/metadata_table_stream_internal.h b/src/iceberg/inspect/metadata_table_stream_internal.h index 079ea9c4da..94df35ded1 100644 --- a/src/iceberg/inspect/metadata_table_stream_internal.h +++ b/src/iceberg/inspect/metadata_table_stream_internal.h @@ -20,7 +20,6 @@ #pragma once #include -#include #include #include #include @@ -40,61 +39,68 @@ namespace iceberg::internal { /// \brief Arrow stream backed by a metadata-table row iterator. -template -class MetadataTableRowsStream { +template +class MetadataTableRowsStream : public Stream { public: - using AppendRow = std::function; - static Result> Make( - const Schema& schema, std::unique_ptr> rows, AppendRow append_row) { + const iceberg::Schema& schema, std::unique_ptr> rows, + AppendRow append_row) { + ICEBERG_PRECHECK(rows != nullptr, "Metadata table row source cannot be null"); + if constexpr (requires { static_cast(append_row); }) { + ICEBERG_PRECHECK(static_cast(append_row), + "Metadata table row callback cannot be empty"); + } ArrowSchema arrow_schema{}; ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); return std::unique_ptr(new MetadataTableRowsStream( std::move(rows), std::move(append_row), std::move(arrow_schema))); } - ~MetadataTableRowsStream() { - auto status = Close(); - static_cast(status); + ~MetadataTableRowsStream() override { + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } } Status Close() { rows_.reset(); - append_row_ = nullptr; + append_row_.reset(); if (arrow_schema_.release != nullptr) { ArrowSchemaRelease(&arrow_schema_); } return {}; } - Result> Next() { + Result Schema() { ICEBERG_PRECHECK(arrow_schema_.release != nullptr, - "Cannot read from a closed metadata table stream"); + "Cannot read schema from a closed metadata table stream"); + ArrowSchema schema_copy{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); + return schema_copy; + } + + protected: + Result> NextImpl() override { + ICEBERG_PRECHECK(rows_ != nullptr, "Cannot read from a closed metadata table stream"); + // Check for exhaustion before allocating a builder, including after a partial batch. + ICEBERG_ASSIGN_OR_RAISE(auto row, rows_->Next()); + if (!row.has_value()) { + return std::nullopt; + } ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_)); + ICEBERG_RETURN_UNEXPECTED((*append_row_)(builder, *row)); while (builder.num_rows() < MetadataTable::kBatchSize) { - ICEBERG_ASSIGN_OR_RAISE(auto row, rows_->Next()); - if (!row.has_value()) { + ICEBERG_ASSIGN_OR_RAISE(auto next_row, rows_->Next()); + if (!next_row.has_value()) { break; } - ICEBERG_RETURN_UNEXPECTED(append_row_(builder, *row)); - } - if (builder.num_rows() == 0) { - return std::nullopt; + ICEBERG_RETURN_UNEXPECTED((*append_row_)(builder, *next_row)); } - ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); return array; } - Result Schema() { - ICEBERG_PRECHECK(arrow_schema_.release != nullptr, - "Cannot read schema from a closed metadata table stream"); - ArrowSchema schema_copy{}; - ICEBERG_NANOARROW_RETURN_UNEXPECTED( - ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); - return schema_copy; - } - private: MetadataTableRowsStream(std::unique_ptr> rows, AppendRow append_row, ArrowSchema arrow_schema) @@ -103,7 +109,7 @@ class MetadataTableRowsStream { arrow_schema_(std::move(arrow_schema)) {} std::unique_ptr> rows_; - AppendRow append_row_; + std::optional append_row_; ArrowSchema arrow_schema_{}; }; @@ -111,11 +117,9 @@ template Result MakeMetadataTableStream(const Schema& schema, std::unique_ptr> rows, AppendRow append_row) { + using RowsStream = MetadataTableRowsStream; ICEBERG_ASSIGN_OR_RAISE( - auto stream, - MetadataTableRowsStream::Make( - schema, std::move(rows), - typename MetadataTableRowsStream::AppendRow(std::move(append_row)))); + auto stream, RowsStream::Make(schema, std::move(rows), std::move(append_row))); return MakeArrowArrayStream(std::move(stream)); } diff --git a/src/iceberg/inspect/refs_table.cc b/src/iceberg/inspect/refs_table.cc index c6815a5233..89c2116163 100644 --- a/src/iceberg/inspect/refs_table.cc +++ b/src/iceberg/inspect/refs_table.cc @@ -20,10 +20,12 @@ #include "iceberg/inspect/refs_table.h" #include +#include #include #include #include #include +#include #include #include "iceberg/arrow_row_builder_internal.h" @@ -90,20 +92,31 @@ Result> RefsTable::Make(std::shared_ptr
table) Result RefsTable::Scan() { std::vector rows; - for (const auto& [name, ref] : source_table()->metadata()->refs) { - ICEBERG_PRECHECK(ref != nullptr, "Snapshot reference '{}' is null", name); + const auto& refs = source_table()->metadata()->refs; + rows.reserve(refs.size()); + for (const auto& [name, ref] : refs) { + if (ref == nullptr) { + return Invalid("Snapshot reference '{}' is null", name); + } + const auto* branch = std::get_if(&ref->retention); RefRow row{.name = name, - .type = ref->type() == SnapshotRefType::kBranch ? "BRANCH" : "TAG", + .type = branch != nullptr ? "BRANCH" : "TAG", .snapshot_id = ref->snapshot_id, - .max_ref_age_ms = ref->max_ref_age_ms()}; - if (const auto* branch = std::get_if(&ref->retention)) { + .max_ref_age_ms = + branch != nullptr + ? branch->max_ref_age_ms + : std::get(ref->retention).max_ref_age_ms}; + if (branch != nullptr) { row.min_snapshots_to_keep = branch->min_snapshots_to_keep; row.max_snapshot_age_ms = branch->max_snapshot_age_ms; } rows.push_back(std::move(row)); } std::ranges::sort(rows, {}, &RefRow::name); - return internal::MakeMetadataTableStream(*schema(), std::move(rows), AppendRef); + return internal::MakeMetadataTableStream( + *schema(), std::move(rows), [](ArrowRowBuilder& builder, const RefRow& ref) { + return AppendRef(builder, ref); + }); } } // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 9e9d7906a1..d7152052a3 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -26,14 +26,10 @@ #include #include -#include - -#include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/arrow_row_builder_internal.h" -#include "iceberg/nanoarrow_status_internal.h" +#include "iceberg/inspect/metadata_table_stream_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_metadata.h" @@ -80,69 +76,25 @@ Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) { return builder.FinishRow(); } -class SnapshotsTableStream { +class SnapshotsIterator : public Stream> { public: - static Result> Make( - std::shared_ptr metadata, const iceberg::Schema& schema) { - ArrowSchema arrow_schema{}; - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); - return std::unique_ptr( - new SnapshotsTableStream(std::move(metadata), std::move(arrow_schema))); - } - - ~SnapshotsTableStream() = default; - - Status Close() { - metadata_.reset(); - if (arrow_schema_.release != nullptr) { - ArrowSchemaRelease(&arrow_schema_); - } - return {}; - } + explicit SnapshotsIterator(std::shared_ptr metadata) + : metadata_(std::move(metadata)) {} - Result> Next() { - if (metadata_ == nullptr) [[unlikely]] { - return InvalidArgument("Cannot read from a closed snapshots table stream"); - } + protected: + Result>> NextImpl() override { const auto& snapshots = metadata_->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) { + while (next_snapshot_ < snapshots.size()) { const auto& snapshot = snapshots[next_snapshot_++]; - if (snapshot == nullptr) [[unlikely]] { - continue; + if (snapshot != nullptr) { + return snapshot; } - 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; + return std::nullopt; } private: - SnapshotsTableStream(std::shared_ptr metadata, - ArrowSchema arrow_schema) - : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema)) {} - std::shared_ptr metadata_; - ArrowSchema arrow_schema_{}; size_t next_snapshot_ = 0; }; @@ -174,9 +126,13 @@ Result> SnapshotsTable::Make( } Result SnapshotsTable::Scan() { - ICEBERG_ASSIGN_OR_RAISE( - auto stream, SnapshotsTableStream::Make(source_table()->metadata(), *schema())); - return MakeArrowArrayStream(std::move(stream)); + std::unique_ptr>> rows = + std::make_unique(source_table()->metadata()); + return internal::MakeMetadataTableStream( + *schema(), std::move(rows), + [](ArrowRowBuilder& builder, const std::shared_ptr& snapshot) { + return AppendSnapshot(builder, *snapshot); + }); } } // namespace iceberg diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index f215198285..e5b76c6e4a 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -191,10 +191,11 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES + inspect/history_table_test.cc + inspect/metadata_table_stream_test.cc + inspect/metadata_table_test.cc inspect/refs_table_test.cc - history_table_test.cc - metadata_table_test.cc - snapshots_table_test.cc) + inspect/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/inspect/history_table_test.cc similarity index 96% rename from src/iceberg/test/history_table_test.cc rename to src/iceberg/test/inspect/history_table_test.cc index 8da311c024..2d1f555127 100644 --- a/src/iceberg/test/history_table_test.cc +++ b/src/iceberg/test/inspect/history_table_test.cc @@ -28,8 +28,8 @@ #include "iceberg/inspect/metadata_table.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" #include "iceberg/test/matchers.h" -#include "iceberg/test/metadata_table_test_base.h" #include "iceberg/type.h" namespace iceberg { diff --git a/src/iceberg/test/inspect/metadata_table_stream_test.cc b/src/iceberg/test/inspect/metadata_table_stream_test.cc new file mode 100644 index 0000000000..e165206d5e --- /dev/null +++ b/src/iceberg/test/inspect/metadata_table_stream_test.cc @@ -0,0 +1,236 @@ +/* + * 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 +#include + +#include +#include +#include +#include +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/inspect/metadata_table_stream_internal.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/stream.h" + +namespace iceberg::internal { +namespace { + +Status AppendRow(ArrowRowBuilder& builder, const int64_t& row) { + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(0), row)); + return builder.FinishRow(); +} + +struct SourceState { + int64_t reads = 0; + int destructions = 0; +}; + +class CountingRows : public Stream { + public: + CountingRows(std::shared_ptr state, int64_t count, + std::optional fail_at = std::nullopt) + : state_(std::move(state)), count_(count), fail_at_(fail_at) {} + + ~CountingRows() override { ++state_->destructions; } + + protected: + Result> NextImpl() override { + ++state_->reads; + if (fail_at_ == next_) { + return IOError("Row source failed"); + } + if (next_ == count_) { + return std::nullopt; + } + return next_++; + } + + private: + std::shared_ptr state_; + int64_t count_; + std::optional fail_at_; + int64_t next_ = 0; +}; + +class MetadataTableStreamTest : public MetadataTableTestBase { + protected: + Schema schema_{std::vector{SchemaField::MakeRequired(1, "id", int64())}}; +}; + +TEST_F(MetadataTableStreamTest, VectorRowsSpanSeveralBatchesWithoutLoss) { + std::vector rows(2 * MetadataTable::kBatchSize + 7); + std::iota(rows.begin(), rows.end(), 0); + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), AppendRow)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 3); + int64_t expected = 0; + for (const auto& batch : batches) { + EXPECT_LE(batch->num_rows(), MetadataTable::kBatchSize); + ASSERT_EQ(batch->num_columns(), 1); + ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::INT64); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(0)); + for (int64_t i = 0; i < batch->num_rows(); ++i) { + EXPECT_EQ(ids->Value(i), expected++); + } + } + EXPECT_EQ(expected, 2 * MetadataTable::kBatchSize + 7); +} + +TEST_F(MetadataTableStreamTest, LazyRowsAreReadOnlyForRequestedBatches) { + auto state = std::make_shared(); + std::unique_ptr> rows = + std::make_unique(state, MetadataTable::kBatchSize + 1); + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), AppendRow)); + nanoarrow::UniqueArrayStream owned(&stream); + EXPECT_EQ(state->reads, 0); + nanoarrow::UniqueArray batch; + ASSERT_EQ(owned->get_next(owned.get(), batch.get()), 0); + ASSERT_NE(batch->release, nullptr); + EXPECT_EQ(batch->length, MetadataTable::kBatchSize); + EXPECT_EQ(state->reads, MetadataTable::kBatchSize); + batch.reset(); + ASSERT_EQ(owned->get_next(owned.get(), batch.get()), 0); + ASSERT_NE(batch->release, nullptr); + EXPECT_EQ(batch->length, 1); + EXPECT_EQ(state->reads, MetadataTable::kBatchSize + 2); + batch.reset(); + for (int i = 0; i < 3; ++i) { + ASSERT_EQ(owned->get_next(owned.get(), batch.get()), 0); + EXPECT_EQ(batch->release, nullptr); + } + EXPECT_EQ(state->reads, MetadataTable::kBatchSize + 2); + owned.reset(); + EXPECT_EQ(state->destructions, 1); +} + +TEST_F(MetadataTableStreamTest, FailedAppendStopsCStreamPermanently) { + auto state = std::make_shared(); + std::unique_ptr> rows = + std::make_unique(state, 3 * MetadataTable::kBatchSize); + int64_t appended = 0; + auto append = [&appended](ArrowRowBuilder& builder, const int64_t& row) -> Status { + ++appended; + if (row == MetadataTable::kBatchSize + 3) { + return InvalidArrowData("Cannot append row"); + } + return AppendRow(builder, row); + }; + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), append)); + nanoarrow::UniqueArrayStream owned(&stream); + nanoarrow::UniqueArray batch; + ASSERT_EQ(owned->get_next(owned.get(), batch.get()), 0); + EXPECT_EQ(batch->length, MetadataTable::kBatchSize); + batch.reset(); + for (int i = 0; i < 3; ++i) { + EXPECT_EQ(owned->get_next(owned.get(), batch.get()), EIO); + EXPECT_EQ(batch->release, nullptr); + EXPECT_STREQ(owned->get_last_error(owned.get()), "Cannot append row"); + } + EXPECT_EQ(state->reads, MetadataTable::kBatchSize + 4); + EXPECT_EQ(appended, MetadataTable::kBatchSize + 4); +} + +TEST_F(MetadataTableStreamTest, FailedSourceStopsCStreamPermanently) { + auto state = std::make_shared(); + std::unique_ptr> rows = std::make_unique(state, 2000, 3); + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), AppendRow)); + nanoarrow::UniqueArrayStream owned(&stream); + nanoarrow::UniqueArray batch; + for (int i = 0; i < 3; ++i) { + EXPECT_EQ(owned->get_next(owned.get(), batch.get()), EIO); + EXPECT_EQ(batch->release, nullptr); + EXPECT_STREQ(owned->get_last_error(owned.get()), "Row source failed"); + } + EXPECT_EQ(state->reads, 4); +} + +TEST_F(MetadataTableStreamTest, RejectsNullSourceAndEmptyCallbacks) { + EXPECT_THAT( + MakeMetadataTableStream(schema_, std::unique_ptr>{}, AppendRow), + IsError(ErrorKind::kInvalidArgument)); + std::function empty; + EXPECT_THAT(MakeMetadataTableStream(schema_, std::vector{}, empty), + IsError(ErrorKind::kInvalidArgument)); + decltype(&AppendRow) null_callback = nullptr; + EXPECT_THAT(MakeMetadataTableStream(schema_, std::vector{}, null_callback), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST_F(MetadataTableStreamTest, EmptySourceRemainsExhausted) { + auto state = std::make_shared(); + std::unique_ptr> rows = std::make_unique(state, 0); + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), AppendRow)); + nanoarrow::UniqueArrayStream owned(&stream); + nanoarrow::UniqueArray batch; + for (int i = 0; i < 3; ++i) { + ASSERT_EQ(owned->get_next(owned.get(), batch.get()), 0); + EXPECT_EQ(batch->release, nullptr); + } + EXPECT_EQ(state->reads, 1); +} + +TEST_F(MetadataTableStreamTest, CloseReleasesSourceAndRejectsReads) { + auto state = std::make_shared(); + using RowsStream = MetadataTableRowsStream; + ICEBERG_UNWRAP_OR_FAIL( + auto stream, + RowsStream::Make(schema_, std::make_unique(state, 1), AppendRow)); + ASSERT_THAT(stream->Close(), IsOk()); + EXPECT_EQ(state->destructions, 1); + EXPECT_THAT(stream->Next(), IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(stream->Schema(), IsError(ErrorKind::kInvalidArgument)); + ASSERT_THAT(stream->Close(), IsOk()); + stream.reset(); + EXPECT_EQ(state->destructions, 1); +} + +TEST_F(MetadataTableStreamTest, EarlyReleaseDestroysSourceOnce) { + auto state = std::make_shared(); + std::unique_ptr> rows = std::make_unique(state, 2000); + ICEBERG_UNWRAP_OR_FAIL(auto stream, + MakeMetadataTableStream(schema_, std::move(rows), AppendRow)); + nanoarrow::UniqueArrayStream owned(&stream); + owned.reset(); + EXPECT_EQ(state->reads, 0); + EXPECT_EQ(state->destructions, 1); +} + +} // namespace +} // namespace iceberg::internal diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/inspect/metadata_table_test.cc similarity index 100% rename from src/iceberg/test/metadata_table_test.cc rename to src/iceberg/test/inspect/metadata_table_test.cc diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/inspect/metadata_table_test_base.h similarity index 99% rename from src/iceberg/test/metadata_table_test_base.h rename to src/iceberg/test/inspect/metadata_table_test_base.h index b9f601cc1c..c78222b6f5 100644 --- a/src/iceberg/test/metadata_table_test_base.h +++ b/src/iceberg/test/inspect/metadata_table_test_base.h @@ -26,7 +26,9 @@ #pragma once +#include #include +#include #include #include #include diff --git a/src/iceberg/test/inspect/refs_table_test.cc b/src/iceberg/test/inspect/refs_table_test.cc index ca7fc75313..ea86e2bfd5 100644 --- a/src/iceberg/test/inspect/refs_table_test.cc +++ b/src/iceberg/test/inspect/refs_table_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -29,16 +30,19 @@ #include #include #include +#include #include #include "iceberg/constants.h" #include "iceberg/inspect/metadata_table.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" #include "iceberg/snapshot.h" #include "iceberg/table.h" -#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" #include "iceberg/test/matchers.h" -#include "iceberg/test/metadata_table_test_base.h" -#include "iceberg/util/macros.h" +#include "iceberg/type.h" namespace iceberg { namespace { @@ -46,17 +50,9 @@ namespace { class RefsTableTest : public MetadataTableTestBase, public ::testing::WithParamInterface { protected: - Result> MakeTable( - std::vector> snapshots, int64_t current_snapshot_id, - std::unordered_map> refs = {}) { - auto metadata = std::make_shared(*metadata_); - metadata->format_version = GetParam(); - metadata->snapshots = std::move(snapshots); - metadata->current_snapshot_id = current_snapshot_id; - metadata->refs = std::move(refs); - return Table::Make( - TableIdentifier{.ns = Namespace{.levels = {"db"}}, .name = "table"}, - std::move(metadata), "s3://bucket/metadata.json", io_, catalog_); + void SetUp() override { + MetadataTableTestBase::SetUp(); + metadata_->format_version = GetParam(); } }; @@ -69,22 +65,11 @@ TEST_P(RefsTableTest, ScansBranchesAndTagsAsRefs) { refs.emplace("dev", std::move(dev_ref)); refs.emplace("release", std::move(release_ref)); - auto first = std::make_shared(Snapshot{ - .snapshot_id = 1, - .sequence_number = 1, - .timestamp_ms = TimePointMsFromUnixMs(1000), - .manifest_list = "unused-1.avro", - }); - auto second = std::make_shared(Snapshot{ - .snapshot_id = 2, - .parent_snapshot_id = 1, - .sequence_number = 2, - .timestamp_ms = TimePointMsFromUnixMs(2000), - .manifest_list = "unused-2.avro", - }); - ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second}, 2, std::move(refs))); - - ICEBERG_UNWRAP_OR_FAIL(auto refs_table, MetadataTable::Make(table)); + auto [first, second] = MakeTestSnapshots(); + metadata_->snapshots = {first, second}; + metadata_->current_snapshot_id = 2; + metadata_->refs = std::move(refs); + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, MetadataTable::Make(table_)); EXPECT_EQ(refs_table->kind(), MetadataTable::Kind::kRefs); EXPECT_FALSE(refs_table->supports_time_travel()); Schema expected_schema({ @@ -100,6 +85,13 @@ TEST_P(RefsTableTest, ScansBranchesAndTagsAsRefs) { ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); ASSERT_EQ(batches.size(), 1); ASSERT_EQ(batches[0]->num_rows(), 3); + ASSERT_EQ(batches[0]->num_columns(), 6); + ASSERT_EQ(batches[0]->column(0)->type_id(), ::arrow::Type::STRING); + ASSERT_EQ(batches[0]->column(1)->type_id(), ::arrow::Type::STRING); + ASSERT_EQ(batches[0]->column(2)->type_id(), ::arrow::Type::INT64); + ASSERT_EQ(batches[0]->column(3)->type_id(), ::arrow::Type::INT64); + ASSERT_EQ(batches[0]->column(4)->type_id(), ::arrow::Type::INT32); + ASSERT_EQ(batches[0]->column(5)->type_id(), ::arrow::Type::INT64); auto names = std::static_pointer_cast<::arrow::StringArray>(batches[0]->column(0)); auto types = std::static_pointer_cast<::arrow::StringArray>(batches[0]->column(1)); auto ids = std::static_pointer_cast<::arrow::Int64Array>(batches[0]->column(2)); @@ -129,14 +121,49 @@ TEST_P(RefsTableTest, ScansBranchesAndTagsAsRefs) { } TEST_P(RefsTableTest, ScanEmpty) { - ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({}, kInvalidSnapshotId)); - ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table_)); ICEBERG_UNWRAP_OR_FAIL(auto stream, metadata_table->Scan()); ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); EXPECT_TRUE(batches.empty()); } -INSTANTIATE_TEST_SUITE_P(FormatVersions, RefsTableTest, ::testing::Values(2, 3)); +TEST_P(RefsTableTest, ScansRefsAcrossBatchBoundary) { + for (int64_t i = 0; i < MetadataTable::kBatchSize + 1; ++i) { + ICEBERG_UNWRAP_OR_FAIL(auto ref, SnapshotRef::MakeTag(i + 1)); + metadata_->refs.emplace("tag-" + std::to_string(i), std::move(ref)); + } + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, RefsTable::Make(table_)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, refs_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 2); + EXPECT_EQ(batches[0]->num_rows(), MetadataTable::kBatchSize); + EXPECT_EQ(batches[1]->num_rows(), 1); + std::string previous; + for (const auto& batch : batches) { + ASSERT_EQ(batch->num_columns(), 6); + ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::STRING); + auto names = std::static_pointer_cast<::arrow::StringArray>(batch->column(0)); + for (int64_t i = 0; i < batch->num_rows(); ++i) { + auto name = names->GetString(i); + EXPECT_LT(previous, name); + previous = std::move(name); + } + } +} + +TEST(RefsTableConstructionTest, RejectsNullTable) { + EXPECT_THAT(RefsTable::Make(nullptr), IsError(ErrorKind::kInvalidArgument)); +} + +TEST_P(RefsTableTest, RejectsNullReferenceMetadata) { + metadata_->refs.emplace("broken", nullptr); + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, RefsTable::Make(table_)); + EXPECT_THAT(refs_table->Scan(), + ::testing::AllOf(IsError(ErrorKind::kInvalid), + HasErrorMessage("Snapshot reference 'broken' is null"))); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, RefsTableTest, ::testing::Values(1, 2, 3)); } // namespace } // namespace iceberg diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/inspect/snapshots_table_test.cc similarity index 99% rename from src/iceberg/test/snapshots_table_test.cc rename to src/iceberg/test/inspect/snapshots_table_test.cc index cddfce69f8..c164eeba1f 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/inspect/snapshots_table_test.cc @@ -33,8 +33,8 @@ #include "iceberg/constants.h" #include "iceberg/inspect/metadata_table.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" #include "iceberg/test/matchers.h" -#include "iceberg/test/metadata_table_test_base.h" namespace iceberg { namespace {