diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index eee720613..c5be41f41 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -63,6 +63,7 @@ set(ICEBERG_SOURCES inspect/metadata_table.cc inspect/metadata_scan_task.cc inspect/metadata_scan_task_reader.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 6bcc5b742..a39ae138e 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/src/iceberg/cpp20_compatibility_internal.h @@ -85,6 +85,7 @@ #include "iceberg/inspect/metadata_scan_task.h" #include "iceberg/inspect/metadata_scan_task_reader.h" #include "iceberg/inspect/metadata_table.h" +#include "iceberg/inspect/refs_table.h" #include "iceberg/inspect/snapshots_table.h" #include "iceberg/labels.h" #include "iceberg/location_provider.h" diff --git a/src/iceberg/inspect/metadata_scan_task.cc b/src/iceberg/inspect/metadata_scan_task.cc index afb3eb106..07195e16f 100644 --- a/src/iceberg/inspect/metadata_scan_task.cc +++ b/src/iceberg/inspect/metadata_scan_task.cc @@ -34,7 +34,8 @@ Result> StaticMetadataTask::Make( ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); ICEBERG_PRECHECK(projected_schema != nullptr, "Projected schema cannot be null"); if (table_kind != MetadataTableKind::kSnapshots && - table_kind != MetadataTableKind::kHistory) { + table_kind != MetadataTableKind::kHistory && + table_kind != MetadataTableKind::kRefs) { return NotImplemented("Static metadata table kind is not implemented"); } return std::shared_ptr(new StaticMetadataTask( diff --git a/src/iceberg/inspect/metadata_scan_task.h b/src/iceberg/inspect/metadata_scan_task.h index b3511e949..6e0dac54e 100644 --- a/src/iceberg/inspect/metadata_scan_task.h +++ b/src/iceberg/inspect/metadata_scan_task.h @@ -35,7 +35,7 @@ class ICEBERG_EXPORT MetadataScanTask { public: /// \brief Task category, determines how the reader opens the task. enum class TaskKind { - /// StaticMetadataTask reads in-memory table metadata (snapshots, history). + /// StaticMetadataTask reads in-memory table metadata (snapshots, history, refs). kStatic, }; @@ -48,7 +48,7 @@ class ICEBERG_EXPORT MetadataScanTask { virtual MetadataTableKind table_kind() const = 0; }; -/// \brief Task for static metadata tables (snapshots, history). +/// \brief Task for static metadata tables (snapshots, history, refs). /// /// Holds the metadata captured for a table scan; readers produce Arrow batches /// on demand. The metadata and its nested objects must not be modified while @@ -57,7 +57,7 @@ class ICEBERG_EXPORT StaticMetadataTask : public MetadataScanTask { public: /// \brief Create a static metadata task. /// - /// \param table_kind The metadata table kind (kSnapshots or kHistory) + /// \param table_kind The metadata table kind (kSnapshots, kHistory, or kRefs) /// \param metadata The table metadata to scan /// \param projected_schema The planned output schema /// \return A task, or an error for null metadata or an unsupported table kind. diff --git a/src/iceberg/inspect/metadata_scan_task_reader.cc b/src/iceberg/inspect/metadata_scan_task_reader.cc index 1ce5b3c1a..74acaa8f3 100644 --- a/src/iceberg/inspect/metadata_scan_task_reader.cc +++ b/src/iceberg/inspect/metadata_scan_task_reader.cc @@ -25,6 +25,7 @@ #include "iceberg/inspect/history_table.h" #include "iceberg/inspect/metadata_scan_task.h" #include "iceberg/inspect/metadata_table.h" +#include "iceberg/inspect/refs_table.h" #include "iceberg/inspect/snapshots_table.h" #include "iceberg/util/checked_cast.h" #include "iceberg/util/macros.h" @@ -53,6 +54,9 @@ Result MetadataScanTaskReader::Open( return SnapshotsTable::ReadRows(static_task.metadata(), static_task.projected_schema(), options_.batch_size); + case MetadataTableKind::kRefs: + return RefsTable::ReadRows(static_task.metadata(), + static_task.projected_schema(), options_.batch_size); case MetadataTableKind::kHistory: return HistoryTable::ReadRows(static_task.metadata(), static_task.projected_schema(), diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 6bd17644f..012ffaa99 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -24,6 +24,7 @@ #include "iceberg/inspect/history_table.h" #include "iceberg/inspect/metadata_scan.h" +#include "iceberg/inspect/refs_table.h" #include "iceberg/inspect/snapshots_table.h" namespace iceberg { @@ -38,6 +39,8 @@ Result> MetadataTable::Make( switch (kind) { case MetadataTableKind::kSnapshots: return SnapshotsTable::Make(std::move(table)); + case MetadataTableKind::kRefs: + return RefsTable::Make(std::move(table)); case MetadataTableKind::kHistory: return HistoryTable::Make(std::move(table)); default: diff --git a/src/iceberg/inspect/refs_table.cc b/src/iceberg/inspect/refs_table.cc new file mode 100644 index 000000000..c57e092bd --- /dev/null +++ b/src/iceberg/inspect/refs_table.cc @@ -0,0 +1,179 @@ +/* + * 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 "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_scan_task.h" +#include "iceberg/inspect/metadata_scan_util_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)); +} + +constexpr std::array, 6> kRefColumns = { + {{.field_id = 1, + .append = [](ArrowArray* column, + const RefRow& row) { return AppendString(column, row.name); }}, + {.field_id = 2, + .append = [](ArrowArray* column, + const RefRow& row) { return AppendString(column, row.type); }}, + {.field_id = 3, + .append = [](ArrowArray* column, + const RefRow& row) { return AppendInt(column, row.snapshot_id); }}, + {.field_id = 4, + .append = + [](ArrowArray* column, const RefRow& row) { + return AppendOptional(column, row.max_ref_age_ms); + }}, + {.field_id = 5, + .append = + [](ArrowArray* column, const RefRow& row) { + return AppendOptional(column, row.min_snapshots_to_keep); + }}, + {.field_id = 6, .append = [](ArrowArray* column, const RefRow& row) { + return AppendOptional(column, row.max_snapshot_age_ms); + }}}}; + +} // namespace + +// RefsTable implementation +RefsTable::RefsTable(std::shared_ptr table) + : MetadataTable(std::move(table)) {} + +RefsTable::~RefsTable() = default; + +const std::shared_ptr& RefsTable::schema() const { return CanonicalSchema(); } + +const std::shared_ptr& RefsTable::CanonicalSchema() { + 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::MakeScan( + internal::MetadataScanContext context) const { + return RefsScan::Make(std::move(context)); +} + +Result RefsTable::ReadRows( + std::shared_ptr metadata, + const std::shared_ptr& projected_schema, int64_t batch_size) { + std::vector rows; + const auto& refs = 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 = branch != nullptr ? "BRANCH" : "TAG", + .snapshot_id = ref->snapshot_id, + .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); + const size_t num_rows = rows.size(); + return internal::MakeMetadataBatchStream( + CanonicalSchema(), projected_schema, kRefColumns, num_rows, batch_size, + [rows = std::move(rows)](size_t index) -> const RefRow& { return rows[index]; }); +} + +// RefsScan implementation +RefsScan::RefsScan(internal::MetadataScanContext context, + std::shared_ptr task) + : MetadataScan(std::move(context)), task_(std::move(task)) {} + +RefsScan::~RefsScan() = default; + +Result> RefsScan::Make(internal::MetadataScanContext context) { + ICEBERG_PRECHECK(context.kind == MetadataTableKind::kRefs, + "Unexpected metadata table kind for refs scan"); + if (context.snapshot_id.has_value() || !context.ref_name.empty() || + context.as_of_time_in_millis.has_value()) { + return NotSupported("Metadata table does not support time travel"); + } + ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot be null"); + ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be null"); + ICEBERG_PRECHECK(context.projected_schema != nullptr, + "Projected schema cannot be null"); + ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns( + *RefsTable::CanonicalSchema(), *context.projected_schema, kRefColumns)); + ICEBERG_ASSIGN_OR_RAISE(auto task, + StaticMetadataTask::Make(context.kind, context.table_metadata, + context.projected_schema)); + return std::unique_ptr(new RefsScan(std::move(context), std::move(task))); +} + +Result RefsScan::PlanTasksStream() const { + return MetadataScanTaskStream::FromVector({task_}); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/refs_table.h b/src/iceberg/inspect/refs_table.h new file mode 100644 index 000000000..baa11f192 --- /dev/null +++ b/src/iceberg/inspect/refs_table.h @@ -0,0 +1,81 @@ +/* + * 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 refs metadata table. + +#include +#include + +#include "iceberg/arrow_c_data.h" +#include "iceberg/iceberg_export.h" +#include "iceberg/inspect/metadata_scan.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Refs metadata table. +class ICEBERG_EXPORT RefsTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr table); + + /// \brief Return the canonical schema of the refs metadata table. + static const std::shared_ptr& CanonicalSchema(); + + ~RefsTable() override; + + MetadataTableKind kind() const noexcept override { return MetadataTableKind::kRefs; } + + const std::shared_ptr& schema() const override; + + protected: + Result> MakeScan( + internal::MetadataScanContext context) const override; + + private: + friend class MetadataScanTaskReader; + + static Result ReadRows( + std::shared_ptr metadata, + const std::shared_ptr& projected_schema, int64_t batch_size); + + explicit RefsTable(std::shared_ptr table); +}; + +/// \brief Scan the refs metadata captured from a table. +class ICEBERG_EXPORT RefsScan final : public MetadataScan { + public: + /// \brief Create a scan from captured metadata table settings. + static Result> Make(internal::MetadataScanContext context); + + ~RefsScan() override; + + Result PlanTasksStream() const override; + + private: + RefsScan(internal::MetadataScanContext context, std::shared_ptr task); + + const std::shared_ptr task_; +}; + +} // namespace iceberg diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index b2fd4f9be..1cbc77e67 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -206,9 +206,10 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES - history_table_test.cc - metadata_table_test.cc - snapshots_table_test.cc) + inspect/history_table_test.cc + inspect/metadata_table_test.cc + inspect/refs_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 99% rename from src/iceberg/test/history_table_test.cc rename to src/iceberg/test/inspect/history_table_test.cc index 1dbcdefbf..308b5353d 100644 --- a/src/iceberg/test/history_table_test.cc +++ b/src/iceberg/test/inspect/history_table_test.cc @@ -32,8 +32,8 @@ #include "iceberg/snapshot.h" #include "iceberg/table_metadata.h" #include "iceberg/test/arrow_test_util.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" #include "iceberg/util/timepoint.h" diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/inspect/metadata_table_test.cc similarity index 99% rename from src/iceberg/test/metadata_table_test.cc rename to src/iceberg/test/inspect/metadata_table_test.cc index 7babd9171..2566180e0 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/inspect/metadata_table_test.cc @@ -40,8 +40,8 @@ #include "iceberg/table.h" #include "iceberg/table_metadata.h" #include "iceberg/test/arrow_test_util.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/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 de44c1d1d..df80b101c 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 new file mode 100644 index 000000000..1d5dba870 --- /dev/null +++ b/src/iceberg/test/inspect/refs_table_test.cc @@ -0,0 +1,275 @@ +/* + * 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 +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/metadata_scan_task.h" +#include "iceberg/inspect/metadata_scan_task_reader.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_metadata.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +constexpr int64_t kBatchSize = MetadataScanTaskReader::Options{}.batch_size; + +class RefsTableTest : public MetadataTableTestBase, + public ::testing::WithParamInterface { + protected: + void SetUp() override { + MetadataTableTestBase::SetUp(); + metadata_->format_version = GetParam(); + } +}; + +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, 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); + 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, Scan(*refs_table)); + 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)); + 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 metadata_table, MetadataTable::Make(table_)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*metadata_table)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); +} + +TEST_P(RefsTableTest, ScansRefsAcrossBatchBoundary) { + for (int64_t i = 0; i < 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, Scan(*refs_table)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 2); + EXPECT_EQ(batches[0]->num_rows(), 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_)); + ICEBERG_UNWRAP_OR_FAIL(auto builder, refs_table->NewScan()); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks()); + ASSERT_EQ(tasks.size(), 1); + ICEBERG_UNWRAP_OR_FAIL(auto reader, MetadataScanTaskReader::Make({})); + EXPECT_THAT(reader->Open(*tasks.front()), + ::testing::AllOf(IsError(ErrorKind::kInvalid), + HasErrorMessage("Snapshot reference 'broken' is null"))); + ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->ToArrowArrayStream()); + 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()), + "Snapshot reference 'broken' is null"); + } +} + +TEST_P(RefsTableTest, ProjectedTasksRetainInputsAndSupportCustomBatchSizes) { + ICEBERG_UNWRAP_OR_FAIL(auto tag, SnapshotRef::MakeTag(1)); + ICEBERG_UNWRAP_OR_FAIL(auto branch, SnapshotRef::MakeBranch(2)); + metadata_->refs.emplace("a-tag", std::move(tag)); + metadata_->refs.emplace("z-branch", std::move(branch)); + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, table_->Inspect(MetadataTableKind::kRefs)); + EXPECT_EQ(refs_table->kind(), MetadataTableKind::kRefs); + ICEBERG_UNWRAP_OR_FAIL(auto builder, refs_table->NewScan()); + auto projection = std::make_shared( + std::vector{SchemaField::MakeRequired(3, "snapshot_id", int64()), + SchemaField::MakeRequired(1, "name", string())}); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Project(projection).Build()); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks()); + ASSERT_EQ(tasks.size(), 1); + EXPECT_EQ(tasks.front()->table_kind(), MetadataTableKind::kRefs); + builder.reset(); + scan.reset(); + refs_table.reset(); + table_.reset(); + metadata_.reset(); + for (int64_t batch_size : {1, 2}) { + ICEBERG_UNWRAP_OR_FAIL(auto reader, + MetadataScanTaskReader::Make({.batch_size = batch_size})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*tasks.front())); + reader.reset(); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), batch_size == 1 ? 2 : 1); + int64_t row = 0; + for (const auto& batch : batches) { + EXPECT_LE(batch->num_rows(), batch_size); + ASSERT_EQ(batch->num_columns(), 2); + EXPECT_EQ(batch->column_name(0), "snapshot_id"); + EXPECT_EQ(batch->column_name(1), "name"); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(0)); + auto names = std::static_pointer_cast<::arrow::StringArray>(batch->column(1)); + for (int64_t i = 0; i < batch->num_rows(); ++i, ++row) { + EXPECT_EQ(ids->Value(i), row + 1); + EXPECT_EQ(names->GetString(i), row == 0 ? "a-tag" : "z-branch"); + } + } + EXPECT_EQ(row, 2); + } +} + +TEST_P(RefsTableTest, SelectsColumnsAndPreservesRowCountWithoutColumns) { + ICEBERG_UNWRAP_OR_FAIL(auto tag, SnapshotRef::MakeTag(1)); + metadata_->refs.emplace("tag", std::move(tag)); + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, RefsTable::Make(table_)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, ScanColumns(*refs_table, {"NAME"}, false)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches.front()->num_columns(), 1); + EXPECT_EQ(batches.front()->column_name(0), "name"); + auto names = std::static_pointer_cast<::arrow::StringArray>(batches.front()->column(0)); + EXPECT_EQ(names->GetString(0), "tag"); + ICEBERG_UNWRAP_OR_FAIL(auto empty_stream, ScanColumns(*refs_table, {})); + ICEBERG_UNWRAP_OR_FAIL(auto empty_batches, ReadAllBatches(std::move(empty_stream))); + ASSERT_EQ(empty_batches.size(), 1); + EXPECT_EQ(empty_batches.front()->num_columns(), 0); + EXPECT_EQ(empty_batches.front()->num_rows(), 1); +} + +TEST_P(RefsTableTest, RejectsTimeTravel) { + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, RefsTable::Make(table_)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, refs_table->NewScan()); + EXPECT_THAT(snapshot_builder->UseSnapshot(1).Build(), + IsError(ErrorKind::kNotSupported)); + ICEBERG_UNWRAP_OR_FAIL(auto ref_builder, refs_table->NewScan()); + EXPECT_THAT(ref_builder->UseRef("main").Build(), IsError(ErrorKind::kNotSupported)); + ICEBERG_UNWRAP_OR_FAIL(auto time_builder, refs_table->NewScan()); + EXPECT_THAT(time_builder->AsOfTime(1000).Build(), IsError(ErrorKind::kNotSupported)); +} + +TEST_P(RefsTableTest, RejectsInvalidProjection) { + ICEBERG_UNWRAP_OR_FAIL(auto refs_table, RefsTable::Make(table_)); + auto required_retention = std::make_shared(std::vector{ + SchemaField::MakeRequired(4, "max_reference_age_in_ms", int64())}); + ICEBERG_UNWRAP_OR_FAIL(auto builder, refs_table->NewScan()); + EXPECT_THAT(builder->Project(required_retention).Build(), + IsError(ErrorKind::kInvalidArgument)); + auto string_id = std::make_shared( + std::vector{SchemaField::MakeRequired(3, "snapshot_id", string())}); + ICEBERG_UNWRAP_OR_FAIL(auto type_builder, refs_table->NewScan()); + EXPECT_THAT(type_builder->Project(string_id).Build(), + IsError(ErrorKind::kNotSupported)); +} + +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 dea10a01c..b5029dc44 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/inspect/snapshots_table_test.cc @@ -35,8 +35,8 @@ #include "iceberg/inspect/metadata_scan_task_reader.h" #include "iceberg/inspect/metadata_table.h" #include "iceberg/test/arrow_test_util.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 { diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index c0230feb2..3502a56c3 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -290,6 +290,7 @@ class MetadataTable; class MetadataScan; class MetadataScanTask; class MetadataScanTaskReader; +class RefsTable; class SnapshotsTable; /// \brief Table encryption