Re: [PR] [fix](group commit) support forward group commit stream load [doris]
gavinchou merged PR #63594: URL: https://github.com/apache/doris/pull/63594 -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
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 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
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
gavinchou commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4896127412 /review -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
github-actions[bot] commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4896016128 PR approved by at least one committer and no changes requested. -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4613538391 TPC-DS: Total hot run time: 168510 ms ``` machine: 'aliyun_ecs.c7a.8xlarge_32C64G' scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools TPC-DS sf100 test result on commit 85a108f1ec6f4269b1737193ec1e0024aa870735, data reload: false query5 4339623 470 470 query6 453 201 173 173 query7 4840588 323 323 query8 370 222 208 208 query9 8801401640084008 query10 461 315 250 250 query11 5874238721902190 query12 160 106 99 99 query13 1252596 445 445 query14 6433542650625062 query14_14396440944134409 query15 209 198 177 177 query16 1041460 468 460 query17 1099679 566 566 query18 2527463 337 337 query19 201 182 138 138 query20 114 107 106 106 query21 210 136 118 118 query22 13672 13552 13383 13383 query23 17370 16480 16111 16111 query23_116231 16378 16308 16308 query24 7520176713261326 query24_11311133613131313 query25 538 448 382 382 query26 1318323 164 164 query27 2652557 352 352 query28 4462202820042004 query29 1097601 471 471 query30 308 238 199 199 query31 11281077938 938 query32 108 63 67 63 query33 535 318 251 251 query34 11991104680 680 query35 756 777 676 676 query36 1391140412651265 query37 162 102 90 90 query38 3199314330543054 query39 945 909 904 904 query39_1881 879 872 872 query40 217 125 100 100 query41 66 62 62 62 query42 94 93 93 93 query43 326 327 277 277 query44 query45 190 189 176 176 query46 10931167744 744 query47 2385240322982298 query48 397 402 289 289 query49 627 468 359 359 query50 1078365 258 258 query51 4395432943304329 query52 91 88 77 77 query53 254 269 193 193 query54 273 221 195 195 query55 79 79 71 71 query56 233 223 223 223 query57 1447143413521352 query58 236 214 220 214 query59 1555161613651365 query60 273 232 234 232 query61 156 159 158 158 query62 693 653 588 588 query63 234 192 184 184 query64 2553793 627 627 query65 query66 1814453 339 339 query67 29145 29556 28934 28934 query68 query69 424 302 268 268 query70 989 953 955 953 query71 292 218 209 209 query72 2933267124052405 query73 850 778 423 423 query74 5185493147684768 query75 2669256522422242 query76 23121155761 761 query77 339 369 296 296 query78 12372 12316 11864 11864 query79 12491065730 730 query80 552 510 411 411 query81 452 294 247 247 query82 241 161 128 128 query83 276 287 258 258 query84 268 147 114 114 query85 899 618 525 525 query86 332 311 275 275 query87 3374336932223222 query88 3608277527532753 query89 428 383 332 332 query90 2146186 186 186 query91 194 188 150 150 query92 69 65 59 59 query93 14301466895 895 query94 565 372 336 336 query95 696 481 380 380 query96 1058820 359 359 query97 2723271225702570 query98 209 207 200 200 query99 1169117210351035 Total cold run time: 249991 ms Total hot run time: 168510 ms ``` -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4613429217 TPC-H: Total hot run time: 29093 ms ``` machine: 'aliyun_ecs.c7a.8xlarge_32C64G' scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools Tpch sf100 test result on commit 85a108f1ec6f4269b1737193ec1e0024aa870735, data reload: false -- Round 1 -- orders Doris NULLNULL0 0 0 NULL0 NULLNULL2023-12-26 18:27:23 2023-12-26 18:42:55 NULLutf-8 NULLNULL q1 17756 404839533953 q2 q3 10757 1428799 799 q4 4682470 349 349 q5 7594843 578 578 q6 184 173 139 139 q7 770 856 641 641 q8 9362158716751587 q9 5812454145364536 q10 6774183615321532 q11 431 271 255 255 q12 629 430 288 288 q13 18094 344727612761 q14 263 258 245 245 q15 q16 815 770 712 712 q17 972 961 917 917 q18 7135585354855485 q19 1533130610991099 q20 500 400 265 265 q21 6487288126172617 q22 467 420 335 335 Total cold run time: 101017 ms Total hot run time: 29093 ms - Round 2, with runtime_filter_mode=off - orders Doris NULLNULL15000 42 6422171781 NULL22778155NULLNULL2023-12-26 18:27:23 2023-12-26 18:42:55 NULLutf-8 NULLNULL q1 5002469747334697 q2 q3 4904530946754675 q4 2096219114021402 q5 4794490445944594 q6 232 174 127 127 q7 1855179515771577 q8 2398214120672067 q9 8026760673617361 q10 4694470642884288 q11 530 385 352 352 q12 730 747 525 525 q13 2963339627942794 q14 277 276 249 249 q15 q16 682 690 611 611 q17 1267125612401240 q18 7246678167776777 q19 1117112311021102 q20 2220223819361936 q21 5253454844474447 q22 531 453 398 398 Total cold run time: 56817 ms Total hot run time: 51219 ms ``` -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4613273261 # FE Regression Coverage Report Increment line coverage ` 3.92% (2/51)` :tada: [Increment coverage report](http://coverage.selectdb-in.cc/coverage/63594_85a108f1ec6f4269b1737193ec1e0024aa870735_merge_fe/increment_report/index.html) [Complete coverage report](http://coverage.selectdb-in.cc/coverage/63594_85a108f1ec6f4269b1737193ec1e0024aa870735_merge_fe/report/index.html) -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
mymeiyi commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4611227706 run buildall -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4533641750 # FE Regression Coverage Report Increment line coverage ` 3.92% (2/51)` :tada: [Increment coverage report](http://coverage.selectdb-in.cc/coverage/63594_e5b7daa5eb1cae8f2f154b483b24b41071d68bd7_merge_fe/increment_report/index.html) [Complete coverage report](http://coverage.selectdb-in.cc/coverage/63594_e5b7daa5eb1cae8f2f154b483b24b41071d68bd7_merge_fe/report/index.html) -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4533088395 TPC-DS: Total hot run time: 172430 ms ``` machine: 'aliyun_ecs.c7a.8xlarge_32C64G' scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools TPC-DS sf100 test result on commit e5b7daa5eb1cae8f2f154b483b24b41071d68bd7, data reload: false query5 4333681 534 534 query6 346 230 198 198 query7 4234554 305 305 query8 325 228 214 214 query9 8813414341394139 query10 457 349 296 296 query11 5817253622402240 query12 182 126 125 125 query13 1261602 431 431 query14 6188556852145214 query14_14522452445854524 query15 212 202 184 184 query16 996 470 421 421 query17 1131747 582 582 query18 2442476 351 351 query19 213 204 164 164 query20 140 127 127 127 query21 210 141 116 116 query22 13557 13584 13385 13385 query23 17329 16460 16339 16339 query23_116356 16477 16435 16435 query24 7619180013521352 query24_11331133913571339 query25 607 503 443 443 query26 1310323 177 177 query27 2685588 349 349 query28 4513204720532047 query29 988 667 514 514 query30 305 240 207 207 query31 11721083951 951 query32 86 82 80 80 query33 543 368 316 316 query34 11961167672 672 query35 782 798 709 709 query36 1450140312871287 query37 152 107 92 92 query38 3238320430693069 query39 946 902 916 902 query39_1901 865 895 865 query40 243 150 128 128 query41 71 69 69 69 query42 114 115 112 112 query43 352 344 299 299 query44 query45 213 210 204 204 query46 11021212737 737 query47 2387239922182218 query48 417 420 302 302 query49 656 523 418 418 query50 999 360 255 255 query51 4373437442744274 query52 109 110 97 97 query53 274 294 214 214 query54 340 298 284 284 query55 97 96 96 96 query56 334 313 309 309 query57 1432144113131313 query58 303 288 287 287 query59 1654171114991499 query60 353 339 320 320 query61 180 210 155 155 query62 692 640 589 589 query63 248 204 204 204 query64 2404829 647 647 query65 query66 1694487 356 356 query67 30282 30067 29921 29921 query68 query69 468 347 306 306 query70 1016101710121012 query71 307 274 269 269 query72 2997268724552455 query73 874 793 432 432 query74 5135495748284828 query75 2694259922892289 query76 22981160780 780 query77 407 413 331 331 query78 12457 12308 11900 11900 query79 14641012742 742 query80 645 520 440 440 query81 449 280 240 240 query82 1349163 124 124 query83 371 290 252 252 query84 266 149 113 113 query85 875 545 442 442 query86 397 325 297 297 query87 3436340632323232 query88 3671276327742763 query89 481 399 345 345 query90 2003187 189 187 query91 180 165 140 140 query92 75 82 76 76 query93 15111506804 804 query94 534 348 322 322 query95 686 479 361 361 query96 1039768 350 350 query97 2713276326042604 query98 238 229 229 229 query99 1153116310301030 Total cold run time: 255425 ms Total hot run time: 172430 ms ``` -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
hello-stephen commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4533028200 TPC-H: Total hot run time: 31418 ms ``` machine: 'aliyun_ecs.c7a.8xlarge_32C64G' scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools Tpch sf100 test result on commit e5b7daa5eb1cae8f2f154b483b24b41071d68bd7, data reload: false -- Round 1 -- orders Doris NULLNULL0 0 0 NULL0 NULLNULL2023-12-26 18:27:23 2023-12-26 18:42:55 NULLutf-8 NULLNULL q1 18004 412040094009 q2 q3 10794 1356837 837 q4 4690471 342 342 q5 7569224120782078 q6 370 178 141 141 q7 937 776 621 621 q8 9419180215801580 q9 7059495349334933 q10 6451220418791879 q11 429 271 244 244 q12 688 422 301 301 q13 18200 339427892789 q14 277 261 234 234 q15 q16 817 774 719 719 q17 1007948 880 880 q18 6950586356825682 q19 1188143310841084 q20 519 416 279 279 q21 6100283524722472 q22 446 376 314 314 Total cold run time: 101914 ms Total hot run time: 31418 ms - Round 2, with runtime_filter_mode=off - orders Doris NULLNULL15000 42 6422171781 NULL22778155NULLNULL2023-12-26 18:27:23 2023-12-26 18:42:55 NULLutf-8 NULLNULL q1 4785485350114853 q2 q3 4853535148394839 q4 2227227314641464 q5 5133488548674867 q6 237 185 132 132 q7 1894188516161616 q8 2439198319771977 q9 7517749874997498 q10 4754471442334233 q11 557 389 370 370 q12 740 780 550 550 q13 3012334027742774 q14 283 293 259 259 q15 q16 684 709 634 634 q17 1338130212991299 q18 7274694568846884 q19 1101108211311082 q20 2240223819671967 q21 5354466145294529 q22 518 487 398 398 Total cold run time: 56940 ms Total hot run time: 52225 ms ``` -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
mymeiyi commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4532703338 run buildall -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
mymeiyi commented on code in PR #63594:
URL: https://github.com/apache/doris/pull/63594#discussion_r3296908189
##
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##
@@ -2779,13 +2784,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 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);
+ok = true;
Review Comment:
StreamLoadHandler.generatePlan() does not call
GroupCommitManager.selectBackendForGroupCommit()
--
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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
github-actions[bot] commented on code in PR #63594:
URL: https://github.com/apache/doris/pull/63594#discussion_r3296749732
##
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##
@@ -2779,13 +2784,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 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);
+ok = true;
Review Comment:
This synchronous forward creates a nested FE RPC cycle for every forwarded
group-commit load: the master RPC worker blocks here waiting for the follower,
and the follower's `StreamLoadHandler.generatePlan()` calls
`GroupCommitManager.selectBackendForGroupCommit()`, whose non-master branch
calls back to the master via `MasterOpExecutor.getGroupCommitLoadBeId()`. With
enough concurrent group-commit stream loads, all master RPC workers can be
occupied waiting on followers while the followers are waiting for free master
RPC workers, causing the loads to stall until RPC timeout. The previous
follower-direct path only used one follower-to-master RPC and did not hold a
master worker while waiting for that callback. Please avoid the
master->follower->master cycle, or make the forwarded request carry the
already-selected group-commit backend/table routing decision so the follower
does not need to call back to the master.
--
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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
mymeiyi commented on PR #63594: URL: https://github.com/apache/doris/pull/63594#issuecomment-4532396614 /review -- 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]
Re: [PR] [fix](group commit) support forward group commit stream load [doris]
Copilot commented on code in PR #63594:
URL: https://github.com/apache/doris/pull/63594#discussion_r3296634988
##
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##
@@ -2779,13 +2784,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 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_schama_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 (DdlException e) {
+status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
+status.addToErrorMsgs(e.getMessage());
+return result;
Review Comment:
The pre-check block only catches `DdlException`, but other runtime
exceptions in this block (e.g. unexpected metadata issues) will propagate and
prevent the intended “forward if possible, otherwise fallback to local”
behavior. Consider catching a broader exception (at least `Exception`) here and
returning `null` so `streamLoadPut` can safely proceed locally.
##
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##
@@ -2779,13 +2784,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 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_schama_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;
Review Comment:
`db.getTableOr
