Skip to content
3 changes: 2 additions & 1 deletion src/Core/ProtocolDefines.h
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_META
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_FILE_BUCKETS_INFO = 4;
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_EXCLUDED_ROWS = 5;
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_FILE_STATS = 6;
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_FILE_STATS;
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH = 9;
static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH;

static constexpr auto DATA_LAKE_TABLE_STATE_SNAPSHOT_PROTOCOL_VERSION = 1;

Expand Down
3 changes: 3 additions & 0 deletions src/Core/Settings.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -641,6 +641,9 @@ Use multiple threads for azure multipart upload.
)", 0) \
DECLARE(Bool, s3_throw_on_zero_files_match, false, R"(
Throw an error, when ListObjects request cannot match any files
)", 0) \
DECLARE(Bool, object_storage_propagate_credentials_to_other_storages, false, R"(
Reuse base-storage credentials for a secondary object storage. For `S3`, credentials are reused when the endpoint matches; when this setting is enabled, they are also reused across different endpoints, including less secure connections (for example, from `https` to plain `http`). For `Azure`, reads stay within the base account.
)", 0) \
DECLARE(Bool, hdfs_throw_on_zero_files_match, false, R"(
Throw an error if matched zero files according to glob expansion rules.
Expand Down
2 changes: 1 addition & 1 deletion src/Core/SettingsChangesHistory.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
addSettingsChanges(settings_changes_history, "26.1.3.20001.altinityantalya",
{
{"iceberg_partition_timezone", "", "", "New setting."},
// {"s3_propagate_credentials_to_other_storages", false, false, "New setting"},
{"object_storage_propagate_credentials_to_other_storages", false, false, "New setting"},
{"export_merge_tree_part_filename_pattern", "", "{part_name}_{checksum}", "New setting"},
// {"use_parquet_metadata_cache", false, true, "Enables cache of parquet file metadata."},
// {"input_format_parquet_use_metadata_cache", true, false, "Obsolete. No-op"}, // https://github.com/Altinity/ClickHouse/pull/586
Expand Down
4 changes: 3 additions & 1 deletion src/Databases/DataLake/DatabaseDataLake.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -635,7 +635,9 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
LOG_DEBUG(log, "Has no credentials");
}
}
else if (!lightweight && table_metadata.requiresCredentials() && std::find(vended_credentials_catalogs.begin(), vended_credentials_catalogs.end(), catalog->getCatalogType()) == vended_credentials_catalogs.end())
else if (!lightweight && table_metadata.requiresCredentials()
&& std::find(vended_credentials_catalogs.begin(), vended_credentials_catalogs.end(), catalog->getCatalogType()) == vended_credentials_catalogs.end()
&& table_metadata.getStorageType() != DatabaseDataLakeStorageType::Local)
{
throw Exception(
ErrorCodes::BAD_ARGUMENTS,
Expand Down
14 changes: 8 additions & 6 deletions src/IO/S3/URI.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,12 @@ namespace DB

struct URIConverter
{
static void modifyURI(Poco::URI & uri, NameToNameMap mapper)
static void modifyURI(Poco::URI & uri, NameToNameMap mapper, bool enable_url_encoding = true)
{
Macros macros({{"bucket", uri.getHost()}});
uri = macros.expand(mapper[uri.getScheme()]).empty() ? uri : Poco::URI(macros.expand(mapper[uri.getScheme()]) + uri.getPathAndQuery());
uri = macros.expand(mapper[uri.getScheme()]).empty()
? uri
: Poco::URI(macros.expand(mapper[uri.getScheme()]) + uri.getPathAndQuery(), enable_url_encoding);
}
};

Expand All @@ -34,7 +36,7 @@ namespace ErrorCodes
namespace S3
{

URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_presigned_query_parameters, S3UriStyle uri_style)
URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_presigned_query_parameters, S3UriStyle uri_style, bool enable_url_encoding)
{
/// Case when AWS Private Link Interface is being used
/// E.g. (bucket.vpce-07a1cd78f1bd55c5f-j3a3vg6w.s3.us-east-1.vpce.amazonaws.com/bucket-name/key)
Expand All @@ -46,9 +48,9 @@ URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_pre
else
uri_str = uri_;

uri = Poco::URI(uri_str);
uri = Poco::URI(uri_str, enable_url_encoding);
/// Keep a copy of how Poco parsed the original string before any mapping
Poco::URI original_uri(uri_str);
Poco::URI original_uri(uri_str, enable_url_encoding);
bool looks_like_presigned = false;
for (const auto & [qk, qv] : original_uri.getQueryParameters())
{
Expand Down Expand Up @@ -93,7 +95,7 @@ URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_pre
}

if (!mapper.empty())
URIConverter::modifyURI(uri, mapper);
URIConverter::modifyURI(uri, mapper, enable_url_encoding);
}

storage_name = "S3";
Expand Down
3 changes: 2 additions & 1 deletion src/IO/S3/URI.h
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@ struct URI
const std::string & uri_,
bool allow_archive_path_syntax = false,
bool keep_presigned_query_parameters = true,
S3UriStyle uri_style = S3UriStyle::AUTO);
S3UriStyle uri_style = S3UriStyle::AUTO,
bool enable_url_encoding = true);
void addRegionToURI(const std::string & region);

