diff --git a/src/iceberg/inspect/history_table.cc b/src/iceberg/inspect/history_table.cc index 03bbcca97f..d3cdbd89a3 100644 --- a/src/iceberg/inspect/history_table.cc +++ b/src/iceberg/inspect/history_table.cc @@ -50,7 +50,8 @@ Result> HistoryTable::Make(std::shared_ptr return std::unique_ptr(new HistoryTable(std::move(table))); } -Result HistoryTable::Scan() { +Result HistoryTable::ScanProjection( + const Schema& /*projected_schema*/) { return NotSupported("Scan is not supported for the history table"); } diff --git a/src/iceberg/inspect/history_table.h b/src/iceberg/inspect/history_table.h index 7d863fc2c0..28258ddf21 100644 --- a/src/iceberg/inspect/history_table.h +++ b/src/iceberg/inspect/history_table.h @@ -42,7 +42,8 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable { const std::shared_ptr& schema() const override; - Result Scan() override; + protected: + Result ScanProjection(const Schema& projected_schema) override; private: explicit HistoryTable(std::shared_ptr
table); diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 7bc94c5115..a8408bf67e 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -22,7 +22,23 @@ #include #include +#include "iceberg/schema.h" +#include "iceberg/util/macros.h" + namespace iceberg { +namespace { + +Result> SelectColumns(const std::shared_ptr& schema, + const ColumnSelection& columns) { + if (!columns.names.has_value()) { + return schema; + } + ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, + schema->Select(*columns.names, columns.case_sensitive)); + return std::shared_ptr(std::move(projected_schema)); +} + +} // namespace MetadataTable::MetadataTable(std::shared_ptr
source_table) : source_table_(std::move(source_table)) {} @@ -35,6 +51,13 @@ const std::shared_ptr
& MetadataTable::source_table() const { return source_table_; } +Result MetadataTable::Scan() { return ScanProjection(*schema()); } + +Result MetadataTable::ScanColumns(const ColumnSelection& columns) { + ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, SelectColumns(schema(), columns)); + return ScanProjection(*projected_schema); +} + TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr
source_table) : MetadataTable(std::move(source_table)) {} @@ -42,13 +65,20 @@ TimeTravelMetadataTable::~TimeTravelMetadataTable() = default; bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; } -Result TimeTravelMetadataTable::Scan() { - return ScanSnapshot(SnapshotSelection{}); -} - Result TimeTravelMetadataTable::Scan( const SnapshotSelection& snapshot_selection) { - return ScanSnapshot(snapshot_selection); + return ScanSnapshot(snapshot_selection, *schema()); +} + +Result TimeTravelMetadataTable::ScanColumns( + const SnapshotSelection& snapshot_selection, const ColumnSelection& columns) { + ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, SelectColumns(schema(), columns)); + return ScanSnapshot(snapshot_selection, *projected_schema); +} + +Result TimeTravelMetadataTable::ScanProjection( + const Schema& projected_schema) { + return ScanSnapshot(SnapshotSelection{}, projected_schema); } } // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 85000ee1e7..a48d6d664f 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -24,9 +24,11 @@ #include #include +#include #include #include #include +#include #include "iceberg/arrow_c_data.h" #include "iceberg/iceberg_export.h" @@ -36,6 +38,20 @@ namespace iceberg { +/// \brief Column selection for a metadata table scan. +struct ColumnSelection { + /// \brief Names of the columns to return. + /// + /// Nested names are dot-concatenated. Columns are returned in schema order, and + /// names that match no column are ignored. std::nullopt or a list containing `*` + /// returns all columns. An empty list returns rows without columns. Selecting a + /// map's value returns the whole map; selecting only its key is an error. + std::optional> names; + + /// \brief Whether names are matched case-sensitively. + bool case_sensitive = true; +}; + /// \brief Base interface for an Iceberg metadata table. class ICEBERG_EXPORT MetadataTable { public: @@ -73,15 +89,33 @@ class ICEBERG_EXPORT MetadataTable { /// \brief Return whether this metadata table supports time travel. virtual bool supports_time_travel() const noexcept; - /// \brief Scan the metadata table without time travel. + /// \brief Scan all columns of the metadata table without time travel. /// /// The caller owns the returned stream and must invoke its `release` callback /// when the stream is no longer needed. - virtual Result Scan() = 0; + Result Scan(); + + /// \brief Scan the selected columns of the metadata table without time travel. + /// + /// The caller owns the returned stream and must invoke its `release` callback + /// when the stream is no longer needed. + /// + /// This is not a Scan() overload so that `Scan({})` on a TimeTravelMetadataTable + /// still resolves to Scan(const SnapshotSelection&). + /// + /// \param columns Names of the columns to return. + /// \return An Arrow stream containing the selected columns, or an error. + Result ScanColumns(const ColumnSelection& columns); protected: explicit MetadataTable(std::shared_ptr
source_table); + /// \brief Implement a scan that returns the columns of a projected schema. + /// + /// \param projected_schema Projection of schema() with the columns to return. It + /// is only valid during this call. + virtual Result ScanProjection(const Schema& projected_schema) = 0; + private: std::shared_ptr
source_table_; }; @@ -110,21 +144,37 @@ class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable { /// \brief Return true because this interface supports time travel. bool supports_time_travel() const noexcept final; - /// \brief Scan using the current snapshot on the main branch. - Result Scan() final; + // Scans without a snapshot selection read the current snapshot on the main branch. + using MetadataTable::Scan; + using MetadataTable::ScanColumns; - /// \brief Scan using the requested snapshot selection. + /// \brief Scan all columns using the requested snapshot selection. /// /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection. /// \return An Arrow stream containing the metadata table rows, or an error. Result Scan(const SnapshotSelection& snapshot_selection); + /// \brief Scan the selected columns using the requested snapshot selection. + /// + /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection. + /// \param columns Names of the columns to return. + /// \return An Arrow stream containing the selected columns, or an error. + Result ScanColumns(const SnapshotSelection& snapshot_selection, + const ColumnSelection& columns); + protected: explicit TimeTravelMetadataTable(std::shared_ptr
source_table); + /// \brief Scan the current snapshot on the main branch. + Result ScanProjection(const Schema& projected_schema) final; + /// \brief Implement a scan for the requested snapshot selection. + /// + /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection. + /// \param projected_schema Projection of schema() with the columns to return. It + /// is only valid during this call. virtual Result ScanSnapshot( - const SnapshotSelection& snapshot_selection) = 0; + const SnapshotSelection& snapshot_selection, const Schema& projected_schema) = 0; }; } // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table_columns_internal.h b/src/iceberg/inspect/metadata_table_columns_internal.h new file mode 100644 index 0000000000..dadecb168e --- /dev/null +++ b/src/iceberg/inspect/metadata_table_columns_internal.h @@ -0,0 +1,94 @@ +/* + * 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/metadata_table_columns_internal.h +/// \brief Build only the selected columns of metadata table rows. + +#include +#include +#include +#include +#include + +#include "iceberg/arrow_c_data.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/result.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg::internal { + +/// \brief Append the value of one metadata table column for a row. +template +using AppendColumnFn = Status (*)(ArrowArray* column, const Row& row); + +/// \brief A metadata table column and the function that appends its values. +template +struct MetadataTableColumn { + int32_t field_id; + AppendColumnFn append; +}; + +/// \brief Return the append functions for the columns of a projected schema. +/// +/// \param schema The metadata table schema. +/// \param projected_schema Projection of `schema` with the columns to return. +/// \param columns The columns of `schema` with their append functions. +/// \return One append function per column of `projected_schema`, in the same order, +/// or NotSupported if only some fields of a nested column are selected. +template +Result>> SelectAppendFns( + const Schema& schema, const Schema& projected_schema, + std::span> columns) { + std::vector> append_fns; + append_fns.reserve(projected_schema.fields().size()); + for (const auto& projected_field : projected_schema.fields()) { + ICEBERG_ASSIGN_OR_RAISE(auto field, schema.FindFieldById(projected_field.field_id())); + ICEBERG_CHECK(field.has_value(), "Metadata table column '{}' is not in the schema", + projected_field.name()); + if (*field->get().type() != *projected_field.type()) { + return NotSupported("Cannot select nested fields of metadata table column '{}'", + projected_field.name()); + } + auto column = std::ranges::find(columns, projected_field.field_id(), + &MetadataTableColumn::field_id); + ICEBERG_CHECK(column != columns.end(), + "Metadata table column '{}' has no append function", + projected_field.name()); + append_fns.push_back(column->append); + } + return append_fns; +} + +/// \brief Append the selected columns of a row and finish the row. +template +Status AppendRow(ArrowRowBuilder& builder, + std::span> append_fns, const Row& row) { + for (size_t index = 0; index < append_fns.size(); ++index) { + ICEBERG_RETURN_UNEXPECTED( + append_fns[index](builder.column(static_cast(index)), row)); + } + return builder.FinishRow(); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 9e9d7906a1..88837a483e 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -19,6 +19,7 @@ #include "iceberg/inspect/snapshots_table.h" +#include #include #include #include @@ -30,6 +31,7 @@ #include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table_columns_internal.h" #include "iceberg/nanoarrow_status_internal.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" @@ -43,51 +45,66 @@ namespace iceberg { namespace { -Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) { - ICEBERG_RETURN_UNEXPECTED( - AppendInt(builder.column(0), std::chrono::duration_cast( - snapshot.timestamp_ms.time_since_epoch()) - .count())); - ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot.snapshot_id)); +Status AppendCommittedAt(ArrowArray* column, const Snapshot& snapshot) { + return AppendInt(column, std::chrono::duration_cast( + snapshot.timestamp_ms.time_since_epoch()) + .count()); +} + +Status AppendSnapshotId(ArrowArray* column, const Snapshot& snapshot) { + return AppendInt(column, snapshot.snapshot_id); +} +Status AppendParentId(ArrowArray* column, const Snapshot& snapshot) { if (snapshot.parent_snapshot_id.has_value()) { - ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), *snapshot.parent_snapshot_id)); - } else { - ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + return AppendInt(column, *snapshot.parent_snapshot_id); } + return AppendNull(column); +} +Status AppendOperation(ArrowArray* column, const Snapshot& snapshot) { auto operation = snapshot.Operation(); if (operation.has_value()) { - ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation)); - } else { - ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + return AppendString(column, *operation); } + return AppendNull(column); +} - ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list)); +Status AppendManifestList(ArrowArray* column, const Snapshot& snapshot) { + return AppendString(column, snapshot.manifest_list); +} - if (!snapshot.summary.empty()) { - if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) { - auto summary = snapshot.summary; - summary.erase(SnapshotSummaryFields::kOperation); - ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary)); - } else { - ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot.summary)); - } - } else { - ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5))); +Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) { + if (snapshot.summary.empty()) { + return AppendNull(column); } - - return builder.FinishRow(); + if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) { + auto summary = snapshot.summary; + summary.erase(SnapshotSummaryFields::kOperation); + return AppendStringMap(column, summary); + } + return AppendStringMap(column, snapshot.summary); } +constexpr std::array, 6> kSnapshotColumns{{ + {.field_id = 1, .append = AppendCommittedAt}, + {.field_id = 2, .append = AppendSnapshotId}, + {.field_id = 3, .append = AppendParentId}, + {.field_id = 4, .append = AppendOperation}, + {.field_id = 5, .append = AppendManifestList}, + {.field_id = 6, .append = AppendSummary}, +}}; + class SnapshotsTableStream { public: static Result> Make( - std::shared_ptr metadata, const iceberg::Schema& schema) { + std::shared_ptr metadata, + const iceberg::Schema& projected_schema, + std::vector> append_fns) { ArrowSchema arrow_schema{}; - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); - return std::unique_ptr( - new SnapshotsTableStream(std::move(metadata), std::move(arrow_schema))); + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(projected_schema, &arrow_schema)); + return std::unique_ptr(new SnapshotsTableStream( + std::move(metadata), std::move(append_fns), std::move(arrow_schema))); } ~SnapshotsTableStream() = default; @@ -116,7 +133,8 @@ class SnapshotsTableStream { if (snapshot == nullptr) [[unlikely]] { continue; } - ICEBERG_RETURN_UNEXPECTED(AppendSnapshot(builder, *snapshot)); + ICEBERG_RETURN_UNEXPECTED( + internal::AppendRow(builder, append_fns_, *snapshot)); } if (builder.num_rows() == 0) { return std::nullopt; @@ -138,10 +156,14 @@ class SnapshotsTableStream { private: SnapshotsTableStream(std::shared_ptr metadata, + std::vector> append_fns, ArrowSchema arrow_schema) - : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema)) {} + : metadata_(std::move(metadata)), + append_fns_(std::move(append_fns)), + arrow_schema_(std::move(arrow_schema)) {} std::shared_ptr metadata_; + std::vector> append_fns_; ArrowSchema arrow_schema_{}; size_t next_snapshot_ = 0; }; @@ -173,9 +195,13 @@ Result> SnapshotsTable::Make( return std::unique_ptr(new SnapshotsTable(std::move(table))); } -Result SnapshotsTable::Scan() { +Result SnapshotsTable::ScanProjection(const Schema& projected_schema) { + ICEBERG_ASSIGN_OR_RAISE( + auto append_fns, + internal::SelectAppendFns(*schema(), projected_schema, kSnapshotColumns)); ICEBERG_ASSIGN_OR_RAISE( - auto stream, SnapshotsTableStream::Make(source_table()->metadata(), *schema())); + auto stream, SnapshotsTableStream::Make(source_table()->metadata(), + projected_schema, std::move(append_fns))); return MakeArrowArrayStream(std::move(stream)); } diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h index d2f0ddf909..bcc130209d 100644 --- a/src/iceberg/inspect/snapshots_table.h +++ b/src/iceberg/inspect/snapshots_table.h @@ -42,10 +42,11 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable { const std::shared_ptr& schema() const override; + protected: /// \brief Scan all snapshots as rows. /// /// The snapshots table always returns every known snapshot. - Result Scan() override; + Result ScanProjection(const Schema& projected_schema) override; private: explicit SnapshotsTable(std::shared_ptr
table); diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc index b014ef9627..8ae9063989 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -19,6 +19,13 @@ #include "iceberg/inspect/metadata_table.h" +#include +#include +#include +#include +#include +#include + #include #include @@ -36,6 +43,43 @@ #include "iceberg/type.h" namespace iceberg { +namespace { + +/// \brief Time-travel table that records the arguments of each scan. +class RecordingTimeTravelTable : public TimeTravelMetadataTable { + public: + explicit RecordingTimeTravelTable(std::shared_ptr
table) + : TimeTravelMetadataTable(std::move(table)) {} + + Kind kind() const noexcept override { return Kind::kSnapshots; } + + const std::shared_ptr& schema() const override { + static const auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeOptional(2, "name", string())}); + return schema; + } + + const SnapshotSelection& scanned_snapshot() const { return scanned_snapshot_; } + const std::vector& scanned_columns() const { return scanned_columns_; } + + protected: + Result ScanSnapshot(const SnapshotSelection& snapshot_selection, + const Schema& projected_schema) override { + scanned_snapshot_ = snapshot_selection; + scanned_columns_.clear(); + for (const auto& field : projected_schema.fields()) { + scanned_columns_.emplace_back(field.name()); + } + return NotSupported("Recording table does not return rows"); + } + + private: + SnapshotSelection scanned_snapshot_; + std::vector scanned_columns_; +}; + +} // namespace class MetadataTableTest : public ::testing::Test { protected: @@ -74,4 +118,39 @@ TEST_F(MetadataTableTest, SupportsTimeTravel) { EXPECT_FALSE(history_table->supports_time_travel()); } +TEST_F(MetadataTableTest, TimeTravelScanSelectsColumns) { + RecordingTimeTravelTable table(table_); + const ColumnSelection name_only{.names = std::vector{"name"}}; + + EXPECT_THAT(table.Scan(), IsError(ErrorKind::kNotSupported)); + EXPECT_TRUE(std::holds_alternative(table.scanned_snapshot().snapshot)); + EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name")); + + EXPECT_THAT(table.ScanColumns(name_only), IsError(ErrorKind::kNotSupported)); + EXPECT_TRUE(std::holds_alternative(table.scanned_snapshot().snapshot)); + EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("name")); + + const SnapshotSelection snapshot_one{.snapshot = int64_t{1}, .ref_name = "dev"}; + EXPECT_THAT(table.Scan(snapshot_one), IsError(ErrorKind::kNotSupported)); + EXPECT_EQ(table.scanned_snapshot().snapshot, snapshot_one.snapshot); + EXPECT_EQ(table.scanned_snapshot().ref_name, "dev"); + EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name")); + + EXPECT_THAT(table.ScanColumns(snapshot_one, name_only), + IsError(ErrorKind::kNotSupported)); + EXPECT_EQ(table.scanned_snapshot().snapshot, snapshot_one.snapshot); + EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("name")); + + EXPECT_THAT(table.Scan({}), IsError(ErrorKind::kNotSupported)); + EXPECT_TRUE(std::holds_alternative(table.scanned_snapshot().snapshot)); + EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name")); +} + +TEST_F(MetadataTableTest, HistoryTableScanWithColumnsIsNotSupported) { + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); + EXPECT_THAT(history_table->ScanColumns( + ColumnSelection{.names = std::vector{"snapshot_id"}}), + IsError(ErrorKind::kNotSupported)); +} + } // namespace iceberg diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc index cddfce69f8..7c300a52df 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/snapshots_table_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -132,6 +133,101 @@ TEST_F(SnapshotsTableTest, Scan) { EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2"))); } +TEST_F(SnapshotsTableTest, ScanSelectedColumns) { + ICEBERG_UNWRAP_OR_FAIL( + auto stream, snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{"operation", "snapshot_id"}})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + const auto& batch = batches.front(); + EXPECT_EQ(batch->num_rows(), 2); + ASSERT_EQ(batch->num_columns(), 2); + EXPECT_EQ(batch->column_name(0), "snapshot_id"); + EXPECT_EQ(batch->column_name(1), "operation"); + ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::INT64); + ASSERT_EQ(batch->column(1)->type_id(), ::arrow::Type::STRING); + + auto snapshot_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(0)); + EXPECT_EQ(snapshot_ids->Value(0), 1); + EXPECT_EQ(snapshot_ids->Value(1), 2); + auto operations = std::static_pointer_cast<::arrow::StringArray>(batch->column(1)); + EXPECT_EQ(operations->GetString(0), "append"); + EXPECT_EQ(operations->GetString(1), "append"); +} + +TEST_F(SnapshotsTableTest, ScanSelectedColumnsIgnoresUnknownNames) { + ICEBERG_UNWRAP_OR_FAIL(auto stream, + snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{"parent_id", "missing"}})); + 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), "parent_id"); +} + +TEST_F(SnapshotsTableTest, ScanSelectedColumnsCaseSensitivity) { + ICEBERG_UNWRAP_OR_FAIL( + auto insensitive_stream, + snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{"SNAPSHOT_ID"}, .case_sensitive = false})); + ICEBERG_UNWRAP_OR_FAIL(auto insensitive_batches, + ReadAllBatches(std::move(insensitive_stream))); + ASSERT_EQ(insensitive_batches.size(), 1); + ASSERT_EQ(insensitive_batches.front()->num_columns(), 1); + EXPECT_EQ(insensitive_batches.front()->column_name(0), "snapshot_id"); + + ICEBERG_UNWRAP_OR_FAIL(auto sensitive_stream, + snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{"SNAPSHOT_ID"}})); + ICEBERG_UNWRAP_OR_FAIL(auto sensitive_batches, + ReadAllBatches(std::move(sensitive_stream))); + ASSERT_EQ(sensitive_batches.size(), 1); + EXPECT_EQ(sensitive_batches.front()->num_columns(), 0); +} + +TEST_F(SnapshotsTableTest, ScanAllColumnsSelection) { + std::vector>> selections = { + std::nullopt, std::vector{"*"}}; + for (const auto& names : selections) { + ICEBERG_UNWRAP_OR_FAIL( + auto stream, snapshots_table_->ScanColumns(ColumnSelection{.names = names})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + EXPECT_EQ(batches.front()->num_rows(), 2); + EXPECT_EQ(batches.front()->num_columns(), 6); + } +} + +TEST_F(SnapshotsTableTest, ScanNoColumns) { + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{}})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + EXPECT_EQ(batches.front()->num_rows(), 2); + EXPECT_EQ(batches.front()->num_columns(), 0); +} + +TEST_F(SnapshotsTableTest, ScanSelectedMapColumn) { + ICEBERG_UNWRAP_OR_FAIL(auto stream, + snapshots_table_->ScanColumns(ColumnSelection{ + .names = std::vector{"summary.value"}})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + const auto& batch = batches.front(); + ASSERT_EQ(batch->num_columns(), 1); + EXPECT_EQ(batch->column_name(0), "summary"); + ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::MAP); + auto summaries = std::static_pointer_cast<::arrow::MapArray>(batch->column(0)); + EXPECT_THAT(GetMapEntries(summaries, 0), + ::testing::Contains(::testing::Pair("total-records", "1"))); + EXPECT_THAT(GetMapEntries(summaries, 1), + ::testing::Contains(::testing::Pair("total-records", "2"))); + + EXPECT_THAT(snapshots_table_->ScanColumns( + ColumnSelection{.names = std::vector{"summary.key"}}), + IsError(ErrorKind::kInvalidArgument)); +} + TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) { // A table with zero snapshots should return zero rows. ICEBERG_UNWRAP_OR_FAIL(