This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 11902c1ca31 branch-4.2: [fix](load) Fix adaptive random bucket routing
for stream and routine load (#68006) (#69146)
11902c1ca31 is described below
commit 11902c1ca31c7a6251b77af00eab5421aea83523
Author: Xin Liao <[email protected]>
AuthorDate: Sun Oct 18 11:44:09 2026 +0800
branch-4.2: [fix](load) Fix adaptive random bucket routing for stream and
routine load (#68006) (#69146)
Pick apache/doris#68006 to branch-4.2.
Branch adaptation: KafkaTaskInfo is in the routineload package on 4.2;
Kinesis routine load is absent, so its source change is omitted. The FE
test uses JUnit 4, matching this branch.
Validation: whitespace check passed with the regression framework's
generated EOF blank line exempted. `DORIS_GCC_HOME=/usr ./run-fe-ut.sh
--run org.apache.doris.load.StreamLoadHandlerTest` was attempted but
generated-code build stopped because `thirdparty/installed/bin/protoc`
is missing from this worktree. Regression tests were not run.
Co-authored-by: Refrain <[email protected]>
---
.../org/apache/doris/load/StreamLoadHandler.java | 74 ++++++++++
.../doris/load/routineload/KafkaTaskInfo.java | 3 +
.../apache/doris/load/StreamLoadHandlerTest.java | 156 +++++++++++++++++++++
.../test_adaptive_random_bucket_routine_load.out | 23 +++
...test_adaptive_random_bucket_routine_load.groovy | 139 ++++++++++++++++++
.../test_adaptive_random_bucket_stream_load.groovy | 84 +++++++++++
6 files changed, 479 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
index 99f516f022e..40f72f309f7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
@@ -42,10 +42,14 @@ import
org.apache.doris.nereids.load.NereidsCloudStreamLoadPlanner;
import org.apache.doris.nereids.load.NereidsStreamLoadPlanner;
import org.apache.doris.nereids.load.NereidsStreamLoadTask;
import org.apache.doris.planner.GroupCommitPlanner;
+import org.apache.doris.planner.OlapTableSink;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TDataSinkType;
+import org.apache.doris.thrift.TOlapTablePartitionParam;
+import org.apache.doris.thrift.TOlapTableSink;
import org.apache.doris.thrift.TPipelineFragmentParams;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
@@ -61,6 +65,7 @@ import org.apache.logging.log4j.Logger;
import java.security.SecureRandom;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -286,6 +291,7 @@ public class StreamLoadHandler {
? multiTableFragmentInstanceIdIndex.getAndIncrement() : 0;
TPipelineFragmentParams result = null;
result = planner.plan(streamLoadTask.getId(), index);
+ assignAdaptiveRandomBucket(result);
result.setTableName(table.getName());
result.query_options.setFeProcessUuid(ExecuteEnv.getInstance().getProcessUUID());
result.setIsMowTable(table.getEnableUniqueKeyMergeOnWrite());
@@ -321,4 +327,72 @@ public class StreamLoadHandler {
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;
+ assignAdaptiveRandomBucket(params, sinkBackendId, request.getDb(),
request.getTbl());
+ }
+
+ public static void assignAdaptiveRandomBucket(TPipelineFragmentParams
params, long sinkBackendId,
+ String dbName, String tableName) {
+ TOlapTableSink sink = getOlapTableSink(params);
+ if (!OlapTableSink.shouldAssignAdaptiveRandomBucket(sink)) {
+ return;
+ }
+ 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. This is a normal compatibility fallback, not an
error.
+ LOG.info("disable adaptive random bucket, stream load sink backend
id is {}, db={}, table={}",
+ sinkBackendId, dbName, tableName);
+ 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);
+ Map<Long, OlapTableSink.AdaptiveBucketAssignment> partitionAssignments
= assignments.get(sinkBackendId);
+ if ((partitionAssignments == null || partitionAssignments.isEmpty())
+ && !isFakePartitionParam(sink.getPartition())) {
+ // Nothing could be assigned, so the sink would stay in adaptive
mode without any
+ // routing and the BE would guess the bucket owner again.
+ LOG.warn("disable adaptive random bucket, no partition could be
assigned for backend {}, "
+ + "db={}, table={}", sinkBackendId, dbName,
tableName);
+ sink.unsetEnableAdaptiveRandomBucket();
+ return;
+ }
+ OlapTableSink.applyAdaptiveRandomBucketAssignments(
+ sink.getPartition().getPartitions(), partitionAssignments);
+ }
+
+ /**
+ * Auto partition tables with no partition yet start with a fake partition
list that is only
+ * used for tablet locations; their real partitions are created and
assigned while the load
+ * runs, so an empty plan time assignment is expected and must keep
adaptive mode enabled.
+ */
+ private static boolean isFakePartitionParam(TOlapTablePartitionParam
partitionParam) {
+ return partitionParam.isSetPartitionsIsFake() &&
partitionParam.isPartitionsIsFake();
+ }
+
+ private static TOlapTableSink getOlapTableSink(TPipelineFragmentParams
params) {
+ if (params == null || !params.isSetFragment() || params.getFragment()
== null
+ || !params.getFragment().isSetOutputSink() ||
params.getFragment().getOutputSink() == null
+ || params.getFragment().getOutputSink().getType() !=
TDataSinkType.OLAP_TABLE_SINK) {
+ return null;
+ }
+ return params.getFragment().getOutputSink().getOlapTableSink();
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index 78d4312812f..375f96c5e98 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -25,6 +25,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.load.StreamLoadHandler;
import org.apache.doris.nereids.load.NereidsLoadTaskInfo;
import org.apache.doris.nereids.load.NereidsStreamLoadPlanner;
import org.apache.doris.qe.ConnectContext;
@@ -187,6 +188,8 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
(OlapTable)
db.getTableOrMetaException(routineLoadJob.getTableId(),
Table.TableType.OLAP), taskInfo);
TPipelineFragmentParams tExecPlanFragmentParams =
routineLoadJob.plan(planner, loadId, txnId);
+ StreamLoadHandler.assignAdaptiveRandomBucket(tExecPlanFragmentParams,
beId,
+ routineLoadJob.getDbFullName(), routineLoadJob.getTableName());
TPlanFragment tPlanFragment = tExecPlanFragmentParams.getFragment();
tPlanFragment.getOutputSink().getOlapTableSink().setTxnId(txnId);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
index 8f266e81ab8..116375ee747 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
@@ -29,9 +29,20 @@ import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TDataSink;
+import org.apache.doris.thrift.TDataSinkType;
+import org.apache.doris.thrift.TOlapTableIndexTablets;
+import org.apache.doris.thrift.TOlapTableLocationParam;
+import org.apache.doris.thrift.TOlapTablePartition;
+import org.apache.doris.thrift.TOlapTablePartitionParam;
+import org.apache.doris.thrift.TOlapTableSink;
+import org.apache.doris.thrift.TPipelineFragmentParams;
+import org.apache.doris.thrift.TPlanFragment;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
+import org.apache.doris.thrift.TTabletLocation;
+import com.google.common.collect.Lists;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.MockedStatic;
@@ -145,6 +156,151 @@ public class StreamLoadHandlerTest {
}
}
+ @Test
+ public void testAssignAdaptiveRandomBucketRoutesPartitionToTabletOwner() {
+ // The load runs on backend 20, but the only tablet of the partition
lives on backend 10,
+ // so the sink has to route the partition to backend 10 instead of
assuming self ownership.
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setBackendId(20L);
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TPipelineFragmentParams params =
buildStreamLoadParams(createSingleBucketSink());
+ handler.assignAdaptiveRandomBucket(params);
+
+ TOlapTablePartition partition = getSinglePartition(params);
+ Assert.assertEquals(10L, partition.getBucketBeId());
+ Assert.assertEquals(0L, partition.getLoadTabletIdx());
+ Assert.assertEquals(Lists.newArrayList(0),
partition.getLocalBucketSeqs());
+ Assert.assertEquals(10L,
partition.getIndexes().get(0).getBucketBeId());
+ Assert.assertEquals(Lists.newArrayList(0),
partition.getIndexes().get(0).getLocalBucketSeqs());
+ }
+
+ @Test
+ public void
testAssignAdaptiveRandomBucketKeepsLocalBucketWhenSinkOwnsTablet() {
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setBackendId(10L);
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TPipelineFragmentParams params =
buildStreamLoadParams(createSingleBucketSink());
+ handler.assignAdaptiveRandomBucket(params);
+
+ TOlapTablePartition partition = getSinglePartition(params);
+ Assert.assertEquals(10L, partition.getBucketBeId());
+ Assert.assertEquals(Lists.newArrayList(0),
partition.getLocalBucketSeqs());
+ }
+
+ @Test
+ public void testAssignAdaptiveRandomBucketFallsBackWithoutBackendId() {
+ // An old client does not report the executing backend, so no
assignment consistent with the
+ // receiver side can be computed. Adaptive mode must be turned off to
keep the legacy routing.
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setDb("test_db");
+ request.setTbl("test_tbl");
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TPipelineFragmentParams params =
buildStreamLoadParams(createSingleBucketSink());
+ handler.assignAdaptiveRandomBucket(params);
+
+ TOlapTableSink sink =
params.getFragment().getOutputSink().getOlapTableSink();
+ Assert.assertFalse(sink.isSetEnableAdaptiveRandomBucket());
+ TOlapTablePartition partition = getSinglePartition(params);
+ Assert.assertFalse(partition.isSetBucketBeId());
+ Assert.assertFalse(partition.getIndexes().get(0).isSetBucketBeId());
+ }
+
+ @Test
+ public void
testAssignAdaptiveRandomBucketFallsBackWhenNothingCanBeAssigned() {
+ // A partition the FE cannot assign (here: without a load tablet
index) would leave the sink
+ // in adaptive mode without any routing, so adaptive mode has to be
turned off.
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setBackendId(20L);
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TOlapTableSink sink = createSingleBucketSink();
+ sink.getPartition().getPartitions().get(0).unsetLoadTabletIdx();
+ TPipelineFragmentParams params = buildStreamLoadParams(sink);
+ handler.assignAdaptiveRandomBucket(params);
+
+ Assert.assertFalse(sink.isSetEnableAdaptiveRandomBucket());
+ }
+
+ @Test
+ public void testAssignAdaptiveRandomBucketKeepsAdaptiveForFakePartitions()
{
+ // Auto partition tables get their real partitions, and their
assignments, while the load
+ // runs, so an empty plan time assignment must keep adaptive mode
enabled.
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setBackendId(20L);
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TOlapTableSink sink = createSingleBucketSink();
+ sink.getPartition().setPartitionsIsFake(true);
+ sink.getPartition().getPartitions().get(0).unsetLoadTabletIdx();
+ TPipelineFragmentParams params = buildStreamLoadParams(sink);
+ handler.assignAdaptiveRandomBucket(params);
+
+ Assert.assertTrue(sink.isSetEnableAdaptiveRandomBucket());
+ Assert.assertTrue(sink.isEnableAdaptiveRandomBucket());
+ }
+
+ @Test
+ public void testAssignAdaptiveRandomBucketIgnoresNonOlapTableSink() {
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setBackendId(20L);
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+
+ TPipelineFragmentParams params = new TPipelineFragmentParams();
+ TPlanFragment fragment = new TPlanFragment();
+ fragment.setOutputSink(new TDataSink(TDataSinkType.DATA_STREAM_SINK));
+ params.setFragment(fragment);
+
+ handler.assignAdaptiveRandomBucket(params);
+
Assert.assertNull(params.getFragment().getOutputSink().getOlapTableSink());
+ }
+
+ /**
+ * Creates a random distribution sink with one bucket, whose only tablet
is located on backend 10.
+ */
+ private static TOlapTableSink createSingleBucketSink() {
+ TOlapTablePartition partition = new TOlapTablePartition();
+ partition.setId(1000L);
+ partition.setNumBuckets(1);
+ partition.setLoadTabletIdx(0);
+ partition.addToIndexes(new TOlapTableIndexTablets(1L,
Lists.newArrayList(100L)));
+ TOlapTablePartitionParam partitionParam = new
TOlapTablePartitionParam();
+ partitionParam.addToPartitions(partition);
+
+ TOlapTableLocationParam locationParam = new TOlapTableLocationParam();
+ locationParam.addToTablets(new TTabletLocation(100L,
Lists.newArrayList(10L)));
+
+ TOlapTableSink sink = new TOlapTableSink();
+ sink.setEnableAdaptiveRandomBucket(true);
+ sink.setLoadToSingleTablet(false);
+ sink.setPartition(partitionParam);
+ sink.setLocation(locationParam);
+ return sink;
+ }
+
+ private static TPipelineFragmentParams
buildStreamLoadParams(TOlapTableSink sink) {
+ TPipelineFragmentParams params = new TPipelineFragmentParams();
+ TPlanFragment fragment = new TPlanFragment();
+ TDataSink tDataSink = new TDataSink(TDataSinkType.OLAP_TABLE_SINK);
+ tDataSink.setOlapTableSink(sink);
+ fragment.setOutputSink(tDataSink);
+ params.setFragment(fragment);
+ return params;
+ }
+
+ private static TOlapTablePartition
getSinglePartition(TPipelineFragmentParams params) {
+ return params.getFragment().getOutputSink().getOlapTableSink()
+ .getPartition().getPartitions().get(0);
+ }
+
private Backend createBackend(long id, String host) {
Backend backend = new Backend(id, host, 9050);
backend.setAlive(true);
diff --git
a/regression-test/data/load_p0/routine_load/test_adaptive_random_bucket_routine_load.out
b/regression-test/data/load_p0/routine_load/test_adaptive_random_bucket_routine_load.out
new file mode 100644
index 00000000000..3bd9915d837
--- /dev/null
+++
b/regression-test/data/load_p0/routine_load/test_adaptive_random_bucket_routine_load.out
@@ -0,0 +1,23 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !first_batch --
+1 value_1
+2 value_2
+3 value_3
+4 value_4
+5 value_5
+6 value_6
+
+-- !second_batch --
+1 value_1
+10 value_10
+11 value_11
+12 value_12
+2 value_2
+3 value_3
+4 value_4
+5 value_5
+6 value_6
+7 value_7
+8 value_8
+9 value_9
+
diff --git
a/regression-test/suites/load_p0/routine_load/test_adaptive_random_bucket_routine_load.groovy
b/regression-test/suites/load_p0/routine_load/test_adaptive_random_bucket_routine_load.groovy
new file mode 100644
index 00000000000..7099ce34797
--- /dev/null
+++
b/regression-test/suites/load_p0/routine_load/test_adaptive_random_bucket_routine_load.groovy
@@ -0,0 +1,139 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.admin.AdminClient
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+
+suite("test_adaptive_random_bucket_routine_load", "docker") {
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+ def topic =
"test_adaptive_random_bucket_routine_load_${UUID.randomUUID()}".toString()
+ def job = "test_adaptive_random_bucket_routine_load_job"
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(2)
+ options.cloudMode = true
+ options.feConfigs += [
+ 'enable_adaptive_random_bucket_load=true',
+ 'max_routine_load_task_num_per_be=1'
+ ]
+
+ docker(options) {
+ sql "DROP TABLE IF EXISTS test_adaptive_random_bucket_routine_load"
+ // Two concurrent Kafka tasks must use different BEs. With only one
tablet,
+ // one task has to route to a BE other than the one executing the task.
+ sql """
+ CREATE TABLE test_adaptive_random_bucket_routine_load (
+ k INT NOT NULL,
+ v STRING
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1")
+ """
+
+ def adminProps = new Properties()
+ adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker)
+ def adminClient = AdminClient.create(adminProps)
+ try {
+ adminClient.createTopics([new NewTopic(topic, 2, (short)
1)]).all().get()
+ try {
+ def producer =
RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+ try {
+ def sendBatch = { int firstId ->
+ for (int partition = 0; partition < 2; partition++) {
+ for (int i = 0; i < 3; i++) {
+ int id = firstId + partition * 3 + i
+ producer.send(new ProducerRecord<>(topic,
partition, null,
+ "${id},value_${id}".toString())).get()
+ }
+ }
+ producer.flush()
+ }
+
+ sendBatch(1)
+ sql """
+ CREATE ROUTINE LOAD ${job} ON
test_adaptive_random_bucket_routine_load
+ COLUMNS TERMINATED BY ","
+ PROPERTIES (
+ "desired_concurrent_number" = "2",
+ "max_batch_interval" = "5",
+ "load_to_single_tablet" = "false"
+ )
+ FROM KAFKA (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${topic}",
+ "kafka_partitions" = "0,1",
+ "kafka_offsets" =
"OFFSET_BEGINNING,OFFSET_BEGINNING",
+ "property.enable.partition.eof" = "false"
+ )
+ """
+ try {
+ def checkJob = {
+ def state = sql_return_maparray "SHOW ROUTINE LOAD
FOR ${job}"
+ logger.info("Routine load status: ${state}")
+ assertTrue(state[0].State in ["NEED_SCHEDULE",
"RUNNING"],
+ "Routine load stopped unexpectedly:
${state}")
+ // Do not let a retry on the tablet owner hide a
routing failure.
+
assertFalse(state[0].OtherMsg.toString().contains("unknown partition channel"),
+ "Adaptive routing failed: ${state}")
+ }
+
+ // Disable early EOF above so both tasks stay active
for the batch interval,
+ // allowing their BE assignments to be observed before
they are renewed.
+ awaitUntil(60) {
+ checkJob()
+ def tasks = sql_return_maparray "SHOW ROUTINE LOAD
TASK WHERE JobName = '${job}'"
+ def backendIds = tasks.collect { it.BeId as long
}.findAll { it > 0 }.toSet()
+ backendIds.size() == 2
+ }
+
+ def waitForRows = { int expected ->
+ awaitUntil(120) {
+ checkJob()
+ def rows = sql "SELECT count(*) FROM
test_adaptive_random_bucket_routine_load"
+ rows[0][0] == expected
+ }
+ }
+ waitForRows(6)
+ order_qt_first_batch "SELECT k, v FROM
test_adaptive_random_bucket_routine_load"
+
+ // Renewed tasks must receive the adaptive assignments
as well.
+ sendBatch(7)
+ waitForRows(12)
+ order_qt_second_batch "SELECT k, v FROM
test_adaptive_random_bucket_routine_load"
+ } finally {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ }
+ } finally {
+ producer.close()
+ }
+ } finally {
+ adminClient.deleteTopics([topic]).all().get()
+ }
+ } finally {
+ adminClient.close()
+ }
+ }
+}
diff --git
a/regression-test/suites/load_p0/stream_load/test_adaptive_random_bucket_stream_load.groovy
b/regression-test/suites/load_p0/stream_load/test_adaptive_random_bucket_stream_load.groovy
new file mode 100644
index 00000000000..c07de6327fa
--- /dev/null
+++
b/regression-test/suites/load_p0/stream_load/test_adaptive_random_bucket_stream_load.groovy
@@ -0,0 +1,84 @@
+// 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 enableAdaptiveRandomBucketConfig =
+ sql """ ADMIN SHOW FRONTEND CONFIG LIKE
'enable_adaptive_random_bucket_load'; """
+ String oldEnableAdaptiveRandomBucket =
enableAdaptiveRandomBucketConfig[0][1]
+
+ try {
+ sql """ ADMIN SET FRONTEND CONFIG
('enable_adaptive_random_bucket_load' = 'true') """
+ sql """ DROP TABLE IF EXISTS test_adaptive_random_bucket_stream_load
"""
+ // With a single bucket the partition has exactly one tablet, so all
but one entry backend
+ // do not own it and have to route the partition to the owning backend.
+ sql """
+ CREATE TABLE test_adaptive_random_bucket_stream_load (
+ k int NOT NULL,
+ v string
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ )
+ """
+
+ Map<String, String> backendIps = [:]
+ Map<String, String> backendHttpPorts = [:]
+ getBackendIpHttpPort(backendIps, backendHttpPorts)
+ assertTrue(backendIps.size() > 1, "adaptive random bucket routing
needs at least two backends")
+
+ int rowsPerLoad = 10
+ int totalRows = 0
+ backendIps.each { backendId, backendIp ->
+ StringBuilder data = new StringBuilder()
+ for (int i = 0; i < rowsPerLoad; i++) {
+ data.append("${totalRows + i},value_${totalRows + i}\n")
+ }
+ totalRows += rowsPerLoad
+ // The entry backend is the sink backend. Only one of them owns
the single tablet, the
+ // others must send the partition to the owning backend instead of
failing with
+ // "unknown partition channel".
+ streamLoad {
+ table "test_adaptive_random_bucket_stream_load"
+ directToBe backendIp, backendHttpPorts.get(backendId) as int
+ set 'column_separator', ','
+ inputText data.toString()
+ time 60000
+
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def json = parseJson(result)
+
assertTrue(json.Status.toString().equalsIgnoreCase("success"), "load failed:
${result}")
+ assertEquals(rowsPerLoad,
json.NumberLoadedRows.toString().toInteger(), "load failed: ${result}")
+ }
+ }
+ sql "sync"
+ }
+
+ def count = sql "SELECT count(*) FROM
test_adaptive_random_bucket_stream_load"
+ assertEquals(totalRows, count[0][0] as int)
+ } finally {
+ sql """ ADMIN SET FRONTEND CONFIG
('enable_adaptive_random_bucket_load' = '${oldEnableAdaptiveRandomBucket}') """
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