static void validateBucket(const std::string & bucket, const Poco::URI & uri);
Expand Down
6 changes: 2 additions & 4 deletions src/Interpreters/ClusterFunctionReadTask.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,8 @@ ClusterFunctionReadTaskResponse::ClusterFunctionReadTaskResponse(ObjectInfoPtr o
data_lake_metadata = object->data_lake_metadata.value();

#if USE_AVRO
if (std::dynamic_pointer_cast<IcebergDataObjectInfo>(object))
{
iceberg_info = dynamic_cast<IcebergDataObjectInfo &>(*object).info;
}
if (auto iceberg_object = std::dynamic_pointer_cast<IcebergDataObjectInfo>(object))
iceberg_info = iceberg_object->info;
#endif

file_meta_info = object->relative_path_with_metadata.file_meta_info;
Expand Down
6 changes: 5 additions & 1 deletion src/Interpreters/IcebergMetadataLog.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -106,12 +106,16 @@ void insertRowToLogTable(
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg metadata log table is not configured");
}

String normalized_table_path = table_path;
while (normalized_table_path.size() > 1 && normalized_table_path.back() == '/')
normalized_table_path.pop_back();

iceberg_metadata_log->add(
DB::IcebergMetadataLogElement{
.current_time = spec.tv_sec,
.query_id = local_context->getCurrentQueryId(),
.content_type = row_log_level,
.table_path = table_path,
.table_path = normalized_table_path,
.file_path = file_path.serialize(),
.metadata_content = get_row(),
.row_in_file = row_in_file,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
}


const auto file_path_key = IcebergPathFromMetadata::deserialize(
const auto file_path_from_metadata = IcebergPathFromMetadata::deserialize(
getValueFromRowByName(row_index, c_data_file_file_path, TypeIndex::String).safeGet<String>());
/// NOTE: This is weird, because in manifest file partition looks like this:
/// {
Expand Down Expand Up @@ -257,7 +257,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
case FileContentType::DATA: {
return std::make_shared<const ParsedManifestFileEntry>(
FileContentType::DATA,
file_path_key,
file_path_from_metadata,
row_index,
status,
sequence_number,
Expand Down Expand Up @@ -304,7 +304,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
}
return std::make_shared<const ParsedManifestFileEntry>(
FileContentType::POSITION_DELETE,
file_path_key,
file_path_from_metadata,
row_index,
status,
sequence_number,
Expand Down Expand Up @@ -335,7 +335,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
c_data_file_equality_ids);
return std::make_shared<const ParsedManifestFileEntry>(
FileContentType::EQUALITY_DELETE,
file_path_key,
file_path_from_metadata,
row_index,
status,
sequence_number,
Expand Down
3 changes: 3 additions & 0 deletions src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,9 @@ class IDataLakeMetadata : boost::noncopyable

virtual bool operator==(const IDataLakeMetadata & other) const = 0;

/// Returns the full table location URI (e.g. `s3a://bucket/prefix/table/`)
virtual std::string getTableLocation() const { return {}; }

/// Return iterator to `data files`.
using FileProgressCallback = std::function<void(FileProgress)>;
virtual ObjectIterator iterate(
Expand Down
Loading
Loading