Copilot commented on code in PR #48720:
URL: https://github.com/apache/doris/pull/48720#discussion_r2719447585
##########
be/src/vec/sink/tablet_sink_hash_partitioner.h:
##########
@@ -46,10 +45,15 @@ class TabletSinkHashPartitioner final : public
PartitionerBase {
Status open(RuntimeState* state) override;
- Status do_partitioning(RuntimeState* state, Block* block, bool eos,
- bool* already_sent) const override;
+ Status do_partitioning(RuntimeState* state, Block* block) const override;
+ // block to create new partition by RPC. return batched data to create.
+ Status try_cut_in_line(Block& prior_block) const;
+ void finish_cut_in_line() const { _row_distribution._deal_batched = false;
}
+ void mark_last_block() const { _row_distribution._deal_batched = true; }
+
+ const std::vector<HashValType>& get_channel_ids() const override { return
_hash_vals; }
+ const std::vector<bool>& get_skipped(int size) const { return _skipped; }
Review Comment:
`get_skipped` takes a `size` parameter that is never used, which will
trigger an unused-parameter warning and makes the API misleading about its
intent. Consider either removing the parameter, marking it as unused (e.g., by
omitting the name or using an attribute), or using it to validate that the
caller-provided size matches `_skipped.size()`.
```suggestion
const std::vector<bool>& get_skipped(int /*size*/) const { return
_skipped; }
```
##########
be/src/vec/sink/tablet_sink_hash_partitioner.cpp:
##########
@@ -81,45 +86,67 @@ Status TabletSinkHashPartitioner::open(RuntimeState* state)
{
return Status::OK();
}
-Status TabletSinkHashPartitioner::do_partitioning(RuntimeState* state, Block*
block, bool eos,
- bool* already_sent) const {
+Status TabletSinkHashPartitioner::do_partitioning(RuntimeState* state, Block*
block) const {
_hash_vals.resize(block->rows());
if (block->empty()) {
return Status::OK();
}
- std::fill(_hash_vals.begin(), _hash_vals.end(), -1);
- int64_t filtered_rows = 0;
- int64_t number_input_rows = _local_state->rows_input_counter()->value();
+
+ // tablet_id_hash % invalid_val never get invalid_val, so we use
invalid_val as sentinel value
+ const auto& invalid_val = _partition_count;
+ std::ranges::fill(_hash_vals, invalid_val);
+
+ int64_t dummy_stats = 0; // _local_state->rows_input_counter() updated in
sink and write.
std::shared_ptr<vectorized::Block> convert_block =
std::make_shared<vectorized::Block>();
+
RETURN_IF_ERROR(_row_distribution.generate_rows_distribution(
- *block, convert_block, filtered_rows, _row_part_tablet_ids,
number_input_rows));
- if (_row_distribution.batching_rows() > 0) {
- SCOPED_TIMER(_local_state->send_new_partition_timer());
- RETURN_IF_ERROR(_send_new_partition_batch(state, block, eos));
- *already_sent = true;
- } else {
- const auto& row_ids = _row_part_tablet_ids[0].row_ids;
- const auto& tablet_ids = _row_part_tablet_ids[0].tablet_ids;
- for (int idx = 0; idx < row_ids.size(); ++idx) {
- const auto& row = row_ids[idx];
- const auto& tablet_id_hash =
- HashUtil::zlib_crc_hash(&tablet_ids[idx],
sizeof(HashValType), 0);
- _hash_vals[row] = tablet_id_hash % _partition_count;
+ *block, convert_block, _row_part_tablet_ids, dummy_stats));
+ _skipped = _row_distribution.get_skipped();
+ const auto& row_ids = _row_part_tablet_ids[0].row_ids;
+ const auto& tablet_ids = _row_part_tablet_ids[0].tablet_ids;
+
+ for (int idx = 0; idx < row_ids.size(); ++idx) {
+ const auto& row = row_ids[idx];
+ const auto& tablet_id_hash =
+ HashUtil::zlib_crc_hash(&tablet_ids[idx], sizeof(HashValType),
0);
+ _hash_vals[row] = tablet_id_hash % invalid_val;
+ }
+
+ // _hash_val == -1 = (_skipped = 1 or filtered = 1)
Review Comment:
This comment still refers to `_hash_val == -1` as the sentinel for
skipped/filtered rows, but the implementation now uses `invalid_val` (i.e.,
`_partition_count`) as the sentinel value. To avoid confusion for future
readers, please update the comment to reflect the current logic (e.g., that
`_hash_vals[i] == invalid_val` corresponds to skipped or filtered rows) or
remove it if it’s no longer accurate.
```suggestion
// _hash_vals[i] == invalid_val => row i is skipped or filtered
```
##########
be/test/pipeline/shuffle/exchange_writer_test.cpp:
##########
@@ -0,0 +1,233 @@
+// 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 "pipeline/shuffle/exchange_writer.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <vector>
+
+#include "gen_cpp/Types_types.h"
+#include "pipeline/exec/exchange_sink_operator.h"
+#include "testutil/column_helper.h"
+#include "testutil/mock/mock_data_stream_sender.h"
+#include "testutil/mock/mock_runtime_state.h"
+
+using doris::MockRuntimeState;
+using doris::Status;
+using doris::vectorized::Block;
+using doris::vectorized::ColumnHelper;
+using doris::vectorized::DataTypeInt32;
+using doris::vectorized::Channel;
+using doris::vectorized::MockChannel;
+using doris::pipeline::ExchangeSinkLocalState;
+
+namespace doris::pipeline {
+
+// Helper: create channels that will never actually send rows (is_receiver_eof
== true),
+// so writer logic can be tested without exercising Channel::add_rows /
BlockSerializer.
+static std::shared_ptr<Channel> make_disabled_channel(ExchangeSinkLocalState*
local_state) {
+ TUniqueId id;
+ id.hi = 0;
+ id.lo = 0;
+ auto ch = std::make_shared<MockChannel>(local_state, id,
/*is_local=*/true);
+ ch->set_receiver_eof(Status::EndOfFile("test eof"));
+ return ch;
+}
+
+static std::vector<std::shared_ptr<Channel>> make_disabled_channels(
+ ExchangeSinkLocalState* local_state, size_t n) {
+ std::vector<std::shared_ptr<Channel>> channels;
+ channels.reserve(n);
+ for (size_t i = 0; i < n; ++i) {
+ channels.emplace_back(make_disabled_channel(local_state));
+ }
+ return channels;
+}
+
+TEST(TrivialExchangeWriterTest, BasicDistribution) {
+ MockRuntimeState state;
+ ExchangeSinkLocalState local_state(&state);
+ ExchangeTrivialWriter writer;
+
+ const size_t channel_count = 2;
+ auto channels = make_disabled_channels(&local_state, channel_count);
+
+ // rows: [1,2,3,4,5], channel_ids: [0,1,0,1,1]
+ Block block = ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4, 5});
+ std::vector<vectorized::PartitionerBase::HashValType> channel_ids = {0, 1,
0, 1, 1};
+ const size_t rows = channel_ids.size();
+
+ Status st = writer._channel_add_rows(&state, channels, channel_count,
channel_ids, rows, &block,
+ /*eos=*/false);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ // Expect histogram: [2,3]
+ ASSERT_EQ(writer._channel_rows_histogram.size(), channel_count);
+ EXPECT_EQ(writer._channel_rows_histogram[0], 2U);
+ EXPECT_EQ(writer._channel_rows_histogram[1], 3U);
+
+ // Expect row index order: [0,2,1,3,4]
+ ASSERT_EQ(writer._origin_row_idx.size(), rows);
+ std::vector<uint32_t> got(rows);
+ for (size_t i = 0; i < rows; ++i) {
+ got[i] = writer._origin_row_idx[i];
+ }
+ std::vector<uint32_t> expected {0, 2, 1, 3, 4};
+ EXPECT_EQ(got, expected);
+}
+
+TEST(TrivialExchangeWriterTest, AllRowsToSingleChannel) {
+ MockRuntimeState state;
+ ExchangeSinkLocalState local_state(&state);
+ ExchangeTrivialWriter writer;
+
+ const size_t channel_count = 3;
+ auto channels = make_disabled_channels(&local_state, channel_count);
+
+ Block block = ColumnHelper::create_block<DataTypeInt32>({10, 20, 30, 40});
+ std::vector<vectorized::PartitionerBase::HashValType> channel_ids = {2, 2,
2, 2};
+ const size_t rows = channel_ids.size();
+
+ Status st = writer._channel_add_rows(&state, channels, channel_count,
channel_ids, rows, &block,
+ /*eos=*/false);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ ASSERT_EQ(writer._channel_rows_histogram.size(), channel_count);
+ EXPECT_EQ(writer._channel_rows_histogram[0], 0U);
+ EXPECT_EQ(writer._channel_rows_histogram[1], 0U);
+ EXPECT_EQ(writer._channel_rows_histogram[2], 4U);
+
+ ASSERT_EQ(writer._origin_row_idx.size(), rows);
+ std::vector<uint32_t> got(rows);
+ for (size_t i = 0; i < rows; ++i) {
+ got[i] = writer._origin_row_idx[i];
+ }
+ std::vector<uint32_t> expected {0, 1, 2, 3};
+ EXPECT_EQ(got, expected);
+}
+
+TEST(TrivialExchangeWriterTest, EmptyInput) {
+ MockRuntimeState state;
+ ExchangeSinkLocalState local_state(&state);
+ ExchangeTrivialWriter writer;
+
+ const size_t channel_count = 4;
+ auto channels = make_disabled_channels(&local_state, channel_count);
+
+ Block block = ColumnHelper::create_block<DataTypeInt32>({});
+ std::vector<vectorized::PartitionerBase::HashValType> channel_ids {};
+ const size_t rows = 0;
+
+ Status st = writer._channel_add_rows(&state, channels, channel_count,
channel_ids, rows, &block,
+ /*eos=*/false);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ ASSERT_EQ(writer._channel_rows_histogram.size(), channel_count);
+ for (size_t i = 0; i < channel_count; ++i) {
+ EXPECT_EQ(writer._channel_rows_histogram[i], 0U);
+ }
+ EXPECT_EQ(writer._origin_row_idx.size(), 0U);
+}
+
+TEST(OlapExchangeWriterTest, NeedCheckSkipsNegativeChannelIds) {
+ MockRuntimeState state;
+ ExchangeSinkLocalState local_state(&state);
+ ExchangeOlapWriter writer;
+
+ const size_t channel_count = 3;
+ auto channels = make_disabled_channels(&local_state, channel_count);
+
+ // channel_ids: [0, x, 2, x, 2]
+ Block block = ColumnHelper::create_block<DataTypeInt32>({10, 20, 30, 40,
50});
+ std::vector<vectorized::PartitionerBase::HashValType> channel_ids = {0,
10, 2, 10, 2};
+ const size_t rows = channel_ids.size();
+
+ Status st = writer._channel_add_rows(&state, channels, channel_count,
channel_ids, rows, &block,
+ /*eos=*/false, 10);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ // Only non-negative ids should be counted: hist = [1,0,2]
Review Comment:
The test name `NeedCheckSkipsNegativeChannelIds` and the inline comment
"Only non-negative ids should be counted" no longer match the behavior being
tested: the code treats channel IDs equal to a specific `invalid_val` (here
`10`) as skipped, not “negative” or “non-negative” values in general. To keep
the test self-documenting, please rename the test and adjust the comment to
describe that only IDs not equal to `invalid_val` are counted.
--
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]
