[feat][store] serial refactor for coprocessor and dingo-sdk.

Author: nokiaMS

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/serial/Buf.java

@@ -24,16 +24,31 @@ public interface Buf {
     void writeInt(int i);
     void writeLong(long l);
 
+    void write(int pos, byte b);
+    void write(int pos, byte[] b);
+    void write(int srcPos, byte[] b, int pos, int length);
+    void writeShort(int pos, short i);
+    void writeInt(int pos, int i);
+    void writeLong(int pos, long l);
+
     byte peek();
     int peekInt();
     long peekLong();
 
     byte read();
     byte[] read(int length);
     void read(byte[] b, int pos, int length);
+    short readShort();
     int readInt();
     long readLong();
 
+    byte readAt(int pos);
+    byte[] readAt(int pos, int length);
+    void readAt(int srcPos, byte[] b, int pos, int length);
+    short readShortAt(int pos);
+    int readIntAt(int pos);
+    long readLongAt(int pos);
+
     void reverseWrite(byte b);
     byte reverseRead();
     void reverseWriteInt(int i);
@@ -44,6 +59,9 @@ public interface Buf {
     void reverseSkipInt();
     void ensureRemainder(int length);
     void resize(int oldSize, int newSize);
+    void setForwardOffset(int pos);
+    int restReadableSize();
+    int readOffset();
 
     boolean isEnd();
 

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/serial/BufImpl.java

@@ -22,6 +22,10 @@ public class BufImpl implements Buf {
     private int forwardPos;
     private int reversePos;
 
+    private void setForwardPos(int pos) {
+        this.forwardPos = pos;
+    }
+
     public BufImpl(int bufSize) {
         this.buf = new byte[bufSize];
         this.forwardPos = 0;
@@ -34,23 +38,50 @@ public BufImpl(byte[] keyBuf) {
         this.reversePos = keyBuf.length - 1;
     }
 
+    public BufImpl(int bufSize, int dataPos) {
+        this.buf = new byte[bufSize];
+        this.forwardPos = dataPos;
+        this.reversePos = bufSize - 1;
+    }
+
     @Override
     public void write(byte b) {
         buf[forwardPos++] = b;
     }
 
+    @Override
+    public void write(int pos, byte b) {
+        buf[pos] = b;
+    }
+
     @Override
     public void write(byte[] b) {
         System.arraycopy(b, 0, buf, forwardPos, b.length);
         forwardPos += b.length;
     }
 
+    @Override
+    public void write(int pos, byte[] b) {
+        System.arraycopy(b, 0, buf, pos, b.length);
+    }
+
     @Override
     public void write(byte[] b, int pos, int length) {
         System.arraycopy(b, pos, buf, forwardPos, length);
         forwardPos += length;
     }
 
+    @Override
+    public void write(int srcPos, byte[] b, int pos, int length) {
+        System.arraycopy(b, pos, buf, srcPos, length);
+    }
+
+    @Override
+    public void writeShort(int pos, short i) {
+        buf[pos] = (byte) (i >>> 8);
+        buf[pos + 1] = (byte) i;
+    }
+
     @Override
     public void writeInt(int i) {
         buf[forwardPos++] = (byte) (i >>> 24);
@@ -59,6 +90,14 @@ public void writeInt(int i) {
         buf[forwardPos++] = (byte) i;
     }
 
+    @Override
+    public void writeInt(int pos, int i) {
+        buf[pos] = (byte) (i >>> 24);
+        buf[pos + 1] = (byte) (i >>> 16);
+        buf[pos + 2] = (byte) (i >>> 8);
+        buf[pos + 3] = (byte) i;
+    }
+
     @Override
     public void writeLong(long l) {
         buf[forwardPos++] = (byte) (l >>> 56);
@@ -71,6 +110,18 @@ public void writeLong(long l) {
         buf[forwardPos++] = (byte) l;
     }
 
+    @Override
+    public void writeLong(int pos, long l) {
+        buf[pos] = (byte) (l >>> 56);
+        buf[pos + 1] = (byte) (l >>> 48);
+        buf[pos + 2] = (byte) (l >>> 40);
+        buf[pos + 3] = (byte) (l >>> 32);
+        buf[pos + 4] = (byte) (l >>> 24);
+        buf[pos + 5] = (byte) (l >>> 16);
+        buf[pos + 6] = (byte) (l >>> 8);
+        buf[pos + 7] = (byte) l;
+    }
+
     @Override
     public byte peek() {
         return buf[forwardPos];
@@ -101,6 +152,11 @@ public byte read() {
         return buf[forwardPos++];
     }
 
+    @Override
+    public byte readAt(int pos) {
+        return buf[pos];
+    }
+
     @Override
     public byte[] read(int length) {
         byte[] b = new byte[length];
@@ -109,12 +165,36 @@ public byte[] read(int length) {
         return b;
     }
 
+    @Override
+    public byte[] readAt(int pos, int length) {
+        byte[] b = new byte[length];
+        System.arraycopy(buf, pos, b, 0, length);
+        return b;
+    }
+
     @Override
     public void read(byte[] b, int pos, int length) {
         System.arraycopy(buf, forwardPos, b, pos, length);
         forwardPos += length;
     }
 
+    @Override
+    public void readAt(int srcPos, byte[] b, int pos, int length) {
+        System.arraycopy(buf, srcPos, b, pos, length);
+    }
+
+    @Override
+    public short readShortAt(int pos) {
+        return (short)(((buf[pos] & 0xFF) << 8)
+                | buf[pos + 1] & 0xFF);
+    }
+
+    @Override
+    public short readShort() {
+        return (short)(((buf[forwardPos++] & 0xFF) << 8)
+                | buf[forwardPos++] & 0xFF);
+    }
+
     @Override
     public int readInt() {
         return (((buf[forwardPos++] & 0xFF) << 24)
@@ -123,6 +203,14 @@ public int readInt() {
                 | buf[forwardPos++] & 0xFF);
     }
 
+    @Override
+    public int readIntAt(int pos) {
+        return (((buf[pos++] & 0xFF) << 24)
+                | ((buf[pos++] & 0xFF) << 16)
+                | ((buf[pos++] & 0xFF) << 8)
+                | buf[pos++] & 0xFF);
+    }
+
     @Override
     public long readLong() {
         long l = buf[forwardPos++]  & 0xFF;
@@ -133,6 +221,16 @@ public long readLong() {
         return l;
     }
 
+    @Override
+    public long readLongAt(int pos) {
+        long l = buf[pos++]  & 0xFF;
+        for (int i = 0; i < 7; i++) {
+            l <<= 8;
+            l |= buf[pos++] & 0xFF;
+        }
+        return l;
+    }
+
     @Override
     public void reverseWrite(byte b) {
         buf[reversePos--] = b;
@@ -213,11 +311,26 @@ public void resize(int oldSize, int newSize) {
         }
     }
 
+    @Override
+    public void setForwardOffset(int pos) {
+        this.forwardPos = pos;
+    }
+
     @Override
     public boolean isEnd() {
         return (reversePos - forwardPos + 1) == 0;
     }
 
+    @Override
+    public int restReadableSize() {
+        return this.buf.length - forwardPos - 1;
+    }
+
+    @Override
+    public int readOffset() {
+        return forwardPos;
+    }
+
     @Override
     public byte[] getBytes() {
         int emptySize = reversePos - forwardPos + 1;

java/dingo-sdk/src/main/java/io/dingodb/sdk/common/serial/Config.java

@@ -22,5 +22,8 @@ public class Config {
     public final static int KEY_REVERSE_TAG_SIZE = 4;
 
     public final static byte CODEC_VERSION = 1;
+    public final static byte CODEC_VERSION_V2 = 2;
 
+    public final static int idUnit = 2;
+    public final static int offsetUnit = 4;
 }