From c76ddc533ceacb81d23eabce6c038abd343e9284 Mon Sep 17 00:00:00 2001 From: Nitin Kashyap Date: Tue, 19 Nov 2024 13:32:59 +0530 Subject: [PATCH] fixed partitioners --- be/src/pipeline/dependency.h | 9 ++- .../pipeline/exec/exchange_sink_operator.cpp | 37 +++++++++-- be/src/pipeline/exec/exchange_sink_operator.h | 4 ++ .../exec/exchange_source_operator.cpp | 7 +- .../pipeline/exec/exchange_source_operator.h | 3 +- .../partitioned_hash_join_sink_operator.cpp | 4 +- .../partitioned_hash_join_sink_operator.h | 3 +- be/src/vec/columns/column.h | 16 ++--- be/src/vec/columns/column_nullable.cpp | 6 +- be/src/vec/runtime/partitioner.cpp | 22 ++++--- be/src/vec/runtime/partitioner.h | 48 +++++++++----- be/src/vec/sink/vdata_stream_sender.h | 7 -- .../datasource/FederationBackendPolicy.java | 10 +-- .../doris/datasource/FileQueryScanNode.java | 66 +++++++++++-------- .../doris/datasource/SplitAssignment.java | 26 ++++++-- .../doris/datasource/SplitToScanRange.java | 1 + .../datasource/hive/HMSExternalTable.java | 2 +- .../doris/datasource/hive/HiveBucketUtil.java | 2 +- .../translator/PhysicalPlanTranslator.java | 10 +-- .../ChildOutputPropertyDeriver.java | 8 ++- .../ChildrenPropertiesRegulator.java | 2 +- .../properties/DistributionSpecHash.java | 6 +- .../LogicalFileScanToPhysicalFileScan.java | 21 ++++-- .../apache/doris/planner/ExchangeNode.java | 9 +++ .../java/org/apache/doris/qe/Coordinator.java | 66 +++++++++++-------- gensrc/thrift/PlanNodes.thrift | 1 + 26 files changed, 256 insertions(+), 140 deletions(-) diff --git a/be/src/pipeline/dependency.h b/be/src/pipeline/dependency.h index 860ed82d06a63e3..058004e37b7979d 100644 --- a/be/src/pipeline/dependency.h +++ b/be/src/pipeline/dependency.h @@ -745,7 +745,14 @@ inline std::string get_exchange_type_name(ExchangeType idx) { struct DataDistribution { DataDistribution(ExchangeType type) : distribution_type(type), hash_type(THashType::CRC32) {} DataDistribution(ExchangeType type, const std::vector& partition_exprs_) - : distribution_type(type), partition_exprs(partition_exprs_) {} + : distribution_type(type), + partition_exprs(partition_exprs_), + hash_type(THashType::CRC32) {} + DataDistribution(ExchangeType type, const THashType::type hash_type) + : distribution_type(type), hash_type(hash_type) {} + DataDistribution(ExchangeType type, const std::vector& partition_exprs_, + const THashType::type hash) + : distribution_type(type), partition_exprs(partition_exprs_), hash_type(hash) {} DataDistribution(const DataDistribution& other) = default; bool need_local_exchange() const { return distribution_type != ExchangeType::NOOP; } DataDistribution& operator=(const DataDistribution& other) = default; diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp index 31ddec74e39536e..0dcbc5a52154dcf 100644 --- a/be/src/pipeline/exec/exchange_sink_operator.cpp +++ b/be/src/pipeline/exec/exchange_sink_operator.cpp @@ -193,15 +193,17 @@ Status ExchangeSinkLocalState::open(RuntimeState* state) { _profile->add_info_string("Partitioner", fmt::format("Crc32HashPartitioner({})", _partition_count)); } else if (_part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED) { - _partition_count = channel_shared_ptrs.size(); + _partition_count = channels.size(); if (_hash_type == THashType::SPARK_MURMUR32) { - _partitioner.reset(new vectorized::Murmur32HashPartitioner( - channel_shared_ptrs.size())); + _partitioner = std::make_unique< + vectorized::Murmur32HashPartitioner>( + channels.size()); _profile->add_info_string("Partitioner", fmt::format("Murmur32HashPartitioner({})", _partition_count)); } else { - _partitioner.reset(new vectorized::Crc32HashPartitioner( - channel_shared_ptrs.size())); + _partitioner = std::make_unique< + vectorized::Crc32HashPartitioner>( + channels.size()); _profile->add_info_string("Partitioner", fmt::format("Crc32HashPartitioner({})", _partition_count)); } @@ -358,6 +360,28 @@ Status ExchangeSinkOperatorX::init(const TDataSink& tsink) { return Status::OK(); } +std::string ExchangeSinkOperatorX::debug_string(int indentation_level) const { + fmt::memory_buffer debug_string_buffer; + fmt::format_to(debug_string_buffer, "{}", Base::debug_string(indentation_level)); + + string dest_names; + for (const auto& dest : _dests) { + if (dest_names.empty()) { + dest_names += print_id(dest.fragment_instance_id); + } else { + dest_names += ", " + print_id(dest.fragment_instance_id); + } + } + + fmt::format_to(debug_string_buffer, + ", Info: (_num_recievers = {}, _dest_node_id = {}," + ", _partition_type = {}, _hash_type = {}," + " _destinations = [{}])", + _dests.size(), _dest_node_id, to_string(_part_type), to_string(_hash_type), + dest_names); + return fmt::to_string(debug_string_buffer); +} + Status ExchangeSinkOperatorX::open(RuntimeState* state) { RETURN_IF_ERROR(DataSinkOperatorX::open(state)); _state = state; @@ -430,7 +454,8 @@ Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block if (serialized) { auto cur_block = local_state._serializer.get_block()->to_block(); if (!cur_block.empty()) { - DCHECK(eos || local_state._serializer.is_local()) << debug_string(state, 0); + DCHECK(eos || local_state._serializer.is_local()) + << Base::debug_string(state, 0); RETURN_IF_ERROR(local_state._serializer.serialize_block( &cur_block, block_holder->get_block(), local_state._rpc_channels_num)); diff --git a/be/src/pipeline/exec/exchange_sink_operator.h b/be/src/pipeline/exec/exchange_sink_operator.h index 8a4f77ffb1c5303..1d0944cf95c820e 100644 --- a/be/src/pipeline/exec/exchange_sink_operator.h +++ b/be/src/pipeline/exec/exchange_sink_operator.h @@ -195,12 +195,16 @@ class ExchangeSinkLocalState final : public PipelineXSinkLocalState<> { }; class ExchangeSinkOperatorX final : public DataSinkOperatorX { + using Base = DataSinkOperatorX; + public: ExchangeSinkOperatorX(RuntimeState* state, const RowDescriptor& row_desc, int operator_id, const TDataStreamSink& sink, const std::vector& destinations); Status init(const TDataSink& tsink) override; + [[nodiscard]] std::string debug_string(int indentation_level) const override; + RuntimeState* state() { return _state; } Status open(RuntimeState* state) override; diff --git a/be/src/pipeline/exec/exchange_source_operator.cpp b/be/src/pipeline/exec/exchange_source_operator.cpp index dbde9abd05dc347..cb84c7b99b3e3b7 100644 --- a/be/src/pipeline/exec/exchange_source_operator.cpp +++ b/be/src/pipeline/exec/exchange_source_operator.cpp @@ -52,8 +52,9 @@ std::string ExchangeSourceOperatorX::debug_string(int indentation_level) const { fmt::memory_buffer debug_string_buffer; fmt::format_to(debug_string_buffer, "{}", OperatorX::debug_string(indentation_level)); - fmt::format_to(debug_string_buffer, ", Info: (_num_senders = {}, _is_merging = {})", - _num_senders, _is_merging); + fmt::format_to(debug_string_buffer, + ", Info: (_num_senders = {}, _is_merging = {}, _hash_type = {})", _num_senders, + _is_merging, to_string(_hash_type)); return fmt::to_string(debug_string_buffer); } @@ -106,6 +107,8 @@ ExchangeSourceOperatorX::ExchangeSourceOperatorX(ObjectPool* pool, const TPlanNo _partition_type(tnode.exchange_node.__isset.partition_type ? tnode.exchange_node.partition_type : TPartitionType::UNPARTITIONED), + _hash_type(tnode.exchange_node.__isset.hash_type ? tnode.exchange_node.hash_type + : THashType::CRC32), _input_row_desc(descs, tnode.exchange_node.input_row_tuples, std::vector(tnode.nullable_tuples.begin(), tnode.nullable_tuples.begin() + diff --git a/be/src/pipeline/exec/exchange_source_operator.h b/be/src/pipeline/exec/exchange_source_operator.h index f938f5007d16430..f52a76c5e6d9c0f 100644 --- a/be/src/pipeline/exec/exchange_source_operator.h +++ b/be/src/pipeline/exec/exchange_source_operator.h @@ -90,7 +90,7 @@ class ExchangeSourceOperatorX final : public OperatorX { return _partition_type == TPartitionType::HASH_PARTITIONED ? DataDistribution(ExchangeType::HASH_SHUFFLE) : _partition_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED - ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE) + ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _hash_type) : DataDistribution(ExchangeType::NOOP); } @@ -99,6 +99,7 @@ class ExchangeSourceOperatorX final : public OperatorX { const int _num_senders; const bool _is_merging; const TPartitionType::type _partition_type; + const THashType::type _hash_type; RowDescriptor _input_row_desc; // use in merge sort diff --git a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp index 55df66ac12fcaa6..ce002286ed56535 100644 --- a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp +++ b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp @@ -402,8 +402,8 @@ PartitionedHashJoinSinkOperatorX::PartitionedHashJoinSinkOperatorX(ObjectPool* p descs), _join_distribution(tnode.hash_join_node.__isset.dist_type ? tnode.hash_join_node.dist_type : TJoinDistributionType::NONE), - _hash_type(tnode.hash_join_node.__isset.__isset.hash_type ? tnode.hash_join_node.hash_type - : THashType::CRC32), + _hash_type(tnode.hash_join_node.__isset.hash_type ? tnode.hash_join_node.hash_type + : THashType::CRC32), _distribution_partition_exprs(tnode.__isset.distribute_expr_lists ? tnode.distribute_expr_lists[1] : std::vector {}), diff --git a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h index 36637dc973e496d..74dba0a4c64372f 100644 --- a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h +++ b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h @@ -109,8 +109,7 @@ class PartitionedHashJoinSinkOperatorX return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || _join_distribution == TJoinDistributionType::COLOCATE ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, - _distribution_partition_exprs, - hash_type) + _distribution_partition_exprs, _hash_type) : DataDistribution(ExchangeType::HASH_SHUFFLE, _distribution_partition_exprs); } diff --git a/be/src/vec/columns/column.h b/be/src/vec/columns/column.h index 86f1040abf05295..5bc16c0ecb04e82 100644 --- a/be/src/vec/columns/column.h +++ b/be/src/vec/columns/column.h @@ -57,16 +57,16 @@ class SipHash; } \ } -#define DO_MURMUR_HASHES_FUNCTION_COLUMN_IMPL() \ - if (null_data == nullptr) { \ - for (size_t i = 0; i < s; i++) { \ +#define DO_MURMUR_HASHES_FUNCTION_COLUMN_IMPL() \ + if (null_data == nullptr) { \ + for (size_t i = 0; i < s; i++) { \ hashes[i] = HashUtil::murmur_hash3_32(&data[i], sizeof(T), hashes[i]); \ - } \ - } else { \ - for (size_t i = 0; i < s; i++) { \ - if (null_data[i] == 0) \ + } \ + } else { \ + for (size_t i = 0; i < s; i++) { \ + if (null_data[i] == 0) \ hashes[i] = HashUtil::murmur_hash3_32(&data[i], sizeof(T), hashes[i]); \ - } \ + } \ } namespace doris::vectorized { diff --git a/be/src/vec/columns/column_nullable.cpp b/be/src/vec/columns/column_nullable.cpp index 007a5addb4dad18..49854b30163fc53 100644 --- a/be/src/vec/columns/column_nullable.cpp +++ b/be/src/vec/columns/column_nullable.cpp @@ -94,8 +94,8 @@ void ColumnNullable::update_murmur_with_value(size_t start, size_t end, int32_t& nested_column->update_murmur_with_value(start, end, hash, nullptr); } else { const auto* __restrict real_null_data = - assert_cast(*null_map).get_data().data(); - for (int i = start; i < end; ++i) { + assert_cast(get_null_map_column()).get_data().data(); + for (size_t i = start; i < end; ++i) { if (real_null_data[i] != 0) { hash = HashUtil::murmur_hash3_32_null(hash); } @@ -140,7 +140,7 @@ void ColumnNullable::update_murmurs_with_value(int32_t* __restrict hashes, auto s = rows; DCHECK(s == size()); const auto* __restrict real_null_data = - assert_cast(*null_map).get_data().data(); + assert_cast(get_null_map_column()).get_data().data(); if (!has_null()) { nested_column->update_murmurs_with_value(hashes, type, rows, offset, nullptr); } else { diff --git a/be/src/vec/runtime/partitioner.cpp b/be/src/vec/runtime/partitioner.cpp index 188918bf9913e8e..c3bf779bd13c563 100644 --- a/be/src/vec/runtime/partitioner.cpp +++ b/be/src/vec/runtime/partitioner.cpp @@ -25,8 +25,8 @@ namespace doris::vectorized { template -Status Partitioner::do_partitioning(RuntimeState* state, Block* block, - MemTracker* mem_tracker) const { +Status Partitioner::do_partitioning(RuntimeState* state, + Block* block) const { int rows = block->rows(); if (rows > 0) { @@ -55,7 +55,7 @@ Status Partitioner::do_partitioning(RuntimeState* sta template void Crc32HashPartitioner::_do_hash(const ColumnPtr& column, uint32_t* __restrict result, int idx) const { - column->update_crcs_with_value(result, _partition_expr_ctxs[idx]->root()->type().type, + column->update_crcs_with_value(result, Base::_partition_expr_ctxs[idx]->root()->type().type, cast_set(column->size())); } @@ -69,12 +69,13 @@ void Murmur32HashPartitioner::_do_hash(const ColumnPtr& column, template Status Crc32HashPartitioner::clone(RuntimeState* state, std::unique_ptr& partitioner) { - auto* new_partitioner = new Crc32HashPartitioner(cast_set(_partition_count)); + auto* new_partitioner = + new Crc32HashPartitioner(cast_set(Base::_partition_count)); partitioner.reset(new_partitioner); - new_partitioner->_partition_expr_ctxs.resize(_partition_expr_ctxs.size()); - for (size_t i = 0; i < _partition_expr_ctxs.size(); i++) { - RETURN_IF_ERROR( - _partition_expr_ctxs[i]->clone(state, new_partitioner->_partition_expr_ctxs[i])); + new_partitioner->_partition_expr_ctxs.resize(Base::_partition_expr_ctxs.size()); + for (size_t i = 0; i < Base::_partition_expr_ctxs.size(); i++) { + RETURN_IF_ERROR(Base::_partition_expr_ctxs[i]->clone( + state, new_partitioner->_partition_expr_ctxs[i])); } return Status::OK(); } @@ -82,7 +83,8 @@ Status Crc32HashPartitioner::clone(RuntimeState* state, template Status Murmur32HashPartitioner::clone(RuntimeState* state, std::unique_ptr& partitioner) { - auto* new_partitioner = new Murmur32HashPartitioner(Base::_partition_count); + auto* new_partitioner = + new Murmur32HashPartitioner(cast_set(Base::_partition_count)); partitioner.reset(new_partitioner); new_partitioner->_partition_expr_ctxs.resize(Base::_partition_expr_ctxs.size()); for (size_t i = 0; i < Base::_partition_expr_ctxs.size(); i++) { @@ -94,7 +96,7 @@ Status Murmur32HashPartitioner::clone(RuntimeState* state, template int32_t Murmur32HashPartitioner::_get_default_seed() const { - return reinterpret_cast(HashUtil::SPARK_MURMUR_32_SEED); + return static_cast(HashUtil::SPARK_MURMUR_32_SEED); } template class Crc32HashPartitioner; diff --git a/be/src/vec/runtime/partitioner.h b/be/src/vec/runtime/partitioner.h index 7dd5d295b4e6623..bf1bd9bb5d2ed7c 100644 --- a/be/src/vec/runtime/partitioner.h +++ b/be/src/vec/runtime/partitioner.h @@ -58,11 +58,11 @@ class PartitionerBase { const size_t _partition_count; }; -template -class Crc32HashPartitioner : public PartitionerBase { +template +class Partitioner : public PartitionerBase { public: - Crc32HashPartitioner(int partition_count) : PartitionerBase(partition_count) {} - ~Crc32HashPartitioner() override = default; + Partitioner(int partition_count) : PartitionerBase(partition_count) {} + ~Partitioner() override = default; Status init(const std::vector& texprs) override { return VExpr::create_expr_trees(texprs, _partition_expr_ctxs); @@ -76,9 +76,9 @@ class Crc32HashPartitioner : public PartitionerBase { Status do_partitioning(RuntimeState* state, Block* block) const override; - ChannelField get_channel_ids() const override { return {_hash_vals.data(), sizeof(uint32_t)}; } - - Status clone(RuntimeState* state, std::unique_ptr& partitioner) override; + ChannelField get_channel_ids() const override { + return {_hash_vals.data(), sizeof(HashValueType)}; + } protected: Status _get_partition_column_result(Block* block, std::vector& result) const { @@ -89,14 +89,12 @@ class Crc32HashPartitioner : public PartitionerBase { return Status::OK(); } - void _do_hash(const ColumnPtr& column, uint32_t* __restrict result, int idx) const; - - HashValueType _get_default_seed() const { - return reinterpret_cast(0); - } + virtual void _do_hash(const ColumnPtr& column, HashValueType* __restrict result, + int idx) const = 0; + virtual HashValueType _get_default_seed() const { return static_cast(0); } VExprContextSPtrs _partition_expr_ctxs; - mutable std::vector _hash_vals; + mutable std::vector _hash_vals; }; struct ShuffleChannelIds { @@ -113,6 +111,27 @@ struct SpillPartitionChannelIds { } }; +struct ShufflePModChannelIds { + template + HashValueType operator()(HashValueType l, int32_t r) { + return (l % r + r) % r; + } +}; + +template +class Crc32HashPartitioner final : public Partitioner { +public: + using Base = Partitioner; + Crc32HashPartitioner(int partition_count) + : Partitioner(partition_count) {} + ~Crc32HashPartitioner() override = default; + + Status clone(RuntimeState* state, std::unique_ptr& partitioner) override; + +private: + void _do_hash(const ColumnPtr& column, uint32_t* __restrict result, int idx) const override; +}; + template class Murmur32HashPartitioner final : public Partitioner { public: @@ -123,10 +142,9 @@ class Murmur32HashPartitioner final : public Partitioner { Status clone(RuntimeState* state, std::unique_ptr& partitioner) override; - int32_t _get_default_seed() const; - private: void _do_hash(const ColumnPtr& column, int32_t* __restrict result, int idx) const override; + int32_t _get_default_seed() const override; }; } // namespace vectorized diff --git a/be/src/vec/sink/vdata_stream_sender.h b/be/src/vec/sink/vdata_stream_sender.h index 5886f52cc43d70f..4999602fdf49a78 100644 --- a/be/src/vec/sink/vdata_stream_sender.h +++ b/be/src/vec/sink/vdata_stream_sender.h @@ -96,13 +96,6 @@ class BlockSerializer { const int _batch_size; }; -struct ShufflePModChannelIds { - template - HashValueType operator()(HashValueType l, int32_t r) { - return (l % r + r) % r; - } -}; - class Channel { public: friend class pipeline::ExchangeSinkBuffer; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java index 36bffa625e9fb15..ec6dedf10e8915f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java @@ -23,6 +23,7 @@ import org.apache.doris.catalog.Env; import org.apache.doris.common.Config; import org.apache.doris.common.IndexedPriorityQueue; +import org.apache.doris.common.Pair; import org.apache.doris.common.ResettableRandomizedIterator; import org.apache.doris.common.UserException; import org.apache.doris.common.util.ConsistentHash; @@ -217,15 +218,16 @@ public void setEnableSplitsRedistribution(boolean enableSplitsRedistribution) { this.enableSplitsRedistribution = enableSplitsRedistribution; } - public Multimap computeBucketAwareScanRangeAssignmentWith(List splits) throws UserException { - ListMultimap assignment = ArrayListMultimap.create(); + public Multimap, Split> computeBucketAwareScanRangeAssignmentWith(List splits) + throws UserException { + ListMultimap, Split> assignment = ArrayListMultimap.create(); int bucketNum = 0; for (Split split : splits) { FileSplit fileSplit = (FileSplit) split; - bucketNum = HiveBucketUtil.getBucketNumberFromPath(fileSplit.getPath().getName()).getAsInt(); + bucketNum = HiveBucketUtil.getBucketNumberFromPath(fileSplit.getPath().getPath().getName()).getAsInt(); List candidateNodes = consistentBucketHash.getNode(bucketNum, 1); - assignment.put(candidateNodes.get(0), split); + assignment.put(Pair.of(candidateNodes.get(0), bucketNum), split); } return assignment; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index 901bf36f5ae3ee4..e9c14358c258729 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -30,13 +30,13 @@ import org.apache.doris.catalog.TableIf; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.NotImplementedException; +import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.hive.AcidInfo; import org.apache.doris.datasource.hive.AcidInfo.DeleteDeltaInfo; import org.apache.doris.datasource.hive.HMSExternalTable; -import org.apache.doris.datasource.hive.HiveBucketUtil; import org.apache.doris.datasource.hive.source.HiveScanNode; import org.apache.doris.datasource.hive.source.HiveSplit; import org.apache.doris.planner.DataPartition; @@ -322,12 +322,19 @@ public void createScanRangeLocations() throws UserException { params.setProperties(locationProperties); } + boolean isSparkBucketedHiveTable = false; + TableIf targetTable = getTargetTable(); + if (targetTable instanceof HMSExternalTable) { + isSparkBucketedHiveTable = ((HMSExternalTable) targetTable).isSparkBucketedTable(); + } + List pathPartitionKeys = getPathPartitionKeys(); if (isBatchMode()) { // File splits are generated lazily, and fetched by backends while scanning. // Only provide the unique ID of split source to backend. splitAssignment = new SplitAssignment( - backendPolicy, this, this::splitToScanRange, locationProperties, pathPartitionKeys); + backendPolicy, this, this::splitToScanRange, locationProperties, pathPartitionKeys, + isSparkBucketedHiveTable); splitAssignment.init(); if (ConnectContext.get().getExecutor() != null) { ConnectContext.get().getExecutor().getSummaryProfile().setGetSplitsFinishTime(); @@ -374,25 +381,33 @@ public void createScanRangeLocations() throws UserException { if (inputSplits.isEmpty() && !isFileStreamType()) { return; } - boolean isSparkBucketedHiveTable = false; - Multimap assignment; - TableIf targetTable = getTargetTable(); - if (targetTable instanceof HMSExternalTable) { - isSparkBucketedHiveTable = ((HMSExternalTable) targetTable).isSparkBucketedTable(); - } if (isSparkBucketedHiveTable) { + Multimap, Split> assignment; + assignment = backendPolicy.computeBucketAwareScanRangeAssignmentWith(inputSplits); + for (Pair backend : assignment.keySet()) { + Collection splits = assignment.get(backend); + for (Split split : splits) { + scanRangeLocations.add(splitToScanRange(backend.first, backend.second, locationProperties, + split, pathPartitionKeys)); + totalFileSize += split.getLength(); + } + scanBackendIds.add(backend.first.getId()); + } } else { + Multimap assignment; + assignment = backendPolicy.computeScanRangeAssignment(inputSplits); - } - for (Backend backend : assignment.keySet()) { - Collection splits = assignment.get(backend); - for (Split split : splits) { - scanRangeLocations.add(splitToScanRange(backend, locationProperties, split, pathPartitionKeys)); - totalFileSize += split.getLength(); + for (Backend backend : assignment.keySet()) { + Collection splits = assignment.get(backend); + for (Split split : splits) { + scanRangeLocations.add(splitToScanRange(backend, 0, locationProperties, split, + pathPartitionKeys)); + totalFileSize += split.getLength(); + } + scanBackendIds.add(backend.getId()); } - scanBackendIds.add(backend.getId()); } } @@ -409,6 +424,7 @@ public void createScanRangeLocations() throws UserException { private TScanRangeLocations splitToScanRange( Backend backend, + Integer bucketNum, Map locationProperties, Split split, List pathPartitionKeys) throws UserException { @@ -426,18 +442,7 @@ private TScanRangeLocations splitToScanRange( false, isACID) : fileSplit.getPartitionValues(); boolean isSparkBucketedHiveTable = false; - int bucketNum = 0; TableIf targetTable = getTargetTable(); - if (targetTable instanceof HMSExternalTable) { - isSparkBucketedHiveTable = ((HMSExternalTable) targetTable).isSparkBucketedTable(); - if (isSparkBucketedHiveTable) { - bucketNum = HiveBucketUtil.getBucketNumberFromPath(fileSplit.getPath().getName()).getAsInt(); - if (!bucketSeq2locations.containsKey(bucketNum)) { - bucketSeq2locations.put(bucketNum, curLocations); - } - curLocations = bucketSeq2locations.get(bucketNum).get(0); - } - } TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys); TFileCompressType fileCompressType = getFileCompressType(fileSplit); @@ -478,6 +483,15 @@ private TScanRangeLocations splitToScanRange( location.setServer(new TNetworkAddress(backend.getHost(), backend.getBePort())); curLocations.addToLocations(location); + if (targetTable instanceof HMSExternalTable) { + isSparkBucketedHiveTable = ((HMSExternalTable) targetTable).isSparkBucketedTable(); + if (isSparkBucketedHiveTable) { + if (!bucketSeq2locations.containsKey(bucketNum)) { + bucketSeq2locations.put(bucketNum, curLocations); + } + } + } + if (LOG.isDebugEnabled()) { LOG.debug("assign to backend {} with table split: {} ({}, {}), location: {}, bucketNum: {}", curLocations.getLocations().get(0).getBackendId(), fileSplit.getPath(), diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java index 928854b91d18102..e64138558c736d8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java @@ -17,11 +17,13 @@ package org.apache.doris.datasource; +import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.spi.Split; import org.apache.doris.system.Backend; import org.apache.doris.thrift.TScanRangeLocations; +import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.Multimap; import java.util.ArrayList; @@ -49,6 +51,7 @@ public class SplitAssignment { private final Map locationProperties; private final List pathPartitionKeys; private final Object assignLock = new Object(); + private final boolean useBucketAssignment; private Split sampleSplit = null; private final AtomicBoolean isStop = new AtomicBoolean(false); private final AtomicBoolean scheduleFinished = new AtomicBoolean(false); @@ -60,12 +63,14 @@ public SplitAssignment( SplitGenerator splitGenerator, SplitToScanRange splitToScanRange, Map locationProperties, - List pathPartitionKeys) { + List pathPartitionKeys, + boolean useBucketedAssignment) { this.backendPolicy = backendPolicy; this.splitGenerator = splitGenerator; this.splitToScanRange = splitToScanRange; this.locationProperties = locationProperties; this.pathPartitionKeys = pathPartitionKeys; + this.useBucketAssignment = useBucketedAssignment; } public void init() throws UserException { @@ -88,14 +93,15 @@ private boolean waitFirstSplit() { return !scheduleFinished.get() && !isStop.get() && exception == null; } - private void appendBatch(Multimap batch) throws UserException { - for (Backend backend : batch.keySet()) { + private void appendBatch(Multimap, Split> batch) throws UserException { + for (Pair backend : batch.keySet()) { Collection splits = batch.get(backend); List locations = new ArrayList<>(splits.size()); for (Split split : splits) { - locations.add(splitToScanRange.getScanRange(backend, locationProperties, split, pathPartitionKeys)); + locations.add(splitToScanRange.getScanRange(backend.first, backend.second, locationProperties, + split, pathPartitionKeys)); } - if (!assignment.computeIfAbsent(backend, be -> new LinkedBlockingQueue<>()).offer(locations)) { + if (!assignment.computeIfAbsent(backend.first, be -> new LinkedBlockingQueue<>()).offer(locations)) { throw new UserException("Failed to offer batch split"); } } @@ -117,14 +123,20 @@ public void addToQueue(List splits) { if (splits.isEmpty()) { return; } - Multimap batch = null; + Multimap, Split> batch = ArrayListMultimap.create(); synchronized (assignLock) { if (sampleSplit == null) { sampleSplit = splits.get(0); assignLock.notify(); } try { - batch = backendPolicy.computeScanRangeAssignment(splits); + if (useBucketAssignment) { + batch = backendPolicy.computeBucketAwareScanRangeAssignmentWith(splits); + } else { + Multimap, Split> finalBatch = batch; + backendPolicy.computeScanRangeAssignment(splits).entries() + .forEach(e -> finalBatch.put(Pair.of(e.getKey(), 0), e.getValue())); + } } catch (UserException e) { exception = e; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitToScanRange.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitToScanRange.java index 0e890252857583c..ea58b6d8d0bb0a4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitToScanRange.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/SplitToScanRange.java @@ -28,6 +28,7 @@ public interface SplitToScanRange { TScanRangeLocations getScanRange( Backend backend, + Integer bucketNum, Map locationProperties, Split split, List pathPartitionKeys) throws UserException; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java index 087a24257d8f03d..ce996ae0299dfc6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java @@ -541,7 +541,7 @@ public Optional initSchema() { } else { columns = getHiveSchema(); } - List partitionColumns = initPartitionColumns(columns); + partitionColumns = initPartitionColumns(columns); initBucketingColumns(columns); return Optional.of(new HMSSchemaCacheValue(columns, partitionColumns)); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveBucketUtil.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveBucketUtil.java index fc0bed8d5e1c252..41d04507a1debaa 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveBucketUtil.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveBucketUtil.java @@ -103,7 +103,7 @@ private static PrimitiveTypeInfo convertToHiveColType(PrimitiveType dorisType) t private static final Iterable BUCKET_PATTERNS = ImmutableList.of( // spark/parquet pattern - // format: f"part-[paritionId]-[tid]-[txnId]-[jobId]-[taskAttemptId]-[fileCount].c000.snappy.parquet" + // format: f"part-[paritionId]-[tid]-[txnId]-[jobId]-[taskAttemptId]_[fileCount].c000.snappy.parquet" Pattern.compile("part-\\d{5}-\\w{8}-\\w{4}-\\w{4}-\\w{4}-\\w{12}_(\\d{5})(?:[-_.].*)?"), // legacy Presto naming pattern (current version matches Hive) Pattern.compile("\\d{8}_\\d{6}_\\d{5}_[a-z0-9]{5}_bucket-(\\d+)(?:[-_.].*)?"), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 7e2416cb2d5eabe..e87a04b8b40316a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -346,6 +346,7 @@ public PlanFragment visitPhysicalDistribute(PhysicalDistribute d } DataPartition dataPartition = toDataPartition(distribute.getDistributionSpec(), validOutputIds, context); exchangeNode.setPartitionType(dataPartition.getType()); + exchangeNode.setHashType(dataPartition.getHashType()); exchangeNode.setChildrenDistributeExprLists(distributeExprLists); PlanFragment parentFragment = new PlanFragment(context.nextFragmentId(), exchangeNode, dataPartition); if (distribute.getDistributionSpec() instanceof DistributionSpecGather) { @@ -555,7 +556,6 @@ public PlanFragment visitPhysicalFileScan(PhysicalFileScan fileScan, PlanTransla // TODO(cmy): determine the needCheckColumnPriv param FileQueryScanNode scanNode; - DataPartition dataPartition = DataPartition.RANDOM; if (table instanceof HMSExternalTable) { switch (((HMSExternalTable) table).getDlaType()) { case ICEBERG: @@ -588,8 +588,8 @@ public PlanFragment visitPhysicalFileScan(PhysicalFileScan fileScan, PlanTransla } else { throw new RuntimeException("do not support table type " + table.getType()); } - if (fileScan.getTableSnapshot().isPresent() && scanNode instanceof FileQueryScanNode) { - ((FileQueryScanNode) scanNode).setQueryTableSnapshot(fileScan.getTableSnapshot().get()); + if (fileScan.getTableSnapshot().isPresent()) { + scanNode.setQueryTableSnapshot(fileScan.getTableSnapshot().get()); } return getPlanFragmentForPhysicalFileScan(fileScan, context, scanNode, table, tupleDescriptor); } @@ -686,9 +686,9 @@ private PlanFragment getPlanFragmentForPhysicalFileScan(PhysicalFileScan fileSca if (fileScan.getDistributionSpec() instanceof DistributionSpecHash) { DistributionSpecHash distributionSpecHash = (DistributionSpecHash) fileScan.getDistributionSpec(); List partitionExprs = distributionSpecHash.getOrderedShuffledColumns().stream() - .map(context::findSlotRef).collect(Collectors.toList()); + .map(context::findSlotRef).collect(Collectors.toList()); dataPartition = new DataPartition(TPartitionType.HASH_PARTITIONED, - partitionExprs, scanNode.getHashType()); + partitionExprs, ((FileQueryScanNode) scanNode).getHashType()); } // Create PlanFragment PlanFragment planFragment = createPlanFragment(scanNode, dataPartition, fileScan); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java index 7b9d696cabc5e9c..a68d44cc3118aad 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java @@ -503,6 +503,8 @@ private PhysicalProperties computeShuffleJoinOutputProperties( ShuffleType outputShuffleType = shuffleSide == ShuffleSide.LEFT ? rightHashSpec.getShuffleType() : leftHashSpec.getShuffleType(); + DistributionSpecHash.StorageBucketHashType outputShuffleFunction = shuffleSide == ShuffleSide.LEFT + ? rightHashSpec.getShuffleFunction() : leftHashSpec.getShuffleFunction(); switch (hashJoin.getJoinType()) { case INNER_JOIN: @@ -522,7 +524,7 @@ private PhysicalProperties computeShuffleJoinOutputProperties( case LEFT_OUTER_JOIN: if (shuffleSide == ShuffleSide.LEFT) { return new PhysicalProperties( - leftHashSpec.withShuffleTypeAndForbidColocateJoin(outputShuffleType) + leftHashSpec.withShuffleTypeAndForbidColocateJoin(outputShuffleType, outputShuffleFunction) ); } else { return new PhysicalProperties(leftHashSpec); @@ -536,7 +538,7 @@ private PhysicalProperties computeShuffleJoinOutputProperties( // retain left shuffle type, since coordinator use left most node to schedule fragment // forbid colocate join, since right table already shuffle return new PhysicalProperties(rightHashSpec.withShuffleTypeAndForbidColocateJoin( - leftHashSpec.getShuffleType())); + leftHashSpec.getShuffleType(), leftHashSpec.getShuffleFunction())); } case FULL_OUTER_JOIN: return PhysicalProperties.createAnyFromHash(leftHashSpec, rightHashSpec); @@ -582,7 +584,7 @@ private PhysicalProperties legacyComputeShuffleJoinOutputProperties( // retain left shuffle type, since coordinator use left most node to schedule fragment // forbid colocate join, since right table already shuffle return new PhysicalProperties(rightHashSpec.withShuffleTypeAndForbidColocateJoin( - leftHashSpec.getShuffleType())); + leftHashSpec.getShuffleType(), leftHashSpec.getShuffleFunction())); } case FULL_OUTER_JOIN: return PhysicalProperties.createAnyFromHash(leftHashSpec); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java index d786215692cbb24..5cca6464afff654 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java @@ -679,7 +679,7 @@ private PhysicalProperties calAnotherSideRequired(ShuffleType shuffleType, notNeedShuffleSideRequired, needShuffleSideRequired); return new PhysicalProperties(new DistributionSpecHash(shuffleSideIds, shuffleType, needShuffleSideOutput.getTableId(), needShuffleSideOutput.getSelectedIndexId(), - needShuffleSideOutput.getPartitionIds(), notShuffleSideOutput.getShuffleFunction())); + needShuffleSideOutput.getPartitionIds(), notNeedShuffleSideOutput.getShuffleFunction())); } private void updateChildEnforceAndCost(int index, PhysicalProperties targetProperties) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java index f5283e7b35eb048..75e5682d9245505 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java @@ -356,8 +356,9 @@ public THashType toThrift() { case STORAGE_BUCKET_SPARK_MURMUR32: return THashType.SPARK_MURMUR32; case STORAGE_BUCKET_XXHASH64: - default: return THashType.XXHASH64; + default: + return THashType.CRC32; } } @@ -371,8 +372,9 @@ public static StorageBucketHashType fromThrift(THashType hashType) { case SPARK_MURMUR32: return STORAGE_BUCKET_SPARK_MURMUR32; case XXHASH64: - default: return STORAGE_BUCKET_XXHASH64; + default: + return STORAGE_BUCKET_CRC32; } } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalFileScanToPhysicalFileScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalFileScanToPhysicalFileScan.java index 5162949ed796d18..10fba90ce5ccb74 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalFileScanToPhysicalFileScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalFileScanToPhysicalFileScan.java @@ -28,11 +28,11 @@ import org.apache.doris.nereids.properties.DistributionSpecStorageAny; import org.apache.doris.nereids.rules.Rule; import org.apache.doris.nereids.rules.RuleType; -import org.apache.doris.nereids.trees.plans.logical.LogicalHudiScan; import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalHudiScan; import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan; import com.google.common.collect.Lists; @@ -68,8 +68,17 @@ private DistributionSpec convertDistribution(LogicalFileScan fileScan) { } HMSExternalTable hmsExternalTable = (HMSExternalTable) table; + if (hmsExternalTable.getDlaType() != HMSExternalTable.DLAType.HIVE + && !hmsExternalTable.isSparkBucketedTable()) { + return DistributionSpecStorageAny.INSTANCE; + } + + boolean isSelectUnpartitioned = !hmsExternalTable.isPartitionedTable() + || hmsExternalTable.getPartitionNames().size() == 1 + || fileScan.getSelectedPartitions().selectedPartitions.size() == 1; + DistributionInfo distributionInfo = hmsExternalTable.getDefaultDistributionInfo(); - if (distributionInfo instanceof HashDistributionInfo) { + if (distributionInfo instanceof HashDistributionInfo && isSelectUnpartitioned) { HashDistributionInfo hashDistributionInfo = (HashDistributionInfo) distributionInfo; List output = fileScan.getOutput(); List hashColumns = Lists.newArrayList(); @@ -80,12 +89,10 @@ private DistributionSpec convertDistribution(LogicalFileScan fileScan) { } } } - StorageBucketHashType function = StorageBucketHashType.STORAGE_BUCKET_CRC32; - if (hmsExternalTable.isSparkBucketedTable()) { - function = StorageBucketHashType.STORAGE_BUCKET_SPARK_MURMUR32; - } + return new DistributionSpecHash(hashColumns, DistributionSpecHash.ShuffleType.NATURAL, - fileScan.getTable().getId(), -1, Collections.emptySet(), function); + fileScan.getTable().getId(), -1, Collections.emptySet(), + StorageBucketHashType.STORAGE_BUCKET_SPARK_MURMUR32); } return DistributionSpecStorageAny.INSTANCE; diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java index cb6628b01c556b9..a13f830692f3855 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java @@ -30,6 +30,7 @@ import org.apache.doris.statistics.StatsRecursiveDerive; import org.apache.doris.thrift.TExchangeNode; import org.apache.doris.thrift.TExplainLevel; +import org.apache.doris.thrift.THashType; import org.apache.doris.thrift.TPartitionType; import org.apache.doris.thrift.TPlanNode; import org.apache.doris.thrift.TPlanNodeType; @@ -67,6 +68,7 @@ public class ExchangeNode extends PlanNode { private boolean isRightChildOfBroadcastHashJoin = false; private TPartitionType partitionType; + private THashType hashType; /** * use for Nereids only. @@ -168,6 +170,10 @@ public void setMergeInfo(SortInfo info) { this.planNodeName = "V" + MERGING_EXCHANGE_NODE; } + public void setHashType(THashType hashType) { + this.hashType = hashType; + } + @Override protected void toThrift(TPlanNode msg) { // If this fragment has another scan node, this exchange node is serial or not should be decided by the scan @@ -182,6 +188,9 @@ protected void toThrift(TPlanNode msg) { if (mergeInfo != null) { msg.exchange_node.setSortInfo(mergeInfo.toThrift()); } + if (hashType != null) { + msg.exchange_node.setHashType(hashType); + } msg.exchange_node.setOffset(offset); msg.exchange_node.setPartitionType(partitionType); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index e9d7fc845c00e79..566563cec6d62dc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -37,8 +37,8 @@ import org.apache.doris.common.util.TimeUtils; import org.apache.doris.datasource.ExternalScanNode; import org.apache.doris.datasource.FileQueryScanNode; -import org.apache.doris.datasource.hive.source.HiveScanNode; import org.apache.doris.datasource.hive.HMSTransaction; +import org.apache.doris.datasource.hive.source.HiveScanNode; import org.apache.doris.datasource.iceberg.IcebergTransaction; import org.apache.doris.load.loadv2.LoadJob; import org.apache.doris.metric.MetricRepo; @@ -2566,7 +2566,7 @@ class BucketShuffleJoinController { private final Map> fragmentIdToSeqToAddressMap = Maps.newHashMap(); // fragment_id -> < be_id -> bucket_count > - private final Map> fragmentIdToBuckendIdBucketCountMap = Maps.newHashMap(); + private final Map> fragmentIdToBackendIdBucketCountMap = Maps.newHashMap(); // fragment_id -> bucket_num protected final Map fragmentIdToBucketNumMap = Maps.newHashMap(); @@ -2621,30 +2621,30 @@ private void getExecHostPortForFragmentIDAndBucketSeq(TScanRangeLocations seqLoc PlanFragmentId fragmentId, Integer bucketSeq, ImmutableMap idToBackend, Map addressToBackendID, Map replicaNumPerHost) throws Exception { - Map buckendIdToBucketCountMap = fragmentIdToBuckendIdBucketCountMap.get(fragmentId); + Map backendIdToBucketCountMap = fragmentIdToBackendIdBucketCountMap.get(fragmentId); int maxBucketNum = Integer.MAX_VALUE; - long buckendId = Long.MAX_VALUE; + long backendId = Long.MAX_VALUE; Long minReplicaNum = Long.MAX_VALUE; for (TScanRangeLocation location : seqLocation.locations) { - if (buckendIdToBucketCountMap.getOrDefault(location.backend_id, 0) < maxBucketNum) { - maxBucketNum = buckendIdToBucketCountMap.getOrDefault(location.backend_id, 0); - buckendId = location.backend_id; + if (backendIdToBucketCountMap.getOrDefault(location.backend_id, 0) < maxBucketNum) { + maxBucketNum = backendIdToBucketCountMap.getOrDefault(location.backend_id, 0); + backendId = location.backend_id; minReplicaNum = replicaNumPerHost.get(location.server); - } else if (buckendIdToBucketCountMap.getOrDefault(location.backend_id, 0) == maxBucketNum + } else if (backendIdToBucketCountMap.getOrDefault(location.backend_id, 0) == maxBucketNum && replicaNumPerHost.get(location.server) < minReplicaNum) { - buckendId = location.backend_id; + backendId = location.backend_id; minReplicaNum = replicaNumPerHost.get(location.server); } } Reference backendIdRef = new Reference<>(); - TNetworkAddress execHostPort = SimpleScheduler.getHost(buckendId, + TNetworkAddress execHostPort = SimpleScheduler.getHost(backendId, seqLocation.locations, idToBackend, backendIdRef); - //the backend with buckendId is not alive, chose another new backend - if (backendIdRef.getRef() != buckendId) { - buckendIdToBucketCountMap.put(backendIdRef.getRef(), - buckendIdToBucketCountMap.getOrDefault(backendIdRef.getRef(), 0) + 1); - } else { //the backend with buckendId is alive, update buckendIdToBucketCountMap directly - buckendIdToBucketCountMap.put(buckendId, buckendIdToBucketCountMap.getOrDefault(buckendId, 0) + 1); + //the backend with backendId is not alive, chose another new backend + if (backendIdRef.getRef() != backendId) { + backendIdToBucketCountMap.put(backendIdRef.getRef(), + backendIdToBucketCountMap.getOrDefault(backendIdRef.getRef(), 0) + 1); + } else { //the backend with backendId is alive, update backendIdToBucketCountMap directly + backendIdToBucketCountMap.put(backendId, backendIdToBucketCountMap.getOrDefault(backendId, 0) + 1); } for (TScanRangeLocation location : seqLocation.locations) { replicaNumPerHost.put(location.server, replicaNumPerHost.get(location.server) - 1); @@ -2676,7 +2676,7 @@ private void computeScanRangeAssignmentByBucketForOlap( fragmentIdToBucketNumMap.put(scanNode.getFragmentId(), bucketNum); fragmentIdToSeqToAddressMap.put(scanNode.getFragmentId(), new HashMap<>()); fragmentIdBucketSeqToScanRangeMap.put(scanNode.getFragmentId(), new BucketSeqToScanRange()); - fragmentIdToBuckendIdBucketCountMap.put(scanNode.getFragmentId(), new HashMap<>()); + fragmentIdToBackendIdBucketCountMap.put(scanNode.getFragmentId(), new HashMap<>()); scanNode.getFragment().setBucketNum(bucketNum); } Map bucketSeqToAddress @@ -2721,7 +2721,7 @@ private void computeScanRangeAssignmentByBucketForHive( fragmentIdToBucketNumMap.put(scanNode.getFragmentId(), bucketNum); fragmentIdToSeqToAddressMap.put(scanNode.getFragmentId(), new HashMap<>()); fragmentIdBucketSeqToScanRangeMap.put(scanNode.getFragmentId(), new BucketSeqToScanRange()); - fragmentIdToBuckendIdBucketCountMap.put(scanNode.getFragmentId(), new HashMap<>()); + fragmentIdToBackendIdBucketCountMap.put(scanNode.getFragmentId(), new HashMap<>()); scanNode.getFragment().setBucketNum(bucketNum); } Map bucketSeqToAddress @@ -2811,12 +2811,11 @@ private void assignScanRanges(PlanFragmentId fragmentId, int parallelExecInstanc for (Pair>> nodeScanRangeMap : scanRange) { for (Map.Entry> nodeScanRange : nodeScanRangeMap.second.entrySet()) { - if (!instanceParam.perNodeScanRanges.containsKey(nodeScanRange.getKey())) { - range.put(nodeScanRange.getKey(), Lists.newArrayList()); - instanceParam.perNodeScanRanges.put(nodeScanRange.getKey(), Lists.newArrayList()); - } - range.get(nodeScanRange.getKey()).addAll(nodeScanRange.getValue()); - instanceParam.perNodeScanRanges.get(nodeScanRange.getKey()) + + range.computeIfAbsent(nodeScanRange.getKey(), ArrayList::new) + .addAll(nodeScanRange.getValue()); + instanceParam.perNodeScanRanges + .computeIfAbsent(nodeScanRange.getKey(), ArrayList::new) .addAll(nodeScanRange.getValue()); } } @@ -3332,10 +3331,25 @@ public void appendScanRange(StringBuilder sb, List params) { } TEsScanRange esScanRange = range.getScanRange().getEsScanRange(); if (esScanRange != null) { + if (idx++ != 0) { + sb.append(","); + } sb.append("{ index=").append(esScanRange.getIndex()) .append(", shardid=").append(esScanRange.getShardId()) .append("}"); } + TExternalScanRange extScanRange = range.getScanRange().getExtScanRange(); + if (extScanRange != null) { + TFileScanRange fileScanRange = extScanRange.getFileScanRange(); + if (fileScanRange != null) { + if (idx++ != 0) { + sb.append(","); + } + sb.append("{path=") + .append(fileScanRange.getRanges().get(0).getPath()) + .append("}"); + } + } } sb.append("]"); } @@ -3352,10 +3366,10 @@ public void appendTo(StringBuilder sb) { } TNetworkAddress address = instanceExecParams.get(i).host; Map> scanRanges = - scanRangeAssignment.get(address); + instanceExecParams.get(i).perNodeScanRanges; sb.append("{"); sb.append("id=").append(DebugUtil.printId(instanceExecParams.get(i).instanceId)); - sb.append(",host=").append(instanceExecParams.get(i).host); + sb.append(",host=").append(address); if (scanRanges == null) { sb.append("}"); continue; diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index 32c82ba0243b3a4..e091e1ff559b624 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -1166,6 +1166,7 @@ struct TExchangeNode { 3: optional i64 offset // Shuffle partition type 4: optional Partitions.TPartitionType partition_type + 5: optional Partitions.THashType hash_type } struct TOlapRewriteNode {