kangkaisen commented on a change in pull request #4677:
URL: https://github.com/apache/incubator-doris/pull/4677#discussion_r495564022
##########
File path: be/src/runtime/data_stream_sender.cpp
##########
@@ -479,7 +495,30 @@ Status DataStreamSender::send(RuntimeState* state,
RowBatch* batch) {
hash_val = RawValue::get_hash_value_fvn(
partition_val, ctx->root()->type(), hash_val);
}
- RETURN_IF_ERROR(_channels[hash_val % num_channels]->add_row(row));
+ auto target_channel_id = hash_val % num_channels;
+ RETURN_IF_ERROR(_channels[target_channel_id]->add_row(row));
+ }
+ } else if (_part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED) {
+ // hash-partition batch's rows across channels
+ int num_channels = _channel_shared_ptrs.size();
+
+ for (int i = 0; i < batch->num_rows(); ++i) {
+ TupleRow* row = batch->get_row(i);
+ size_t hash_val = 0;
+
+ for (auto ctx : _partition_expr_ctxs) {
+ void* partition_val = ctx->get_value(row);
+ // We can't use the crc hash function here because it does not
result
+ // in uncorrelated hashes with different seeds. Instead we
must use
+ // fvn hash.
+ // TODO: fix crc hash/GetHashValue()
+ //hash_val = RawValue::get_hash_value_fvn(
+ // partition_val, ctx->root()->type(), hash_val);
+ hash_val = RawValue::zlib_crc32(
Review comment:
Why comment is fvn hash but code is crc32 hash?
##########
File path:
fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
##########
@@ -498,6 +520,72 @@ private boolean canColocateJoin(HashJoinNode node,
PlanFragment leftChildFragmen
return false;
}
+ private boolean canBucketShuffleJoin(HashJoinNode node, PlanFragment
leftChildFragment,
+ List<Expr> rhsHashExprs) {
+ if
(!ConnectContext.get().getSessionVariable().isEnableBucketShuffleJoin()) {
+ return false;
+ }
+ // If user have a join hint to use proper way of join, can not be
colocate join
Review comment:
```suggestion
// If user have a join hint to use proper way of join, can not be
bucket shuffle join
```
##########
File path: fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
##########
@@ -783,13 +784,39 @@ private void computeFragmentExecParams() throws Exception
{
params.instanceExecParams.size() +
destParams.perExchNumSenders.get(exchId.asInt()));
}
- // add destination host to this fragment's destination
- for (int j = 0; j < destParams.instanceExecParams.size(); ++j) {
- TPlanFragmentDestination dest = new TPlanFragmentDestination();
- dest.fragment_instance_id =
destParams.instanceExecParams.get(j).instanceId;
- dest.server =
toRpcHost(destParams.instanceExecParams.get(j).host);
-
dest.setBrpcServer(toBrpcHost(destParams.instanceExecParams.get(j).host));
- params.destinations.add(dest);
+ if
(bucketShuffleJoinController.isBucketShuffleJoin(destFragment.getFragmentId().asInt()))
{
Review comment:
If you could abstract the query schedule strategy like `Presto` and
refactor this class, which would be very great.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
