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
17 changes: 17 additions & 0 deletions src/iceberg/inspect/metadata_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,23 @@ MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,

MetadataTable::~MetadataTable() = default;

bool MetadataTable::supports_time_travel() const noexcept {
// Time travel is supported for tables that read from a single snapshot's
// manifests. Tables that scan all snapshots or return in-memory history do
// not.
switch (kind()) {
case Kind::kSnapshots:
case Kind::kHistory:
return false;
}
return false;
}

Result<ArrowArray> MetadataTable::Scan(
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
36 changes: 36 additions & 0 deletions src/iceberg/inspect/metadata_table.h
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,28 @@
#pragma once

#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;
/// \brief Read the snapshot referenced by this named ref (branch or tag).
std::optional<std::string> ref_name;
};

/// \brief Base class for Iceberg metadata tables.
class ICEBERG_EXPORT MetadataTable {
public:
Expand All @@ -43,6 +57,28 @@ class ICEBERG_EXPORT MetadataTable {

virtual Kind kind() const noexcept = 0;

/// \brief Whether this metadata table supports time-travel queries.
///
/// Time travel is supported for tables that read from a single snapshot's
/// manifests (e.g., Entries, Files, Manifests, Partitions). Tables that
/// scan all snapshots (All*) or return in-memory history (Snapshots,
/// History, Refs) 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(std::optional<SnapshotSelection> snapshot_selection);

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

const std::shared_ptr<Schema>& schema() const { return schema_; }
Expand Down
45 changes: 45 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,46 @@ Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
}

Result<ArrowArray> SnapshotsTable::Scan(
std::optional<SnapshotSelection> /*snapshot_selection*/) {
ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema()));

for (const auto& snapshot : source_table()->snapshots()) {
// 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));

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 @@ -37,6 +37,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(
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 @@ -188,7 +188,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
80 changes: 16 additions & 64 deletions src/iceberg/test/metadata_table_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -33,88 +33,40 @@
#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>(
auto 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);
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