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
66 changes: 66 additions & 0 deletions src/iceberg/catalog/rest/json_serde.cc
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
#include "iceberg/expression/json_serde_internal.h"
#include "iceberg/file_format.h"
#include "iceberg/json_serde_internal.h"
#include "iceberg/labels.h"
#include "iceberg/manifest/manifest_entry.h"
#include "iceberg/partition_spec.h"
#include "iceberg/schema.h"
Expand Down Expand Up @@ -72,6 +73,10 @@ constexpr std::string_view kDestination = "destination";
constexpr std::string_view kMetadata = "metadata";
constexpr std::string_view kConfig = "config";
constexpr std::string_view kStorageCredentials = "storage-credentials";
constexpr std::string_view kLabels = "labels";
constexpr std::string_view kObjectLabels = "object-labels";
constexpr std::string_view kFields = "fields";
constexpr std::string_view kFieldId = "field-id";
constexpr std::string_view kPrefix = "prefix";
constexpr std::string_view kIdentifiers = "identifiers";
constexpr std::string_view kOverrides = "overrides";
Expand Down Expand Up @@ -712,6 +717,59 @@ Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json
return request;
}

nlohmann::json ToJson(const FieldLabel& field_label) {
nlohmann::json json;
json[kFieldId] = field_label.field_id;
json[kLabels] = field_label.labels;
return json;
}

Result<FieldLabel> FieldLabelFromJson(const nlohmann::json& json) {
if (!json.is_object()) {
return JsonParseError("Cannot parse field label from non-object: {}",
SafeDumpJson(json));
}
FieldLabel field_label;
ICEBERG_ASSIGN_OR_RAISE(field_label.field_id, GetJsonInt32Strict(json, kFieldId));
ICEBERG_ASSIGN_OR_RAISE(field_label.labels,
GetJsonValue<decltype(field_label.labels)>(json, kLabels));
return field_label;
}

nlohmann::json ToJson(const Labels& labels) {
nlohmann::json json = nlohmann::json::object();
if (!labels.object_labels.empty()) {
json[kObjectLabels] = labels.object_labels;
}
if (!labels.fields.empty()) {
nlohmann::json fields = nlohmann::json::array();
for (const auto& field_label : labels.fields) {
fields.push_back(ToJson(field_label));
}
json[kFields] = std::move(fields);
}
return json;
}

// A non-object value is treated as absent labels rather than failing the load.
Result<Labels> LabelsFromJson(const nlohmann::json& json) {
Labels labels;
ICEBERG_ASSIGN_OR_RAISE(
labels.object_labels,
GetJsonValueOrDefault<decltype(labels.object_labels)>(json, kObjectLabels));
if (auto it = json.find(kFields); it != json.end() && !it->is_null()) {
if (!it->is_array()) {
return JsonParseError("Cannot parse '{}' from non-array: {}", kFields,
SafeDumpJson(*it));
}
for (const auto& entry : *it) {
ICEBERG_ASSIGN_OR_RAISE(auto field_label, FieldLabelFromJson(entry));
labels.fields.push_back(std::move(field_label));
}
}
return labels;
}

// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
Result<nlohmann::json> ToJson(const LoadTableResult& result) {
nlohmann::json json;
Expand All @@ -726,6 +784,9 @@ Result<nlohmann::json> ToJson(const LoadTableResult& result) {
}
json[kStorageCredentials] = std::move(creds);
}
if (!result.labels.empty()) {
json[kLabels] = ToJson(result.labels);
}
return json;
}

