utafrali commented on code in PR #68006:
URL: https://github.com/apache/doris/pull/68006#discussion_r4013192430
##########
fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java:
##########
@@ -321,4 +326,44 @@ public void generatePlan() throws UserException,
MetaNotFoundException {
public List getFragmentParams() {
return fragmentParams;
}
+
+ /**
+ * Stream load sends the whole plan to the BE that issued this request, so
the sink routing
+ * has to be resolved for exactly that BE. Without it the BE only knows
the sink is in
+ * adaptive random bucket mode, assumes it owns every bucket and fails with
+ * "unknown partition channel" as soon as a partition has no tablet on it.
+ */
+ void assignAdaptiveRandomBucket(TPipelineFragmentParams params) {
+ TOlapTableSink sink = getOlapTableSink(params);
+ if (!OlapTableSink.shouldAssignAdaptiveRandomBucket(sink)) {
+ return;
+ }
+ long sinkBackendId = request.isSetBackendId() ? request.getBackendId()
: -1L;
+ if (sinkBackendId <= 0 || !sink.isSetLocation() || sink.getLocation()
== null) {
+ // Old clients do not report the executing BE, and without the
sink backend id or the
+ // tablet locations no assignment consistent with the receiver
side can be computed
+ // here. Fall back to the non-adaptive per-batch routing, which
never depends on the
+ // bucket owner.
+ LOG.warn("disable adaptive random bucket, stream load sink backend
id is {}, db={}, table={}",
+ sinkBackendId, request.getDb(), request.getTbl());
+ sink.unsetEnableAdaptiveRandomBucket();
+ return;
+ }
+ int sinkInstanceNum = Math.max(params.getLocalParamsSize(), 1);
+ Map<Long, Map<Long, OlapTableSink.AdaptiveBucketAssignment>>
assignments =
+ OlapTableSink.computeAdaptiveRandomBucketAssignments(
+ Lists.newArrayList(sinkBackendId),
sink.getPartition().getPartitions(),
+ sink.getLocation().getTablets(), sinkInstanceNum);
+ OlapTableSink.applyAdaptiveRandomBucketAssignments(
Review Comment:
`assignments.get(sinkBackendId)` is always non-null here because
`computeAdaptiveRandomBucketAssignments` pre-populates the map for every ID in
the input list, and `applyAdaptiveRandomBucketAssignments` guards against a
null argument anyway. However, if `computeAdaptiveRandomBucketAssignments`
returns an empty inner map (all partitions skipped because
`!partition.isSetLoadTabletIdx()` or `numBuckets <= 0`),
`applyAdaptiveRandomBucketAssignments` exits early without touching any
partition, so the sink remains in adaptive mode with no assignments. The BE
would then fall into the same broken state the PR is trying to fix. A defensive
check here — if the returned inner map is empty, call
`sink.unsetEnableAdaptiveRandomBucket()` — would close that gap. In practice
`loadTabletIdx` is always set by `NereidsStreamLoadPlanner`, but making the
guard explicit costs one line and prevents silent regression if that assumption
ever changes.
##########
regression-test/suites/load_p0/stream_load/test_adaptive_random_bucket_stream_load.groovy:
##########
@@ -0,0 +1,85 @@
+// 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.
+
+suite("test_adaptive_random_bucket_stream_load", "p0,nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+
+ def tableName = "test_adaptive_random_bucket_stream_load"
Review Comment:
Per project test conventions, table names in simple regression tests should
be hardcoded rather than assigned via `def tableName`. Change `def tableName =
"test_adaptive_random_bucket_stream_load"` to just use the string literal
inline, or at a minimum drop the `def` indirection since the name is already a
constant literal that matches the suite name.
##########
fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java:
##########
@@ -321,4 +326,44 @@ public void generatePlan() throws UserException,
MetaNotFoundException {
public List getFragmentParams() {
return fragmentParams;
}
+
+ /**
+ * Stream load sends the whole plan to the BE that issued this request, so
the sink routing
+ * has to be resolved for exactly that BE. Without it the BE only knows
the sink is in
+ * adaptive random bucket mode, assumes it owns every bucket and fails with
+ * "unknown partition channel" as soon as a partition has no tablet on it.
+ */
+ void assignAdaptiveRandomBucket(TPipelineFragmentParams params) {
+ TOlapTableSink sink = getOlapTableSink(params);
+ if (!OlapTableSink.shouldAssignAdaptiveRandomBucket(sink)) {
+ return;
+ }
+ long sinkBackendId = request.isSetBackendId() ? request.getBackendId()
: -1L;
+ if (sinkBackendId <= 0 || !sink.isSetLocation() || sink.getLocation()
== null) {
+ // Old clients do not report the executing BE, and without the
sink backend id or the
+ // tablet locations no assignment consistent with the receiver
side can be computed
+ // here. Fall back to the non-adaptive per-batch routing, which
never depends on the
+ // bucket owner.
+ LOG.warn("disable adaptive random bucket, stream load sink backend
id is {}, db={}, table={}",
Review Comment:
This `LOG.warn` fires for every stream load from a client that does not set
`backendId` (any pre-fix client or any non-cloud deployment that somehow
reaches this path). In a busy cloud cluster with mixed client versions that
could produce a warn per request and quickly flood logs. Consider `LOG.info`
here, or guard it behind `LOG.isInfoEnabled()` similar to the pattern used in
`applyAdaptiveRandomBucketAssignments`. The condition itself is correct, but
warn-level is disproportionate for a normal compatibility fallback.
--
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]
