Skip to content

Commit 28c4513

Browse files
committed
feat(inspect): implement SnapshotsTable scanning
- Add Scan() virtual method and Scan() convenience overload to MetadataTable - Add SnapshotSelection struct for time-travel snapshot resolution - Add supports_time_travel() concrete method driven by kind() - Implement SnapshotsTable::Scan() to materialize snapshot rows via ArrowRowBuilder
1 parent cd4ca42 commit 28c4513

8 files changed

Lines changed: 433 additions & 8 deletions

File tree

‎src/iceberg/inspect/metadata_table.cc‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,23 @@ MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,
3535

3636
MetadataTable::~MetadataTable() = default;
3737

38+
bool MetadataTable::supports_time_travel() const noexcept {
39+
// Time travel is supported for tables that read from a single snapshot's
40+
// manifests. Tables that scan all snapshots or return in-memory history do
41+
// not.
42+
switch (kind()) {
43+
case Kind::kSnapshots:
44+
case Kind::kHistory:
45+
return false;
46+
}
47+
return false;
48+
}
49+
50+
Result<ArrowArray> MetadataTable::Scan(
51+
std::optional<SnapshotSelection> /*snapshot_selection*/) {
52+
return NotSupported("Scan is not supported for this metadata table type");
53+
}
54+
3855
Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(std::shared_ptr<Table> table,
3956
Kind kind) {
4057
if (table == nullptr) [[unlikely]] {

‎src/iceberg/inspect/metadata_table.h‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,28 @@
2020
#pragma once
2121

2222
#include <memory>
23+
#include <optional>
24+
#include <string>
2325

26+
#include "iceberg/arrow_c_data.h"
2427
#include "iceberg/iceberg_export.h"
2528
#include "iceberg/result.h"
2629
#include "iceberg/table_identifier.h"
2730
#include "iceberg/type_fwd.h"
31+
#include "iceberg/util/timepoint.h"
2832

2933
namespace iceberg {
3034

35+
/// \brief Parameters for snapshot selection (time travel).
36+
struct SnapshotSelection {
37+
/// \brief The snapshot ID to read.
38+
std::optional<int64_t> snapshot_id;
39+
/// \brief Read the snapshot that was current at this timestamp.
40+
std::optional<TimePointMs> as_of_timestamp;
41+
/// \brief Read the snapshot referenced by this named ref (branch or tag).
42+
std::optional<std::string> ref_name;
43+
};
44+
3145
/// \brief Base class for Iceberg metadata tables.
3246
class ICEBERG_EXPORT MetadataTable {
3347
public:
@@ -43,6 +57,28 @@ class ICEBERG_EXPORT MetadataTable {
4357

4458
virtual Kind kind() const noexcept = 0;
4559

60+
/// \brief Whether this metadata table supports time-travel queries.
61+
///
62+
/// Time travel is supported for tables that read from a single snapshot's
63+
/// manifests (e.g., Entries, Files, Manifests, Partitions). Tables that
64+
/// scan all snapshots (All*) or return in-memory history (Snapshots,
65+
/// History, Refs) do not support time travel.
66+
bool supports_time_travel() const noexcept;
67+
68+
/// \brief Scan the metadata table using the current snapshot.
69+
///
70+
/// Convenience overload — delegates to Scan(std::nullopt).
71+
Result<ArrowArray> Scan() { return Scan(std::nullopt); }
72+
73+
/// \brief Scan the metadata table and return all rows as an Arrow struct array.
74+
///
75+
/// The returned ArrowArray is a struct array where each element is one row.
76+
/// The caller takes ownership and must call ArrowArrayRelease when done.
77+
///
78+
/// The default implementation returns NotSupported. Subclasses override this
79+
/// to materialize their data.
80+
virtual Result<ArrowArray> Scan(std::optional<SnapshotSelection> snapshot_selection);
81+
4682
const TableIdentifier& name() const { return identifier_; }
4783

4884
const std::shared_ptr<Schema>& schema() const { return schema_; }

‎src/iceberg/inspect/snapshots_table.cc‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,18 @@
1919

2020
#include "iceberg/inspect/snapshots_table.h"
2121

22+
#include <chrono>
2223
#include <memory>
2324
#include <utility>
2425
#include <vector>
2526

27+
#include "iceberg/arrow_row_builder_internal.h"
2628
#include "iceberg/schema.h"
2729
#include "iceberg/schema_field.h"
2830
#include "iceberg/table.h"
2931
#include "iceberg/table_identifier.h"
3032
#include "iceberg/type.h"
33+
#include "iceberg/util/macros.h"
3134

3235
namespace iceberg {
3336
namespace {
@@ -65,4 +68,46 @@ Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
6568
return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
6669
}
6770

71+
Result<ArrowArray> SnapshotsTable::Scan(
72+
std::optional<SnapshotSelection> /*snapshot_selection*/) {
73+
ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema()));
74+
75+
for (const auto& snapshot : source_table()->snapshots()) {
76+
// column 0: committed_at (timestamptz → int64 micros)
77+
ICEBERG_RETURN_UNEXPECTED(AppendInt(
78+
builder.column(0), std::chrono::duration_cast<std::chrono::microseconds>(
79+
snapshot->timestamp_ms.time_since_epoch())
80+
.count()));
81+
82+
// column 1: snapshot_id (long)
83+
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot->snapshot_id));
84+
85+
// column 2: parent_id (long, optional)
86+
if (snapshot->parent_snapshot_id.has_value()) {
87+
ICEBERG_RETURN_UNEXPECTED(
88+
AppendInt(builder.column(2), *snapshot->parent_snapshot_id));
89+
} else {
90+
ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2)));
91+
}
92+
93+
// column 3: operation (string, optional)
94+
auto op = snapshot->Operation();
95+
if (op.has_value()) {
96+
ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *op));
97+
} else {
98+
ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
99+
}
100+
101+
// column 4: manifest_list (string, optional)
102+
ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot->manifest_list));
103+
104+
// column 5: summary (map<string,string>)
105+
ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot->summary));
106+
107+
ICEBERG_RETURN_UNEXPECTED(builder.FinishRow());
108+
}
109+
110+
return std::move(builder).Finish();
111+
}
112+
68113
} // namespace iceberg

