Changed-lines coverage: PR changed C/C++ lines covered by tests: 33.33% (126/378) Uncovered changed code (with context): ================================================================================ src/Databases/DataLake/AWSV4Signer.cpp ================================================================================ --- uncovered block 35-50 --- 33 | Aws::Http::HttpMethod mapPocoMethodToAws(const String & method) 34 | { >> 35 | using Poco::Net::HTTPRequest; 36 | >> 37 | static const std::pair supported_methods[] = { >> 38 | {HTTPRequest::HTTP_GET, Aws::Http::HttpMethod::HTTP_GET}, >> 39 | {HTTPRequest::HTTP_POST, Aws::Http::HttpMethod::HTTP_POST}, >> 40 | {HTTPRequest::HTTP_PUT, Aws::Http::HttpMethod::HTTP_PUT}, >> 41 | {HTTPRequest::HTTP_DELETE, Aws::Http::HttpMethod::HTTP_DELETE}, >> 42 | {HTTPRequest::HTTP_HEAD, Aws::Http::HttpMethod::HTTP_HEAD}, >> 43 | {HTTPRequest::HTTP_PATCH, Aws::Http::HttpMethod::HTTP_PATCH}, 44 | }; 45 | >> 46 | for (const auto & [poco_method, aws_method] : supported_methods) >> 47 | if (method == poco_method) >> 48 | return aws_method; 49 | >> 50 | throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unsupported HTTP method for AWS SigV4 signing: {}", method); 51 | } 52 | --- uncovered block 65-103 --- 63 | DB::HTTPHeaderEntries & out_headers) 64 | { >> 65 | const Aws::Http::URI aws_uri(uri.toString().c_str()); >> 66 | Aws::Http::Standard::StandardHttpRequest request(aws_uri, mapPocoMethodToAws(method)); 67 | >> 68 | for (const auto & h : extra_headers) 69 | { >> 70 | if (Poco::icompare(h.name, "authorization") == 0) >> 71 | continue; >> 72 | request.SetHeaderValue(Aws::String(h.name.c_str(), h.name.size()), Aws::String(h.value.c_str(), h.value.size())); 73 | } 74 | >> 75 | if (!payload.empty()) 76 | { >> 77 | auto body_stream = Aws::MakeShared("AWSV4Signer"); >> 78 | body_stream->write(payload.data(), static_cast(payload.size())); >> 79 | body_stream->seekg(0); >> 80 | request.AddContentBody(body_stream); 81 | } 82 | >> 83 | static constexpr bool sign_body = true; >> 84 | if (!signer.SignRequest(request, region.c_str(), service.c_str(), sign_body)) >> 85 | throw DB::Exception(DB::ErrorCodes::S3_ERROR, "AWS SigV4 signing failed"); 86 | >> 87 | bool has_authorization = false; >> 88 | for (const auto & [key, value] : request.GetHeaders()) 89 | { >> 90 | if (Poco::icompare(key, "authorization") == 0 && !value.empty()) >> 91 | has_authorization = true; 92 | } >> 93 | if (!has_authorization) >> 94 | throw DB::Exception( >> 95 | DB::ErrorCodes::BAD_ARGUMENTS, >> 96 | "AWS credentials are missing or incomplete; cannot sign S3 Tables REST request"); 97 | >> 98 | out_headers.clear(); >> 99 | for (const auto & [key, value] : request.GetHeaders()) 100 | { >> 101 | if (Poco::icompare(key, "host") == 0) >> 102 | continue; >> 103 | out_headers.emplace_back(String(key.c_str(), key.size()), String(value.c_str(), value.size())); 104 | } 105 | } ================================================================================ src/Databases/DataLake/DatabaseDataLake.cpp ================================================================================ --- uncovered block 206-224 --- 204 | { 205 | if (settings[DatabaseDataLakeSetting::region].value.empty()) >> 206 | throw Exception( >> 207 | ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for Glue catalog. " >> 208 | "Please specify 'SETTINGS region=' in the CREATE DATABASE query"); 209 | } 210 | else if (settings[DatabaseDataLakeSetting::catalog_type].value == DB::DatabaseDataLakeCatalogType::S3_TABLES) 211 | { >> 212 | if (settings[DatabaseDataLakeSetting::region].value.empty()) >> 213 | throw Exception( >> 214 | ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for S3 Tables catalog. " >> 215 | "Please specify 'SETTINGS region=' in the CREATE DATABASE query"); 216 | >> 217 | if (settings[DatabaseDataLakeSetting::warehouse].value.empty()) >> 218 | throw Exception( >> 219 | ErrorCodes::BAD_ARGUMENTS, "`warehouse` setting cannot be empty for S3 Tables catalog. " >> 220 | "Please specify 'SETTINGS warehouse=' in the CREATE DATABASE query"); 221 | } 222 | else if (settings[DatabaseDataLakeSetting::warehouse].value.empty()) 223 | { >> 224 | throw Exception( 225 | ErrorCodes::BAD_ARGUMENTS, "`warehouse` setting cannot be empty. " 226 | "Please specify 'SETTINGS warehouse=' in the CREATE DATABASE query"); --- uncovered block 393-409 --- 391 | throw Exception(ErrorCodes::BAD_ARGUMENTS, "Paimon catalog requires either catalog_credential or (dlf_access_key_id, dlf_access_key_secret and region)"); 392 | } >> 393 | break; 394 | } >> 395 | case DB::DatabaseDataLakeCatalogType::S3_TABLES: 396 | { >> 397 | #if USE_AWS_S3 && USE_SSL >> 398 | catalog_impl = std::make_shared( >> 399 | settings[DatabaseDataLakeSetting::warehouse].value, >> 400 | url, >> 401 | settings[DatabaseDataLakeSetting::region].value, >> 402 | catalog_parameters, >> 403 | Context::getGlobalContextInstance()); 404 | #else 405 | throw Exception( 406 | ErrorCodes::SUPPORT_IS_DISABLED, 407 | "Amazon S3 Tables catalog requires ClickHouse built with USE_AWS_S3 and USE_SSL"); 408 | #endif >> 409 | break; 410 | } 411 | } --- uncovered block 496-496 --- 494 | } 495 | } >> 496 | case DatabaseDataLakeCatalogType::ICEBERG_HIVE: 497 | case DatabaseDataLakeCatalogType::ICEBERG_REST: 498 | case DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE: --- uncovered block 1446-1446 --- 1444 | engine_args = function_define->arguments->children; 1445 | if (requires_arguments && engine_args.empty()) >> 1446 | throw Exception(ErrorCodes::BAD_ARGUMENTS, "Engine `{}` must have arguments", database_engine_name); 1447 | } 1448 | --- uncovered block 1455-1456 --- 1453 | && catalog_type != DatabaseDataLakeCatalogType::ICEBERG_ONELAKE) 1454 | { >> 1455 | throw Exception(ErrorCodes::BAD_ARGUMENTS, >> 1456 | "Engine `Iceberg` must use `rest`, `s3tables`, `biglake`, or `onelake` catalog type only"); 1457 | } 1458 | --- uncovered block 1557-1571 --- 1555 | } 1556 | >> 1557 | engine_func->name = "Paimon"; >> 1558 | break; 1559 | } >> 1560 | case DatabaseDataLakeCatalogType::S3_TABLES: 1561 | { >> 1562 | if (!args.create_query.attach >> 1563 | && !args.context->getSettingsRef()[Setting::allow_experimental_database_iceberg]) 1564 | { >> 1565 | throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, >> 1566 | "DatabaseDataLake with S3 Tables catalog (Iceberg REST) is beta. " >> 1567 | "To allow its usage, enable setting allow_database_iceberg"); 1568 | } 1569 | >> 1570 | engine_func->name = "Iceberg"; >> 1571 | break; 1572 | } 1573 | case DatabaseDataLakeCatalogType::NONE: ================================================================================ src/Databases/DataLake/ICatalog.cpp ================================================================================ --- uncovered block 306-306 --- 304 | bool TableMetadata::hasDataLakeSpecificProperties() const 305 | { >> 306 | return data_lake_specific_metadata.has_value(); 307 | } 308 | ================================================================================ src/Databases/DataLake/RestCatalog.cpp ================================================================================ --- uncovered block 1068-1068 --- 1066 | Poco::URI::QueryParameters params = base_params; 1067 | if (!page_token.empty()) >> 1068 | params.push_back({"pageToken", page_token}); 1069 | 1070 | auto buf = createReadBuffer( --- uncovered block 1219-1219 --- 1217 | Poco::URI::QueryParameters params; 1218 | if (!page_token.empty()) >> 1219 | params.push_back({"pageToken", page_token}); 1220 | 1221 | auto buf = createReadBuffer( --- uncovered block 1776-1781 --- 1774 | 1775 | const auto state_snapshot = state.get(); >> 1776 | const auto & table = storage_id.getTableName(); >> 1777 | auto [namespace_name, table_name] = DataLake::parseTableName(table); >> 1778 | const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name; >> 1779 | auto buf = createReadBuffer(*state_snapshot, state_snapshot->config.prefix / endpoint, /* params */ {}, headers, /* auth_headers */ std::nullopt); 1780 | >> 1781 | if (buf->eof()) 1782 | { 1783 | LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint); ================================================================================ src/Databases/DataLake/S3TablesCatalog.cpp ================================================================================ --- uncovered block 65-127 --- 63 | const CatalogSettings & catalog_settings_, 64 | DB::ContextPtr context_) >> 65 | : RestCatalog(warehouse_, base_url_, "", "", false, context_) >> 66 | , region(region_) >> 67 | , storage_endpoint(catalog_settings_.storage_endpoint) >> 68 | , signing_service("s3tables") 69 | { >> 70 | if (region.empty()) >> 71 | throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "S3 Tables catalog requires non-empty `region` setting"); 72 | >> 73 | DB::S3::CredentialsConfiguration creds_config; >> 74 | creds_config.use_environment_credentials = true; >> 75 | creds_config.role_arn = catalog_settings_.aws_role_arn; >> 76 | creds_config.role_session_name = catalog_settings_.aws_role_session_name; 77 | >> 78 | const auto & server_settings = getContext()->getGlobalContext()->getServerSettings(); >> 79 | const DB::Settings & global_settings = getContext()->getGlobalContext()->getSettingsRef(); 80 | >> 81 | int s3_max_redirects = static_cast(server_settings[DB::ServerSetting::s3_max_redirects]); >> 82 | if (global_settings.isChanged("s3_max_redirects")) >> 83 | s3_max_redirects = static_cast(global_settings[DB::Setting::s3_max_redirects]); 84 | >> 85 | int s3_retry_attempts = static_cast(server_settings[DB::ServerSetting::s3_retry_attempts]); >> 86 | if (global_settings.isChanged("s3_retry_attempts")) >> 87 | s3_retry_attempts = static_cast(global_settings[DB::Setting::s3_retry_attempts]); 88 | >> 89 | bool s3_slow_all_threads_after_network_error = global_settings[DB::Setting::s3_slow_all_threads_after_network_error]; >> 90 | bool s3_slow_all_threads_after_retryable_error = false; >> 91 | bool enable_s3_requests_logging = global_settings[DB::Setting::enable_s3_requests_logging]; 92 | >> 93 | DB::S3::PocoHTTPClientConfiguration poco_config = DB::S3::ClientFactory::instance().createClientConfiguration( >> 94 | region, >> 95 | getContext()->getRemoteHostFilter(), >> 96 | s3_max_redirects, >> 97 | DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = static_cast(s3_retry_attempts)}, >> 98 | s3_slow_all_threads_after_network_error, >> 99 | s3_slow_all_threads_after_retryable_error, >> 100 | enable_s3_requests_logging, >> 101 | /* for_disk_s3 = */ false, >> 102 | /* opt_disk_name = */ {}, >> 103 | /* request_throttler = */ {}); 104 | >> 105 | Aws::Auth::AWSCredentials credentials(catalog_settings_.aws_access_key_id, catalog_settings_.aws_secret_access_key); >> 106 | credentials_provider = DB::S3::getCredentialsProvider(poco_config, credentials, creds_config); 107 | >> 108 | signer = std::make_unique( >> 109 | credentials_provider, >> 110 | "s3tables", >> 111 | Aws::String(region.data(), region.size()), >> 112 | Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always, >> 113 | /* urlEscapePath = */ false); 114 | 115 | /// The signer is fully initialised above, so the virtual `createReadBuffer` reached by 116 | /// `loadConfig` dispatches to this class's SigV4 implementation. >> 117 | CatalogState initial_state; >> 118 | initial_state.config = loadConfig(initial_state); 119 | >> 120 | if (initial_state.config.prefix.empty()) 121 | { >> 122 | String encoded_warehouse; >> 123 | Poco::URI::encode(warehouse_, "", encoded_warehouse); >> 124 | initial_state.config.prefix = encoded_warehouse; 125 | } 126 | >> 127 | state.set(std::make_unique(std::move(initial_state))); 128 | } 129 | --- uncovered block 135-159 --- 133 | CatalogTables S3TablesCatalog::getTables() const 134 | { >> 135 | auto namespaces = listChildNamespaces(""); 136 | >> 137 | auto & pool = getContext()->getIcebergCatalogThreadpool(); >> 138 | DB::ThreadPoolCallbackRunnerLocal runner(pool, DB::ThreadName::DATALAKE_REST_CATALOG); 139 | >> 140 | DB::Names tables; >> 141 | std::mutex mutex; >> 142 | for (const auto & ns : namespaces) 143 | { >> 144 | runner.enqueueAndKeepTrack( >> 145 | [&, ns] 146 | { >> 147 | auto tables_in_ns = listTablesInNamespace(ns); >> 148 | std::lock_guard lock(mutex); >> 149 | std::move(tables_in_ns.begin(), tables_in_ns.end(), std::back_inserter(tables)); >> 150 | }); 151 | } >> 152 | runner.waitForAllToFinishAndRethrowFirstError(); 153 | 154 | /// A REST catalog is Iceberg-only, so every listed table is readable. >> 155 | CatalogTables result; >> 156 | result.reserve(tables.size()); >> 157 | for (auto & name : tables) >> 158 | result.push_back(CatalogTable{.name = std::move(name)}); >> 159 | return result; 160 | } 161 | --- uncovered block 167-168 --- 165 | TableMetadata & result) const 166 | { >> 167 | if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, result)) >> 168 | return false; 169 | 170 | /// For S3 Tables the catalog and the underlying data live in AWS S3 under the same --- uncovered block 182-244 --- 180 | /// the IAM fallback calls `credentials_provider->GetAWSCredentials`, which the AWS SDK refreshes 181 | /// on expiry. >> 182 | bool need_credentials = true; >> 183 | if (result.hasStorageCredentials()) 184 | { >> 185 | auto creds = std::dynamic_pointer_cast(result.getStorageCredentials()); >> 186 | if (creds && !creds->isEmpty()) >> 187 | need_credentials = false; 188 | } 189 | >> 190 | if (need_credentials) 191 | { >> 192 | LOG_DEBUG(log, "S3 Tables: no vended credentials for {}.{}, injecting catalog IAM credentials", namespace_name, table_name); >> 193 | auto aws_creds = credentials_provider->GetAWSCredentials(); >> 194 | result.withStorageCredentials(); >> 195 | result.setStorageCredentials(std::make_shared( >> 196 | aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken())); 197 | } 198 | >> 199 | if (result.getEndpoint().empty()) 200 | { >> 201 | String endpoint = storage_endpoint.empty() >> 202 | ? DB::S3::expandRegionToAmazonPath(region) >> 203 | : storage_endpoint; >> 204 | LOG_DEBUG(log, "S3 Tables: no endpoint for {}.{}, injecting: {}", namespace_name, table_name, endpoint); >> 205 | result.setEndpoint(endpoint); 206 | } 207 | >> 208 | if (auto props = result.getDataLakeSpecificProperties(); >> 209 | props && !props->iceberg_metadata_file_location.empty()) 210 | { >> 211 | const String & loc = props->iceberg_metadata_file_location; >> 212 | auto scheme_end = loc.find("://"); >> 213 | if (scheme_end != String::npos) 214 | { >> 215 | auto path_start = loc.find('/', scheme_end + 3); >> 216 | if (path_start != String::npos) >> 217 | props->iceberg_metadata_file_location = loc.substr(path_start + 1); 218 | } >> 219 | result.setDataLakeSpecificProperties(std::move(props)); 220 | } 221 | >> 222 | return true; 223 | } 224 | 225 | void S3TablesCatalog::dropTable(const String & namespace_name, const String & table_name) const 226 | { >> 227 | const auto state_snapshot = state.get(); >> 228 | const std::string endpoint >> 229 | = (base_url / state_snapshot->config.prefix / "namespaces" / namespace_name / "tables" / table_name).string() >> 230 | + "?purgeRequested=True"; 231 | >> 232 | Poco::JSON::Object::Ptr request_body = nullptr; >> 233 | try 234 | { >> 235 | sendRequest(*state_snapshot, endpoint, request_body, Poco::Net::HTTPRequest::HTTP_DELETE, true); >> 236 | LOG_INFO(log, "S3 Tables: dropped table {}.{} (purgeRequested=True)", namespace_name, table_name); 237 | } >> 238 | catch (const DB::HTTPException & ex) 239 | { >> 240 | if (ex.getHTTPStatus() == Poco::Net::HTTPResponse::HTTP_NOT_FOUND) 241 | // 404 is returned by the API when the table does not exist >> 242 | LOG_DEBUG(log, "S3 Tables: table {}.{} already does not exist (404 on purge-delete)", namespace_name, table_name); 243 | else >> 244 | throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Failed to drop table {}", ex.displayText()); 245 | } 246 | } --- uncovered block 258-273 --- 256 | bool isSigV4AuthHeader(const String & name) 257 | { >> 258 | if (Poco::icompare(name, "authorization") == 0) >> 259 | return true; >> 260 | static constexpr size_t x_amz_prefix_len = 6; >> 261 | return name.size() >= x_amz_prefix_len >> 262 | && Poco::icompare(name, 0, x_amz_prefix_len, "x-amz-") == 0; 263 | } 264 | 265 | DB::HTTPHeaderEntries extractSigV4AuthHeaders(DB::HTTPHeaderEntries && all_signed) 266 | { >> 267 | DB::HTTPHeaderEntries auth_headers; >> 268 | for (auto & h : all_signed) 269 | { >> 270 | if (isSigV4AuthHeader(h.name)) >> 271 | auth_headers.push_back(std::move(h)); 272 | } >> 273 | return auth_headers; 274 | } 275 | --- uncovered block 285-285 --- 283 | const std::optional & /* auth_headers */) const 284 | { >> 285 | const auto & context = getContext(); 286 | 287 | /// enable_url_encoding=false to allow using tables with encoded sequences in names like 'foo%2Fbar' WARNING: Failed to get start time for [Print Uncovered Code] - start time and duration won't be set --- Coverage counts --- Lines : baseline 936,613/1,082,719 -> current 936,591/1,083,087 (delta -22 / +368) Functions : baseline 806,733/878,152 -> current 806,732/878,195 (delta -1 / +43) Branches : baseline 307,028/389,982 -> current 307,069/390,126 (delta +41 / +144)