[improvement](meta) serialize meta via gson part3

Switch meta serialization to gson
Contains the following classes：

Database
SmallFileMgr$SmallFile
catalog.Table
loadv2.LoadJob
loadv2.LoadJobFinalOperation

Co-authored-by: iszhangpch

fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java

@@ -92,8 +92,10 @@ public final class FeMetaVersion {
  public static final int VERSION_134 = 134;
  // For mate gson
  public static final int VERSION_135 = 135;
+ // For mate gson
+ public static final int VERSION_136 = 136;
  // note: when increment meta version, should assign the latest version to VERSION_CURRENT
- public static final int VERSION_CURRENT = VERSION_135;
+ public static final int VERSION_CURRENT = VERSION_136;
 
  // all logs meta version should >= the minimum version, so that we could remove many if clause, for example
  // if (FE_METAVERSION < VERSION_94) ...

fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java

@@ -28,6 +28,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
 import com.google.gson.Gson;
+import com.google.gson.annotations.SerializedName;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -68,16 +69,24 @@ public class AggregateFunction extends Function {
  public static ImmutableSet<String> SUPPORT_ORDER_BY_AGGREGATE_FUNCTION_NAME_SET = ImmutableSet.of("group_concat");
 
  // Set if different from retType_, null otherwise.
+ @SerializedName("it")
  private Type intermediateType;
 
  // The symbol inside the binary at location_ that contains this particular.
  // They can be null if it is not required.
+ @SerializedName("ufs")
  private String updateFnSymbol;
+ @SerializedName("ifs")
  private String initFnSymbol;
+ @SerializedName("sfs")
  private String serializeFnSymbol;
+ @SerializedName("mfs")
  private String mergeFnSymbol;
+ @SerializedName("gvfs")
  private String getValueFnSymbol;
+ @SerializedName("rfs")
  private String removeFnSymbol;
+ @SerializedName("ffs")
  private String finalizeFnSymbol;
 
  private static String BE_BUILTINS_CLASS = "AggregateFunctions";
@@ -86,26 +95,31 @@ public class AggregateFunction extends Function {
  // e.g. min(distinct col) == min(col).
  // TODO: currently it is not possible for user functions to specify this. We should
  // extend the create aggregate function stmt to allow additional metadata like this.
+ @SerializedName("igd")
  private boolean ignoresDistinct;
 
  // True if this function can appear within an analytic expr (fn() OVER(...)).
  // TODO: Instead of manually setting this flag for all builtin aggregate functions
  // we should identify this property from the function itself (e.g., based on which
  // functions of the UDA API are implemented).
  // Currently, there is no reliable way of doing that.
+ @SerializedName("isAn")
  private boolean isAnalyticFn;
 
  // True if this function can be used for aggregation (without an OVER() clause).
+ @SerializedName("isAg")
  private boolean isAggregateFn;
 
  // True if this function returns a non-null value on an empty input. It is used
  // primarily during the equal of scalar subqueries.
  // TODO: Instead of manually setting this flag, we should identify this
  // property from the function itself (e.g. evaluating the function on an
  // empty input in BE).
+ @SerializedName("rnno")
  private boolean returnsNonNullOnEmpty;
 
  // use for java-udaf to point the class of user define
+ @SerializedName("sn")
  private String symbolName;
 
  // only used for serialization

fe/fe-core/src/main/java/org/apache/doris/catalog/AliasFunction.java

@@ -35,6 +35,7 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.gson.Gson;
+import com.google.gson.annotations.SerializedName;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -59,7 +60,9 @@ public class AliasFunction extends Function {
 
  private static final String DIGITAL_MASKING = "digital_masking";
 
+ @SerializedName("of")
  private Expr originFunction;
+ @SerializedName("pm")
  private List<String> parameters = new ArrayList<>();
  private List<String> typeDefParams = new ArrayList<>();
 

fe/fe-core/src/main/java/org/apache/doris/catalog/AuthorizationInfo.java

@@ -22,6 +22,7 @@
 
 import com.google.common.collect.Sets;
 import com.google.gson.Gson;
+import com.google.gson.annotations.SerializedName;
 
 import java.io.DataInput;
 import java.io.DataOutput;
@@ -39,7 +40,9 @@
  * The job checks the privilege by auth info.
  */
 public class AuthorizationInfo implements Writable {
+ @SerializedName(value = "dn")
  private String dbName;
+ @SerializedName(value = "tn")
  private Set<String> tableNameList;
 
  // only for persist

fe/fe-core/src/main/java/org/apache/doris/catalog/BrokerTable.java

@@ -26,12 +26,12 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.gson.annotations.SerializedName;
 import org.apache.commons.lang3.StringEscapeUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.net.URLDecoder;
@@ -46,11 +46,16 @@ public class BrokerTable extends Table {
  private static final String COLUMN_SEPARATOR = "column_separator";
  private static final String LINE_DELIMITER = "line_delimiter";
  private static final String FILE_FORMAT = "format";
+ @SerializedName("bn")
  private String brokerName;
+ @SerializedName("ps")
  private List<String> paths;
+ @SerializedName("cs")
  private String columnSeparator;
+ @SerializedName("ld")
  private String lineDelimiter;
  private String fileFormat;
+ @SerializedName("bp")
  private Map<String, String> brokerProperties;
 
  public BrokerTable() {
@@ -209,24 +214,7 @@ public TTableDescriptor toThrift() {
  return tTableDescriptor;
  }
 
- @Override
- public void write(DataOutput out) throws IOException {
- super.write(out);
-
- Text.writeString(out, brokerName);
- out.writeInt(paths.size());
- for (String path : paths) {
- Text.writeString(out, path);
- }
- Text.writeString(out, columnSeparator);
- Text.writeString(out, lineDelimiter);
- out.writeInt(brokerProperties.size());
- for (Map.Entry<String, String> prop : brokerProperties.entrySet()) {
- Text.writeString(out, prop.getKey());
- Text.writeString(out, prop.getValue());
- }
- }
-
+ @Deprecated
  public void readFields(DataInput in) throws IOException {
  super.readFields(in);
 

fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java

@@ -36,12 +36,12 @@
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.persist.CreateTableInfo;
 import org.apache.doris.persist.gson.GsonUtils;
-import org.apache.doris.system.SystemInfoService;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.gson.JsonObject;
 import com.google.gson.annotations.SerializedName;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.logging.log4j.LogManager;
@@ -108,6 +108,7 @@ public class Database extends MetaObject implements Writable, DatabaseIf<Table>
  @SerializedName(value = "replicaQuotaSize")
  private volatile long replicaQuotaSize;
 
+ @SerializedName(value = "tq")
  private volatile long transactionQuotaSize;
 
  private volatile boolean isDropped;
@@ -124,6 +125,7 @@ public enum DbState {
  @SerializedName(value = "dbProperties")
  private DatabaseProperty dbProperties = new DatabaseProperty();
 
+ @SerializedName(value = "bc")
  private BinlogConfig binlogConfig = new BinlogConfig();
 
  public Database() {
@@ -587,9 +589,24 @@ public int getMaxReplicationNum() {
  }
 
  public static Database read(DataInput in) throws IOException {
- Database db = new Database();
- db.readFields(in);
- return db;
+ LOG.info("read db from journal {}", in);
+ if (Env.getCurrentEnvJournalVersion() < FeMetaVersion.VERSION_136) {
+ Database db = new Database();
+ db.readFields(in);
+ return db;
+ } else {
+ Database db = GsonUtils.GSON.fromJson(Text.readString(in), Database.class);
+ db.fullQualifiedName = ClusterNamespace.getNameFromFullName(db.fullQualifiedName);
+ db.nameToTable.forEach((tn, tb) -> {
+ tb.setQualifiedDbName(db.fullQualifiedName);
+ if (tb instanceof MTMV) {
+ Env.getCurrentEnv().getMtmvService().registerMTMV((MTMV) tb, db.id);
+ }
+ db.idToTable.put(tb.getId(), tb);
+ db.lowerCaseToTableName.put(tn.toLowerCase(), tn);
+ });
+ return db;
+ }
  }
 
  @Override
@@ -605,33 +622,16 @@ public String getSignature(int signatureVersion) {
 
  @Override
  public void write(DataOutput out) throws IOException {
- super.write(out);
-
- out.writeLong(id);
- Text.writeString(out, ClusterNamespace.getNameFromFullName(fullQualifiedName));
- // write tables
  discardHudiTable();
- int numTables = nameToTable.size();
- out.writeInt(numTables);
- for (Map.Entry<String, Table> entry : nameToTable.entrySet()) {
- entry.getValue().write(out);
- }
-
- out.writeLong(dataQuotaBytes);
- Text.writeString(out, SystemInfoService.DEFAULT_CLUSTER);
- Text.writeString(out, dbState.name());
- Text.writeString(out, attachDbName);
-
- // write functions
- FunctionUtil.write(out, name2Function);
-
- // write encryptKeys
- dbEncryptKey.write(out);
-
- out.writeLong(replicaQuotaSize);
- dbProperties.write(out);
+ String json = GsonUtils.GSON.toJson(this);
+ JsonObject jsonObject = GsonUtils.GSON.fromJson(json, JsonObject.class);
+ String fqn = ClusterNamespace.getNameFromFullName(jsonObject.get("fullQualifiedName").getAsString());
+ jsonObject.remove("fullQualifiedName");
+ jsonObject.addProperty("fullQualifiedName", fqn);
+ Text.writeString(out, GsonUtils.GSON.toJson(jsonObject));
  }
 
+
  private void discardHudiTable() {
  Iterator<Entry<String, Table>> iterator = nameToTable.entrySet().iterator();
  while (iterator.hasNext()) {
@@ -650,6 +650,7 @@ public void analyze() {
  }
  }
 
+ @Deprecated
  @Override
  public void readFields(DataInput in) throws IOException {
  super.readFields(in);

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -2082,10 +2082,12 @@ public long loadFrontends(DataInputStream dis, long checksum) throws IOException
  }
 
  public long loadBackends(DataInputStream dis, long checksum) throws IOException {
+ LOG.info("start loading backends from image");
  return systemInfo.loadBackends(dis, checksum);
  }
 
  public long loadDb(DataInputStream dis, long checksum) throws IOException, DdlException {
+ LOG.info("start loading db from image");
  return getInternalCatalog().loadDb(dis, checksum);
  }
 
@@ -2312,6 +2314,7 @@ public long loadPolicy(DataInputStream in, long checksum) throws IOException {
  * Load catalogs through file.
  **/
  public long loadCatalog(DataInputStream in, long checksum) throws IOException {
+ LOG.info("start loading catalog from image");
  CatalogMgr mgr = CatalogMgr.read(in);
  // When enable the multi catalog in the first time, the "mgr" will be a null value.
  // So ignore it to use default catalog manager.

fe/fe-core/src/main/java/org/apache/doris/catalog/EsTable.java

@@ -28,6 +28,7 @@
 import org.apache.doris.thrift.TTableDescriptor;
 import org.apache.doris.thrift.TTableType;
 
+import com.google.gson.annotations.SerializedName;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.commons.codec.digest.DigestUtils;
@@ -36,7 +37,6 @@
 import org.apache.logging.log4j.Logger;
 
 import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -65,6 +65,7 @@ public class EsTable extends Table {
  // Here we have a slightly conservative value of 20, but at the same time
  // we also provide configurable parameters for expert-using
  // @see `MAX_DOCVALUE_FIELDS`
+ @Getter
  private static final int DEFAULT_MAX_DOCVALUE_FIELDS = 20;
  private String hosts;
  private String[] seeds;
@@ -77,6 +78,7 @@ public class EsTable extends Table {
  private String mappingType = null;
  // only save the partition definition, save the partition key,
  // partition list is got from es cluster dynamically and is saved in esTableState
+ @SerializedName("pi")
  private PartitionInfo partitionInfo;
  private EsTablePartitions esTablePartitions;
 
@@ -101,6 +103,7 @@ public class EsTable extends Table {
  private boolean includeHiddenIndex = Boolean.parseBoolean(EsResource.INCLUDE_HIDDEN_INDEX_DEFAULT_VALUE);
 
  // tableContext is used for being convenient to persist some configuration parameters uniformly
+ @SerializedName("tc")
  private Map<String, String> tableContext = new HashMap<>();
 
  // record the latest and recently exception when sync ES table metadata (mapping, shard location)
@@ -261,18 +264,7 @@ public String getSignature(int signatureVersion) {
  return md5;
  }
 
- @Override
- public void write(DataOutput out) throws IOException {
- super.write(out);
- out.writeInt(tableContext.size());
- for (Map.Entry<String, String> entry : tableContext.entrySet()) {
- Text.writeString(out, entry.getKey());
- Text.writeString(out, entry.getValue());
- }
- Text.writeString(out, partitionInfo.getType().name());
- partitionInfo.write(out);
- }
-
+ @Deprecated
  @Override
  public void readFields(DataInput in) throws IOException {
  super.readFields(in);