[common] Improve the implementation of ARRAY type

Author: wuchong

fluss-client/pom.xml

@@ -113,15 +113,6 @@
                                     <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,7 +10,6 @@ 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)
@@ -21,10 +20,4 @@ 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
-
-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
+- com.github.luben:zstd-jni:1.5.7-1

fluss-common/pom.xml

@@ -62,14 +62,7 @@
             <artifactId>fluss-shaded-arrow</artifactId>
         </dependency>
 
-        <!-- 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>
-
+        <!-- TODO: these two dependencies need to be shaded. -->
         <dependency>
             <groupId>org.lz4</groupId>
             <artifactId>lz4-java</artifactId>
@@ -116,12 +109,6 @@
             <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,17 +45,11 @@ 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. */
-    protected MemorySegment currentSegment;
+    /** The current memory segment to write to. */
+    private MemorySegment currentSegment;
 
     /** the offset in the current segment. */
-    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;
+    private int positionInSegment;
 
     protected AbstractPagedOutputView(MemorySegment initialSegment, int pageSize) {
         if (initialSegment == null) {
@@ -307,62 +301,6 @@ 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,13 +23,8 @@
 import java.io.IOException;
 
 /**
- * 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.
+ * A very similar interface to {@link java.io.DataInput} but reading multi-bytes primitive types in
+ * little endian.
  *
  * @since 0.2
  */
@@ -59,8 +54,8 @@ public interface InputView {
     byte readByte() throws IOException;
 
     /**
-     * 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)}.
+     * 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)}.
      *
      * @return the 16-bit value read.
      * @exception EOFException if this stream reaches the end before reading all the bytes.
@@ -69,8 +64,8 @@ public interface InputView {
     short readShort() throws IOException;
 
     /**
-     * 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)}.
+     * Reads four input bytes and returns an {@code int} value. 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.
@@ -79,28 +74,8 @@ public interface InputView {
     int readInt() throws IOException;
 
     /**
-     * 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)}.
+     * Reads eight input bytes and returns a {@code long} value. 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.
@@ -186,28 +161,4 @@ 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,7 +18,6 @@
 package org.apache.fluss.memory;
 
 import java.io.IOException;
-import java.io.UTFDataFormatException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -48,94 +47,4 @@ 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,7 +24,6 @@
 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;
@@ -66,7 +65,7 @@
  * implementations on invocations of abstract methods.
  */
 @Internal
-public final class MemorySegment implements Serializable {
+public final class MemorySegment {
 
     /** The unsafe handle for transparent memory copied (heap / off-heap). */
     private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;