github-actions[bot] commented on code in PR #63594:
URL: https://github.com/apache/doris/pull/63594#discussion_r3531184769
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -2792,13 +2797,86 @@ private void recordFinishedLoadJobRequestImpl(String
label, long txnId, String d
EtlJobType.INSERT, createTime, failMsg, trackingUrl,
firstErrorMsg, userIdentity, -1);
}
+ private static int nextGroupCommitFollowerIndex(int followerCount) {
+ return Math.floorMod(GROUP_COMMIT_FOLLOWER_INDEX.getAndIncrement(),
followerCount);
+ }
+
+ private TStreamLoadPutResult
forwardGroupCommitStreamLoad(TStreamLoadPutRequest request) {
+ HostInfo selfNode = Env.getCurrentEnv().getSelfNode();
+ List<Frontend> followers =
Env.getCurrentEnv().getFrontends(FrontendNodeType.FOLLOWER).stream()
+ .filter(fe -> fe.isAlive() &&
!(fe.getHost().equals(selfNode.getHost())
+ && fe.getEditLogPort() == selfNode.getPort())).collect(
+ Collectors.toList());
+ if (CollectionUtils.isEmpty(followers)) {
+ return null;
+ }
+
+ // check table enable light_schema_change and group commit does not
block for schema change
+ TStreamLoadPutResult result = new TStreamLoadPutResult();
+ TStatus status = new TStatus(TStatusCode.OK);
+ result.setStatus(status);
+ try {
+ Database db =
Env.getCurrentInternalCatalog().getDbOrDdlException(request.getDb());
+ OlapTable table = (OlapTable)
db.getTableOrDdlException(request.getTbl());
+ if (!table.getTableProperty().getUseSchemaLightChange()) {
+ status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
+ status.addToErrorMsgs(
+ "table light_schema_change is false, can't do stream
load with group commit mode");
+ return result;
+ }
+ if
(Env.getCurrentEnv().getGroupCommitManager().isBlock(table.getId())) {
+ String msg = "insert table " + table.getId() +
GroupCommitPlanner.SCHEMA_CHANGE;
+ LOG.info(msg);
+ status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
+ status.addToErrorMsgs(msg);
+ return result;
+ }
+ } catch (Exception e) {
+ LOG.warn("failed to pre-check group commit stream load, fallback
to local. db={}, tbl={}",
+ request.getDb(), request.getTbl(), e);
+ return null;
+ }
+
+ int idx = nextGroupCommitFollowerIndex(followers.size());
+ Frontend follower = followers.get(idx);
+ TNetworkAddress address = new TNetworkAddress(follower.getHost(),
follower.getRpcPort());
+ LOG.info("forward group commit stream load put to follower {}, db={},
tbl={}, groupCommitMode={}",
+ address, request.getDb(), request.getTbl(),
request.getGroupCommitMode());
+ FrontendService.Client client = null;
+ boolean ok = false;
+ try {
+ client = ClientPool.frontendPool.borrowObject(address);
+ TStreamLoadPutResult streamLoadPutResult =
client.streamLoadPut(request);
Review Comment:
The forward path makes planning depend on whichever follower is selected,
but this method only checks liveness before returning the follower's
`streamLoadPut` result. That follower resolves the DB/table from its own
catalog and fills plan defaults from its own static `Config` values, including
master-only mutable settings such as `be_exec_version` and stream-load memtable
defaults. In the window after the master commits a CREATE/ALTER/schema-change
unblock, or after an operator changes a master-only planner config, an alive
follower may not have replayed that state yet; the old master-local path would
plan with the current master metadata/config, while this path can return
NOT_FOUND/ANALYSIS_ERROR or a plan with stale schema/defaults. Please either
keep the planning on the master or ensure the selected follower has replayed
the required journal/config state before using its plan.
##########
regression-test/suites/load_p0/stream_load/test_group_commit_stream_load_multi_follower.groovy:
##########
@@ -0,0 +1,123 @@
+// 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.codehaus.groovy.runtime.IOGroovyMethods
+
+suite('test_group_commit_stream_load_multi_follower', 'docker') {
+ def databaseName = context.config.getDbNameByFile(context.file)
+ def tableName = "tbl"
+
+ def groupCommitStreamLoad = { fe ->
+ def feIp = fe.getHttpAddress()[0]
+ def fePort = fe.getHttpAddress()[1]
+ def command = """ curl -sS --location-trusted -u root:
+ -H group_commit:async_mode
+ -H column_separator:,
+ -H columns:id,name
+ -T
${context.config.dataPath}/load_p0/stream_load/test_stream_load1.csv
+
http://${feIp}:${fePort}/api/${databaseName}/${tableName}/_stream_load """
+ log.info("group commit command: ${command}")
+
+ def process = command.execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new
InputStreamReader(process.getErrorStream())))
+ def out = process.getText()
+ logger.info("load through fe {}:{} master={} code={}, out={}, err={}",
+ feIp, fePort, fe.isMaster, code, out, err)
+ assertEquals(code, 0)
+
+ def json = parseJson(out)
+ assertEquals("success", json.Status.toLowerCase())
+ assertTrue(json.GroupCommit)
+ assertTrue(json.Label.startsWith("group_commit_"))
+ assertEquals(2, json.NumberTotalRows)
+ assertEquals(2, json.NumberLoadedRows)
+ assertEquals(0, json.NumberFilteredRows)
+ assertEquals(0, json.NumberUnselectedRows)
+ assertTrue(json.ErrorURL == null || json.ErrorURL.isEmpty())
+ }
+
+ def getRowCount = { expectedRowCount ->
+ def retry = 0
+ while (retry < 30) {
+ sleep(1000)
+ def rowCount = sql "select count(*) from
${databaseName}.${tableName}"
+ logger.info("rowCount: " + rowCount + ", retry: " + retry)
+ if (rowCount[0][0] >= expectedRowCount) {
+ break
+ }
+ retry++
+ }
+ }
+
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.beNum = 1
+ options.cloudMode = true
+ options.useFollowersMode = true
+ options.beConfigs.add('enable_java_support=false')
+
options.feConfigs.add('enable_forward_group_commit_stream_load_to_follower=true')
Review Comment:
This test can still pass without exercising the new forwarding branch. The
production code catches follower-forwarding failures and falls back to local
master planning, and the test only checks that group-commit stream loads
eventually succeed and the final row count matches. If this config were ignored
or every follower RPC returned `null`, the existing master path would satisfy
the same assertions. Please add a deterministic signal that the master actually
called a follower, for example a debug point/counter on the follower
`streamLoadPut` path or a setup where local fallback cannot satisfy the test.
--
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]
