w41ter commented on code in PR #48014:
URL: https://github.com/apache/doris/pull/48014#discussion_r1967224197
##########
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java:
##########
@@ -1745,6 +1777,50 @@ public void
logColocateModifyRepliaAlloc(ColocatePersistInfo info) {
logEdit(OperationType.OP_COLOCATE_MOD_REPLICA_ALLOC, info);
}
+ public void logRestoreAndColocateAddTable(RestoreJob job,
List<ColocatePersistInfo> infos) {
+ long start = System.currentTimeMillis();
+
+ List<Pair<Short, Writable>> list = new ArrayList<>();
+ list.add(Pair.of(OperationType.OP_RESTORE_JOB, job));
+
+ for (ColocatePersistInfo info : infos) {
+ list.add(Pair.of(OperationType.OP_COLOCATE_ADD_TABLE, info));
+ }
+
+ try {
+ logEdit(list);
+ } catch (Exception e) {
+ LOG.warn("failed to log ColocateAddTable", e);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("log ColocateAddTable {} . cost: {} ms", infos.size(),
+ (System.currentTimeMillis() - start));
+ }
+ }
Review Comment:
And then instance a `JournalBatch` here instead of `List<Pair<Short,
Writeable>>`.
##########
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java:
##########
@@ -1745,6 +1777,50 @@ public void
logColocateModifyRepliaAlloc(ColocatePersistInfo info) {
logEdit(OperationType.OP_COLOCATE_MOD_REPLICA_ALLOC, info);
}
+ public void logRestoreAndColocateAddTable(RestoreJob job,
List<ColocatePersistInfo> infos) {
+ long start = System.currentTimeMillis();
+
+ List<Pair<Short, Writable>> list = new ArrayList<>();
+ list.add(Pair.of(OperationType.OP_RESTORE_JOB, job));
+
+ for (ColocatePersistInfo info : infos) {
+ list.add(Pair.of(OperationType.OP_COLOCATE_ADD_TABLE, info));
+ }
+
+ try {
+ logEdit(list);
+ } catch (Exception e) {
+ LOG.warn("failed to log ColocateAddTable", e);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("log ColocateAddTable {} . cost: {} ms", infos.size(),
+ (System.currentTimeMillis() - start));
+ }
+ }
+
+ public void logRestoreAndColocateRemoveTable(RestoreJob job,
List<ColocatePersistInfo> infos) {
+ long start = System.currentTimeMillis();
+
+ List<Pair<Short, Writable>> list = new ArrayList<>();
+ list.add(Pair.of(OperationType.OP_RESTORE_JOB, job));
+
+ for (ColocatePersistInfo info : infos) {
+ list.add(Pair.of(OperationType.OP_COLOCATE_REMOVE_TABLE, info));
+ }
+
+ try {
+ logEdit(list);
+ } catch (Exception e) {
+ LOG.warn("failed to log ColocateRemoveTable", e);
+ }
Review Comment:
The exception throws by `logEdit` is not recoverable, rethrow it.
##########
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java:
##########
@@ -1345,6 +1347,36 @@ private <T extends Writable> void logEdit(short op,
List<T> entries) throws IOEx
}
}
+ // NOTICE: No guarantee atomicity of entries
+ private <T extends Writable> void logEdit(List<Pair<Short, T>> entries)
throws IOException {
+ int itemNum = Math.max(1, Math.min(Config.batch_edit_log_max_item_num,
entries.size()));
+ JournalBatch batch = new JournalBatch(itemNum);
+ long batchCount = 0;
+ for (Pair<Short, T> entry : entries) {
+ if (batch.getJournalEntities().size() >=
Config.batch_edit_log_max_item_num
+ || batch.getSize() >= Config.batch_edit_log_max_byte_size)
{
+ journal.write(batch);
+ batch = new JournalBatch(itemNum);
Review Comment:
Please don't split the batch if you need to get an atomic guarantee.
You can change the signature of this method to
```
void logEdit(JournalBatch batch) throws IOException
```
--
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]
