Skip to content

Commit fdb6927

Browse files
meegoomeegoo
committed
[Feature] Support dynamic overwrite for insert overwrite
Signed-off-by: meegoo <meegoo.sr@gmail.com>
1 parent 3b3a67f commit fdb6927

24 files changed

+277
-38
lines changed

‎be/src/exec/tablet_sink.cpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,9 @@ Status OlapTableSink::init(const TDataSink& t_sink, RuntimeState* state) {
138138
_load_channel_timeout_s = config::streaming_load_rpc_max_alive_time_sec;
139139
}
140140

141+
if (table_sink.__isset.dynamic_overwrite) {
142+
_dynamic_overwrite = table_sink.dynamic_overwrite;
143+
}
141144
if (table_sink.__isset.ignore_out_of_partition) {
142145
_ignore_out_of_partition = table_sink.ignore_out_of_partition;
143146
}
@@ -175,6 +178,7 @@ void OlapTableSink::_prepare_profile(RuntimeState* state) {
175178
_profile->add_info_string("ReplicatedStorage", fmt::format("{}", _enable_replicated_storage));
176179
_profile->add_info_string("AutomaticPartition", fmt::format("{}", _enable_automatic_partition));
177180
_profile->add_info_string("AutomaticBucketSize", fmt::format("{}", _automatic_bucket_size));
181+
_profile->add_info_string("DynamicOverwrite", fmt::format("{}", _dynamic_overwrite));
178182

179183
_ts_profile = state->obj_pool()->add(new TabletSinkProfile());
180184
_ts_profile->runtime_profile = _profile;
@@ -407,6 +411,9 @@ Status OlapTableSink::_automatic_create_partition() {
407411
request.__set_db_id(_vectorized_partition->db_id());
408412
request.__set_table_id(_vectorized_partition->table_id());
409413
request.__set_partition_values(_partition_not_exist_row_values);
414+
if (_dynamic_overwrite) {
415+
request.__set_is_temp(true);
416+
}
410417

411418
LOG(INFO) << "load_id=" << print_id(_load_id) << ", txn_id: " << std::to_string(_txn_id)
412419
<< " automatic partition rpc begin request " << request;

‎be/src/exec/tablet_sink.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -244,6 +244,7 @@ class OlapTableSink : public AsyncDataSink {
244244
std::unique_ptr<ThreadPoolToken> _automatic_partition_token;
245245
std::vector<std::vector<std::string>> _partition_not_exist_row_values;
246246
bool _enable_automatic_partition = false;
247+
bool _dynamic_overwrite = false;
247248
bool _has_automatic_partition = false;
248249
std::atomic<bool> _is_automatic_partition_running = false;
249250
Status _automatic_partition_status;

‎fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2721,6 +2721,33 @@ public void dropTempPartition(String partitionName, boolean needDropTablet) {
27212721
}
27222722
}
27232723

2724+
public void replaceMatchPartitions(List<String> tempPartitionNames) {
2725+
for (String partitionName : tempPartitionNames) {
2726+
Partition partition = tempPartitions.getPartition(partitionName);
2727+
if (partition != null) {
2728+
String oldPartitionName = partitionName.substring(
2729+
partitionName.indexOf(AnalyzerUtils.PARTITION_NAME_PREFIX_SPLIT) + 1);
2730+
Partition oldPartition = nameToPartition.get(oldPartitionName);
2731+
if (oldPartition != null) {
2732+
// drop old partition
2733+
dropPartition(-1, oldPartitionName, true);
2734+
}
2735+
// add new partition
2736+
addPartition(partition);
2737+
// drop temp partition
2738+
tempPartitions.dropPartition(partitionName, false);
2739+
// move the range from idToTempRange to idToRange
2740+
partitionInfo.moveRangeFromTempToFormal(partition.getId());
2741+
// rename partition
2742+
renamePartition(partitionName, oldPartitionName);
2743+
}
2744+
}
2745+
2746+
for (Column column : getColumns()) {
2747+
IDictManager.getInstance().removeGlobalDict(this.getId(), column.getColumnId());
2748+
}
2749+
}
2750+
27242751
/*
27252752
* replace partitions in 'partitionNames' with partitions in
27262753
* 'tempPartitionNames'.

‎fe/fe-core/src/main/java/com/starrocks/load/InsertOverwriteJob.java‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,27 +45,35 @@ public class InsertOverwriteJob {
4545
@SerializedName(value = "sourcePartitionNames")
4646
private List<String> sourcePartitionNames;
4747

48+
@SerializedName(value = "dynamicOverwrite")
49+
private boolean dynamicOverwrite = false;
4850

4951
private transient InsertStmt insertStmt;
5052

53+
public InsertOverwriteJob() {
54+
}
55+
5156
public InsertOverwriteJob(long jobId, InsertStmt insertStmt, long targetDbId,
52-
long targetTableId, long warehouseId) {
57+
long targetTableId, long warehouseId, boolean dynamicOverwrite) {
5358
this.jobId = jobId;
5459
this.insertStmt = insertStmt;
5560
this.sourcePartitionIds = insertStmt.getTargetPartitionIds();
5661
this.jobState = InsertOverwriteJobState.OVERWRITE_PENDING;
5762
this.targetDbId = targetDbId;
5863
this.targetTableId = targetTableId;
5964
this.warehouseId = warehouseId;
65+
this.dynamicOverwrite = dynamicOverwrite;
6066
}
6167

6268
// used to replay InsertOverwriteJob
63-
public InsertOverwriteJob(long jobId, long targetDbId, long targetTableId, List<Long> sourcePartitionIds) {
69+
public InsertOverwriteJob(long jobId, long targetDbId, long targetTableId,
70+
List<Long> sourcePartitionIds, boolean dynamicOverwrite) {
6471
this.jobId = jobId;
6572
this.targetDbId = targetDbId;
6673
this.targetTableId = targetTableId;
6774
this.sourcePartitionIds = sourcePartitionIds;
6875
this.jobState = InsertOverwriteJobState.OVERWRITE_PENDING;
76+
this.dynamicOverwrite = dynamicOverwrite;
6977
}
7078

7179
public long getJobId() {
@@ -128,4 +136,8 @@ public InsertStmt getInsertStmt() {
128136
public long getWarehouseId() {
129137
return warehouseId;
130138
}
139+
140+
public boolean isDynamicOverwrite() {
141+
return dynamicOverwrite;
142+
}
131143
}

‎fe/fe-core/src/main/java/com/starrocks/load/InsertOverwriteJobMgr.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ public boolean hasRunningOverwriteJob(long tableId) {
149149

150150
public void replayCreateInsertOverwrite(CreateInsertOverwriteJobLog jobInfo) {
151151
InsertOverwriteJob insertOverwriteJob = new InsertOverwriteJob(jobInfo.getJobId(),
152-
jobInfo.getDbId(), jobInfo.getTableId(), jobInfo.getTargetPartitionIds());
152+
jobInfo.getDbId(), jobInfo.getTableId(), jobInfo.getTargetPartitionIds(), jobInfo.isDynamicOverwrite());
153153
boolean registered = registerOverwriteJob(insertOverwriteJob);
154154
if (!registered) {
155155
LOG.warn("register insert overwrite job failed. jobId:{}", insertOverwriteJob.getJobId());

‎fe/fe-core/src/main/java/com/starrocks/load/InsertOverwriteJobRunner.java‎

Lines changed: 40 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,9 @@ private void createPartitionByValue(InsertStmt insertStmt) {
260260
if (insertStmt.getTargetPartitionNames() == null) {
261261
return;
262262
}
263+
if (insertStmt.isDynamicOverwrite()) {
264+
return;
265+
}
263266
OlapTable olapTable = (OlapTable) insertStmt.getTargetTable();
264267
List<List<String>> partitionValues = Lists.newArrayList();
265268
if (!olapTable.getPartitionInfo().isAutomaticPartition()) {
@@ -305,7 +308,7 @@ private void createPartitionByValue(InsertStmt insertStmt) {
305308
}
306309

307310
try {
308-
addPartitionClause = AnalyzerUtils.getAddPartitionClauseFromPartitionValues(olapTable, partitionValues);
311+
addPartitionClause = AnalyzerUtils.getAddPartitionClauseFromPartitionValues(olapTable, partitionValues, false, null);
309312
} catch (AnalysisException ex) {
310313
LOG.warn(ex.getMessage(), ex);
311314
throw new RuntimeException(ex);
@@ -363,6 +366,9 @@ private void executeInsert() throws Exception {
363366
}
364367

365368
private void createTempPartitions() throws DdlException {
369+
if (job.isDynamicOverwrite()) {
370+
return;
371+
}
366372
long createPartitionStartTimestamp = System.currentTimeMillis();
367373
Database db = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb(dbId);
368374
if (db == null) {
@@ -421,6 +427,24 @@ private void gc(boolean isReplay) {
421427
}
422428
}
423429
}
430+
// if dynamic overwrite, drop all runtime created partitions
431+
if (job.isDynamicOverwrite()) {
432+
TransactionState txnState = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr()
433+
.getTransactionState(dbId, insertStmt.getTxnId());
434+
if (txnState == null) {
435+
throw new DmlException("transaction state is null for dbId:%s, txnId:%s", dbId, insertStmt.getTxnId());
436+
}
437+
List<String> tmpPartitionNames = txnState.getCreatedPartitionNames();
438+
for (String partitionName : tmpPartitionNames) {
439+
Partition partition = targetTable.getPartition(partitionName, true);
440+
if (partition != null) {
441+
for (MaterializedIndex index : partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
442+
sourceTablets.addAll(index.getTablets());
443+
}
444+
targetTable.dropTempPartition(partitionName, true);
445+
}
446+
}
447+
}
424448
if (!isReplay) {
425449
// mark all source tablet ids force delete to drop it directly on BE,
426450
// not to move it to trash
@@ -486,7 +510,17 @@ private void doCommit(boolean isReplay) {
486510

487511
PartitionInfo partitionInfo = targetTable.getPartitionInfo();
488512
if (partitionInfo.isRangePartition() || partitionInfo.getType() == PartitionType.LIST) {
489-
targetTable.replaceTempPartitions(sourcePartitionNames, tmpPartitionNames, true, false);
513+
if (job.isDynamicOverwrite()) {
514+
TransactionState txnState = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr()
515+
.getTransactionState(dbId, insertStmt.getTxnId());
516+
if (txnState == null) {
517+
throw new DmlException("transaction state is null for dbId:%s, txnId:%s", dbId, insertStmt.getTxnId());
518+
}
519+
tmpPartitionNames = txnState.getCreatedPartitionNames();
520+
targetTable.replaceMatchPartitions(tmpPartitionNames);
521+
} else {
522+
targetTable.replaceTempPartitions(sourcePartitionNames, tmpPartitionNames, true, false);
523+
}
490524
} else if (partitionInfo instanceof SinglePartitionInfo) {
491525
targetTable.replacePartition(sourcePartitionNames.get(0), tmpPartitionNames.get(0));
492526
} else {
@@ -559,8 +593,10 @@ private void prepareInsert() {
559593
if (insertStmt.getTargetPartitionNames() == null) {
560594
insertStmt.setPartitionNotSpecifiedInOverwrite(true);
561595
}
562-
insertStmt.setTargetPartitionNames(partitionNames);
563-
insertStmt.setTargetPartitionIds(job.getTmpPartitionIds());
596+
if (!job.isDynamicOverwrite()) {
597+
insertStmt.setTargetPartitionNames(partitionNames);
598+
insertStmt.setTargetPartitionIds(job.getTmpPartitionIds());
599+
}
564600
insertStmt.setOverwrite(false);
565601
insertStmt.setSystem(true);
566602

‎fe/fe-core/src/main/java/com/starrocks/persist/CreateInsertOverwriteJobLog.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,14 @@ public class CreateInsertOverwriteJobLog implements Writable {
3737
@SerializedName(value = "targetPartitionIds")
3838
private List<Long> targetPartitionIds;
3939

40-
public CreateInsertOverwriteJobLog(long jobId, long dbId, long tableId, List<Long> targetPartitionIds) {
40+
@SerializedName(value = "dynamicOverwrite")
41+
private boolean dynamicOverwrite = false;
42+
43+
public CreateInsertOverwriteJobLog() {
44+
}
45+
46+
public CreateInsertOverwriteJobLog(long jobId, long dbId, long tableId,
47+
List<Long> targetPartitionIds, boolean dynamicOverwrite) {
4148
this.jobId = jobId;
4249
this.dbId = dbId;
4350
this.tableId = tableId;
@@ -60,6 +67,10 @@ public List<Long> getTargetPartitionIds() {
6067
return targetPartitionIds;
6168
}
6269

70+
public boolean isDynamicOverwrite() {
71+
return dynamicOverwrite;
72+
}
73+
6374
@Override
6475
public String toString() {
6576
return "CreateInsertOverwriteJobInfo{" +

‎fe/fe-core/src/main/java/com/starrocks/planner/OlapTableSink.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,7 @@ public class OlapTableSink extends DataSink {
143143
private TPartialUpdateMode partialUpdateMode;
144144
private long warehouseId = WarehouseManager.DEFAULT_WAREHOUSE_ID;
145145
private long automaticBucketSize = 0;
146+
private boolean enableDynamicOverwrite = false;
146147

147148
public OlapTableSink(OlapTable dstTable, TupleDescriptor tupleDescriptor, List<Long> partitionIds,
148149
TWriteQuorumType writeQuorum, boolean enableReplicatedStorage,
@@ -229,6 +230,10 @@ public void setPartialUpdateMode(TPartialUpdateMode mode) {
229230
this.partialUpdateMode = mode;
230231
}
231232

233+
public void setDynamicOverwrite(boolean enableDynamicOverwrite) {
234+
this.enableDynamicOverwrite = enableDynamicOverwrite;
235+
}
236+
232237
public void complete(String mergeCondition) throws UserException {
233238
TOlapTableSink tSink = tDataSink.getOlap_table_sink();
234239
if (mergeCondition != null && !mergeCondition.isEmpty()) {
@@ -238,6 +243,10 @@ public void complete(String mergeCondition) throws UserException {
238243
}
239244

240245
public List<Long> getOpenPartitions() {
246+
if (enableAutomaticPartition && enableDynamicOverwrite) {
247+
return new ArrayList<>(Collections.singletonList(
248+
dstTable.getPartition(ExpressionRangePartitionInfo.AUTOMATIC_SHADOW_PARTITION_NAME).getId()));
249+
}
241250
if (!enableAutomaticPartition || Config.max_load_initial_open_partition_number <= 0
242251
|| partitionIds.size() < Config.max_load_initial_open_partition_number) {
243252
return partitionIds;
@@ -267,6 +276,9 @@ public void complete() throws UserException {
267276
long partitionId = optionalPartition.get().getId();
268277
numReplicas = dstTable.getPartitionInfo().getReplicationNum(partitionId);
269278
}
279+
if (enableAutomaticPartition && enableDynamicOverwrite) {
280+
tSink.setDynamic_overwrite(true);
281+
}
270282
tSink.setNum_replicas(numReplicas);
271283
tSink.setNeed_gen_rollup(dstTable.shouldLoadToNewRollup());
272284
tSink.setSchema(createSchema(tSink.getDb_id(), dstTable, tupleDescriptor));

‎fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -191,6 +191,7 @@ public class SessionVariable implements Serializable, Writable, Cloneable {
191191
public static final String MAX_PARALLEL_SCAN_INSTANCE_NUM = "max_parallel_scan_instance_num";
192192
public static final String ENABLE_INSERT_STRICT = "enable_insert_strict";
193193
public static final String INSERT_MAX_FILTER_RATIO = "insert_max_filter_ratio";
194+
public static final String DYNAMIC_OVERWRITE = "dynamic_overwrite";
194195
public static final String ENABLE_SPILL = "enable_spill";
195196
public static final String ENABLE_SPILL_TO_REMOTE_STORAGE = "enable_spill_to_remote_storage";
196197
public static final String DISABLE_SPILL_TO_LOCAL_DISK = "disable_spill_to_local_disk";
@@ -1151,6 +1152,9 @@ public static MaterializedViewRewriteMode parse(String str) {
11511152
@VariableMgr.VarAttr(name = ENABLE_INSERT_STRICT)
11521153
private boolean enableInsertStrict = true;
11531154

1155+
@VariableMgr.VarAttr(name = DYNAMIC_OVERWRITE)
1156+
private boolean dynamicOverwrite = false;
1157+
11541158
@VariableMgr.VarAttr(name = INSERT_MAX_FILTER_RATIO)
11551159
private double insertMaxFilterRatio = 0;
11561160

@@ -2817,6 +2821,14 @@ public void setEnableInsertStrict(boolean enableInsertStrict) {
28172821
this.enableInsertStrict = enableInsertStrict;
28182822
}
28192823

2824+
public boolean isDynamicOverwrite() {
2825+
return dynamicOverwrite;
2826+
}
2827+
2828+
public void setDynamicOverwrite(boolean dynamicOverwrite) {
2829+
this.dynamicOverwrite = dynamicOverwrite;
2830+
}
2831+
28202832
public double getInsertMaxFilterRatio() {
28212833
return insertMaxFilterRatio;
28222834
}

‎fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2083,14 +2083,16 @@ public void handleInsertOverwrite(InsertStmt insertStmt) throws Exception {
20832083
}
20842084
OlapTable olapTable = (OlapTable) insertStmt.getTargetTable();
20852085
InsertOverwriteJob job = new InsertOverwriteJob(GlobalStateMgr.getCurrentState().getNextId(),
2086-
insertStmt, db.getId(), olapTable.getId(), context.getCurrentWarehouseId());
2086+
insertStmt, db.getId(), olapTable.getId(), context.getCurrentWarehouseId(),
2087+
insertStmt.isDynamicOverwrite());
20872088
if (!locker.lockDatabaseAndCheckExist(db, LockType.WRITE)) {
20882089
throw new DmlException("database:%s does not exist.", db.getFullName());
20892090
}
20902091
try {
20912092
// add an edit log
20922093
CreateInsertOverwriteJobLog info = new CreateInsertOverwriteJobLog(job.getJobId(),
2093-
job.getTargetDbId(), job.getTargetTableId(), job.getSourcePartitionIds());
2094+
job.getTargetDbId(), job.getTargetTableId(), job.getSourcePartitionIds(),
2095+
job.isDynamicOverwrite());
20942096
GlobalStateMgr.getCurrentState().getEditLog().logCreateInsertOverwrite(info);
20952097
} finally {
20962098
locker.unLockDatabase(db.getId(), LockType.WRITE);

0 commit comments

Comments
 (0)