Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/iceberg/inspect/metadata_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,13 @@ MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,

MetadataTable::~MetadataTable() = default;

bool MetadataTable::supports_time_travel() const noexcept { return false; }

Result<ArrowArray> MetadataTable::Scan(
const std::optional<SnapshotSelection>& /*snapshot_selection*/) {
return NotSupported("Scan is not supported for this metadata table type");
}

Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(std::shared_ptr<Table> table,
Kind kind) {
if (table == nullptr) [[unlikely]] {
Expand Down
35 changes: 35 additions & 0 deletions src/iceberg/inspect/metadata_table.h
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,28 @@
/// \brief Define base APIs for metadata tables.

#include <memory>
#include <optional>
#include <string>

#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
#include "iceberg/table_identifier.h"
#include "iceberg/type_fwd.h"
#include "iceberg/util/timepoint.h"

namespace iceberg {

/// \brief Parameters for snapshot selection (time travel).
struct SnapshotSelection {
/// \brief The snapshot ID to read.
std::optional<int64_t> snapshot_id;
/// \brief Read the snapshot that was current at this timestamp.
std::optional<TimePointMs> as_of_timestamp;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should it be std::variant<std::monostate, int64_t, TimePointMs>? Otherwise we need to define the priority if both snapshot_id and as_of_timestamp are set.

/// \brief Read the snapshot referenced by this named ref (branch or tag).
std::optional<std::string> ref_name;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is std::string enough? An empty one is unset.

};

/// \brief Base class for Iceberg metadata tables.
class ICEBERG_EXPORT MetadataTable {
public:
Expand All @@ -46,6 +60,27 @@ class ICEBERG_EXPORT MetadataTable {

virtual Kind kind() const noexcept = 0;

/// \brief Whether this metadata table supports time-travel queries.
///
/// The currently supported snapshots and history metadata tables do not
/// support time travel.
bool supports_time_travel() const noexcept;

/// \brief Scan the metadata table using the current snapshot.
///
/// Convenience overload — delegates to Scan(std::nullopt).
Result<ArrowArray> Scan() { return Scan(std::nullopt); }

/// \brief Scan the metadata table and return all rows as an Arrow struct array.
///
/// The returned ArrowArray is a struct array where each element is one row.
/// The caller takes ownership and must call ArrowArrayRelease when done.
///
/// The default implementation returns NotSupported. Subclasses override this
/// to materialize their data.
virtual Result<ArrowArray> Scan(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we return ArrowArrayStream instead? My concern is that a single ArrowArray may lead to OOM if the result is too large. ArrowArrayStream is an iterator of ArrowArrays which is more scalable.

const std::optional<SnapshotSelection>& snapshot_selection);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It doesn't seem necessary to add a std::optional on top of SnapshotSelection if all its values are optional.


const TableIdentifier& name() const { return identifier_; }

const std::shared_ptr<Schema>& schema() const { return schema_; }
Expand Down
49 changes: 49 additions & 0 deletions src/iceberg/inspect/snapshots_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,18 @@

#include "iceberg/inspect/snapshots_table.h"

#include <chrono>
#include <memory>
#include <utility>
#include <vector>

#include "iceberg/arrow_row_builder_internal.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/table.h"
#include "iceberg/table_identifier.h"
#include "iceberg/type.h"
#include "iceberg/util/macros.h"

namespace iceberg {
namespace {
Expand Down Expand Up @@ -65,4 +68,50 @@ Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
}

Result<ArrowArray> SnapshotsTable::Scan(
const std::optional<SnapshotSelection>& /*snapshot_selection*/) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Silently ignoring an invalid selection diverges from Java. SnapshotScan.useSnapshot, useRef, and asOfTime validate or resolve the requested snapshot before SnapshotsTable returns all known snapshots. With this implementation, Scan({.snapshot_id = 999}) succeeds and returns unrelated rows. Please validate the selector (or reject a non-null selection as unsupported) instead of treating invalid input as an unqualified scan.

ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema()));

for (const auto& snapshot : source_table()->snapshots()) {
if (snapshot == nullptr) [[unlikely]] {
continue;
}

// column 0: committed_at (timestamptz -> int64 micros)
ICEBERG_RETURN_UNEXPECTED(AppendInt(
builder.column(0), std::chrono::duration_cast<std::chrono::microseconds>(
snapshot->timestamp_ms.time_since_epoch())
.count()));

// column 1: snapshot_id (long)
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot->snapshot_id));

// column 2: parent_id (long, optional)
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)));
}

// column 3: operation (string, optional)
auto op = snapshot->Operation();
if (op.has_value()) {
ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *op));
} else {
ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
}

// column 4: manifest_list (string, optional)
ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot->manifest_list));

// column 5: summary (map<string,string>)
ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot->summary));
Comment thread
WZhuo marked this conversation as resolved.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Snapshot::summary is not equivalent to Java’s Snapshot.summary() here. In C++, the map still contains the operation entry, while Java stores the operation separately and SnapshotsTable.snapshotToRow emits the remaining summary map. As a result, every normal row duplicates operation in both columns (the copied Java fixture has 10 summary entries, not 11). Please omit SnapshotSummaryFields::kOperation when materializing this map, while preserving null versus empty-map semantics.


ICEBERG_RETURN_UNEXPECTED(builder.FinishRow());
}

