[common] Introduce ARRAY type for ARROW, COMPACTED and INDEXED formats

Author: XuQianJin-Stars

fluss-client/pom.xml

@@ -113,6 +113,15 @@
                                     <include>*:*</include>
                                 </includes>
                             </artifactSet>
+                            <filters>
+                                <filter>
+                                    <artifact>*</artifact>
+                                    <excludes>
+                                        <exclude>LICENSE-EPL-1.0.txt</exclude>
+                                        <exclude>LICENSE-EDL-1.0.txt</exclude>
+                                    </excludes>
+                                </filter>
+                            </filters>
                         </configuration>
                     </execution>
                 </executions>

fluss-client/src/main/resources/META-INF/NOTICE

@@ -10,6 +10,7 @@ This project bundles the following dependencies under the Apache Software Licens
 - com.ververica:frocksdbjni:6.20.3-ververica-2.0
 - org.apache.commons:commons-lang3:3.18.0
 - org.apache.commons:commons-math3:3.6.1
+- org.jetbrains:annotations:26.0.1
 - org.lz4:lz4-java:1.8.0
 
 This project bundles the following dependencies under the MIT (https://opensource.org/licenses/MIT)
@@ -20,4 +21,10 @@ See bundled license files for details.
 This project bundles the following dependencies under BSD License (https://opensource.org/licenses/bsd-license.php).
 See bundled license files for details.
 
-- com.github.luben:zstd-jni:1.5.7-1
+- com.github.luben:zstd-jni:1.5.7-1
+
+This project bundles the following dependencies under the Eclipse Distribution License 1.0 (https://www.eclipse.org/org/documents/edl-v10.php) and Eclipse Public License 1.0 (https://www.eclipse.org/legal/epl-v10.html).
+See bundled license files for details.
+
+- org.eclipse.collections:eclipse-collections-api:11.1.0
+- org.eclipse.collections:eclipse-collections:11.1.0

fluss-common/pom.xml

@@ -62,7 +62,14 @@
             <artifactId>fluss-shaded-arrow</artifactId>
         </dependency>
 
-        <!-- TODO: these two dependencies need to be shaded. -->
+        <!-- TODO: these three dependencies need to be shaded. -->
+        <!-- Use the Arrow compatible version -->
+        <dependency>
+            <groupId>org.eclipse.collections</groupId>
+            <artifactId>eclipse-collections</artifactId>
+            <version>11.1.0</version>
+        </dependency>
+
         <dependency>
             <groupId>org.lz4</groupId>
             <artifactId>lz4-java</artifactId>
@@ -109,6 +116,12 @@
             <version>${iceberg.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.jetbrains</groupId>
+            <artifactId>annotations</artifactId>
+            <version>26.0.1</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

fluss-common/src/main/java/org/apache/fluss/memory/AbstractPagedOutputView.java

@@ -45,11 +45,17 @@ public abstract class AbstractPagedOutputView implements OutputView, MemorySegme
     /** The list of memory segments that have been fully written and are immutable now. */
     private List<MemorySegmentBytesView> finishedPages;
 
-    /** The current memory segment to write to. */
-    private MemorySegment currentSegment;
+    /** the current memory segment to write to. */
+    protected MemorySegment currentSegment;
 
     /** the offset in the current segment. */
-    private int positionInSegment;
+    protected int positionInSegment;
+
+    /** Flag indicating whether throw BufferExhaustedException or wait for available segment. */
+    protected boolean waitingSegment;
+
+    /** the reusable array for UTF encodings. */
+    protected byte[] utfBuffer;
 
     protected AbstractPagedOutputView(MemorySegment initialSegment, int pageSize) {
         if (initialSegment == null) {
@@ -301,6 +307,62 @@ public void writeDouble(double v) throws IOException {
         }
     }
 
+    @Override
+    public void write(int b) throws IOException {
+        writeByte(b);
+    }
+
+    @Override
+    public void writeChar(int v) throws IOException {
+        if (positionInSegment < pageSize - 1) {
+            currentSegment.putChar(positionInSegment, (char) v);
+            positionInSegment += 2;
+        } else if (positionInSegment == pageSize) {
+            advance();
+            writeChar(v);
+        } else {
+            writeByte(v);
+            writeByte(v >> 8);
+        }
+    }
+
+    @Override
+    public void writeBytes(String s) throws IOException {
+        if (s == null) {
+            throw new NullPointerException();
+        }
+        for (int i = 0; i < s.length(); i++) {
+            writeByte(s.charAt(i));
+        }
+    }
+
+    @Override
+    public void writeChars(String s) throws IOException {
+        if (s == null) {
+            throw new NullPointerException();
+        }
+        for (int i = 0; i < s.length(); i++) {
+            writeChar(s.charAt(i));
+        }
+    }
+
+    @Override
+    public void writeUTF(String str) throws IOException {
+        if (str == null) {
+            throw new NullPointerException();
+        }
+
+        byte[] bytes = str.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+        int length = bytes.length;
+
+        if (length > 65535) {
+            throw new IOException("String too long for UTF encoding: " + length);
+        }
+
+        writeShort(length);
+        write(bytes);
+    }
+
     @Override
     public void write(MemorySegment segment, int off, int len) throws IOException {
         int remaining = pageSize - positionInSegment;

fluss-common/src/main/java/org/apache/fluss/memory/InputView.java

@@ -3,8 +3,8 @@
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
+ * (the "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
  *
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
@@ -23,8 +23,13 @@
 import java.io.IOException;
 
 /**
- * A very similar interface to {@link java.io.DataInput} but reading multi-bytes primitive types in
- * little endian.
+ * A specialized interface for reading multi-byte primitive types in little endian byte order. This
+ * interface is designed to be independent of Java's standard DataInput interface to avoid
+ * endianness confusion and API semantic violations.
+ *
+ * <p>Note: This interface uses little endian byte order, which is different from Java's standard
+ * DataInput interface that uses big endian. If you need big endian compatibility, use the {@link
+ * #asDataInput()} method to get a DataInput wrapper.
  *
  * @since 0.2
  */
@@ -54,8 +59,8 @@ public interface InputView {
     byte readByte() throws IOException;
 
     /**
-     * Reads two input bytes and returns a {@code short} value. This method is suitable for reading
-     * the bytes written by the {@link OutputView#writeShort(int)}.
+     * Reads two input bytes and returns a {@code short} value in little endian byte order. This
+     * method is suitable for reading the bytes written by the {@link OutputView#writeShort(int)}.
      *
      * @return the 16-bit value read.
      * @exception EOFException if this stream reaches the end before reading all the bytes.
@@ -64,8 +69,8 @@ public interface InputView {
     short readShort() throws IOException;
 
     /**
-     * Reads four input bytes and returns an {@code int} value. This method is suitable for reading
-     * bytes written by the {@link OutputView#writeInt(int)}.
+     * Reads four input bytes and returns an {@code int} value in little endian byte order. This
+     * method is suitable for reading bytes written by the {@link OutputView#writeInt(int)}.
      *
      * @return the {@code int} value read.
      * @exception EOFException if this stream reaches the end before reading all the bytes.
@@ -74,8 +79,28 @@ public interface InputView {
     int readInt() throws IOException;
 
     /**
-     * Reads eight input bytes and returns a {@code long} value. This method is suitable for reading
-     * bytes written by the {@link OutputView#writeLong(long)}.
+     * Reads two input bytes and returns a {@code char} value in little endian byte order. This
+     * method is suitable for reading bytes written by the {@link OutputView#writeChar(int)}.
+     *
+     * @return the {@code char} value read.
+     * @exception EOFException if this stream reaches the end before reading all the bytes.
+     * @exception IOException if an I/O error occurs.
+     */
+    char readChar() throws IOException;
+
+    /**
+     * Reads two input bytes and returns an unsigned 16-bit integer in little endian byte order.
+     * This method is suitable for reading bytes written by the {@link OutputView#writeShort(int)}.
+     *
+     * @return the unsigned 16-bit value read.
+     * @exception EOFException if this stream reaches the end before reading all the bytes.
+     * @exception IOException if an I/O error occurs.
+     */
+    int readUnsignedShort() throws IOException;
+
+    /**
+     * Reads eight input bytes and returns a {@code long} value in little endian byte order. This
+     * method is suitable for reading bytes written by the {@link OutputView#writeLong(long)}.
      *
      * @return the {@code long} value read.
      * @exception EOFException if this stream reaches the end before reading all the bytes.
@@ -161,4 +186,28 @@ public interface InputView {
      * @exception IOException if an I/O error occurs.
      */
     void readFully(byte[] b, int offset, int len) throws IOException;
+
+    /**
+     * Skips over and discards {@code n} bytes of data from the input stream. The skip method may,
+     * for a variety of reasons, end up skipping over some smaller number of bytes, possibly 0. This
+     * may result from any of a number of conditions; reaching end of file before {@code n} bytes
+     * have been skipped is only one possibility. The actual number of bytes skipped is returned. If
+     * {@code n} is negative, no bytes are skipped.
+     *
+     * @param n the number of bytes to be skipped.
+     * @return the actual number of bytes skipped.
+     * @exception IOException if an I/O error occurs.
+     */
+    int skipBytes(int n) throws IOException;
+
+    /**
+     * Reads a string that has been encoded using a modified UTF-8 format in little endian byte
+     * order. This method is suitable for reading strings written by the {@link
+     * OutputView#writeUTF(String)}.
+     *
+     * @return a Unicode string.
+     * @exception EOFException if this stream reaches the end before reading all the bytes.
+     * @exception IOException if an I/O error occurs.
+     */
+    String readUTF() throws IOException;
 }

fluss-common/src/main/java/org/apache/fluss/memory/ManagedPagedOutputView.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.memory;
 
 import java.io.IOException;
+import java.io.UTFDataFormatException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -47,4 +48,94 @@ protected MemorySegment nextSegment() throws IOException {
     public List<MemorySegment> allocatedPooledSegments() {
         return pooledSegments;
     }
+
+    @Override
+    public void write(int b) throws IOException {
+        writeByte(b);
+    }
+
+    @Override
+    public void writeChar(int v) throws IOException {
+        if (this.positionInSegment < this.pageSize - 1) {
+            this.currentSegment.putCharBigEndian(this.positionInSegment, (char) v);
+            this.positionInSegment += 2;
+        } else if (this.positionInSegment == this.pageSize) {
+            advance();
+            writeChar(v);
+        } else {
+            writeByte(v >> 8);
+            writeByte(v);
+        }
+    }
+
+    @Override
+    public void writeBytes(String s) throws IOException {
+        for (int i = 0; i < s.length(); i++) {
+            writeByte(s.charAt(i));
+        }
+    }
+
+    @Override
+    public void writeChars(String s) throws IOException {
+        for (int i = 0; i < s.length(); i++) {
+            writeChar(s.charAt(i));
+        }
+    }
+
+    @Override
+    public void writeUTF(String str) throws IOException {
+        int strlen = str.length();
+        int utflen = 0;
+        int c, count = 0;
+
+        /* use charAt instead of copying String to char array */
+        for (int i = 0; i < strlen; i++) {
+            c = str.charAt(i);
+            if ((c >= 0x0001) && (c <= 0x007F)) {
+                utflen++;
+            } else if (c > 0x07FF) {
+                utflen += 3;
+            } else {
+                utflen += 2;
+            }
+        }
+
+        if (utflen > 65535) {
+            throw new UTFDataFormatException("encoded string too long: " + utflen + " memory");
+        }
+
+        if (this.utfBuffer == null || this.utfBuffer.length < utflen + 2) {
+            this.utfBuffer = new byte[utflen + 2];
+        }
+        final byte[] bytearr = this.utfBuffer;
+
+        bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF);
+        bytearr[count++] = (byte) (utflen & 0xFF);
+
+        int i;
+        for (i = 0; i < strlen; i++) {
+            c = str.charAt(i);
+            if (!((c >= 0x0001) && (c <= 0x007F))) {
+                break;
+            }
+            bytearr[count++] = (byte) c;
+        }
+
+        for (; i < strlen; i++) {
+            c = str.charAt(i);
+            if ((c >= 0x0001) && (c <= 0x007F)) {
+                bytearr[count++] = (byte) c;
+
+            } else if (c > 0x07FF) {
+                bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
+                bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F));
+                bytearr[count++] = (byte) (0x80 | (c & 0x3F));
+            } else {
+                bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
+                bytearr[count++] = (byte) (0x80 | (c & 0x3F));
+            }
+        }
+
+        write(bytearr, 0, utflen + 2);
+    }
 }

fluss-common/src/main/java/org/apache/fluss/memory/MemorySegment.java

@@ -24,6 +24,7 @@
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.io.Serializable;
 import java.nio.BufferOverflowException;
 import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
@@ -65,7 +66,7 @@
  * implementations on invocations of abstract methods.
  */
 @Internal
-public final class MemorySegment {
+public final class MemorySegment implements Serializable {
 
     /** The unsafe handle for transparent memory copied (heap / off-heap). */
     private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;