Skip to content
Merged
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
11 changes: 9 additions & 2 deletions example/demo_example.cc
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/

#include <iostream>
#include <utility>

#include "iceberg/arrow/arrow_io_util.h"
#include "iceberg/avro/avro_register.h"
Expand All @@ -42,8 +43,14 @@ int main(int argc, char** argv) {
iceberg::avro::RegisterAll();
iceberg::parquet::RegisterAll();

auto catalog = iceberg::InMemoryCatalog::Make("test", iceberg::arrow::MakeLocalFileIO(),
warehouse_location, properties);
auto catalog_result = iceberg::InMemoryCatalog::Make(
"test", iceberg::arrow::MakeLocalFileIO(), warehouse_location, properties);
if (!catalog_result.has_value()) {
std::cerr << "Failed to create catalog: " << catalog_result.error().message
<< std::endl;
return 1;
}
auto catalog = std::move(catalog_result.value());

auto register_result = catalog->RegisterTable({.name = table_name}, table_location);
if (!register_result.has_value()) {
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ set(ICEBERG_SOURCES
arrow_c_data_guard_internal.cc
arrow_c_data_util.cc
arrow_row_builder.cc
catalog/catalog_util.cc
catalog/memory/in_memory_catalog.cc
catalog/session_catalog.cc
catalog/session_context.cc
Expand All @@ -39,6 +40,7 @@ set(ICEBERG_SOURCES
expression/projections.cc
expression/residual_evaluator.cc
expression/rewrite_not.cc
expression/sanitize_expression.cc
expression/strict_metrics_evaluator.cc
expression/term.cc
file_io.cc
Expand Down
49 changes: 49 additions & 0 deletions src/iceberg/catalog/catalog_util.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* 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/catalog/catalog_util.h"

#include "iceberg/table_identifier.h"

namespace iceberg {

std::string CatalogUtil::FullTableName(std::string_view catalog_name,
const TableIdentifier& identifier) {
if (catalog_name.empty()) {
return identifier.ToString();
}

std::string result;
if (catalog_name.contains('/') || catalog_name.contains(':')) {
result = catalog_name;
if (!catalog_name.ends_with('/')) {
result += '/';
}
} else {
result = std::string(catalog_name) + '.';
}

for (const auto& level : identifier.ns.levels) {
result += level + '.';
}
result += identifier.name;
return result;
}

} // namespace iceberg
38 changes: 38 additions & 0 deletions src/iceberg/catalog/catalog_util.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
/*
* 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

#include <string>
#include <string_view>

#include "iceberg/iceberg_export.h"
#include "iceberg/type_fwd.h"

namespace iceberg {

/// \brief Utilities for working with catalogs.
class ICEBERG_EXPORT CatalogUtil {
public:
/// \brief Return the fully-qualified name for a table in a catalog.
static std::string FullTableName(std::string_view catalog_name,
const TableIdentifier& identifier);
};

} // namespace iceberg
42 changes: 29 additions & 13 deletions src/iceberg/catalog/memory/in_memory_catalog.cc
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@
#include <algorithm>
#include <iterator>

#include "iceberg/catalog/catalog_util.h"
#include "iceberg/file_io.h"
#include "iceberg/metrics/metrics_reporters.h"
#include "iceberg/table.h"
#include "iceberg/table_identifier.h"
#include "iceberg/table_metadata.h"
Expand Down Expand Up @@ -337,22 +339,30 @@ Status InMemoryNamespace::UpdateTableMetadataLocation(
return {};
}

std::shared_ptr<InMemoryCatalog> InMemoryCatalog::Make(
Result<std::shared_ptr<InMemoryCatalog>> InMemoryCatalog::Make(
const std::string& name, const std::shared_ptr<FileIO>& file_io,
const std::string& warehouse_location,
const std::unordered_map<std::string, std::string>& properties) {
return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties);
std::shared_ptr<MetricsReporter> reporter;
auto it = properties.find(std::string(kMetricsReporterImpl));
if (it != properties.end() && !it->second.empty() &&
it->second != kMetricsReporterTypeNoop) {
ICEBERG_ASSIGN_OR_RAISE(reporter, MetricsReporters::Load(properties));
}
return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties,
std::move(reporter));
}

InMemoryCatalog::InMemoryCatalog(
const std::string& name, const std::shared_ptr<FileIO>& file_io,
const std::string& warehouse_location,
const std::unordered_map<std::string, std::string>& properties)
InMemoryCatalog::InMemoryCatalog(std::string name, std::shared_ptr<FileIO> file_io,
std::string warehouse_location,
std::unordered_map<std::string, std::string> properties,
std::shared_ptr<MetricsReporter> reporter)
: catalog_name_(std::move(name)),
properties_(std::move(properties)),
file_io_(std::move(file_io)),
warehouse_location_(std::move(warehouse_location)),
root_namespace_(std::make_unique<InMemoryNamespace>()) {}
root_namespace_(std::make_unique<InMemoryNamespace>()),
reporter_(std::move(reporter)) {}

InMemoryCatalog::~InMemoryCatalog() = default;

Expand Down Expand Up @@ -429,7 +439,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::CreateTable(
ICEBERG_RETURN_UNEXPECTED(
root_namespace_->UpdateTableMetadataLocation(identifier, metadata_file_location));
return Table::Make(identifier, std::move(table_metadata),
std::move(metadata_file_location), file_io_, shared_from_this());
std::move(metadata_file_location), file_io_, shared_from_this(),
CatalogUtil::FullTableName(name(), identifier), reporter_);
}

Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
Expand Down Expand Up @@ -480,7 +491,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
TableMetadataUtil::DeleteRemovedMetadataFiles(*file_io_, base.get(), *updated);

return Table::Make(identifier, std::move(updated), std::move(new_metadata_location),
file_io_, shared_from_this());
file_io_, shared_from_this(),
CatalogUtil::FullTableName(name(), identifier), reporter_);
}

Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
Expand All @@ -500,8 +512,10 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
auto table_metadata,
TableMetadata::Make(*schema, *spec, *order, base_location, properties));
ICEBERG_ASSIGN_OR_RAISE(
auto table, StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
shared_from_this()));
auto table,
StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
shared_from_this(),
CatalogUtil::FullTableName(name(), identifier), reporter_));
return Transaction::Make(std::move(table), TransactionKind::kCreate);
}

Expand Down Expand Up @@ -581,7 +595,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::LoadTable(
ICEBERG_ASSIGN_OR_RAISE(auto metadata,
TableMetadataUtil::Read(*file_io_, metadata_location));
return Table::Make(identifier, std::move(metadata), std::move(metadata_location),
file_io_, shared_from_this());
file_io_, shared_from_this(),
CatalogUtil::FullTableName(name(), identifier), reporter_);
}

Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
Expand All @@ -601,7 +616,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
return UnknownError("The registry failed.");
}
return Table::Make(identifier, std::move(metadata), metadata_file_location, file_io_,
shared_from_this());
shared_from_this(), CatalogUtil::FullTableName(name(), identifier),
reporter_);
}

} // namespace iceberg
11 changes: 7 additions & 4 deletions src/iceberg/catalog/memory/in_memory_catalog.h
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
/// \file iceberg/catalog/memory/in_memory_catalog.h
/// \brief Provide an in-memory catalog implementation.

#include <memory>
#include <shared_mutex>

#include "iceberg/catalog.h"
Expand All @@ -42,12 +43,13 @@ class ICEBERG_EXPORT InMemoryCatalog
: public Catalog,
public std::enable_shared_from_this<InMemoryCatalog> {
public:
InMemoryCatalog(std::string const& name, std::shared_ptr<FileIO> const& file_io,
std::string const& warehouse_location,
std::unordered_map<std::string, std::string> const& properties);
InMemoryCatalog(std::string name, std::shared_ptr<FileIO> file_io,
std::string warehouse_location,
std::unordered_map<std::string, std::string> properties,
std::shared_ptr<MetricsReporter> reporter = nullptr);
~InMemoryCatalog() override;

static std::shared_ptr<InMemoryCatalog> Make(
static Result<std::shared_ptr<InMemoryCatalog>> Make(
std::string const& name, std::shared_ptr<FileIO> const& file_io,
std::string const& warehouse_location,
std::unordered_map<std::string, std::string> const& properties);
Expand Down Expand Up @@ -109,6 +111,7 @@ class ICEBERG_EXPORT InMemoryCatalog
std::string warehouse_location_;
std::unique_ptr<class InMemoryNamespace> root_namespace_;
mutable std::shared_mutex mutex_;
std::shared_ptr<MetricsReporter> reporter_;
};

} // namespace iceberg
2 changes: 1 addition & 1 deletion src/iceberg/catalog/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
subdir('memory')

install_headers(
['session_catalog.h', 'session_context.h'],
['catalog_util.h', 'session_catalog.h', 'session_context.h'],
subdir: 'iceberg/catalog',
)

Expand Down
1 change: 1 addition & 0 deletions src/iceberg/catalog/rest/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ set(ICEBERG_REST_SOURCES
resource_paths.cc
rest_catalog.cc
rest_file_io.cc
rest_metrics_reporter.cc
rest_util.cc
types.cc)

Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/catalog/rest/catalog_properties.h
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
inline static Entry<std::string> kNamespaceSeparator{"namespace-separator", "%1F"};
/// \brief The snapshot loading mode (ALL or REFS).
inline static Entry<std::string> kSnapshotLoadingMode{"snapshot-loading-mode", "ALL"};
/// \brief Whether to report metrics to the REST catalog server (default: true).
inline static Entry<std::string> kMetricsReportingEnabled{
"rest-metrics-reporting-enabled", "true"};
/// \brief The prefix for HTTP headers.
inline static constexpr std::string_view kHeaderPrefix = "header.";

Expand Down
1 change: 1 addition & 0 deletions src/iceberg/catalog/rest/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ iceberg_rest_sources = files(
'resource_paths.cc',
'rest_catalog.cc',
'rest_file_io.cc',
'rest_metrics_reporter.cc',
Comment thread
wgtmac marked this conversation as resolved.
'rest_util.cc',
'types.cc',
)
Expand Down
Loading
Loading