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..94df35ded1 --- /dev/null +++ b/src/iceberg/inspect/metadata_table_stream_internal.h @@ -0,0 +1,153 @@ +/* + * 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 "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 Stream { + public: + static Result> Make( + 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() override { + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + } + + Status Close() { + rows_.reset(); + append_row_.reset(); + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + return {}; + } + + 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; + } + + 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 next_row, rows_->Next()); + if (!next_row.has_value()) { + break; + } + ICEBERG_RETURN_UNEXPECTED((*append_row_)(builder, *next_row)); + } + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } + + 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_; + std::optional append_row_; + ArrowSchema arrow_schema_{}; +}; + +template +Result MakeMetadataTableStream(const Schema& schema, + std::unique_ptr> rows, + AppendRow append_row) { + using RowsStream = MetadataTableRowsStream; + ICEBERG_ASSIGN_OR_RAISE( + auto stream, RowsStream::Make(schema, std::move(rows), 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..89c2116163 --- /dev/null +++ b/src/iceberg/inspect/refs_table.cc @@ -0,0 +1,122 @@ +/* + * 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 "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; + 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 = 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); + return internal::MakeMetadataTableStream( + *schema(), std::move(rows), [](ArrowRowBuilder& builder, const RefRow& ref) { + return AppendRef(builder, ref); + }); +} + +} // 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/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 058a9dd3cc..e5b76c6e4a 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -191,9 +191,11 @@ 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_stream_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 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 new file mode 100644 index 0000000000..ea86e2bfd5 --- /dev/null +++ b/src/iceberg/test/inspect/refs_table_test.cc @@ -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. + */ + +#include "iceberg/inspect/refs_table.h" + +#include +#include +#include +#include +#include +#include +#include + +#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_metadata.h" +#include "iceberg/test/inspect/metadata_table_test_base.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +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); + 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); + 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, metadata_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); +} + +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 { 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