Add edit log of insert stmt

apache · May 23, 2019 · 7130352 · 7130352
1 parent 7f73f17
commit 7130352
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 12 deletions.
diff --git a/fe/src/main/java/org/apache/doris/catalog/Catalog.java b/fe/src/main/java/org/apache/doris/catalog/Catalog.java
@@ -1751,9 +1751,9 @@ public long loadRoutineLoadJobs(DataInputStream dis, long checksum) throws IOExc
     }
 
     public long loadLoadJobsV2(DataInputStream in, long checksum) throws IOException {
-//        if (Catalog.getCurrentCatalogJournalVersion() >= FeMetaVersion.VERSION_50) {
-//            Catalog.getCurrentCatalog().getLoadManager().readFields(in);
-//        }
+        if (Catalog.getCurrentCatalogJournalVersion() >= FeMetaVersion.VERSION_50) {
+            Catalog.getCurrentCatalog().getLoadManager().readFields(in);
+        }
         return checksum;
     }
 
@@ -2057,7 +2057,7 @@ public void replayGlobalVariable(SessionVariable variable) throws IOException, D
     }
 
     public long saveLoadJobsV2(DataOutputStream out, long checksum) throws IOException {
-//        Catalog.getCurrentCatalog().getLoadManager().write(out);
+        Catalog.getCurrentCatalog().getLoadManager().write(out);
         return checksum;
     }
 

diff --git a/fe/src/main/java/org/apache/doris/common/FeConstants.java b/fe/src/main/java/org/apache/doris/common/FeConstants.java
@@ -35,5 +35,5 @@ public class FeConstants {
 
     // general model
     // Current meta data version. Use this version to write journals and image
-    public static int meta_version = FeMetaVersion.VERSION_49;
+    public static int meta_version = FeMetaVersion.VERSION_50;
 }
diff --git a/fe/src/main/java/org/apache/doris/common/FeMetaVersion.java b/fe/src/main/java/org/apache/doris/common/FeMetaVersion.java
@@ -109,5 +109,5 @@ public final class FeMetaVersion {
     // routine load job
     public static final int VERSION_49 = 49;
     // load job v2 for broker load
-//    public static final int VERSION_50 = 50;
+    public static final int VERSION_50 = 50;
 }
diff --git a/fe/src/main/java/org/apache/doris/load/BrokerFileGroup.java b/fe/src/main/java/org/apache/doris/load/BrokerFileGroup.java
@@ -21,12 +21,14 @@
 import org.apache.doris.analysis.DataDescription;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.catalog.BrokerTable;
+import org.apache.doris.catalog.Catalog;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Partition;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 
@@ -57,7 +59,6 @@ public class BrokerFileGroup implements Writable {
     private String valueSeparator;
     private String lineDelimiter;
     // fileFormat may be null, which means format will be decided by file's suffix
-    // TODO(zc): we need to persist fileFormat, this should be done in next META_VERSION increase
     private String fileFormat;
     private boolean isNegative;
     private List<Long> partitionIds;
@@ -264,7 +265,13 @@ public void write(DataOutput out) throws IOException {
                 Expr.writeTo(entry.getValue(), out);
             }
         }
-        //
+        // fileFormat
+        if (fileFormat == null) {
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            Text.writeString(out, fileFormat);
+        }
     }
 
     @Override
@@ -312,6 +319,12 @@ public void readFields(DataInput in) throws IOException {
                 }
             }
         }
+        // file format
+        if (Catalog.getCurrentCatalogJournalVersion() >= FeMetaVersion.VERSION_50) {
+            if (in.readBoolean()) {
+                fileFormat = Text.readString(in);
+            }
+        }
     }
 
     public static BrokerFileGroup read(DataInput in) throws IOException {

diff --git a/fe/src/main/java/org/apache/doris/load/loadv2/InsertLoadJob.java b/fe/src/main/java/org/apache/doris/load/loadv2/InsertLoadJob.java
@@ -50,8 +50,7 @@ public InsertLoadJob() {
     }
 
     public InsertLoadJob(String label, long dbId, long tableId, long createTimestamp) {
-        this.label = label;
-        this.dbId = dbId;
+        super(dbId, label);
         this.tableId = tableId;
         this.createTimestamp = createTimestamp;
         this.loadStartTimestamp = createTimestamp;

diff --git a/fe/src/main/java/org/apache/doris/persist/EditLog.java b/fe/src/main/java/org/apache/doris/persist/EditLog.java
@@ -1195,10 +1195,10 @@ public void logRemoveRoutineLoadJob(RoutineLoadOperation operation) {
     }
 
     public void logCreateLoadJob(org.apache.doris.load.loadv2.LoadJob loadJob) {
-//        logEdit(OperationType.OP_CREATE_LOAD_JOB, loadJob);
+        logEdit(OperationType.OP_CREATE_LOAD_JOB, loadJob);
     }
 
     public void logEndLoadJob(LoadJobFinalOperation loadJobFinalOperation) {
-//        logEdit(OperationType.OP_END_LOAD_JOB, loadJobEndOperation);
+        logEdit(OperationType.OP_END_LOAD_JOB, loadJobFinalOperation);
     }
 }