diff --git a/src/Interpreters/GlobalSubqueriesVisitor.cpp b/src/Interpreters/GlobalSubqueriesVisitor.cpp index 7ac4337fd92be..a1261e64f0060 100644 --- a/src/Interpreters/GlobalSubqueriesVisitor.cpp +++ b/src/Interpreters/GlobalSubqueriesVisitor.cpp @@ -54,13 +54,16 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names if (const auto * ast_table_expr = ast->as()) { - subquery_or_table_name = ast_table_expr->subquery; - + /// `interpretSubquery` reads a table function. A `GLOBAL JOIN s3(...)` is that shape. if (ast_table_expr->database_and_table_name) { subquery_or_table_name = ast_table_expr->database_and_table_name; is_table = true; } + else if (ast_table_expr->table_function) + subquery_or_table_name = ast_table_expr->table_function; + else + subquery_or_table_name = ast_table_expr->subquery; } else if (ast->as()) { @@ -87,14 +90,14 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names } String alias = subquery_or_table_name->tryGetAlias(); - String external_table_name; - if (alias.empty()) - { - auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true); - external_table_name = fmt::format("_data_{}", toString(hash)); - } - else - external_table_name = alias; + const auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true); + const String hashed_name = fmt::format("_data_{}", toString(hash)); + /// The alias is the SQL name. A session temporary table with that name must stay untouched, + /// so the broadcast is stored under `_data_{hash}` and the alias is only `AS alias`. + const bool alias_taken = !alias.empty() + && (external_tables.contains(alias) + || static_cast(getContext()->tryResolveStorageID(StorageID("", alias), Context::ResolveExternal))); + const String external_table_name = (alias.empty() || alias_taken) ? hashed_name : alias; /** We replace the subquery with the name of the temporary table. * It is in this form, the request will go to the remote server. @@ -115,6 +118,7 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names if (auto * ast_table_expr = ast->as()) { ast_table_expr->subquery.reset(); + ast_table_expr->table_function.reset(); ast_table_expr->database_and_table_name = database_and_table_name; ast_table_expr->children.clear(); diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 931b5b08d6638..500a3516e5bf2 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -1491,16 +1491,14 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres auto table_function_ast = table_function_node->toAST(); table_function_ast->setAlias({}); - WriteBufferFromOwnString out; - IAST::FormatSettings format_settings( - /*one_line=*/true, - IdentifierQuotingRule::WhenNecessary, - IdentifierQuotingStyle::Backticks, - /*show_secrets_=*/false); - - table_function_ast->format(out, format_settings); - - auto table_function_serialized_ast = std::move(out.str()); + /// This serialized AST only travels inside the cluster, over the native + /// protocol, to build the same table function on the executing shard (see + /// `ReadFromTableFunctionStep` / `resolveStorages`). It is never shown to the + /// user, so keep secrets, same as the plain-SQL distributed query path + /// (`IStorageCluster::readFromRemoteInitiator` / `ClusterProxy::rewriteSelectQuery` + /// use `formatWithSecretsOneLine` for the same reason). Hiding them here would + /// send a masked literal (e.g. a redacted S3 secret key) to the remote shard. + auto table_function_serialized_ast = table_function_ast->formatWithSecretsOneLine(); auto reading_from_table_function = std::make_unique( sample_block, diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 134ceaa4237f8..315f26adf2636 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -42,6 +42,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -306,33 +309,6 @@ void stripObjectStorageRemoteInitiatorSettings(ASTPtr & query) select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {}); } -void rewriteASTInFunctionsToGlobalIn(ASTPtr & node) -{ - if (!node) - return; - - if (auto * function = node->as()) - { - if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2) - { - const auto & rhs = function->arguments->children[1]; - if (rhs - && (rhs->as() || rhs->as() || rhs->as() - || rhs->as() || rhs->as())) - { - function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name); - } - } - } - - for (auto & child : node->children) - { - if (astIsNestedSelect(child)) - continue; - rewriteASTInFunctionsToGlobalIn(child); - } -} - bool queryHasRightOrFullJoin(const ASTPtr & query) { if (!query) @@ -368,6 +344,59 @@ bool queryHasRightOrFullJoin(const ASTPtr & query) return false; } +bool astContainsTableIdentifier(const ASTPtr & node) +{ + if (!node) + return false; + + if (node->as()) + return true; + + for (const auto & child : node->children) + if (astContainsTableIdentifier(child)) + return true; + + return false; +} + +/// A named table in `IN` is read where the client connected. `IN (SELECT 1)` has no table and stays on the replica. +bool inRightSideIsInitiatorLocal(const ASTPtr & rhs) +{ + if (!rhs) + return false; + + if (rhs->as() || rhs->as()) + return true; + + if (rhs->as() || rhs->as() || rhs->as()) + return astContainsTableIdentifier(rhs); + + return false; +} + +void rewriteInitiatorLocalInToGlobalIn(ASTPtr & node) +{ + if (!node) + return; + + if (auto * function = node->as()) + { + if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2 + && inRightSideIsInitiatorLocal(function->arguments->children[1])) + function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name); + } + + for (auto & child : node->children) + { + if (astIsNestedSelect(child)) + continue; + rewriteInitiatorLocalInToGlobalIn(child); + } +} + +/// `IN some_table` is read on the initiator. `IN (SELECT 1)` stays on the replica. +/// `getSampleBlockImpl` sends this same tree, so the interpreter pass must not +/// turn every subquery into `globalIn` before the broadcast pass runs. void rewriteASTJoinsToGlobal(ASTPtr & query) { ASTSelectQuery * select_query = query->as(); @@ -399,7 +428,7 @@ void rewriteASTJoinsToGlobal(ASTPtr & query) } } - rewriteASTInFunctionsToGlobalIn(query); + rewriteInitiatorLocalInToGlobalIn(query); } } @@ -409,6 +438,47 @@ void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query) rewriteASTJoinsToGlobal(query); } +/// The initiator interpreter usually does this already: `IStorageCluster::isRemote` is true, so +/// `GlobalSubqueriesVisitor` replaces a `GLOBAL` right side with an external table. This pass +/// covers a query that still names a local table when `convertToRemote` runs. A second pass +/// is a no-op once that external table is in the context. Sets are added to the interpreter's +/// `prepared_sets`, which `DelayedCreatingSetsStep` builds before the rows are sent. +namespace +{ +void broadcastOldAnalyzerGlobalJoin(ASTPtr & query_to_send, const SelectQueryInfo & query_info, const ContextPtr & context) +{ + /// Idempotent with `rewriteASTForGlobalJoin`: `IN (SELECT 1)` is still local here. + rewriteASTJoinsToGlobal(query_to_send); + + auto prepared_sets = query_info.prepared_sets ? query_info.prepared_sets : std::make_shared(); + TemporaryTablesMapping external_tables; + bool has_global_subqueries = false; + + /// `table_join` stays null: the right side is read with all of its columns. Passing the + /// analyzed join here would rebuild its actions against an AST the first pass may already + /// have rewritten to an external table. + GlobalSubqueriesVisitor::Data data( + context, + /* subquery_depth */ 0, + /* is_remote */ true, + /* is_explain */ false, + external_tables, + prepared_sets, + has_global_subqueries, + /* table_join */ nullptr); + GlobalSubqueriesVisitor(data).visit(query_to_send); + + /// The read context is const. External tables live on the query context, and a later + /// `Context::createCopy` still sees them through `getExternalTables`. + auto mutable_context = context->getQueryContext(); + for (auto & [name, holder] : external_tables) + { + if (!mutable_context->tryResolveStorageID(StorageID{"", name}, Context::ResolveExternal)) + mutable_context->addExternalTable(name, std::move(*holder)); + } +} +} + /* Try to make subquery to send on nodes Converts @@ -455,14 +525,18 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } case ObjectStorageClusterJoinMode::GLOBAL: { - /// Old analyzer: ship the left scan only. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`. + /// Old analyzer. `RIGHT`/`FULL`/`PASTE` stay on the initiator. `INNER`/`LEFT`/`CROSS` + /// stay in `query_to_send`, with the right side replaced by an external table. if (!query_info.query_tree) { if (queryHasRightOrFullJoin(query_to_send)) { auto info = getQueryJoinInfoFromAST(query_to_send); rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); + return; } + + broadcastOldAnalyzerGlobalJoin(query_to_send, query_info, context); return; } @@ -853,6 +927,12 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote( if (!table_expression->table_function) throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table function in table expression"); + /// `AS alias` is stored on the inner table function. `JoinedTables` reads only the outer + /// table function, so `remote()` must carry the same alias. Leave it on the inner function: + /// `addTableFunction` sends that node to the remote initiator and does not copy a non-empty + /// outer alias onto the replacement. + const String table_alias = table_expression->table_function->tryGetAlias(); + boost::intrusive_ptr remote_query; if (shard_addresses[0].user_specified) @@ -868,6 +948,11 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote( remote_query = makeASTFunction(remote_function_name, make_intrusive(host_name), table_expression->table_function); } + if (!table_alias.empty()) + { + remote_query->setAlias(table_alias); + } + table_expression->table_function = remote_query; auto remote_function = TableFunctionFactory::instance().get(remote_query, new_context); diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp index 562e7ec40938c..79a8e4b722769 100644 --- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp @@ -3,6 +3,8 @@ #include #include #include +#include +#include #include #include #include @@ -155,10 +157,21 @@ DataFileMetaInfo::DataFileMetaInfo( String right_str; if (i_bounds->second.first.tryGet(left_str) && i_bounds->second.second.tryGet(right_str)) { - auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true); - auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false); - if (left && right && !DB::Range::less(*right, *left)) - hyperrectangle = DB::Range(*left, true, *right, true); + const auto non_nullable_type = removeNullable(type); + if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6) + { + /// Two ranges are the middle band, so this column is not a constant. + const auto timestamp_ranges = Iceberg::deserializeDateTime64MinMaxRanges(left_str, right_str, type); + if (timestamp_ranges.size() == 1) + hyperrectangle = DB::Range(timestamp_ranges[0].lower, true, timestamp_ranges[0].upper, true); + } + else + { + auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true); + auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false); + if (left && right && !DB::Range::less(*right, *left)) + hyperrectangle = DB::Range(*left, true, *right, true); + } } } } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index b8bb69a04a6ef..81dca86944d0b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -26,15 +26,8 @@ namespace Iceberg namespace { -/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`. -constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; constexpr Int64 nanoseconds_per_microsecond = 1000; -bool magnitudeExceeds(Int64 value, Int64 threshold) -{ - return value > threshold || value < -threshold; -} - /// Convert ns ticks to us ticks without shrinking a [lower, upper] range. Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) { @@ -57,9 +50,8 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) /// Iceberg timestamps are 8-byte little-endian (spec Appendix D). /// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`). -/// Convert only when `|ticks| > 1e18`. Otherwise fail open: `|ticks| <= 1e16` -/// collides near-epoch nanoseconds (~115 days) with 1970–2286 microseconds, and -/// `(1e16, 1e18]` collides Spark `9999-12-31` sentinels with 1970–2001 nanoseconds. +/// `|ticks| >= 1e18` is nanoseconds. A smaller value stays microseconds here; +/// the min/max pruner also tries nanoseconds for `|ticks| >= 1e16`. std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) { if (str.size() != sizeof(Int64)) @@ -70,15 +62,37 @@ std::optional deserializeDateTime64FromBinaryRepr(const std::string & str if (scale == 6) { - if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) - return std::nullopt; - - unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); + if (!icebergTimestampMagnitudeBelow(unscaled, iceberg_timestamp_nanoseconds_lower_threshold)) + unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); } return DecimalField(DateTime64(unscaled), scale); } +std::optional loadDateTime64Ticks(const std::string & str) +{ + if (str.size() != sizeof(Int64)) + return std::nullopt; + return unalignedLoadLittleEndian(str.data()); +} + +void appendRange(std::vector & ranges, Int64 lower_ticks, Int64 upper_ticks, UInt32 scale) +{ + if (lower_ticks > upper_ticks) + return; + + ranges.push_back(DateTime64BoundRange{ + DecimalField(DateTime64(lower_ticks), scale), + DecimalField(DateTime64(upper_ticks), scale)}); +} + +} + +std::string serializeDateTime64IcebergBound(Int64 ticks) +{ + std::string bytes(sizeof(Int64), '\0'); + unalignedStoreLittleEndian(bytes.data(), ticks); + return bytes; } Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name) @@ -209,6 +223,55 @@ static std::optional deserializeDecimalBound(const std::string & str, UIn return DecimalField(unscaled_value, scale); } +std::vector deserializeDateTime64MinMaxRanges( + const std::string & lower_str, + const std::string & upper_str, + DataTypePtr type) +{ + std::vector ranges; + const auto lower_ticks = loadDateTime64Ticks(lower_str); + const auto upper_ticks = loadDateTime64Ticks(upper_str); + if (!lower_ticks || !upper_ticks) + return ranges; + + const UInt32 scale = getDecimalScale(*removeNullable(type)); + if (scale != 6) + { + appendRange(ranges, *lower_ticks, *upper_ticks, scale); + return ranges; + } + + /// One precision for the pair. Below `1e16` keeps `1e15` and `1e18` as microseconds. + const bool one_below_1e16 = icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_microseconds_only_threshold) + || icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_microseconds_only_threshold); + const bool one_at_or_above_1e18 = !icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_nanoseconds_lower_threshold) + || !icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_nanoseconds_lower_threshold); + + if (one_below_1e16) + { + appendRange(ranges, *lower_ticks, *upper_ticks, scale); + } + else if (one_at_or_above_1e18) + { + appendRange( + ranges, + nanosecondsToMicrosecondsForBound(*lower_ticks, true), + nanosecondsToMicrosecondsForBound(*upper_ticks, false), + scale); + } + else + { + appendRange(ranges, *lower_ticks, *upper_ticks, scale); + appendRange( + ranges, + nanosecondsToMicrosecondsForBound(*lower_ticks, true), + nanosecondsToMicrosecondsForBound(*upper_ticks, false), + scale); + } + + return ranges; +} + std::optional deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound) { auto non_nullable_type = removeNullable(expected_type); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h index 21156647fd4b4..e2af51f6b3b63 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h @@ -16,6 +16,16 @@ namespace DB::Iceberg { +/// `iceberg_timestamp_microseconds_only_threshold` microseconds is `2286-11-20`. +/// `iceberg_timestamp_nanoseconds_lower_threshold` nanoseconds is `2001-09-09`. +constexpr Int64 iceberg_timestamp_microseconds_only_threshold = 10'000'000'000'000'000LL; +constexpr Int64 iceberg_timestamp_nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; + +inline bool icebergTimestampMagnitudeBelow(Int64 value, Int64 threshold) +{ + return value > -threshold && value < threshold; +} + /// Convert a Field to Int64, accepting Int64 and UInt64 (with overflow check). Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name); @@ -32,6 +42,26 @@ std::vector fieldToInt64Array(const Field & value, std::string_view conte /// See https://iceberg.apache.org/spec/#appendix-d-single-value-serialization std::optional deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound); +/// Appendix D stores `timestamp` and `timestamp_ns` as an 8-byte little-endian `long`. +std::string serializeDateTime64IcebergBound(Int64 ticks); + +/// One interpretation of an Iceberg `timestamp` lower/upper pair. +struct DateTime64BoundRange +{ + Field lower; + Field upper; +}; + +/// `DateTime64(6)` min/max used for file pruning. +/// Both bounds use one precision. If either has `|ticks| < 1e16`, both are +/// microseconds. Otherwise, if either has `|ticks| >= 1e18`, both are nanoseconds. +/// When both sit in `[1e16, 1e18)`, both ranges are returned. +/// `DateTime64(9)` returns the raw nanosecond range. +std::vector deserializeDateTime64MinMaxRanges( + const std::string & lower_str, + const std::string & upper_str, + DataTypePtr type); + } #endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index e07dff8ff28f8..2277bcae28ee5 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -37,6 +37,7 @@ #include #include #include +#include #include #include #include @@ -163,6 +164,8 @@ bool isNaNPartitionValue(const Field & field, DataTypePtr type) } } +/// Integers, floats, dates and times are copied in host order. Their reader uses `insertData`, which +/// expects that layout. `DateTime64` is the exception: it is loaded with `unalignedLoadLittleEndian`. template std::vector dumpValue(T value) { @@ -349,7 +352,10 @@ std::vector dumpFieldToBytes(const Field & field, DataTypePtr type) case TypeIndex::UInt64: return dumpValue(applyVisitor(FieldVisitorConvertToNumber(), field)); case TypeIndex::DateTime64: - return dumpValue(getDateTime64ValueForIcebergBounds(field, type)); + { + const auto bytes = Iceberg::serializeDateTime64IcebergBound(getDateTime64ValueForIcebergBounds(field, type)); + return std::vector(bytes.begin(), bytes.end()); + } case TypeIndex::String: { auto value = field.safeGet(); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 5ef953b9a2086..1d86787154055 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -87,7 +87,9 @@ namespace return false; } - void addRowLineageHyperrectangles(std::unordered_map & hyperrectangles, const ProcessedManifestFileEntry & entry) + void addRowLineageHyperrectangles( + std::unordered_map> & hyperrectangles, + const ProcessedManifestFileEntry & entry) { const auto & parsed_entry = *entry.parsed_entry; if (!entry.first_row_id.has_value() || parsed_entry.record_count <= 0 || entry.sequence_number < 0) @@ -110,17 +112,17 @@ namespace { if (column_presence_is_known) { - hyperrectangles.emplace(field_id, DB::Range(inherited_lower_bound, true, inherited_upper_bound, true)); + hyperrectangles[field_id].emplace_back(inherited_lower_bound, true, inherited_upper_bound, true); continue; } } else if (auto range = getMaterializedRowLineageRange(parsed_entry, field_id)) { - hyperrectangles.emplace(field_id, *range); + hyperrectangles[field_id].emplace_back(*range); continue; } - hyperrectangles.emplace(field_id, DB::Range(UInt64(0), true, inherited_upper_bound, true)); + hyperrectangles[field_id].emplace_back(UInt64(0), true, inherited_upper_bound, true); } } } @@ -500,7 +502,7 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) if (filter_dag) { /// Compute per-column hyperrectangles for DATA files - std::unordered_map hyperrectangles; + std::unordered_map> hyperrectangles; if (parsed_entry->content_type == FileContentType::DATA) { for (const auto & [column_id, bounds] : parsed_entry->value_bounds) @@ -525,6 +527,19 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) || type_id == DB::TypeIndex::Variant) continue; + /// Optional Iceberg fields are `Nullable(DateTime64)`. The outer type is not `DateTime64`. + if (WhichDataType(removeNullable(name_and_type.type)).isDateTime64()) + { + const auto timestamp_ranges = deserializeDateTime64MinMaxRanges(left_str, right_str, name_and_type.type); + if (timestamp_ranges.empty()) + continue; + + auto & ranges = hyperrectangles[column_id]; + for (const auto & timestamp_range : timestamp_ranges) + ranges.emplace_back(timestamp_range.lower, true, timestamp_range.upper, true); + continue; + } + auto left = deserializeFieldFromBinaryRepr(left_str, name_and_type.type, true); auto right = deserializeFieldFromBinaryRepr(right_str, name_and_type.type, false); if (!left || !right) @@ -533,7 +548,7 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) if (DB::Range::less(*right, *left)) continue; - hyperrectangles.emplace(column_id, DB::Range(*left, true, *right, true)); + hyperrectangles[column_id].emplace_back(*left, true, *right, true); } addRowLineageHyperrectangles(hyperrectangles, *entry); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp index c2d400056a47b..662671c352a51 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp @@ -274,7 +274,8 @@ Field reinterpretPartitionDateTime64(const Field & field, const IDataType & type } PruningReturnStatus ManifestFilesPruner::canBePruned( - const ProcessedManifestFileEntryPtr & entry, const std::unordered_map & entry_hyperrectangles) const + const ProcessedManifestFileEntryPtr & entry, + const std::unordered_map> & entry_hyperrectangles) const { if (partition_key_condition.has_value()) { @@ -344,7 +345,21 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( if (rect_it == entry_hyperrectangles.end()) continue; - if (has_no_nulls && !key_condition.mayBeTrueInRange(1, &rect_it->second.left, &rect_it->second.right, {name_and_type->type})) + if (!has_no_nulls) + continue; + + /// A `timestamp` column in `[1e16, 1e18)` has two ranges. Prune only when every range misses. + bool may_be_true = false; + for (const auto & range : rect_it->second) + { + if (key_condition.mayBeTrueInRange(1, &range.left, &range.right, {name_and_type->type})) + { + may_be_true = true; + break; + } + } + + if (!may_be_true) { return PruningReturnStatus::MIN_MAX_INDEX_PRUNED; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h index 732edb21b58da..3a228fc6b981c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -66,7 +67,9 @@ class ManifestFilesPruner const ManifestFileIterator & manifest_file, DB::ContextPtr context); - PruningReturnStatus canBePruned(const ProcessedManifestFileEntryPtr & entry, const std::unordered_map & entry_hyperrectangles) const; + PruningReturnStatus canBePruned( + const ProcessedManifestFileEntryPtr & entry, + const std::unordered_map> & entry_hyperrectangles) const; }; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp index 069a01bed01c1..211ecaa9134bd 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include using namespace DB; @@ -29,49 +30,167 @@ Int64 decimal64Ticks(const Field & field) } -TEST(IcebergFieldParseHelpers, NearEpochNanosecondDateTime64BoundsFailOpen) +TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept) { auto type = std::make_shared(6); - /// One day after the epoch as nanoseconds: kept as µs this is ~1972, not 1970-01-02. - constexpr Int64 one_day_ns = 86'400'000'000'000LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_day_ns), type, true).has_value()); - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_day_ns), type, false).has_value()); - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(-one_day_ns), type, true).has_value()); + /// 2024-02-20 10:00:00 UTC as microseconds — spec-correct Iceberg `timestamp`. + constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL; + auto lower = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_02_20), type, true); + auto upper = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_02_20), type, false); + ASSERT_TRUE(lower.has_value()); + ASSERT_TRUE(upper.has_value()); + EXPECT_EQ(decimal64Ticks(*lower), us_2024_02_20); + EXPECT_EQ(decimal64Ticks(*upper), us_2024_02_20); + + /// 1960-01-01 00:00:00 UTC as microseconds. + constexpr Int64 us_1960_01_01 = -315'619'200'000'000LL; + auto negative = deserializeFieldFromBinaryRepr(littleEndianInt64(us_1960_01_01), type, true); + ASSERT_TRUE(negative.has_value()); + EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01); + + /// Spark `9999-12-31` sentinel is spec-correct microseconds, below `1e18`. + constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL; + auto sentinel = deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false); + ASSERT_TRUE(sentinel.has_value()); + EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us); } -TEST(IcebergFieldParseHelpers, AmbiguousScaleDateTime64BoundsFailOpen) +TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18) { auto type = std::make_shared(6); - /// 2024-01-01 UTC µs is also ~19.7 days as ns. No integer test splits those scales. - constexpr Int64 us_2024_01_01 = 1'704'067'200'000'000LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01), type, true).has_value()); - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(-us_2024_01_01), type, true).has_value()); - - /// 50 days after the epoch as ns (`4.32e15`) sits in the same band as 2024–2286 µs. - constexpr Int64 fifty_days_ns = 4'320'000'000'000'000LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(fifty_days_ns), type, true).has_value()); - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(fifty_days_ns), type, false).has_value()); - - /// Band upper edge: 1e16 ns is ~1970-04-26, 1e16 µs is ~year 2286. - constexpr Int64 one_e16 = 10'000'000'000'000'000LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, true).has_value()); + constexpr Int64 middle = 200'000'000'000'000'000LL; + const auto bytes = littleEndianInt64(middle); + auto ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, type); + ASSERT_EQ(ranges.size(), 2); + EXPECT_EQ(decimal64Ticks(ranges[0].lower), middle); + EXPECT_EQ(decimal64Ticks(ranges[0].upper), middle); + EXPECT_EQ(decimal64Ticks(ranges[1].lower), middle / 1000); + EXPECT_EQ(decimal64Ticks(ranges[1].upper), middle / 1000); + + /// Optional Iceberg `timestamp` is `Nullable(DateTime64(6))` and uses the same two ranges. + auto nullable_ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, makeNullable(type)); + ASSERT_EQ(nullable_ranges.size(), 2); + EXPECT_EQ(decimal64Ticks(nullable_ranges[0].lower), middle); + EXPECT_EQ(decimal64Ticks(nullable_ranges[1].lower), middle / 1000); + + /// `2024-02-20` is below `1e16`, so it has one microsecond range. + constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL; + const auto small = littleEndianInt64(us_2024_02_20); + auto microseconds = deserializeDateTime64MinMaxRanges(small, small, type); + ASSERT_EQ(microseconds.size(), 1); + EXPECT_EQ(decimal64Ticks(microseconds[0].lower), us_2024_02_20); + + /// A small lower bound keeps the Spark sentinel as microseconds only. + constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL; + auto sentinel = deserializeDateTime64MinMaxRanges(small, littleEndianInt64(spark_9999_us), type); + ASSERT_EQ(sentinel.size(), 1); + EXPECT_EQ(decimal64Ticks(sentinel[0].lower), us_2024_02_20); + EXPECT_EQ(decimal64Ticks(sentinel[0].upper), spark_9999_us); } -TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted) +TEST(IcebergFieldParseHelpers, BoundsBelow1e16StayMicroseconds) { auto type = std::make_shared(6); - constexpr Int64 ns_2024_01_01 = 1'704'067'200'123'456'789LL; - auto field = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2024_01_01), type, true); - ASSERT_TRUE(field.has_value()); - EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01 / 1000); + /// `86400000000000` is one day in nanoseconds (`1970-01-02`) and 1000 days in + /// microseconds (`1972-09-27`). It is below `1e16`, so Iceberg `timestamp` keeps + /// the integer and does not add a nanosecond range. The constant-column shortcut + /// may use this point: it is not the ambiguous middle band. + constexpr Int64 ticks = 86'400'000'000'000LL; + const auto bytes = littleEndianInt64(ticks); + auto lower = deserializeFieldFromBinaryRepr(bytes, type, true); + auto upper = deserializeFieldFromBinaryRepr(bytes, type, false); + ASSERT_TRUE(lower.has_value()); + ASSERT_TRUE(upper.has_value()); + EXPECT_EQ(decimal64Ticks(*lower), ticks); + EXPECT_EQ(decimal64Ticks(*upper), ticks); + + auto ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, type); + ASSERT_EQ(ranges.size(), 1); + EXPECT_EQ(decimal64Ticks(ranges[0].lower), ticks); + EXPECT_EQ(decimal64Ticks(ranges[0].upper), ticks); } -TEST(IcebergFieldParseHelpers, FarFutureMicrosecondDateTime64BoundsFailOpen) +TEST(IcebergFieldParseHelpers, BoundsThatDifferBy1000ShareOnePrecision) { auto type = std::make_shared(6); - /// Spark 9999-12-31 sentinel sits in (1e16, 1e18]. - constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false).has_value()); + /// `1e15` microseconds and `1e18` nanoseconds are the same instant. The lower bound is + /// below `1e16`, so both stay microseconds and the constant-column shortcut sees a range. + constexpr Int64 lower_us = 1'000'000'000'000'000LL; + constexpr Int64 upper_times_1000 = 1'000'000'000'000'000'000LL; + auto ranges = deserializeDateTime64MinMaxRanges(littleEndianInt64(lower_us), littleEndianInt64(upper_times_1000), type); + ASSERT_EQ(ranges.size(), 1); + EXPECT_EQ(decimal64Ticks(ranges[0].lower), lower_us); + EXPECT_EQ(decimal64Ticks(ranges[0].upper), upper_times_1000); + EXPECT_NE(decimal64Ticks(ranges[0].lower), decimal64Ticks(ranges[0].upper)); + + /// The same gap before the epoch. `|upper| < 1e16`, so the lower bound is not divided. + constexpr Int64 negative_lower = -1'000'000'000'000'000'000LL; + constexpr Int64 negative_upper = -1'000'000'000'000'000LL; + auto negative = deserializeDateTime64MinMaxRanges( + littleEndianInt64(negative_lower), littleEndianInt64(negative_upper), type); + ASSERT_EQ(negative.size(), 1); + EXPECT_EQ(decimal64Ticks(negative[0].lower), negative_lower); + EXPECT_EQ(decimal64Ticks(negative[0].upper), negative_upper); + + /// Neither bound is below `1e16`, and one is at `1e18`, so both are nanoseconds. + constexpr Int64 middle = iceberg_timestamp_microseconds_only_threshold; + auto nanoseconds = deserializeDateTime64MinMaxRanges( + littleEndianInt64(middle), littleEndianInt64(upper_times_1000), type); + ASSERT_EQ(nanoseconds.size(), 1); + EXPECT_EQ(decimal64Ticks(nanoseconds[0].lower), middle / 1000); + EXPECT_EQ(decimal64Ticks(nanoseconds[0].upper), upper_times_1000 / 1000); +} + +TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted) +{ + auto type = std::make_shared(6); + /// #2240: `2026-08-11 19:02:06.576815622` stored as nanoseconds on `timestamp`. + constexpr Int64 ns_2026_08_11 = 1'786'474'926'576'815'622LL; + auto lower = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2026_08_11), type, true); + auto upper = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2026_08_11), type, false); + ASSERT_TRUE(lower.has_value()); + ASSERT_TRUE(upper.has_value()); + /// Division truncates toward zero; the upper bound widens by one microsecond. + EXPECT_EQ(decimal64Ticks(*lower), 1'786'474'926'576'815LL); + EXPECT_EQ(decimal64Ticks(*upper), 1'786'474'926'576'816LL); + + /// `1e18` itself is nanoseconds. The pair has only the converted range. + constexpr Int64 exactly_1e18 = 1'000'000'000'000'000'000LL; + auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(exactly_1e18), type, true); + ASSERT_TRUE(edge.has_value()); + EXPECT_EQ(decimal64Ticks(*edge), exactly_1e18 / 1000); + const auto edge_bytes = littleEndianInt64(exactly_1e18); + auto edge_ranges = deserializeDateTime64MinMaxRanges(edge_bytes, edge_bytes, type); + ASSERT_EQ(edge_ranges.size(), 1); + EXPECT_EQ(decimal64Ticks(edge_ranges[0].lower), exactly_1e18 / 1000); +} + +TEST(IcebergFieldParseHelpers, DateTime64BoundsAreStoredLittleEndian) +{ + auto type = std::make_shared(6); + /// 2024-02-20 10:00:00 UTC. The written bytes must be little-endian, then read back unchanged. + constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL; + const auto bytes = serializeDateTime64IcebergBound(us_2024_02_20); + EXPECT_EQ(bytes, littleEndianInt64(us_2024_02_20)); + auto field = deserializeFieldFromBinaryRepr(bytes, type, true); + ASSERT_TRUE(field.has_value()); + EXPECT_EQ(decimal64Ticks(*field), us_2024_02_20); + + constexpr Int64 us_1960_01_01 = -315'619'200'000'000LL; + const auto negative_bytes = serializeDateTime64IcebergBound(us_1960_01_01); + EXPECT_EQ(negative_bytes, littleEndianInt64(us_1960_01_01)); + auto negative = deserializeFieldFromBinaryRepr(negative_bytes, type, false); + ASSERT_TRUE(negative.has_value()); + EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01); +} + +TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept) +{ + auto type = std::make_shared(9); + constexpr Int64 ns_2024_01_01 = 1'704'067'200'123'456'789LL; + auto field = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2024_01_01), type, true); + ASSERT_TRUE(field.has_value()); + EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01); } #endif diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py index 3922f8d37d891..1c3a942c55c56 100644 --- a/tests/integration/helpers/iceberg_timestamp_ns.py +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -164,57 +164,13 @@ def check_ns_pruning_datetime64_microseconds(node, table, column, pruned_files, ) == "3\n4\n" -def check_minmax_ns_bounds_on_timestamp_schema(node, table, column, pruned_files): - """Customer min/max over-prune: Iceberg `timestamp` with nanosecond min/max bytes. - - OTEL `time_unix_nano` tables are often declared as Iceberg `timestamp` - (`DateTime64(6)`) while 8-byte lower/upper bounds stay in nanoseconds. - ClickHouse then deserializes those ns bytes as microseconds (year 2299), - so `WHERE col <= toDateTime64('2026-...', 6)` prunes every file while the - parquet scan (microseconds) with `use_iceberg_partition_pruning = 0` is correct. - """ - type_name = node.query( - f"SELECT toTypeName({column}) FROM {table} LIMIT 1", - settings=PRUNE_OFF, - ) - assert "DateTime64(6" in type_name, type_name - assert "DateTime64(9" not in type_name, type_name - - ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF) - assert ids == "1\n2\n3\n4\n" - # Parquet is timestamp[us] (correct 2024 wall times). Iceberg lower/upper bounds - # are still the original nanosecond int64 bytes, so DateTime64(6) min/max - # interprets them as year 2299 and over-prunes. - micros = node.query( - f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1", - settings=PRUNE_OFF, - ).strip() - assert micros == "1704067200123456", micros - - # Same predicate shape as the customer query: scale-6 literal near a file's - # nanosecond upper bound. Rows 1 and 2 must survive; pruning must not drop them. - le_after_row2 = datetime64_literal( - "2024-03-15 12:00:00.000001000", scale=6, timezone=None - ) - query = f"SELECT id FROM {table} WHERE {column} <= {le_after_row2} ORDER BY id" - # One data file covering all four timestamps; the predicate overlaps it, so - # min/max pruning must keep the file. Over-pruning currently returns 0 rows. - assert pruned_files(query) == 0 - assert node.query(query, settings=PRUNE_ON) == "1\n2\n" - - count_query = f"SELECT count() FROM {table} WHERE {column} <= {le_after_row2}" - assert pruned_files(count_query) == 0 - assert node.query(count_query, settings=PRUNE_ON) == "2\n" - - def _utc_datetime_to_micros(dt): """Integer microseconds from Unix epoch. Do not use float `timestamp` * 1e6.""" delta = dt - datetime(1970, 1, 1, tzinfo=timezone.utc) return delta.days * 86_400_000_000 + delta.seconds * 1_000_000 + delta.microseconds -# Spec-correct Iceberg `timestamp` microseconds that sit in the ambiguous band -# (1e16, 1e18]: converting them as nanoseconds over-prunes. +# Spec-correct Iceberg `timestamp` microseconds, including the Spark `9999-12-31` sentinel. SPARK_TIMESTAMP_SENTINEL_US = _utc_datetime_to_micros( datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=timezone.utc) ) @@ -223,79 +179,12 @@ def _utc_datetime_to_micros(dt): ) -# One day after the Unix epoch as nanoseconds. Kept as microseconds this is ~1972, -# so a predicate on 1970-01-02 over-prunes unless the decoder fails open. -NEAR_EPOCH_NS_ROWS = [ - (86_400_000_000_000, 1, "1970-01-02 00:00:00.000000000"), -] - -# 50 days after the epoch as nanoseconds (`4.32e15`). Same integer as ~2106 µs, -# so a predicate on 1970-02-20 over-prunes unless the decoder fails open. -OVERLAP_BAND_NS_ROWS = [ - (4_320_000_000_000_000, 1, "1970-02-20 00:00:00.000000000"), -] - - -def check_minmax_near_epoch_ns_bounds_on_timestamp_schema(node, table, column, pruned_files): - """Iceberg `timestamp` whose min/max bytes are near-epoch nanoseconds must not over-prune. - - One day after the epoch is `8.64e13` ns (`|ticks| <= 1e16`). The inverted-range - check does not fire when both bounds stay in that band. - """ - type_name = node.query( - f"SELECT toTypeName({column}) FROM {table} LIMIT 1", - settings=PRUNE_OFF, - ) - assert "DateTime64(6" in type_name, type_name - assert "DateTime64(9" not in type_name, type_name - - ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF) - assert ids == "1\n" - micros = node.query( - f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1", - settings=PRUNE_OFF, - ).strip() - assert micros == "86400000000", micros - - eq_epoch_day = datetime64_literal("1970-01-02 00:00:00.000000000", scale=6, timezone=None) - query = f"SELECT id FROM {table} WHERE {column} = {eq_epoch_day} ORDER BY id" - assert pruned_files(query) == 0 - assert node.query(query, settings=PRUNE_ON) == "1\n" - - -def check_minmax_overlap_band_ns_bounds_on_timestamp_schema(node, table, column, pruned_files): - """Iceberg `timestamp` whose min/max bytes are leftover-band nanoseconds must not over-prune. - - 50 days after the epoch is `4.32e15` ns, which is also a 2024–2286 microsecond - timestamp. The decoder cannot tell the scales apart and must fail open. - """ - type_name = node.query( - f"SELECT toTypeName({column}) FROM {table} LIMIT 1", - settings=PRUNE_OFF, - ) - assert "DateTime64(6" in type_name, type_name - assert "DateTime64(9" not in type_name, type_name - - ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF) - assert ids == "1\n" - micros = node.query( - f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1", - settings=PRUNE_OFF, - ).strip() - assert micros == "4320000000000", micros - - eq_day = datetime64_literal("1970-02-20 00:00:00.000000000", scale=6, timezone=None) - query = f"SELECT id FROM {table} WHERE {column} = {eq_day} ORDER BY id" - assert pruned_files(query) == 0 - assert node.query(query, settings=PRUNE_ON) == "1\n" - - def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files): - """Iceberg `timestamp` with a far-future microsecond *upper* bound must not over-prune. + """Iceberg `timestamp` with a far-future microsecond upper bound must keep matching rows. - Lower bound stays a real 2024 microsecond value. If the decoder treats the - upper bound as nanoseconds, max becomes ~1978, the [min, max] range inverts, - and `WHERE col <= toDateTime64('2024-03-15 ...', 6)` drops the file. + Lower bound stays a real 2024 microsecond value. The upper bound is the Spark + `9999-12-31` sentinel or the `DateTime64` maximum. A predicate on 2024 must + still return the matching rows. """ type_name = node.query( f"SELECT toTypeName({column}) FROM {table} LIMIT 1", @@ -320,117 +209,6 @@ def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files): assert node.query(query, settings=PRUNE_ON) == "1\n2\n" -def _scale_bound_bytes(raw, factor=1000): - if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8: - return raw - value = struct.unpack(" 0, f"no timestamp min/max bounds scaled under {table_location}" - - def _set_bound_bytes(raw, value): if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8: return raw diff --git a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py index 5a78ef2a69005..c7a9beb9a6b5b 100644 --- a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py @@ -24,20 +24,14 @@ DATETIME64_MAX_US, FILTER_LITERAL_IDS, FILTER_LITERALS, - NEAR_EPOCH_NS_ROWS, - OVERLAP_BAND_NS_ROWS, PRUNE_OFF, PRUNE_ON, ROWS, SPARK_TIMESTAMP_SENTINEL_US, check_minmax_far_future_us_upper_bound, - check_minmax_near_epoch_ns_bounds_on_timestamp_schema, - check_minmax_ns_bounds_on_timestamp_schema, - check_minmax_overlap_band_ns_bounds_on_timestamp_schema, check_ns_pruning, check_ns_pruning_datetime64_microseconds, patch_iceberg_table_long_to_timestamp_ns, - patch_iceberg_timestamp_bounds_us_to_ns, patch_iceberg_timestamp_upper_bounds, ) from helpers.iceberg_utils import check_validity_and_get_prunned_files_general @@ -123,13 +117,6 @@ def _create_timestamp_table(started_cluster, table_name, patch, rows=ROWS): shutil.rmtree(work, ignore_errors=True) -def _create_timestamp_table_with_ns_bounds(started_cluster, table_name, rows=ROWS): - """Iceberg v2 `timestamp` (microseconds) whose manifest min/max bytes are nanoseconds.""" - return _create_timestamp_table( - started_cluster, table_name, patch_iceberg_timestamp_bounds_us_to_ns, rows=rows - ) - - def _create_timestamp_table_with_upper_bound_us(started_cluster, table_name, micros): """Iceberg v2 `timestamp` whose upper bounds are spec-correct far-future microseconds.""" def patch(table_location, old_prefix, new_prefix): @@ -244,73 +231,6 @@ def pruned_files(select_expression): check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone) -def test_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark): - """Iceberg `timestamp` (`DateTime64(6)`) with nanosecond min/max bytes must not over-prune. - - Customer: `WHERE time_unix_nano <= toDateTime64('2026-08-11 20:02:07.897356', 6)` - returns 0 with min/max pruning and the correct row count without it. - """ - node = started_cluster_iceberg_no_spark.instances["node1"] - s3_filename = _create_timestamp_table_with_ns_bounds( - started_cluster_iceberg_no_spark, "minmax_ns_on_ts" - ) - table = _iceberg_table(s3_filename) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, - "minmax_ns_on_ts", - PRUNE_OFF, - PRUNE_ON, - "IcebergMinMaxIndexPrunedFiles", - select_expression, - ) - - check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files) - - -def test_minmax_pruning_near_epoch_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark): - """Near-epoch ns min/max on Iceberg `timestamp` must not be decoded as microseconds.""" - node = started_cluster_iceberg_no_spark.instances["node1"] - s3_filename = _create_timestamp_table_with_ns_bounds( - started_cluster_iceberg_no_spark, "minmax_near_epoch_ns", rows=NEAR_EPOCH_NS_ROWS - ) - table = _iceberg_table(s3_filename) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, - "minmax_near_epoch_ns", - PRUNE_OFF, - PRUNE_ON, - "IcebergMinMaxIndexPrunedFiles", - select_expression, - ) - - check_minmax_near_epoch_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files) - - -def test_minmax_pruning_overlap_band_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark): - """Ns min/max in `[1.704e15, 1e16]` on Iceberg `timestamp` must not be decoded as microseconds.""" - node = started_cluster_iceberg_no_spark.instances["node1"] - s3_filename = _create_timestamp_table_with_ns_bounds( - started_cluster_iceberg_no_spark, "minmax_overlap_band_ns", rows=OVERLAP_BAND_NS_ROWS - ) - table = _iceberg_table(s3_filename) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, - "minmax_overlap_band_ns", - PRUNE_OFF, - PRUNE_ON, - "IcebergMinMaxIndexPrunedFiles", - select_expression, - ) - - check_minmax_overlap_band_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files) - - @pytest.mark.parametrize( "upper_bound_us", [SPARK_TIMESTAMP_SENTINEL_US, DATETIME64_MAX_US], diff --git a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py index c62e6029b8513..5265d08769e1c 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py @@ -81,6 +81,15 @@ def execute_spark_query(query: str): )) creation_expression = TABLE_NAME + spark_timestamps = spark.sql( + f"SELECT unix_micros(ts), unix_micros(time_struct.b) FROM {TABLE_NAME} ORDER BY tag" + ).collect() + expected_timestamps = "".join(f"{row[0]}\t{row[1]}\n" for row in spark_timestamps) + assert instance.query( + f"SELECT toUnixTimestamp64Micro(ts), toUnixTimestamp64Micro(time_struct.b) " + f"FROM {creation_expression} ORDER BY tag" + ) == expected_timestamps + def check_validity_and_get_prunned_files(select_expression): settings1 = { "use_iceberg_partition_pruning": 0, diff --git a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py index 229d0b4b239b8..387e1272831bd 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py @@ -18,11 +18,9 @@ PRUNE_OFF, PRUNE_ON, ROWS, - check_minmax_ns_bounds_on_timestamp_schema, check_ns_pruning, check_ns_pruning_datetime64_microseconds, patch_iceberg_table_long_to_timestamp_ns, - patch_iceberg_timestamp_bounds_us_to_ns, ) from helpers.iceberg_utils import ( check_validity_and_get_prunned_files_general, @@ -72,36 +70,6 @@ def _create_spark_ns_table( ) -def _create_spark_timestamp_table_with_ns_bounds(spark, started_cluster, table_name): - """Spark Iceberg `TIMESTAMP` (microseconds) whose manifest min/max bytes are nanoseconds.""" - spark.sql( - f""" - CREATE TABLE {table_name} ( - ts TIMESTAMP, - value TIMESTAMP, - id INT - ) - USING iceberg - OPTIONS('format-version'='2') - """ - ) - values = ", ".join( - f"(TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', " - f"TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', {row_id})" - for _ts_ns, row_id, utc in ROWS - ) - spark.sql(f"INSERT INTO {table_name} VALUES {values}") - - table_location = Path(f"{WAREHOUSE_PREFIX}/{table_name}") - patch_iceberg_timestamp_bounds_us_to_ns(table_location) - default_upload_directory( - started_cluster, - "s3", - f"/iceberg_data/default/{table_name}/", - f"/iceberg_data/default/{table_name}/", - ) - - @pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) def test_spark_partition_pruning_datetime64_nanoseconds(started_cluster_iceberg_with_spark, make_literal): """Identity partition pruning on Spark-written Iceberg timestamp_ns must not drop matching rows.""" @@ -159,23 +127,3 @@ def pruned_files(select_expression): ) check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone) - - -def test_spark_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_with_spark): - """Nanosecond parquet/bounds with Iceberg `timestamp` (DateTime64(6)) must not over-prune.""" - node = started_cluster_iceberg_with_spark.instances["node1"] - spark = started_cluster_iceberg_with_spark.spark_session - table_name = "test_spark_minmax_ns_on_ts_" + get_uuid_str() - _create_spark_timestamp_table_with_ns_bounds( - spark, started_cluster_iceberg_with_spark, table_name - ) - table = get_creation_expression( - "s3", table_name, started_cluster_iceberg_with_spark, table_function=True - ) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, table_name, PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", select_expression - ) - - check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files) diff --git a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py index 77891e65fb449..c8de48a059184 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py @@ -144,14 +144,103 @@ def execute_spark_query(query: str): instance.query(f"CREATE TABLE {TABLE2_NAME} (tag INT, number2 INT) ENGINE=Memory") instance.query(f"INSERT INTO {TABLE2_NAME} VALUES (1, 2)") + # `convertToRemote` picks one replica of `cluster_simple`. The Memory table exists only on + # node1, so a single run can pass by landing there. Repeat until every replica is likely. + for _ in range(9): + res = instance.query(f""" + SELECT t1.tag, t1.number, t2.number2 + FROM {TABLE_NAME} AS t1 + JOIN {TABLE2_NAME} AS t2 USING (tag) + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_simple', + object_storage_cluster_join_mode='global', + allow_experimental_analyzer=0 + """) + assert res == "1\t1\t2\n" + + # The alias `t2` is already a temporary table with a different row. The join must use the Memory row. + # One client session, so the temporary table is visible to the SELECT. + # The integration image has `/usr/bin/clickhouse` and no `clickhouse-client` on `PATH`. + res = instance.exec_in_container( + [ + "/usr/bin/clickhouse", + "client", + "--multiquery", + "--query", + f""" + CREATE TEMPORARY TABLE t2 (tag INT, number2 INT); + INSERT INTO t2 VALUES (1, 9); + SELECT t1.tag, t1.number, t2.number2 + FROM {TABLE_NAME} AS t1 + JOIN {TABLE2_NAME} AS t2 USING (tag) + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_simple', + object_storage_cluster_join_mode='global', + allow_experimental_analyzer=0 + """, + ] + ) + assert res == "1\t1\t2\n" + res = instance.query(f""" - SELECT t1.tag, t1.number, t2.number2 + SELECT t1.tag, t1.number FROM {TABLE_NAME} AS t1 - JOIN {TABLE2_NAME} AS t2 USING (tag) + JOIN numbers(2) AS t2 ON t1.tag = toInt32(t2.number + 1) SETTINGS object_storage_remote_initiator=1, object_storage_remote_initiator_cluster='cluster_simple', object_storage_cluster_join_mode='global', allow_experimental_analyzer=0 """) - assert res == "1\t1\t2\n" + assert res == "1\t1\n" + + res = instance.query(f""" + SELECT tag, number + FROM {TABLE_NAME} + WHERE tag IN (SELECT tag FROM {TABLE2_NAME}) + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_simple', + object_storage_cluster_join_mode='global', + allow_experimental_analyzer=0 + """) + assert res == "1\t1\n" + + # `IN (SELECT 1)` has no table, so a replica must still see that subquery. + # The sent text is `IN ((SELECT 1) AS _subqueryN)`, not `GLOBAL IN (_subqueryN)`. + # Choosing node1 runs in-process: the client row keeps the original text and the same + # `query_id`, so it does not count. A remote row has a different `query_id`. + saw_remote_subquery = False + for _ in range(9): + query_id = uuid.uuid4().hex + res = instance.query( + f""" + SELECT tag, number + FROM {TABLE_NAME} + WHERE number IN (SELECT 1) + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_simple', + object_storage_cluster_join_mode='global', + allow_experimental_analyzer=0 + """, + query_id=query_id, + ) + assert res == "1\t1\n" + for node in started_cluster_iceberg_with_spark.instances.values(): + node.query("SYSTEM FLUSH LOGS") + sent_with_subquery = instance.query(f""" + SELECT count() + FROM clusterAllReplicas('cluster_simple', system.query_log) + WHERE initial_query_id = '{query_id}' + AND type = 'QueryFinish' + AND query_id != initial_query_id + AND position(query, '(SELECT 1)') > 0 + AND position(query, 'GLOBAL IN') = 0 + """) + if sent_with_subquery != "0\n": + saw_remote_subquery = True + break + assert saw_remote_subquery