Expand All @@ -747,6 +808,9 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
result.storage_credentials.push_back(std::move(cred));
}
}
if (auto it = json.find(kLabels); it != json.end() && !it->is_null()) {
ICEBERG_ASSIGN_OR_RAISE(result.labels, LabelsFromJson(*it));
}
ICEBERG_RETURN_UNEXPECTED(result.Validate());
return result;
}
Expand Down Expand Up @@ -1216,5 +1280,7 @@ ICEBERG_DEFINE_FROM_JSON(CommitTableResponse)
ICEBERG_DEFINE_FROM_JSON(OAuthTokenResponse)
ICEBERG_DEFINE_FROM_JSON(PlanTableScanRequest)
ICEBERG_DEFINE_FROM_JSON(FetchScanTasksRequest)
ICEBERG_DEFINE_FROM_JSON(FieldLabel)
ICEBERG_DEFINE_FROM_JSON(Labels)

} // namespace iceberg::rest
2 changes: 2 additions & 0 deletions src/iceberg/catalog/rest/json_serde_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@ ICEBERG_DECLARE_JSON_SERDE(ListTablesResponse)
ICEBERG_DECLARE_JSON_SERDE(RegisterTableRequest)
ICEBERG_DECLARE_JSON_SERDE(RenameTableRequest)
ICEBERG_DECLARE_JSON_SERDE(OAuthTokenResponse)
ICEBERG_DECLARE_JSON_SERDE(FieldLabel)
ICEBERG_DECLARE_JSON_SERDE(Labels)

#undef ICEBERG_DECLARE_JSON_SERDE

