github-actions[bot] commented on code in PR #67768:
URL: https://github.com/apache/doris/pull/67768#discussion_r3979478892
##########
be/src/exec/operator/hashjoin_build_sink.cpp:
##########
@@ -209,11 +210,18 @@ size_t
HashJoinBuildSinkLocalState::get_reserve_mem_size(RuntimeState* state, bo
null_map_val = ColumnUInt8::create();
null_map_val->get_data().assign(build_block_rows, (uint8_t)0);
- // Get the key column that needs to be built
- Status st = _extract_join_column(block, null_map_val, raw_ptrs,
_build_col_ids);
+ // Get the key column that needs to be built. The normalized float
key copies made
+ // here only estimate their own size; process_build_block() makes
the ones the hash
+ // table points into.
+ std::vector<ColumnPtr> normalized_key_columns;
+ Status st = _extract_join_column(block, null_map_val, raw_ptrs,
_build_col_ids,
Review Comment:
[P1] Keep reservation estimation allocation- and side-effect-free. This runs
before PipelineTask can revoke/spill or reserve memory, but
`_extract_join_column()` now COW-copies every float-bearing build key merely to
measure it. For a non-nullable key that must serialize nulls, it also appends
the wrapper to the member `_key_columns_holder`; a denied reservation retries
this path and retains another full copy each time. A large mixed-nullability
`<=>` join can therefore consume the memory it is trying to reserve (and grow
on every retry) before spill/admission can act. Please estimate the copy from
the original columns without materializing it, keep estimator holders local,
and cover repeated reserve retries.
##########
be/src/exec/common/hash_table/hash_key_normalize.h:
##########
@@ -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 "core/column/column.h"
+#include "core/data_type/data_type.h"
+
+namespace doris {
+
+/// Whether `type` is FLOAT/DOUBLE or a nested type (array/map/struct) holding
one.
+/// Nullable wrappers are looked through.
+bool contains_float_or_double(const DataTypePtr& type);
+
+/// Hash tables, fixed/serialized keys and the shuffle partitioners compare
floating point
+/// keys by their raw bits, while Doris equality treats -0.0 == +0.0 and all
NaN payloads as
+/// one value. Collapse such values (see NormalizeFloat) so equal keys always
hash and compare
+/// equal. Columns without a floating point leaf are left untouched and never
copied; a float
+/// column is normalized in place when `column` is its only owner and replaced
by a normalized
+/// copy otherwise. Operators pass a second reference to the block's column,
so the block keeps
+/// the stored row values and only the copy they hash is canonical.
+void normalize_float_hash_key(ColumnPtr& column, const DataTypePtr& type);
Review Comment:
[P1] Normalize recursive `UNION DISTINCT` keys too.
`RecCTESharedState::emplace_block()` is another `DistinctDataVariants`
consumer, but it feeds raw result columns directly into the hash method. A
single DOUBLE uses the UInt64 one-number key, so an anchor `-0.0` and recursive
`+0.0` are admitted as two rows even though Doris equality treats them as one;
alternate NaN payloads and composite float rows have the same gap. Please hash
detached normalized copies there while preserving the original output block,
and add a recursive `UNION DISTINCT` regression.
##########
be/src/exec/operator/partition_sort_sink_operator.cpp:
##########
@@ -189,6 +190,10 @@ Status
PartitionSortSinkOperatorX::_split_block_by_partition(
Columns key_columns(_partition_exprs_num);
for (int i = 0; i < _partition_exprs_num; ++i) {
RETURN_IF_ERROR(_partition_expr_ctxs[i]->execute(input_block,
key_columns[i]));
+ // Same rule as aggregation / hash join keys: -0.0 == +0.0 and NaN
payloads must hash
+ // alike. `key_columns[i]` shares the block column, so a float key is
copied and the
+ // partitioned rows keep their stored values.
+ normalize_float_hash_key(key_columns[i],
_partition_expr_ctxs[i]->root()->data_type());
Review Comment:
[P2] Include the normalized key copy in this sink's reservation.
`normalize_float_hash_key()` COW-copies every float-bearing expression result
here (recursively cloning nested children), but `get_reserve_mem_size()` still
reserves only hash-table growth plus the hash-value array. Under workload-group
pressure a wide/nested Partition TopN key can therefore pass admission and
allocate materially beyond the approved amount. Please make the block-aware
estimate include this transient copy, or normalize while hashing without
materializing it.
##########
be/test/exec/partitioner/partitioner_float_key_test.cpp:
##########
@@ -0,0 +1,191 @@
+// 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 <gtest/gtest.h>
+
+#include <bit>
+#include <cmath>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <type_traits>
+#include <vector>
+
+#include "agent/be_exec_version_manager.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_array.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/field.h"
+#include "exec/partitioner/partitioner.h"
+#include "testutil/column_helper.h"
+#include "testutil/mock/mock_runtime_state.h"
+#include "testutil/mock/mock_slot_ref.h"
+#include "util/hash_util.hpp"
+
+namespace doris {
+
+namespace {
+
+constexpr uint32_t kPartitions = 1U << 20;
+
+const double kQuietNaN = std::numeric_limits<double>::quiet_NaN();
+const double kPayloadNaN =
std::bit_cast<double>(std::bit_cast<uint64_t>(kQuietNaN) | 0x1234ULL);
+
+// Legacy zlib CRC32 convention of Crc32HashPartitioner<ShuffleChannelIds>:
hash of the raw
+// bytes seeded with 0, then modulo partition count.
+uint32_t legacy_channel(double value) {
+ return HashUtil::zlib_crc32_fixed(value, 0) % kPartitions;
+}
+
+std::vector<uint32_t> partition(Crc32HashPartitioner<ShuffleChannelIds>&
partitioner,
+ RuntimeState* state, Block& block) {
+ EXPECT_TRUE(partitioner.do_partitioning(state, &block).ok());
+ return partitioner.get_channel_ids();
+}
+
+} // namespace
+
+// From NORMALIZE_FLOAT_HASH_KEY_VERSION on, -0.0 / +0.0 and every NaN reach
the same channel;
+// with an older query version the legacy raw-bit convention is kept so that
senders of mixed
+// BE versions agree during a rolling upgrade. The shuffled rows themselves
are never rewritten.
+TEST(PartitionerFloatKeyTest, DoubleKeyIsNormalizedOnlyFromTheNewVersion) {
+ Crc32HashPartitioner<ShuffleChannelIds> partitioner(kPartitions);
+ partitioner._partition_expr_ctxs =
+ MockSlotRef::create_mock_contexts(DataTypes
{std::make_shared<DataTypeFloat64>()});
+ MockRuntimeState state;
+
+ Block block = ColumnHelper::create_block<DataTypeFloat64>(
+ {-0.0, 0.0, kPayloadNaN, kQuietNaN, 1.5, -1.5});
+
+ state.set_be_exec_version(NORMALIZE_FLOAT_HASH_KEY_VERSION);
Review Comment:
[P1] Set the Thrift presence bit before exercising the partitioner.
`RuntimeState::set_be_exec_version()` currently assigns only the numeric field,
while the newly reached `be_exec_version()` accessor DCHECKs
`__isset.be_exec_version`; ASAN_UT keeps DCHECKs enabled, so these new tests
abort before their channel assertions. Please make the setter (and the test
`TQueryOptions` builder) use `__set_be_exec_version()`, or initialize the test
options through an API that sets both value and presence, and cover the
absent-option compatibility behavior explicitly.
##########
be/src/exec/pipeline/dependency.h:
##########
@@ -645,6 +645,10 @@ struct HashJoinSharedState : public JoinSharedState {
const std::vector<TupleDescriptor*> build_side_child_desc;
size_t build_exprs_size = 0;
std::shared_ptr<Block> build_block;
+ // Normalized copies of the float build keys (-0.0 -> +0.0, NaN -> quiet
NaN). The hash
+ // table keeps raw pointers into them for the whole probe phase, while
`build_block` still
+ // holds the stored values for the join output.
+ std::vector<ColumnPtr> normalized_build_key_columns;
Review Comment:
[P2] Avoid retaining normalized inputs after owning hash methods have
persisted their keys. Fixed and serialized contexts copy the canonical keys
into `build_stored_keys`/`build_arena`, yet this shared vector keeps the
recursively cloned input columns alive through the entire probe phase.
Composite float keys therefore retain a second full representation in addition
to the original build block and packed hash keys, which can materially raise
peak join memory. Please retain these columns only for methods that actually
alias input storage (such as one-number), or transfer ownership into the hash
context and release redundant copies after build.
##########
be/src/exec/operator/hashjoin_probe_operator.cpp:
##########
@@ -356,7 +357,14 @@ Status
HashJoinProbeLocalState::_extract_join_column(Block& block,
auto& shared_state = *_shared_state;
for (size_t i = 0; i < shared_state.build_exprs_size; ++i) {
- const auto& column_ptr = block.get_by_position(res_col_ids[i]).column;
+ ColumnPtr column_ptr = block.get_by_position(res_col_ids[i]).column;
+ // Probe keys are normalized the same way as the build keys (see
+ // HashJoinBuildSinkLocalState::_extract_join_column): a float key is
hashed from a copy
+ // kept in `_key_columns_holder` for this batch and the probe block
stays untouched.
+ normalize_float_hash_key(column_ptr,
_probe_expr_ctxs[i]->root()->data_type());
Review Comment:
[P2] Account for the holder-owned normalized copy in probe reservation and
reporting. `push()` learns only the input block's allocation delta and reports
only `_probe_block.allocated_bytes()`, but a direct-slot clone created here
lives solely in `_key_columns_holder`; it contributes zero to both measures and
remains live for the whole probe batch. Wide nested float keys can therefore
allocate several MiB beyond the learned/admitted amount on every batch while
`MemoryUsage` omits it. Please add the uniquely owned holder bytes to the
adaptive estimate/counter (without double-counting nullable wrappers), or
measure the probe operation with the standard peak-memory scope, and add a
multi-batch wide-key test.
##########
be/src/exec/partitioner/partitioner.cpp:
##########
@@ -39,12 +41,25 @@ Status
Crc32HashPartitioner<ChannelIds>::do_partitioning(RuntimeState* state, Bl
_initialize_hash_vals(rows);
auto* __restrict hashes = _hash_vals.data();
RETURN_IF_ERROR(_get_partition_column_result(block, result));
+ // Equal float keys (-0.0 == +0.0, any NaN) must reach the same
channel. Every BE of a
+ // query hashes with the same be_exec_version, so mixed-version
senders during a rolling
+ // upgrade keep the legacy raw-bit convention until FE raises the
version.
+ const bool normalize_float_keys =
+ state->be_exec_version() >= NORMALIZE_FLOAT_HASH_KEY_VERSION;
for (int j = 0; j < result_size; ++j) {
const auto& [col, is_const] =
unpack_if_const(block->get_by_position(result[j]).column);
if (is_const) {
continue;
}
- _do_hash(col, hashes, j);
+ if (normalize_float_keys) {
+ // The block keeps its own reference, so a float column is
hashed from a
+ // normalized copy and the rows sent downstream are unchanged.
+ ColumnPtr key = col;
+ normalize_float_hash_key(key,
_partition_expr_ctxs[j]->root()->data_type());
Review Comment:
[P2] Reserve the normalization copy before running the common partitioner.
At version 15 this call COW-detaches every float-bearing expression result, and
recursive columns clone offsets and all struct/map siblings because the input
block still owns the source. Exchange and local-exchange sinks invoke this
after admission but do not override `get_reserve_mem_size()`, so a wide
`STRUCT<STRING, DOUBLE>` key can allocate several MiB beyond the default
minimum; the spill build/probe callers likewise reserve only their inner/I/O
baselines. Under a tight workload-group or query limit, a batch can be admitted
and then fail here. Please expose a no-copy estimate from the partitioner and
include it in every caller's block-aware reserve, or normalize while hashing
without materializing the full composite, with a wide nested-key reservation
test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
