database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](broker load) pass loadToSingleTablet to olapTableSink (#26680)

Browse Source
This commit is contained in:
qiye
2023-11-13 14:14:25 +08:00
committed by GitHub
parent 7e62c3c2de
commit 5a7c0ec9dc
2 changed files with 6 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
View File

@ -111,7 +111,7 @@ public class LoadLoadingTask extends LoadTask {
this.loadId = loadId;
planner = new LoadingTaskPlanner(callback.getCallbackId(), txnId, db.getId(), table, brokerDesc, fileGroups,
strictMode, isPartialUpdate, timezone, this.timeoutS, this.loadParallelism, this.sendBatchParallelism,
this.useNewLoadScanNode, userInfo);
this.useNewLoadScanNode, userInfo, singleTabletLoadPerSink);
planner.plan(loadId, fileStatusList, fileNum);
}

7
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadingTaskPlanner.java
View File

@ -71,6 +71,7 @@ public class LoadingTaskPlanner {
private final int loadParallelism;
private final int sendBatchParallelism;
private final boolean useNewLoadScanNode;
private final boolean singleTabletLoadPerSink;
private UserIdentity userInfo;
// Something useful
// ConnectContext here is just a dummy object to avoid some NPE problem, like ctx.getDatabase()
@ -86,7 +87,8 @@ public class LoadingTaskPlanner {
public LoadingTaskPlanner(Long loadJobId, long txnId, long dbId, OlapTable table,
BrokerDesc brokerDesc, List<BrokerFileGroup> brokerFileGroups,
boolean strictMode, boolean isPartialUpdate, String timezone, long timeoutS, int loadParallelism,
int sendBatchParallelism, boolean useNewLoadScanNode, UserIdentity userInfo) {
int sendBatchParallelism, boolean useNewLoadScanNode, UserIdentity userInfo,
boolean singleTabletLoadPerSink) {
this.loadJobId = loadJobId;
this.txnId = txnId;
this.dbId = dbId;
@ -100,6 +102,7 @@ public class LoadingTaskPlanner {
this.loadParallelism = loadParallelism;
this.sendBatchParallelism = sendBatchParallelism;
this.useNewLoadScanNode = useNewLoadScanNode;
this.singleTabletLoadPerSink = singleTabletLoadPerSink;
this.userInfo = userInfo;
if (Env.getCurrentEnv().getAccessManager()
.checkDbPriv(userInfo, Env.getCurrentInternalCatalog().getDbNullable(dbId).getFullName(),
@ -202,7 +205,7 @@ public class LoadingTaskPlanner {
List<Long> partitionIds = getAllPartitionIds();
OlapTableSink olapTableSink = new OlapTableSink(table, destTupleDesc, partitionIds,
Config.enable_single_replica_load);
olapTableSink.init(loadId, txnId, dbId, timeoutS, sendBatchParallelism, false, strictMode);
olapTableSink.init(loadId, txnId, dbId, timeoutS, sendBatchParallelism, singleTabletLoadPerSink, strictMode);
olapTableSink.setPartialUpdateInputColumns(isPartialUpdate, partialUpdateInputColumns);
olapTableSink.complete(analyzer);