[common] Introduce AlignedRow for row data encoding (apache#1620)

The AlignedRow is inspired by Flink BinaryRowData.

Author: platinumhamburg

LICENSE

@@ -242,6 +242,8 @@ Apache Flink
 ./fluss-common/src/main/java/org/apache/fluss/row/BinarySegmentUtils.java
 ./fluss-common/src/main/java/org/apache/fluss/row/BinaryString.java
 ./fluss-common/src/main/java/org/apache/fluss/row/Decimal.java
+./fluss-common/src/main/java/org/apache/fluss/row/aligned/AlignedRow.java
+./fluss-common/src/main/java/org/apache/fluss/row/aligned/AlignedRowWriter.java
 ./fluss-common/src/main/java/org/apache/fluss/types/DataType.java
 ./fluss-common/src/main/java/org/apache/fluss/utils/AbstractAutoCloseableRegistry.java
 ./fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java

fluss-common/src/main/java/org/apache/fluss/row/BinarySection.java

@@ -31,10 +31,38 @@
 
 /** Describe a section of memory. */
 @Internal
-abstract class BinarySection implements MemoryAwareGetters, Serializable {
+public abstract class BinarySection implements MemoryAwareGetters, Serializable {
 
     private static final long serialVersionUID = 1L;
 
+    /**
+     * It decides whether to put data in FixLenPart or VarLenPart. See more in {@link BinaryRow}.
+     *
+     * <p>If len is less than 8, its binary format is: 1-bit mark(1) = 1, 7-bits len, and 7-bytes
+     * data. Data is stored in fix-length part.
+     *
+     * <p>If len is greater or equal to 8, its binary format is: 1-bit mark(1) = 0, 31-bits offset
+     * to the data, and 4-bytes length of data. Data is stored in variable-length part.
+     */
+    public static final int MAX_FIX_PART_DATA_SIZE = 7;
+
+    /**
+     * To get the mark in highest bit of long. Form: 10000000 00000000 ... (8 bytes)
+     *
+     * <p>This is used to decide whether the data is stored in fixed-length part or variable-length
+     * part. see {@link #MAX_FIX_PART_DATA_SIZE} for more information.
+     */
+    public static final long HIGHEST_FIRST_BIT = 0x80L << 56;
+
+    /**
+     * To get the 7 bits length in second bit to eighth bit out of a long. Form: 01111111 00000000
+     * ... (8 bytes)
+     *
+     * <p>This is used to get the length of the data which is stored in this long. see {@link
+     * #MAX_FIX_PART_DATA_SIZE} for more information.
+     */
+    public static final long HIGHEST_SECOND_TO_EIGHTH_BIT = 0x7FL << 56;
+
     protected MemorySegment[] segments;
     protected int offset;
     protected int sizeInBytes;

fluss-common/src/main/java/org/apache/fluss/row/BinarySegmentUtils.java

@@ -23,8 +23,11 @@
 import org.apache.fluss.utils.MurmurHashUtils;
 
 import java.io.IOException;
+import java.nio.ByteOrder;
 
 import static org.apache.fluss.memory.MemoryUtils.UNSAFE;
+import static org.apache.fluss.row.BinarySection.HIGHEST_FIRST_BIT;
+import static org.apache.fluss.row.BinarySection.HIGHEST_SECOND_TO_EIGHTH_BIT;
 
 /* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -34,6 +37,9 @@
 @Internal
 public final class BinarySegmentUtils {
 
+    public static final boolean LITTLE_ENDIAN =
+            (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);
+
     private static final int ADDRESS_BITS_PER_WORD = 3;
 
     private static final int BIT_BYTE_INDEX_MASK = 7;
@@ -368,6 +374,20 @@ private static int byteIndex(int bitIndex) {
         return bitIndex >>> ADDRESS_BITS_PER_WORD;
     }
 
+    /**
+     * unset bit.
+     *
+     * @param segment target segment.
+     * @param baseOffset bits base offset.
+     * @param index bit index from base offset.
+     */
+    public static void bitUnSet(MemorySegment segment, int baseOffset, int index) {
+        int offset = baseOffset + byteIndex(index);
+        byte current = segment.get(offset);
+        current &= ~(1 << (index & BIT_BYTE_INDEX_MASK));
+        segment.put(offset, current);
+    }
+
     /**
      * read bit.
      *
@@ -386,4 +406,262 @@ private static int hashMultiSeg(MemorySegment[] segments, int offset, int numByt
         copyMultiSegmentsToBytes(segments, offset, bytes, 0, numBytes);
         return MurmurHashUtils.hashUnsafeBytes(bytes, BYTE_ARRAY_BASE_OFFSET, numBytes);
     }
+
+    /**
+     * get long from segments.
+     *
+     * @param segments target segments.
+     * @param offset value offset.
+     */
+    public static long getLong(MemorySegment[] segments, int offset) {
+        if (inFirstSegment(segments, offset, 8)) {
+            return segments[0].getLong(offset);
+        } else {
+            return getLongMultiSegments(segments, offset);
+        }
+    }
+
+    private static long getLongMultiSegments(MemorySegment[] segments, int offset) {
+        int segSize = segments[0].size();
+        int segIndex = offset / segSize;
+        int segOffset = offset - segIndex * segSize; // equal to %
+
+        if (segOffset < segSize - 7) {
+            return segments[segIndex].getLong(segOffset);
+        } else {
+            return getLongSlowly(segments, segSize, segIndex, segOffset);
+        }
+    }
+
+    private static long getLongSlowly(
+            MemorySegment[] segments, int segSize, int segNum, int segOffset) {
+        MemorySegment segment = segments[segNum];
+        long ret = 0;
+        for (int i = 0; i < 8; i++) {
+            if (segOffset == segSize) {
+                segment = segments[++segNum];
+                segOffset = 0;
+            }
+            long unsignedByte = segment.get(segOffset) & 0xff;
+            if (LITTLE_ENDIAN) {
+                ret |= (unsignedByte << (i * 8));
+            } else {
+                ret |= (unsignedByte << ((7 - i) * 8));
+            }
+            segOffset++;
+        }
+        return ret;
+    }
+
+    /**
+     * set long from segments.
+     *
+     * @param segments target segments.
+     * @param offset value offset.
+     */
+    public static void setLong(MemorySegment[] segments, int offset, long value) {
+        if (inFirstSegment(segments, offset, 8)) {
+            segments[0].putLong(offset, value);
+        } else {
+            setLongMultiSegments(segments, offset, value);
+        }
+    }
+
+    private static void setLongMultiSegments(MemorySegment[] segments, int offset, long value) {
+        int segSize = segments[0].size();
+        int segIndex = offset / segSize;
+        int segOffset = offset - segIndex * segSize; // equal to %
+
+        if (segOffset < segSize - 7) {
+            segments[segIndex].putLong(segOffset, value);
+        } else {
+            setLongSlowly(segments, segSize, segIndex, segOffset, value);
+        }
+    }
+
+    private static void setLongSlowly(
+            MemorySegment[] segments, int segSize, int segNum, int segOffset, long value) {
+        MemorySegment segment = segments[segNum];
+        for (int i = 0; i < 8; i++) {
+            if (segOffset == segSize) {
+                segment = segments[++segNum];
+                segOffset = 0;
+            }
+            long unsignedByte;
+            if (LITTLE_ENDIAN) {
+                unsignedByte = value >> (i * 8);
+            } else {
+                unsignedByte = value >> ((7 - i) * 8);
+            }
+            segment.put(segOffset, (byte) unsignedByte);
+            segOffset++;
+        }
+    }
+
+    /**
+     * Copy target segments from source byte[].
+     *
+     * @param segments target segments.
+     * @param offset target segments offset.
+     * @param bytes source byte[].
+     * @param bytesOffset source byte[] offset.
+     * @param numBytes the number bytes to copy.
+     */
+    public static void copyFromBytes(
+            MemorySegment[] segments, int offset, byte[] bytes, int bytesOffset, int numBytes) {
+        if (segments.length == 1) {
+            segments[0].put(offset, bytes, bytesOffset, numBytes);
+        } else {
+            copyMultiSegmentsFromBytes(segments, offset, bytes, bytesOffset, numBytes);
+        }
+    }
+
+    private static void copyMultiSegmentsFromBytes(
+            MemorySegment[] segments, int offset, byte[] bytes, int bytesOffset, int numBytes) {
+        int remainSize = numBytes;
+        for (MemorySegment segment : segments) {
+            int remain = segment.size() - offset;
+            if (remain > 0) {
+                int nCopy = Math.min(remain, remainSize);
+                segment.put(offset, bytes, numBytes - remainSize + bytesOffset, nCopy);
+                remainSize -= nCopy;
+                // next new segment.
+                offset = 0;
+                if (remainSize == 0) {
+                    return;
+                }
+            } else {
+                // remain is negative, let's advance to next segment
+                // now the offset = offset - segmentSize (-remain)
+                offset = -remain;
+            }
+        }
+    }
+
+    /** Gets an instance of {@link Decimal} from underlying {@link MemorySegment}. */
+    public static Decimal readDecimalData(
+            MemorySegment[] segments,
+            int baseOffset,
+            long offsetAndSize,
+            int precision,
+            int scale) {
+        final int size = ((int) offsetAndSize);
+        int subOffset = (int) (offsetAndSize >> 32);
+        byte[] bytes = new byte[size];
+        copyToBytes(segments, baseOffset + subOffset, bytes, 0, size);
+        return Decimal.fromUnscaledBytes(bytes, precision, scale);
+    }
+
+    /**
+     * Get binary, if len less than 8, will be include in variablePartOffsetAndLen.
+     *
+     * <p>Note: Need to consider the ByteOrder.
+     *
+     * @param baseOffset base offset of composite binary format.
+     * @param fieldOffset absolute start offset of 'variablePartOffsetAndLen'.
+     * @param variablePartOffsetAndLen a long value, real data or offset and len.
+     */
+    public static byte[] readBinary(
+            MemorySegment[] segments,
+            int baseOffset,
+            int fieldOffset,
+            long variablePartOffsetAndLen) {
+        long mark = variablePartOffsetAndLen & HIGHEST_FIRST_BIT;
+        if (mark == 0) {
+            final int subOffset = (int) (variablePartOffsetAndLen >> 32);
+            final int len = (int) variablePartOffsetAndLen;
+            return copyToBytes(segments, baseOffset + subOffset, len);
+        } else {
+            int len = (int) ((variablePartOffsetAndLen & HIGHEST_SECOND_TO_EIGHTH_BIT) >>> 56);
+            if (LITTLE_ENDIAN) {
+                return copyToBytes(segments, fieldOffset, len);
+            } else {
+                // fieldOffset + 1 to skip header.
+                return copyToBytes(segments, fieldOffset + 1, len);
+            }
+        }
+    }
+
+    /**
+     * Get binary string, if len less than 8, will be include in variablePartOffsetAndLen.
+     *
+     * <p>Note: Need to consider the ByteOrder.
+     *
+     * @param baseOffset base offset of composite binary format.
+     * @param fieldOffset absolute start offset of 'variablePartOffsetAndLen'.
+     * @param variablePartOffsetAndLen a long value, real data or offset and len.
+     */
+    public static BinaryString readBinaryString(
+            MemorySegment[] segments,
+            int baseOffset,
+            int fieldOffset,
+            long variablePartOffsetAndLen) {
+        long mark = variablePartOffsetAndLen & HIGHEST_FIRST_BIT;
+        if (mark == 0) {
+            final int subOffset = (int) (variablePartOffsetAndLen >> 32);
+            final int len = (int) variablePartOffsetAndLen;
+            return BinaryString.fromAddress(segments, baseOffset + subOffset, len);
+        } else {
+            int len = (int) ((variablePartOffsetAndLen & HIGHEST_SECOND_TO_EIGHTH_BIT) >>> 56);
+            if (BinarySegmentUtils.LITTLE_ENDIAN) {
+                return BinaryString.fromAddress(segments, fieldOffset, len);
+            } else {
+                // fieldOffset + 1 to skip header.
+                return BinaryString.fromAddress(segments, fieldOffset + 1, len);
+            }
+        }
+    }
+
+    /**
+     * hash segments to int, numBytes must be aligned to 4 bytes.
+     *
+     * @param segments Source segments.
+     * @param offset Source segments offset.
+     * @param numBytes the number bytes to hash.
+     */
+    public static int hashByWords(MemorySegment[] segments, int offset, int numBytes) {
+        if (inFirstSegment(segments, offset, numBytes)) {
+            return MurmurHashUtils.hashBytesByWords(segments[0], offset, numBytes);
+        } else {
+            return hashMultiSegByWords(segments, offset, numBytes);
+        }
+    }
+
+    private static int hashMultiSegByWords(MemorySegment[] segments, int offset, int numBytes) {
+        byte[] bytes = allocateReuseBytes(numBytes);
+        copyMultiSegmentsToBytes(segments, offset, bytes, 0, numBytes);
+        return MurmurHashUtils.hashUnsafeBytesByWords(bytes, BYTE_ARRAY_BASE_OFFSET, numBytes);
+    }
+
+    /**
+     * Gets an instance of {@link TimestampLtz} from underlying {@link MemorySegment}.
+     *
+     * @param segments the underlying MemorySegments
+     * @param baseOffset the base offset of current instance of {@code TimestampLtz}
+     * @param offsetAndNanos the offset of milli-seconds part and nanoseconds
+     * @return an instance of {@link TimestampLtz}
+     */
+    public static TimestampLtz readTimestampLtzData(
+            MemorySegment[] segments, int baseOffset, long offsetAndNanos) {
+        final int nanoOfMillisecond = (int) offsetAndNanos;
+        final int subOffset = (int) (offsetAndNanos >> 32);
+        final long millisecond = getLong(segments, baseOffset + subOffset);
+        return TimestampLtz.fromEpochMillis(millisecond, nanoOfMillisecond);
+    }
+
+    /**
+     * Gets an instance of {@link TimestampNtz} from underlying {@link MemorySegment}.
+     *
+     * @param segments the underlying MemorySegments
+     * @param baseOffset the base offset of current instance of {@code TimestampNtz}
+     * @param offsetAndNanos the offset of milli-seconds part and nanoseconds
+     * @return an instance of {@link TimestampNtz}
+     */
+    public static TimestampNtz readTimestampNtzData(
+            MemorySegment[] segments, int baseOffset, long offsetAndNanos) {
+        final int nanoOfMillisecond = (int) offsetAndNanos;
+        final int subOffset = (int) (offsetAndNanos >> 32);
+        final long millisecond = getLong(segments, baseOffset + subOffset);
+        return TimestampNtz.fromMillis(millisecond, nanoOfMillisecond);
+    }
 }