This is an automated email from the ASF dual-hosted git repository.
sollhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f5995f402b3 [fix](load) support compute_group in FE stream load
routing and planning (#65571)
f5995f402b3 is described below
commit f5995f402b322fe65ed59bd62a450275c9d1d28e
Author: hui lai <[email protected]>
AuthorDate: Tue Jul 21 10:20:36 2026 +0800
[fix](load) support compute_group in FE stream load routing and planning
(#65571)
### What problem does this PR solve?
PR #53031 added support for the `compute_group` header on the BE side,
but the FE Stream Load redirect path still only recognized the legacy
`cloud_cluster` header.
As a result, clients sending Stream Load requests through FE could not
use `compute_group` to select the target compute group.
There was also an ambiguity during planning: if a request was sent
directly to a BE in compute group A while specifying compute group B in
the header, propagating the header to the planning request could make
the receiving BE and the execution compute group inconsistent.
### What is changed?
This PR adds `compute_group` support to the FE Stream Load redirect path
and defines the planning behavior based on the receiving BE.
- Make FE recognize the `compute_group` header when selecting the Stream
Load redirect target.
- Give `compute_group` precedence over the legacy `cloud_cluster`
header.
- Keep `cloud_cluster` as a compatibility fallback.
- Make ordinary Stream Load pass the receiving BE's `backend_id` to FE.
- Resolve the planning compute group from the receiving BE's
`backend_id`.
- Override the planning request's `cloud_cluster` with the receiving
BE's compute group.
- Keep the existing fallback behavior when `backend_id` is unavailable.
HTTP Stream already passes the receiving BE's `backend_id`, so no
additional BE-side behavior change is required for HTTP Stream.
### Behavior
- Request through FE with `compute_group=B`:
- FE redirects the request to a BE in B.
- The receiving BE passes its `backend_id` to FE.
- The load is planned and executed in B.
- Request sent directly to a BE in A with `compute_group=B`:
- The receiving BE passes its own `backend_id` to FE.
- The load is planned and executed in A.
- The conflicting `compute_group` header does not change the planning
compute group.
### Tests
- Added FE unit tests for:
- `compute_group` header precedence during redirect.
- Fallback to the legacy `cloud_cluster` header.
- Resolving the planning compute group from `backend_id`.
- Added cloud regression coverage for:
- FE Stream Load redirect using `compute_group`.
- Direct-to-BE Stream Load with a conflicting `compute_group` header.
---
be/src/service/http/action/stream_load.cpp | 8 ++
.../org/apache/doris/httpv2/rest/LoadAction.java | 7 +-
.../org/apache/doris/load/StreamLoadHandler.java | 12 ++-
.../apache/doris/httpv2/rest/LoadActionTest.java | 31 +++++++
.../apache/doris/load/StreamLoadHandlerTest.java | 102 ++++++++++++++++++++
.../multi_cluster/stream_load/stream_load.out | 5 +
.../multi_cluster/stream_load/stream_load.groovy | 103 +++++++++++++++++++++
7 files changed, 265 insertions(+), 3 deletions(-)
diff --git a/be/src/service/http/action/stream_load.cpp
b/be/src/service/http/action/stream_load.cpp
index 3b312a6d153..92cdb5844fd 100644
--- a/be/src/service/http/action/stream_load.cpp
+++ b/be/src/service/http/action/stream_load.cpp
@@ -786,12 +786,20 @@ Status StreamLoadAction::_process_put(HttpRequest*
http_req,
request.__set_group_commit_mode(ctx->group_commit_mode);
}
+ // Keep cloud_cluster for compatibility with old FEs during rolling
upgrade. New FEs use
+ // backend_id below to bind planning to the compute group of the receiving
BE.
if (!http_req->header(HTTP_COMPUTE_GROUP).empty()) {
request.__set_cloud_cluster(http_req->header(HTTP_COMPUTE_GROUP));
} else if (!http_req->header(HTTP_CLOUD_CLUSTER).empty()) {
request.__set_cloud_cluster(http_req->header(HTTP_CLOUD_CLUSTER));
}
+ if (_exec_env->cluster_info()->backend_id != 0) {
+ request.__set_backend_id(_exec_env->cluster_info()->backend_id);
+ } else {
+ LOG(WARNING) << "_exec_env->cluster_info not set backend_id";
+ }
+
if (!http_req->header(HTTP_EMPTY_FIELD_AS_NULL).empty()) {
if (iequal(http_req->header(HTTP_EMPTY_FIELD_AS_NULL), "true")) {
request.__set_empty_field_as_null(true);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
index 6f522d3af72..12405ef558a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
@@ -367,7 +367,12 @@ public class LoadAction extends RestBaseController {
}
private String getCloudClusterName(HttpServletRequest request) {
- String cloudClusterName =
request.getHeader(SessionVariable.CLOUD_CLUSTER);
+ String cloudClusterName =
request.getHeader(SessionVariable.COMPUTE_GROUP);
+ if (!Strings.isNullOrEmpty(cloudClusterName)) {
+ return cloudClusterName;
+ }
+
+ cloudClusterName = request.getHeader(SessionVariable.CLOUD_CLUSTER);
if (!Strings.isNullOrEmpty(cloudClusterName)) {
return cloudClusterName;
}
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 e4971a96b7b..60249b8df51 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
@@ -137,11 +137,19 @@ public class StreamLoadHandler {
if (!request.isSetToken() && !request.isSetAuthCode() &&
!Strings.isNullOrEmpty(userName)) {
ctx.setCurrentUserIdentity(resolveCloudLoadUserIdentity(userName));
}
- if ((request.isSetToken() || request.isSetAuthCode()) &&
request.isSetBackendId()) {
+ if (request.isSetBackendId()) {
long backendId = request.getBackendId();
Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
Preconditions.checkNotNull(backend);
- ctx.setCloudCluster(backend.getCloudClusterName());
+ String computeGroup = backend.getCloudClusterName();
+ // Token/auth-code and user-less internal loads keep their
existing trusted path. Regular
+ // stream loads must still validate compute group privilege,
existence, and status.
+ if (request.isSetToken() || request.isSetAuthCode() ||
Strings.isNullOrEmpty(userName)) {
+ ctx.setCloudCluster(computeGroup);
+ } else {
+ ((CloudEnv)
Env.getCurrentEnv()).changeCloudCluster(computeGroup, ctx);
+ }
+ request.setCloudCluster(computeGroup);
return;
}
if (!Strings.isNullOrEmpty(request.getCloudCluster())) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java
index 2c8c59d413a..5226aa66a64 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.qe.SessionVariable;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TNetworkAddress;
@@ -253,6 +254,30 @@ public class LoadActionTest {
Assertions.assertTrue(headers.contains("label:load_label"));
}
+ @Test
+ public void testGetCloudClusterNamePrefersComputeGroupHeader() throws
Exception {
+ LoadAction loadAction = new LoadAction();
+ HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
+
Mockito.when(request.getHeader(SessionVariable.COMPUTE_GROUP)).thenReturn("compute_group_1");
+
Mockito.when(request.getHeader(SessionVariable.CLOUD_CLUSTER)).thenReturn("cloud_cluster_1");
+
+ String cloudClusterName = invokeGetCloudClusterName(loadAction,
request);
+
+ Assertions.assertEquals("compute_group_1", cloudClusterName);
+ }
+
+ @Test
+ public void testGetCloudClusterNameFallsBackToCloudClusterHeader() throws
Exception {
+ LoadAction loadAction = new LoadAction();
+ HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
+
Mockito.when(request.getHeader(SessionVariable.COMPUTE_GROUP)).thenReturn("");
+
Mockito.when(request.getHeader(SessionVariable.CLOUD_CLUSTER)).thenReturn("cloud_cluster_1");
+
+ String cloudClusterName = invokeGetCloudClusterName(loadAction,
request);
+
+ Assertions.assertEquals("cloud_cluster_1", cloudClusterName);
+ }
+
@Test
public void testExecuteWithoutPasswordRedirectsToBackend() throws
Exception {
Config.enable_debug_points = true;
@@ -453,6 +478,12 @@ public class LoadActionTest {
return (RedirectView) method.invoke(loadAction, request, addr,
forwardTarget);
}
+ private String invokeGetCloudClusterName(LoadAction loadAction,
HttpServletRequest request) throws Exception {
+ Method method =
LoadAction.class.getDeclaredMethod("getCloudClusterName",
HttpServletRequest.class);
+ method.setAccessible(true);
+ return (String) method.invoke(loadAction, request);
+ }
+
private TNetworkAddress invokeSelectEndpointByRedirectPolicy(LoadAction
loadAction, HttpServletRequest request,
Backend backend) throws Exception {
Method method =
LoadAction.class.getDeclaredMethod("selectEndpointByRedirectPolicy",
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 20ecd88538c..8f266e81ab8 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
@@ -17,14 +17,25 @@
package org.apache.doris.load;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.InternalCatalog;
+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.TStreamLoadPutRequest;
+import org.apache.doris.thrift.TStreamLoadPutResult;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.util.Arrays;
import java.util.List;
@@ -48,6 +59,92 @@ public class StreamLoadHandlerTest {
}
}
+ @Test
+ public void testSetCloudClusterUsesBackendComputeGroup() throws Exception {
+ SystemInfoService originalSystemInfoService =
Env.getCurrentSystemInfo();
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ Backend backend = createBackend(10001L, "127.0.0.1");
+ backend.setCloudClusterName("backend_compute_group");
+ CloudSystemInfoService systemInfoService =
+ new TestCloudSystemInfoService(Arrays.asList(backend));
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setUser("");
+ request.setBackendId(backend.getId());
+ request.setCloudCluster("header_compute_group");
+
+ try {
+ Config.cloud_unique_id = "test_cloud_unique_id";
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
systemInfoService);
+ ConnectContext.remove();
+
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+ handler.setCloudCluster();
+
+ Assert.assertEquals("backend_compute_group",
+
ConnectContext.get().getSessionVariable().getCloudCluster());
+ Assert.assertEquals("backend_compute_group",
request.getCloudCluster());
+ } finally {
+ ConnectContext.remove();
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
originalSystemInfoService);
+ }
+ }
+
+ @Test
+ public void testGroupCommitValidatesBackendComputeGroupPrivilege() throws
Exception {
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ Backend backend = createBackend(10001L, "127.0.0.1");
+ backend.setCloudClusterName("backend_compute_group");
+ CloudSystemInfoService systemInfoService =
+ new TestCloudSystemInfoService(Arrays.asList(backend));
+ CloudEnv cloudEnv = Mockito.mock(CloudEnv.class);
+ InternalCatalog internalCatalog = Mockito.mock(InternalCatalog.class);
+ Auth auth = Mockito.mock(Auth.class);
+
Mockito.when(cloudEnv.getInternalCatalog()).thenReturn(internalCatalog);
+
Mockito.when(internalCatalog.getName()).thenReturn(InternalCatalog.INTERNAL_CATALOG_NAME);
+ Mockito.when(cloudEnv.getAuth()).thenReturn(auth);
+ Mockito.doAnswer(invocation -> {
+ List<UserIdentity> currentUser = invocation.getArgument(3);
+
currentUser.add(UserIdentity.createAnalyzedUserIdentWithIp("test_user", "%"));
+ return null;
+ }).when(auth).checkPlainPassword(Mockito.eq("test_user"),
Mockito.anyString(),
+ Mockito.anyString(), Mockito.anyList());
+ Mockito.doThrow(new DdlException("USAGE denied"))
+ .when(cloudEnv).changeCloudCluster(
+ Mockito.eq("backend_compute_group"),
Mockito.any(ConnectContext.class));
+
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setUser("test_user");
+ request.setUserIp("127.0.0.1");
+ request.setPasswd("test_password");
+ request.setBackendId(backend.getId());
+ request.setCloudCluster("header_compute_group");
+ request.setGroupCommitMode("sync_mode");
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "test_cloud_unique_id";
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(cloudEnv);
+
mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+ ConnectContext.remove();
+
+ StreamLoadHandler handler = new StreamLoadHandler(
+ request, null, new TStreamLoadPutResult(), "127.0.0.1");
+ try {
+ handler.setCloudCluster();
+ Assert.fail("group commit should validate compute group
privilege");
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("USAGE denied"));
+ }
+
+ Mockito.verify(cloudEnv).changeCloudCluster(
+ Mockito.eq("backend_compute_group"),
Mockito.any(ConnectContext.class));
+ } finally {
+ ConnectContext.remove();
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+ }
+
private Backend createBackend(long id, String host) {
Backend backend = new Backend(id, host, 9050);
backend.setAlive(true);
@@ -65,5 +162,10 @@ public class StreamLoadHandlerTest {
public List<Backend> getBackendsByClusterName(final String
clusterName) {
return backends;
}
+
+ @Override
+ public Backend getBackend(long backendId) {
+ return backends.stream().filter(backend -> backend.getId() ==
backendId).findFirst().orElse(null);
+ }
}
}
diff --git
a/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
index adc5cddaaad..126e5359d67 100644
--- a/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
+++ b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
@@ -11,3 +11,8 @@
-- !all12 --
11
+-- !http_stream_compute_group --
+40
+
+-- !direct_be_compute_group --
+60
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
index 6e702b96c29..f2bec9bb352 100644
---
a/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
+++
b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
@@ -312,6 +312,109 @@ suite("stream_load") {
assertTrue(before_cluster0_load_rows < after_cluster0_load_rows)
assertTrue(before_cluster0_flush < after_cluster0_flush)
+ assertTrue(before_cluster1_load_rows == after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush == after_cluster1_flush)
+
+ // case4 FE redirects HTTP stream by compute_group and planning follows
the receiving backend
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0],
"load_rows");
+ log.info("before_cluster0_load_rows :
${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0],
"memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1],
"load_rows");
+ log.info("before_cluster1_load_rows :
${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1],
"memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ streamLoad {
+ set 'version', '1'
+ set 'sql', """
+ insert into ${context.dbName}.${tableName4}
+ select * from http_stream("format"="csv",
"column_separator"=",")
+ """
+ set 'compute_group', 'stream_load_cluster_name1'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("HTTP stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ order_qt_http_stream_compute_group "SELECT count(*) FROM ${tableName4}"
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0],
"load_rows");
+ log.info("after_cluster0_load_rows :
${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0],
"memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1],
"load_rows");
+ log.info("after_cluster1_load_rows :
${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1],
"memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows < after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+
+ // case5 direct-to-BE stream load ignores compute_group and uses the
receiving backend's compute group
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0],
"load_rows");
+ log.info("before_cluster0_load_rows :
${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0],
"memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1],
"load_rows");
+ log.info("before_cluster1_load_rows :
${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1],
"memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ streamLoad {
+ table "${tableName4}"
+ directToBe ipList[0], httpPortList[0].toInteger()
+
+ set 'column_separator', ','
+ set 'compute_group', 'stream_load_cluster_name1'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Direct-to-BE stream load result:
${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ order_qt_direct_be_compute_group "SELECT count(*) FROM ${tableName4}"
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0],
"load_rows");
+ log.info("after_cluster0_load_rows :
${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0],
"memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1],
"load_rows");
+ log.info("after_cluster1_load_rows :
${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1],
"memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows < after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush < after_cluster0_flush)
+
assertTrue(before_cluster1_load_rows == after_cluster1_load_rows)
assertTrue(before_cluster1_flush == after_cluster1_flush)
} finally {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