‎src/iceberg/inspect/snapshots_table.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,13 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {
3737

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

40+
/// \brief Scan all snapshots as rows.
41+
///
42+
/// The snapshots table always returns every known snapshot, so the
43+
/// snapshot_selection parameter is ignored.
44+
Result<ArrowArray> Scan(
45+
std::optional<SnapshotSelection> /*snapshot_selection*/) override;
46+
4047
private:
4148
explicit SnapshotsTable(std::shared_ptr<Table> table);
4249
};

‎src/iceberg/test/CMakeLists.txt‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -188,7 +188,11 @@ if(ICEBERG_BUILD_BUNDLE)
188188

189189
add_iceberg_test(catalog_test USE_BUNDLE SOURCES in_memory_catalog_test.cc)
190190

191-
add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc)
191+
add_iceberg_test(metadata_table_test
192+
USE_BUNDLE
193+
SOURCES
194+
metadata_table_test.cc
195+
snapshots_scan_test.cc)
192196

193197
add_iceberg_test(eval_expr_test
194198
USE_BUNDLE

‎src/iceberg/test/metadata_table_test.cc‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,7 @@
2828
#include "iceberg/table_identifier.h"
2929
#include "iceberg/table_metadata.h"
3030
#include "iceberg/test/matchers.h"
31-
#include "iceberg/test/mock_catalog.h"
32-
#include "iceberg/test/mock_io.h"
31+
#include "iceberg/test/metadata_table_test_base.h"
3332
#include "iceberg/type.h"
3433

3534
namespace iceberg {
@@ -58,11 +57,10 @@ std::shared_ptr<Schema> MakeHistorySchema() {
5857

5958
} // namespace
6059

61-
class MetadataTableTest : public ::testing::Test {
60+
class MetadataTableTest : public MetadataTableTestBase {
6261
protected:
6362
void SetUp() override {
64-
io_ = std::make_shared<MockFileIO>();
65-
catalog_ = std::make_shared<MockCatalog>();
63+
MetadataTableTestBase::SetUp();
6664

6765
auto schema = std::make_shared<Schema>(
6866
std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
@@ -84,8 +82,6 @@ class MetadataTableTest : public ::testing::Test {
8482
snapshots_table_ = std::move(*snapshots_table_result);
8583
}
8684

87-
std::shared_ptr<MockFileIO> io_;
88-
std::shared_ptr<MockCatalog> catalog_;
8985
std::shared_ptr<TableMetadata> metadata_;
9086
std::shared_ptr<Table> source_table_;
9187
std::unique_ptr<MetadataTable> snapshots_table_;
Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
/// \file metadata_table_test_base.h
21+
/// Shared test base for all metadata table tests.
22+
///
23+
/// Provides common helpers (FinishAndImport, MakeTestSnapshots,
24+
/// MakeTableWithSnapshots) and the MockFileIO + MockCatalog fixture that
25+
/// every metadata table test needs.
26+
27+
#pragma once
28+
29+
#include <memory>
30+
#include <unordered_map>
31+
#include <utility>
32+
#include <vector>
33+
34+
#include <arrow/array.h>
35+
#include <arrow/c/bridge.h>
36+
#include <arrow/record_batch.h>
37+
#include <gmock/gmock.h>
38+
#include <gtest/gtest.h>
39+
40+
#include "iceberg/schema.h"
41+
#include "iceberg/schema_field.h"
42+
#include "iceberg/schema_internal.h"
43+
#include "iceberg/snapshot.h"
44+
#include "iceberg/table.h"
45+
#include "iceberg/table_identifier.h"
46+
#include "iceberg/table_metadata.h"
47+
#include "iceberg/test/matchers.h"
48+
#include "iceberg/test/mock_catalog.h"
49+
#include "iceberg/test/mock_io.h"
50+
#include "iceberg/type.h"
51+
#include "iceberg/util/timepoint.h"
52+
53+
namespace iceberg {
54+
55+
/// \brief Base class for all metadata table tests.
56+
///
57+
/// Provides MockFileIO and MockCatalog instances plus helpers shared across
58+
/// metadata table tests (SnapshotsTable, HistoryTable, RefsTable, ...).
59+
class MetadataTableTestBase : public ::testing::Test {
60+
protected:
61+
void SetUp() override {
62+
io_ = std::make_shared<MockFileIO>();
63+
catalog_ = std::make_shared<MockCatalog>();
64+
}
65+
66+
/// \brief Import a Scan()-produced ArrowArray into an Arrow RecordBatch.
67+
static std::shared_ptr<::arrow::RecordBatch> FinishAndImport(ArrowArray array,
68+
const Schema& schema) {
69+
ArrowSchema c_schema;
70+
EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk());
71+
auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie();
72+
73+
// ImportRecordBatch takes ownership of the array and releases it.
74+
return ::arrow::ImportRecordBatch(&array, arrow_schema).ValueOrDie();
75+
}
76+
77+
/// \brief Create two snapshots matching the Java TestDataTaskParser test data.
78+
///
79+
/// Snapshot 1: id=1, no parent, timestamp=1234567890000, operation="append"
80+
/// Snapshot 2: id=2, parent=1, timestamp=9876543210000, operation="append"
81+
static std::pair<std::shared_ptr<Snapshot>, std::shared_ptr<Snapshot>>
82+
MakeTestSnapshots() {
83+
std::unordered_map<std::string, std::string> summary1{
84+
{"added-data-files", "1"}, {"added-records", "1"},
85+
{"added-files-size", "10"}, {"changed-partition-count", "1"},
86+
{"total-records", "1"}, {"total-files-size", "10"},
87+
{"total-data-files", "1"}, {"total-delete-files", "0"},
88+
{"total-position-deletes", "0"}, {"total-equality-deletes", "0"},
89+
{"operation", "append"},
90+
};
91+
92+
std::unordered_map<std::string, std::string> summary2{
93+
{"added-data-files", "1"}, {"added-records", "1"},
94+
{"added-files-size", "10"}, {"changed-partition-count", "1"},
95+
{"total-records", "2"}, {"total-files-size", "20"},
96+
{"total-data-files", "2"}, {"total-delete-files", "0"},
97+
{"total-position-deletes", "0"}, {"total-equality-deletes", "0"},
98+
{"operation", "append"},
99+
};
100+
101+
auto snap1 = std::make_shared<Snapshot>(Snapshot{
102+
.snapshot_id = 1,
103+
.parent_snapshot_id = std::nullopt,
104+
.sequence_number = 1,
105+
.timestamp_ms = TimePointMsFromUnixMs(1234567890000),
106+
.manifest_list = "file:/tmp/manifest1.avro",
107+
.summary = std::move(summary1),
108+
.schema_id = 1,
109+
});
110+
111+
auto snap2 = std::make_shared<Snapshot>(Snapshot{
112+
.snapshot_id = 2,
113+
.parent_snapshot_id = 1,
114+
.sequence_number = 2,
115+
.timestamp_ms = TimePointMsFromUnixMs(9876543210000),
116+
.manifest_list = "file:/tmp/manifest2.avro",
117+
.summary = std::move(summary2),
118+
.schema_id = 1,
119+
});
120+
121+
return {snap1, snap2};
122+
}
123+
124+
/// \brief Create a Table with the given snapshots.
125+
Result<std::shared_ptr<Table>> MakeTableWithSnapshots(
126+
std::vector<std::shared_ptr<Snapshot>> snapshots, int64_t current_snapshot_id) {
127+
auto schema = std::make_shared<Schema>(
128+
std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
129+
SchemaField::MakeOptional(2, "name", string())},
130+
1);
131+
auto metadata = std::make_shared<TableMetadata>(TableMetadata{
132+
.format_version = 2,
133+
.schemas = {schema},
134+
.current_schema_id = 1,
135+
.current_snapshot_id = current_snapshot_id,
136+
.snapshots = std::move(snapshots),
137+
});
138+
139+
TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "test_table"};
140+
return Table::Make(ident, metadata, "s3://bucket/meta.json", io_, catalog_);
141+
}
142+
143+
std::shared_ptr<MockFileIO> io_;
144+
std::shared_ptr<MockCatalog> catalog_;
145+
};
146+
147+
} // namespace iceberg

0 commit comments

Comments
 (0)