Expand Down
2 changes: 1 addition & 1 deletion src/iceberg/catalog/rest/rest_catalog.cc
Original file line number Diff line number Diff line change
Expand Up @@ -902,7 +902,7 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
return Table::Make(identifier, std::move(result.metadata),
std::move(result.metadata_location), std::move(table_io),
std::move(table_catalog), RestTableName(name_, identifier),
std::move(reporter));
std::move(reporter), std::move(result.labels));
}

Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromCommitResponse(
Expand Down
2 changes: 1 addition & 1 deletion src/iceberg/catalog/rest/types.cc
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {

bool LoadTableResult::operator==(const LoadTableResult& other) const {
if (metadata_location != other.metadata_location || config != other.config ||
storage_credentials != other.storage_credentials) {
storage_credentials != other.storage_credentials || labels != other.labels) {
return false;
}

Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/catalog/rest/types.h
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@

#include "iceberg/catalog/rest/endpoint.h"
#include "iceberg/catalog/rest/iceberg_rest_export.h"
#include "iceberg/labels.h"
#include "iceberg/result.h"
#include "iceberg/storage_credential.h"
#include "iceberg/table_identifier.h"
Expand Down Expand Up @@ -188,6 +189,8 @@ struct ICEBERG_REST_EXPORT LoadTableResult {
std::unordered_map<std::string, std::string> config;
/// \brief Vended storage credentials, one per URI prefix; empty if none.
std::vector<StorageCredential> storage_credentials;
/// \brief Catalog-provided labels; empty if none.
Labels labels;

/// \brief Validates the LoadTableResult.
Status Validate() const {
Expand Down
56 changes: 56 additions & 0 deletions src/iceberg/labels.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
/*
* 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/labels.h
/// \brief Catalog-provided labels for tables and their fields.

#include <cstdint>
#include <string>
#include <unordered_map>
#include <vector>

#include "iceberg/iceberg_export.h"

namespace iceberg {

/// \brief Catalog-provided labels for a single field, identified by its field ID.
struct ICEBERG_EXPORT FieldLabel {
int32_t field_id = 0;
std::unordered_map<std::string, std::string> labels;

bool operator==(const FieldLabel&) const = default;
};

/// \brief Optional catalog-provided labels returned when a table is loaded.
///
/// Labels are advisory enrichment, not table state: they are not persisted with the
/// table and may be absent.
struct ICEBERG_EXPORT Labels {
std::unordered_map<std::string, std::string> object_labels;
std::vector<FieldLabel> fields;

/// \brief Returns true when there are neither object-level nor field-level labels.
bool empty() const { return object_labels.empty() && fields.empty(); }

bool operator==(const Labels&) const = default;
};

} // namespace iceberg
26 changes: 14 additions & 12 deletions src/iceberg/table.cc
Original file line number Diff line number Diff line change
Expand Up @@ -53,13 +53,11 @@

namespace iceberg {

Result<std::shared_ptr<Table>> Table::Make(TableIdentifier identifier,
std::shared_ptr<TableMetadata> metadata,
std::string metadata_location,
std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog,
std::string full_name,
std::shared_ptr<MetricsReporter> reporter) {
Result<std::shared_ptr<Table>> Table::Make(
TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name,
std::shared_ptr<MetricsReporter> reporter, Labels labels) {
if (metadata == nullptr) [[unlikely]] {
return InvalidArgument("Metadata cannot be null");
}
Expand All @@ -72,25 +70,27 @@ Result<std::shared_ptr<Table>> Table::Make(TableIdentifier identifier,
if (catalog == nullptr) [[unlikely]] {
return InvalidArgument("Catalog cannot be null");
}
return std::shared_ptr<Table>(new Table(
std::move(identifier), std::move(metadata), std::move(metadata_location),
std::move(io), std::move(catalog), std::move(full_name), std::move(reporter)));
return std::shared_ptr<Table>(new Table(std::move(identifier), std::move(metadata),
std::move(metadata_location), std::move(io),
std::move(catalog), std::move(full_name),
std::move(reporter), std::move(labels)));
}

Table::~Table() = default;

Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name,
std::shared_ptr<MetricsReporter> reporter)
std::shared_ptr<MetricsReporter> reporter, Labels labels)
: identifier_(std::move(identifier)),
full_name_(full_name.empty() ? identifier_.ToString() : std::move(full_name)),
metadata_(std::move(metadata)),
metadata_location_(std::move(metadata_location)),
io_(std::move(io)),
catalog_(std::move(catalog)),
reporter_(std::move(reporter)),
metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())) {}
metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())),
labels_(std::move(labels)) {}

const std::string& Table::uuid() const { return metadata_->table_uuid; }

Expand Down Expand Up @@ -164,6 +164,8 @@ const std::shared_ptr<Catalog>& Table::catalog() const { return catalog_; }

const std::shared_ptr<MetricsReporter>& Table::reporter() const { return reporter_; }

const Labels& Table::labels() const { return labels_; }

Result<std::unique_ptr<LocationProvider>> Table::location_provider() const {
return LocationProvider::Make(metadata_->location, metadata_->properties);
}
Expand Down
14 changes: 12 additions & 2 deletions src/iceberg/table.h
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
#include <vector>

#include "iceberg/iceberg_export.h"
#include "iceberg/labels.h"
#include "iceberg/snapshot.h"
#include "iceberg/table_identifier.h"
#include "iceberg/type_fwd.h"
Expand All @@ -50,11 +51,12 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
/// string representation of identifier when empty.
/// \param[in] reporter Optional metrics reporter for this table. Defaults to nullptr
/// (noop).
/// \param[in] labels Catalog-provided labels for this table. Defaults to empty.
static Result<std::shared_ptr<Table>> Make(
TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name = "",
std::shared_ptr<MetricsReporter> reporter = nullptr);
std::shared_ptr<MetricsReporter> reporter = nullptr, Labels labels = {});

virtual ~Table();

Expand Down Expand Up @@ -130,6 +132,13 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
/// \brief Returns the metrics reporter for this table.
const std::shared_ptr<MetricsReporter>& reporter() const;

/// \brief Returns the catalog-provided labels for this table.
///
/// Labels come only from the catalog's load, create or register response and are fixed
/// for the life of this table: Refresh() does not update them, and a table returned by
/// Transaction::Commit() has none. Load the table again to get current labels.
const Labels& labels() const;

/// \brief Returns a LocationProvider for this table
Result<std::unique_ptr<LocationProvider>> location_provider() const;

Expand Down Expand Up @@ -216,7 +225,7 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name,
std::shared_ptr<MetricsReporter> reporter = nullptr);
std::shared_ptr<MetricsReporter> reporter = nullptr, Labels labels = {});

const TableIdentifier identifier_;
const std::string full_name_;
Expand All @@ -226,6 +235,7 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
std::shared_ptr<Catalog> catalog_;
std::shared_ptr<MetricsReporter> reporter_;
std::unique_ptr<class TableMetadataCache> metadata_cache_;
const Labels labels_;
};

/// \brief A table created by stage-create and not yet committed.
Expand Down
Loading
Loading