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
1 change: 1 addition & 0 deletions src/iceberg/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ set(ICEBERG_SOURCES
inspect/metadata_table.cc
inspect/metadata_scan_task.cc
inspect/metadata_scan_task_reader.cc
inspect/refs_table.cc
inspect/snapshots_table.cc
inheritable_metadata.cc
json_serde.cc
Expand Down
1 change: 1 addition & 0 deletions src/iceberg/cpp20_compatibility_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@
#include "iceberg/inspect/metadata_scan_task.h"
#include "iceberg/inspect/metadata_scan_task_reader.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/inspect/refs_table.h"
#include "iceberg/inspect/snapshots_table.h"
#include "iceberg/labels.h"
#include "iceberg/location_provider.h"
Expand Down
3 changes: 2 additions & 1 deletion src/iceberg/inspect/metadata_scan_task.cc
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,8 @@ Result<std::shared_ptr<StaticMetadataTask>> StaticMetadataTask::Make(
ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
ICEBERG_PRECHECK(projected_schema != nullptr, "Projected schema cannot be null");
if (table_kind != MetadataTableKind::kSnapshots &&
table_kind != MetadataTableKind::kHistory) {
table_kind != MetadataTableKind::kHistory &&
table_kind != MetadataTableKind::kRefs) {
return NotImplemented("Static metadata table kind is not implemented");
}
return std::shared_ptr<StaticMetadataTask>(new StaticMetadataTask(
Expand Down
6 changes: 3 additions & 3 deletions src/iceberg/inspect/metadata_scan_task.h
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ class ICEBERG_EXPORT MetadataScanTask {
public:
/// \brief Task category, determines how the reader opens the task.
enum class TaskKind {
/// StaticMetadataTask reads in-memory table metadata (snapshots, history).
/// StaticMetadataTask reads in-memory table metadata (snapshots, history, refs).
kStatic,
};

Expand All @@ -48,7 +48,7 @@ class ICEBERG_EXPORT MetadataScanTask {
virtual MetadataTableKind table_kind() const = 0;
};

/// \brief Task for static metadata tables (snapshots, history).
/// \brief Task for static metadata tables (snapshots, history, refs).
///
/// Holds the metadata captured for a table scan; readers produce Arrow batches
/// on demand. The metadata and its nested objects must not be modified while
Expand All @@ -57,7 +57,7 @@ class ICEBERG_EXPORT StaticMetadataTask : public MetadataScanTask {
public:
/// \brief Create a static metadata task.
///
/// \param table_kind The metadata table kind (kSnapshots or kHistory)
/// \param table_kind The metadata table kind (kSnapshots, kHistory, or kRefs)
/// \param metadata The table metadata to scan
/// \param projected_schema The planned output schema
/// \return A task, or an error for null metadata or an unsupported table kind.
Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/inspect/metadata_scan_task_reader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
#include "iceberg/inspect/history_table.h"
#include "iceberg/inspect/metadata_scan_task.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/inspect/refs_table.h"
#include "iceberg/inspect/snapshots_table.h"
#include "iceberg/util/checked_cast.h"
#include "iceberg/util/macros.h"
Expand Down Expand Up @@ -53,6 +54,9 @@ Result<ArrowArrayStream> MetadataScanTaskReader::Open(
return SnapshotsTable::ReadRows(static_task.metadata(),
static_task.projected_schema(),
options_.batch_size);
case MetadataTableKind::kRefs:
return RefsTable::ReadRows(static_task.metadata(),
static_task.projected_schema(), options_.batch_size);
case MetadataTableKind::kHistory:
return HistoryTable::ReadRows(static_task.metadata(),
static_task.projected_schema(),
Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/inspect/metadata_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

#include "iceberg/inspect/history_table.h"
#include "iceberg/inspect/metadata_scan.h"
#include "iceberg/inspect/refs_table.h"
#include "iceberg/inspect/snapshots_table.h"

namespace iceberg {
Expand All @@ -38,6 +39,8 @@ Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(
switch (kind) {
case MetadataTableKind::kSnapshots:
return SnapshotsTable::Make(std::move(table));
case MetadataTableKind::kRefs:
return RefsTable::Make(std::move(table));
case MetadataTableKind::kHistory:
return HistoryTable::Make(std::move(table));
default:
Expand Down
179 changes: 179 additions & 0 deletions src/iceberg/inspect/refs_table.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,179 @@
/*
* 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 <algorithm>
#include <array>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <variant>
#include <vector>

#include "iceberg/arrow_row_builder_internal.h"
#include "iceberg/inspect/metadata_scan_task.h"
#include "iceberg/inspect/metadata_scan_util_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<int64_t> max_ref_age_ms;
std::optional<int32_t> min_snapshots_to_keep;
std::optional<int64_t> max_snapshot_age_ms;
};

Status AppendOptional(ArrowArray* array, const auto& value) {
Comment thread
manuzhang marked this conversation as resolved.
Resolved
if (!value.has_value()) {
return AppendNull(array);
}
return AppendInt(array, static_cast<int64_t>(*value));
}

constexpr std::array<internal::MetadataTableColumn<RefRow>, 6> kRefColumns = {
{{.field_id = 1,
.append = [](ArrowArray* column,
const RefRow& row) { return AppendString(column, row.name); }},
{.field_id = 2,
.append = [](ArrowArray* column,
const RefRow& row) { return AppendString(column, row.type); }},
{.field_id = 3,
.append = [](ArrowArray* column,
const RefRow& row) { return AppendInt(column, row.snapshot_id); }},
{.field_id = 4,
.append =
[](ArrowArray* column, const RefRow& row) {
return AppendOptional(column, row.max_ref_age_ms);
}},
{.field_id = 5,
.append =
[](ArrowArray* column, const RefRow& row) {
return AppendOptional(column, row.min_snapshots_to_keep);
}},
{.field_id = 6, .append = [](ArrowArray* column, const RefRow& row) {
return AppendOptional(column, row.max_snapshot_age_ms);
}}}};

} // namespace

// RefsTable implementation
RefsTable::RefsTable(std::shared_ptr<const Table> table)
: MetadataTable(std::move(table)) {}

RefsTable::~RefsTable() = default;

const std::shared_ptr<Schema>& RefsTable::schema() const { return CanonicalSchema(); }

const std::shared_ptr<Schema>& RefsTable::CanonicalSchema() {
static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
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<std::unique_ptr<RefsTable>> RefsTable::Make(std::shared_ptr<const Table> table) {
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
return std::unique_ptr<RefsTable>(new RefsTable(std::move(table)));
}

Result<std::unique_ptr<MetadataScan>> RefsTable::MakeScan(
internal::MetadataScanContext context) const {
return RefsScan::Make(std::move(context));
}

Result<ArrowArrayStream> RefsTable::ReadRows(
std::shared_ptr<const TableMetadata> metadata,
const std::shared_ptr<Schema>& projected_schema, int64_t batch_size) {
std::vector<RefRow> rows;
const auto& refs = 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<SnapshotRef::Branch>(&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<SnapshotRef::Tag>(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);
Comment thread
manuzhang marked this conversation as resolved.
Resolved
const size_t num_rows = rows.size();
return internal::MakeMetadataBatchStream<RefRow>(
CanonicalSchema(), projected_schema, kRefColumns, num_rows, batch_size,
[rows = std::move(rows)](size_t index) -> const RefRow& { return rows[index]; });
}

// RefsScan implementation
RefsScan::RefsScan(internal::MetadataScanContext context,
std::shared_ptr<MetadataScanTask> task)
: MetadataScan(std::move(context)), task_(std::move(task)) {}

RefsScan::~RefsScan() = default;

Result<std::unique_ptr<RefsScan>> RefsScan::Make(internal::MetadataScanContext context) {
ICEBERG_PRECHECK(context.kind == MetadataTableKind::kRefs,
"Unexpected metadata table kind for refs scan");
if (context.snapshot_id.has_value() || !context.ref_name.empty() ||
context.as_of_time_in_millis.has_value()) {
return NotSupported("Metadata table does not support time travel");
}
ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot be null");
ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be null");
ICEBERG_PRECHECK(context.projected_schema != nullptr,
"Projected schema cannot be null");
ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns<RefRow>(
*RefsTable::CanonicalSchema(), *context.projected_schema, kRefColumns));
ICEBERG_ASSIGN_OR_RAISE(auto task,
StaticMetadataTask::Make(context.kind, context.table_metadata,
context.projected_schema));
return std::unique_ptr<RefsScan>(new RefsScan(std::move(context), std::move(task)));
}

Result<MetadataScanTaskStreamPtr> RefsScan::PlanTasksStream() const {
return MetadataScanTaskStream::FromVector({task_});
}

} // namespace iceberg
81 changes: 81 additions & 0 deletions src/iceberg/inspect/refs_table.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
/*
* 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 refs metadata table.

#include <cstdint>
#include <memory>

#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/inspect/metadata_scan.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"

namespace iceberg {

/// \brief Refs metadata table.
class ICEBERG_EXPORT RefsTable : public MetadataTable {
public:
static Result<std::unique_ptr<RefsTable>> Make(std::shared_ptr<const Table> table);

/// \brief Return the canonical schema of the refs metadata table.
static const std::shared_ptr<Schema>& CanonicalSchema();

~RefsTable() override;

MetadataTableKind kind() const noexcept override { return MetadataTableKind::kRefs; }

const std::shared_ptr<Schema>& schema() const override;

protected:
Result<std::unique_ptr<MetadataScan>> MakeScan(
internal::MetadataScanContext context) const override;

private:
friend class MetadataScanTaskReader;

static Result<ArrowArrayStream> ReadRows(
std::shared_ptr<const TableMetadata> metadata,
const std::shared_ptr<Schema>& projected_schema, int64_t batch_size);

explicit RefsTable(std::shared_ptr<const Table> table);
};

/// \brief Scan the refs metadata captured from a table.
class ICEBERG_EXPORT RefsScan final : public MetadataScan {
public:
/// \brief Create a scan from captured metadata table settings.
static Result<std::unique_ptr<RefsScan>> Make(internal::MetadataScanContext context);

~RefsScan() override;

Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override;

private:
RefsScan(internal::MetadataScanContext context, std::shared_ptr<MetadataScanTask> task);

const std::shared_ptr<MetadataScanTask> task_;
};

} // namespace iceberg
7 changes: 4 additions & 3 deletions src/iceberg/test/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -206,9 +206,10 @@ 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_test.cc
inspect/refs_table_test.cc
inspect/snapshots_table_test.cc)

add_iceberg_test(eval_expr_test
USE_BUNDLE
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,8 @@
#include "iceberg/snapshot.h"
#include "iceberg/table_metadata.h"
#include "iceberg/test/arrow_test_util.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"
#include "iceberg/util/timepoint.h"

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,8 +40,8 @@
#include "iceberg/table.h"
#include "iceberg/table_metadata.h"
#include "iceberg/test/arrow_test_util.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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,9 @@

#pragma once

#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <utility>
Expand Down
Loading
Loading