[feature][dingo-sdk] Add codecVersion to tableDefinition

Author: guojn1

dingo-store-proto

@@ -46,8 +46,16 @@ public DingoKeyValueCodec(int schemaVersion, long id, List<DingoSchema> schemas)
         rd = new RecordDecoder(schemaVersion, schemas, id);
     }
 
+    public DingoKeyValueCodec(int codecVersion, int schemaVersion, long id, List<DingoSchema> schemas) {
+        this.schemas = schemas;
+        this.id = id;
+        re = new RecordEncoder(codecVersion, schemaVersion, schemas, id);
+        rd = new RecordDecoder(schemaVersion, schemas, id);
+    }
+
     public static DingoKeyValueCodec of(Table table) {
         return of(
+                table.getCodecVersion(),
                 table.getVersion(),
                 Optional.mapOrGet(table.id(), DingoCommonId::entityId, () -> 0L),
                 table.getColumns()
@@ -58,16 +66,16 @@ public static DingoKeyValueCodec of(long id, Table table) {
         return of(table.getVersion(), id, table);
     }
 
-    public static DingoKeyValueCodec of(long id, List<Column> columns) {
-        return of(1, id, columns);
+    public static DingoKeyValueCodec of(int codecVersion, long id, List<Column> columns) {
+        return of(codecVersion, 1, id, columns);
     }
 
     public static DingoKeyValueCodec of(int schemaVersion,long id, Table table) {
-        return of(schemaVersion, id, table.getColumns());
+        return of(table.getCodecVersion(), schemaVersion, id, table.getColumns());
     }
 
-    public static DingoKeyValueCodec of(int schemaVersion, long id, List<Column> columns) {
-        return new DingoKeyValueCodec(schemaVersion, id, CodecUtils.createSchemaForColumns(columns));
+    public static DingoKeyValueCodec of(int codecVersion, int schemaVersion, long id, List<Column> columns) {
+        return new DingoKeyValueCodec(codecVersion, schemaVersion, id, CodecUtils.createSchemaForColumns(columns));
     }
 
     @Override

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/codec/DingoKeyValueCodec.java

@@ -59,6 +59,10 @@ default DingoCommonId id() {
 
     String getCollate();
 
+    default int getCodecVersion() {
+        return 2;
+    }
+
     default String getTableType() {
         return "BASE TABLE";
     }

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/table/Table.java

@@ -53,5 +53,6 @@ public class TableDefinition implements Table {
     private String rowFormat;
     private long createTime;
     private long updateTime;
+    private int codecVersion;
 
 }

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/table/TableDefinition.java

@@ -160,7 +160,7 @@ public static Partition mapping(long id, Meta.TableDefinition tableDefinition, L
         if (partition.getPartitionsCount() < 1) {
             return null;
         }
-        DingoKeyValueCodec codec = DingoKeyValueCodec.of(id, columns);
+        DingoKeyValueCodec codec = DingoKeyValueCodec.of(tableDefinition.getCodecVersion(), id, columns);
         List<PartitionDetail> details = partition.getPartitionsList().stream()
                 .map(Meta.Partition::getRange)
                 .map(Common.Range::getStartKey)
@@ -863,7 +863,7 @@ public static Meta.PartitionRule calcRange(
         DingoCommonId tableId1 = mapping(tableId);
         List<Column> keyColumns = table.getKeyColumns();
         keyColumns.sort(Comparator.comparingInt(Column::getPrimary));
-        KeyValueCodec codec = DingoKeyValueCodec.of(tableId1.entityId(), keyColumns);
+        KeyValueCodec codec = DingoKeyValueCodec.of(table.getCodecVersion(), tableId1.entityId(), keyColumns);
         byte[] minKeyPrefix = codec.encodeMinKeyPrefix();
         byte[] maxKeyPrefix = codec.encodeMaxKeyPrefix();
         Meta.PartitionRule.Builder builder = Meta.PartitionRule.newBuilder();

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/utils/EntityConversion.java

@@ -426,7 +426,7 @@ public void addDistribution(String tableName, PartitionDetail partitionDetail) {
         );
         Table table = mapping(definitionWithId);
         DingoCommonId tableId = mapping(definitionWithId.getTableId());
-        DingoKeyValueCodec codec = DingoKeyValueCodec.of(tableId.entityId(), table.getKeyColumns());
+        DingoKeyValueCodec codec = DingoKeyValueCodec.of(table.getCodecVersion(), tableId.entityId(), table.getKeyColumns());
         try {
             byte[] key = codec.encodeKeyPrefix(partitionDetail.getOperand(), partitionDetail.getOperand().length);
             RangeDistribution distribution = getRangeDistribution(tableName, new ComparableByteArray(key, POS));