return std::move(builder).Finish();
}

} // namespace iceberg
7 changes: 7 additions & 0 deletions src/iceberg/inspect/snapshots_table.h
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,13 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {

Kind kind() const noexcept override { return Kind::kSnapshots; }

/// \brief Scan all snapshots as rows.
///
/// The snapshots table always returns every known snapshot, so the
/// snapshot_selection parameter is ignored.
Result<ArrowArray> Scan(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This override hides the zero-argument MetadataTable::Scan() overload under C++ name lookup. A caller using the public SnapshotsTable::Make() result cannot write snapshots_table->Scan(); only callers typed as MetadataTable compile, which is why the current tests miss it. Please add using MetadataTable::Scan; (or an explicit zero-argument overload).

const std::optional<SnapshotSelection>& /*snapshot_selection*/) override;

private:
explicit SnapshotsTable(std::shared_ptr<Table> table);
};
Expand Down
7 changes: 6 additions & 1 deletion src/iceberg/test/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,12 @@ if(ICEBERG_BUILD_BUNDLE)

add_iceberg_test(catalog_test USE_BUNDLE SOURCES in_memory_catalog_test.cc)

add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc)
add_iceberg_test(metadata_table_test
USE_BUNDLE
SOURCES
history_table_test.cc
metadata_table_test.cc
snapshots_table_test.cc)

add_iceberg_test(eval_expr_test
USE_BUNDLE
Expand Down
54 changes: 54 additions & 0 deletions src/iceberg/test/history_table_test.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

/// \file history_table_test.cc
/// Unit tests for HistoryTable.

#include <gmock/gmock.h>
#include <gtest/gtest.h>

#include "iceberg/inspect/metadata_table.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/test/matchers.h"
#include "iceberg/test/metadata_table_test_base.h"
#include "iceberg/type.h"

namespace iceberg {
namespace {

std::shared_ptr<Schema> MakeHistorySchema() {
return std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
SchemaField::MakeRequired(2, "snapshot_id", int64()),
SchemaField::MakeOptional(3, "parent_id", int64()),
SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
}

} // namespace

class HistoryTableTest : public MetadataTableTestBase {};

TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) {
ICEBERG_UNWRAP_OR_FAIL(auto history_table,
MetadataTable::Make(table_, MetadataTable::Kind::kHistory));
EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema());
}

} // namespace iceberg
88 changes: 22 additions & 66 deletions src/iceberg/test/metadata_table_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>

#include "iceberg/constants.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/table.h"
Expand All @@ -33,88 +34,43 @@
#include "iceberg/type.h"

namespace iceberg {
namespace {

std::shared_ptr<Schema> MakeSnapshotsSchema() {
return std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
SchemaField::MakeRequired(2, "snapshot_id", int64()),
SchemaField::MakeOptional(3, "parent_id", int64()),
SchemaField::MakeOptional(4, "operation", string()),
SchemaField::MakeOptional(5, "manifest_list", string()),
SchemaField::MakeOptional(
6, "summary",
std::make_shared<MapType>(SchemaField::MakeRequired(7, "key", string()),
SchemaField::MakeRequired(8, "value", string())))});
}

std::shared_ptr<Schema> MakeHistorySchema() {
return std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
SchemaField::MakeRequired(2, "snapshot_id", int64()),
SchemaField::MakeOptional(3, "parent_id", int64()),
SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
}

} // namespace

class MetadataTableTest : public ::testing::Test {
protected:
void SetUp() override {
io_ = std::make_shared<MockFileIO>();
catalog_ = std::make_shared<MockCatalog>();

auto schema = std::make_shared<Schema>(
std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
SchemaField::MakeOptional(2, "name", string())},
1);
metadata_ = std::make_shared<TableMetadata>(
TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1});

TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}},
.name = "source_table"};
auto source_table_result =
Table::Make(source_ident, metadata_, "s3://bucket/meta.json", io_, catalog_);
EXPECT_THAT(source_table_result, IsOk());
source_table_ = *source_table_result;

auto snapshots_table_result =
MetadataTable::Make(source_table_, MetadataTable::Kind::kSnapshots);
EXPECT_THAT(snapshots_table_result, IsOk());
snapshots_table_ = std::move(*snapshots_table_result);
auto metadata = std::make_shared<TableMetadata>(
TableMetadata{.format_version = 2,
.schemas = {schema},
.current_schema_id = 1,
.current_snapshot_id = kInvalidSnapshotId});

TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"};
ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata, "s3://bucket/meta.json",
std::make_shared<MockFileIO>(),
std::make_shared<MockCatalog>()));
}

std::shared_ptr<MockFileIO> io_;
std::shared_ptr<MockCatalog> catalog_;
std::shared_ptr<TableMetadata> metadata_;
std::shared_ptr<Table> source_table_;
std::unique_ptr<MetadataTable> snapshots_table_;
std::shared_ptr<Table> table_;
};

TEST_F(MetadataTableTest, Constructor) {
EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots);
EXPECT_EQ(snapshots_table_->source_table(), source_table_);
EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots");
EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector<std::string>{"db"}));
EXPECT_NE(snapshots_table_->schema(), nullptr);
}

TEST_F(MetadataTableTest, SnapshotsSchemaMatchesIcebergSchema) {
EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema());
}

TEST_F(MetadataTableTest, HistorySchemaMatchesIcebergSchema) {
auto history_table_result =
MetadataTable::Make(source_table_, MetadataTable::Kind::kHistory);
ASSERT_THAT(history_table_result, IsOk());

EXPECT_TRUE(*(*history_table_result)->schema() == *MakeHistorySchema());
}

TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) {
auto result = MetadataTable::Make(nullptr, MetadataTable::Kind::kSnapshots);
EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
EXPECT_THAT(result, HasErrorMessage("Table cannot be null"));
}

TEST_F(MetadataTableTest, SupportsTimeTravel) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots));
EXPECT_FALSE(snapshots_table->supports_time_travel());

ICEBERG_UNWRAP_OR_FAIL(auto history_table,
MetadataTable::Make(table_, MetadataTable::Kind::kHistory));
EXPECT_FALSE(history_table->supports_time_travel());
}

} // namespace iceberg
Loading
Loading