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
2 changes: 2 additions & 0 deletions src/iceberg/arrow_row_builder.cc
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,8 @@ ArrowRowBuilder::~ArrowRowBuilder() {

int64_t ArrowRowBuilder::num_columns() const { return array_.n_children; }

int64_t ArrowRowBuilder::num_rows() const { return array_.length; }

ArrowArray* ArrowRowBuilder::column(int64_t index) {
if (index < 0 || index >= array_.n_children) {
return nullptr;
Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/arrow_row_builder_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,9 @@ class ICEBERG_EXPORT ArrowRowBuilder {
/// \brief The number of top-level columns in the batch.
int64_t num_columns() const;

/// \brief The number of completed rows in the batch.
int64_t num_rows() const;

/// \brief Access the nanoarrow child builder for a top-level column.
///
/// \param index Zero-based column index. Returns nullptr if out of range.
Expand Down
33 changes: 14 additions & 19 deletions src/iceberg/inspect/history_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -26,37 +26,32 @@
#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 {

std::shared_ptr<Schema> MakeHistoryTableSchema() {
return std::make_shared<Schema>(std::vector<SchemaField>{
HistoryTable::HistoryTable(std::shared_ptr<Table> table)
: MetadataTable(std::move(table)) {}

HistoryTable::~HistoryTable() = default;

const std::shared_ptr<Schema>& HistoryTable::schema() const {
static const auto schema = 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())});
return schema;
}

TableIdentifier MakeHistoryTableName(const TableIdentifier& source_name) {
return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".history"};
}

} // namespace

HistoryTable::HistoryTable(std::shared_ptr<Table> table)
: MetadataTable(table, MakeHistoryTableName(table->name()),
MakeHistoryTableSchema()) {}

HistoryTable::~HistoryTable() = default;

Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(std::shared_ptr<Table> table) {
if (table == nullptr) [[unlikely]] {
return InvalidArgument("Table cannot be null");
}
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
}

Result<ArrowArrayStream> HistoryTable::Scan() {
return NotSupported("Scan is not supported for the history table");
}

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

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

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

Result<ArrowArrayStream> Scan() override;

private:
explicit HistoryTable(std::shared_ptr<Table> table);
};
Expand Down
40 changes: 20 additions & 20 deletions src/iceberg/inspect/metadata_table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -22,33 +22,33 @@
#include <memory>
#include <utility>

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

namespace iceberg {

MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,
TableIdentifier identifier, std::shared_ptr<Schema> schema)
: identifier_(std::move(identifier)),
schema_(std::move(schema)),
source_table_(std::move(source_table)) {}
MetadataTable::MetadataTable(std::shared_ptr<Table> source_table)
: source_table_(std::move(source_table)) {}

MetadataTable::~MetadataTable() = default;

Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(std::shared_ptr<Table> table,
Kind kind) {
if (table == nullptr) [[unlikely]] {
return InvalidArgument("Table cannot be null");
}
bool MetadataTable::supports_time_travel() const noexcept { return false; }

const std::shared_ptr<Table>& MetadataTable::source_table() const {
return source_table_;
}

TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr<Table> source_table)
: MetadataTable(std::move(source_table)) {}

switch (kind) {
case Kind::kSnapshots:
return SnapshotsTable::Make(table);
case Kind::kHistory:
return HistoryTable::Make(table);
}
TimeTravelMetadataTable::~TimeTravelMetadataTable() = default;

bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; }

Result<ArrowArrayStream> TimeTravelMetadataTable::Scan() {
return ScanSnapshot(SnapshotSelection{});
}

return NotSupported("Unsupported metadata table type");
Result<ArrowArrayStream> TimeTravelMetadataTable::Scan(
const SnapshotSelection& snapshot_selection) {
return ScanSnapshot(snapshot_selection);
}

} // namespace iceberg
86 changes: 74 additions & 12 deletions src/iceberg/inspect/metadata_table.h
Original file line number Diff line number Diff line change
Expand Up @@ -20,46 +20,108 @@
#pragma once

/// \file iceberg/inspect/metadata_table.h
/// \brief Define base APIs for metadata tables.
/// \brief Base APIs for inspecting Iceberg metadata tables.

#include <concepts>
#include <memory>
#include <string>
#include <utility>
#include <variant>

#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 Base class for Iceberg metadata tables.
/// \brief Base interface for an Iceberg metadata table.
class ICEBERG_EXPORT MetadataTable {
public:
/// \brief Supported metadata table kinds.
enum class Kind {
kSnapshots,
kHistory,
};

static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<Table> table,
Kind kind);
/// \brief Maximum number of rows emitted in each Arrow batch.
static constexpr int64_t kBatchSize = 1024;

/// \brief Create a metadata table of the requested concrete type.
///
/// \tparam MetadataTableType Concrete class derived from MetadataTable.
/// \param table Source table whose metadata will be exposed.
/// \return The constructed metadata table, or an error.
template <typename MetadataTableType>
requires std::derived_from<MetadataTableType, MetadataTable>
static Result<std::unique_ptr<MetadataTableType>> Make(std::shared_ptr<Table> table) {
return MetadataTableType::Make(std::move(table));
}
Comment thread
WZhuo marked this conversation as resolved.

virtual ~MetadataTable();

/// \brief Return this metadata table's kind.
virtual Kind kind() const noexcept = 0;

const TableIdentifier& name() const { return identifier_; }
/// \brief Return the schema of rows emitted by scans.
virtual const std::shared_ptr<Schema>& schema() const = 0;

/// \brief Return the source table whose metadata is exposed.
const std::shared_ptr<Table>& source_table() const;

const std::shared_ptr<Schema>& schema() const { return schema_; }
/// \brief Return whether this metadata table supports time travel.
virtual bool supports_time_travel() const noexcept;

const std::shared_ptr<Table>& source_table() const { return source_table_; }
/// \brief Scan the metadata table without time travel.
///
/// The caller owns the returned stream and must release it with
/// ArrowArrayStreamRelease.
virtual Result<ArrowArrayStream> Scan() = 0;

protected:
explicit MetadataTable(std::shared_ptr<Table> source_table, TableIdentifier identifier,
std::shared_ptr<Schema> schema);
explicit MetadataTable(std::shared_ptr<Table> source_table);

private:
TableIdentifier identifier_;
std::shared_ptr<Schema> schema_;
std::shared_ptr<Table> source_table_;
};

/// \brief Snapshot selection parameters for a time-travel scan.
struct SnapshotSelection {
/// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
///
/// std::monostate selects the current snapshot.
std::variant<std::monostate, int64_t, TimePointMs> snapshot;

/// \brief Resolve the snapshot relative to this branch or tag.
///
/// An empty string uses the main branch.
std::string ref_name;
};

/// \brief Base interface for metadata tables that support time travel.
class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
public:
~TimeTravelMetadataTable() override;

/// \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<ArrowArrayStream> Scan() final;

/// \brief Scan 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<ArrowArrayStream> Scan(const SnapshotSelection& snapshot_selection);

protected:
explicit TimeTravelMetadataTable(std::shared_ptr<Table> source_table);

/// \brief Implement a scan for the requested snapshot selection.
virtual Result<ArrowArrayStream> ScanSnapshot(
const SnapshotSelection& snapshot_selection) = 0;
};

} // namespace iceberg
Loading
Loading