Fluss vector datatype POC

fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java

@@ -136,8 +136,20 @@ public DataTypeRoot[] getSupportedDataTypeRoots() {
             case LAST_VALUE_IGNORE_NULLS:
             case FIRST_VALUE:
             case FIRST_VALUE_IGNORE_NULLS:
-                // all data types are supported
-                return DataTypeRoot.values();
+                // All data types are supported except VECTOR. VECTOR has no well-defined
+                // value-selection semantics in an aggregation context (dense vectors are
+                // typically compared by distance, not equality), so it is excluded here to
+                // prevent accidental misuse.
+                DataTypeRoot[] allRoots = DataTypeRoot.values();
+                int vectorOrdinal = DataTypeRoot.VECTOR.ordinal();
+                DataTypeRoot[] nonVectorRoots = new DataTypeRoot[allRoots.length - 1];
+                int idx = 0;
+                for (int i = 0; i < allRoots.length; i++) {
+                    if (i != vectorOrdinal) {
+                        nonVectorRoots[idx++] = allRoots[i];
+                    }
+                }
+                return nonVectorRoots;
             default:
                 throw new IllegalStateException("Unsupported aggregation function type: " + this);
         }

fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java

@@ -812,6 +812,17 @@ private static List<Column> normalizeColumns(
                         "The data type of auto increment column must be INT or BIGINT.");
             }
 
+            // VECTOR columns cannot be primary keys: equality comparisons on dense vectors are
+            // not supported (floating-point precision issues; semantically users want similarity
+            // distance, not bit-exact equality). See VECTOR design decision in the type system.
+            if (pkSet.contains(column.getName()) && column.getDataType().is(DataTypeRoot.VECTOR)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Column '%s' of type VECTOR cannot be used as a primary key. "
+                                        + "VECTOR columns do not support equality comparisons.",
+                                column.getName()));
+            }
+
             // primary key and auto increment column should not nullable
             if (pkSet.contains(column.getName()) && column.getDataType().isNullable()) {
                 newColumns.add(

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

@@ -90,6 +90,7 @@ public static int calculateFixLengthPartSize(DataType type) {
             case ARRAY:
             case MAP:
             case ROW:
+            case VECTOR:
                 // long and double are 8 bytes;
                 // otherwise it stores the length and offset of the variable-length part for types
                 // such as is string, map, etc.

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

@@ -220,6 +220,7 @@ public static NullSetter createNullSetter(DataType elementType) {
             case ARRAY:
             case MAP:
             case ROW:
+            case VECTOR:
                 return BinaryArrayWriter::setNullLong;
             case BOOLEAN:
                 return BinaryArrayWriter::setNullBoolean;

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

@@ -24,6 +24,7 @@
 import org.apache.fluss.row.serializer.RowSerializer;
 import org.apache.fluss.types.ArrayType;
 import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.FloatType;
 import org.apache.fluss.types.MapType;
 import org.apache.fluss.types.RowType;
 
@@ -184,6 +185,12 @@ static BinaryWriter.ValueWriter createNotNullValueWriter(
                                 rowType.getFieldTypes().toArray(new DataType[0]), rowFormat);
                 return (writer, pos, value) ->
                         writer.writeRow(pos, (InternalRow) value, rowSerializer);
+            case VECTOR:
+                // VECTOR is serialized as an array of non-nullable FLOAT32 elements.
+                final ArraySerializer vectorSerializer =
+                        new ArraySerializer(new FloatType(false), rowFormat);
+                return (writer, pos, value) ->
+                        writer.writeArray(pos, (InternalArray) value, vectorSerializer);
             default:
                 String msg =
                         String.format(

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

@@ -143,6 +143,10 @@ static ElementGetter createElementGetter(DataType fieldType) {
                 final int rowFieldCount = ((RowType) fieldType).getFieldCount();
                 elementGetter = (array, pos) -> array.getRow(pos, rowFieldCount);
                 break;
+            case VECTOR:
+                // VECTOR values are represented as InternalArray of FLOAT32 elements.
+                elementGetter = InternalArray::getArray;
+                break;
             default:
                 String msg =
                         String.format(
@@ -224,9 +228,17 @@ static ElementGetter createDeepElementGetter(DataType fieldType) {
                             return genericRow;
                         };
                 break;
+            case VECTOR:
+                elementGetter =
+                        (array, pos) -> {
+                            InternalArray inner = array.getArray(pos);
+                            return new GenericArray(inner.toFloatArray());
+                        };
+                break;
             default:
                 // for primitive types, we can directly return the element getter
                 elementGetter = createElementGetter(fieldType);
+                break;
         }
         if (!fieldType.isNullable()) {
             return elementGetter;

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

@@ -135,6 +135,9 @@ static Class<?> getDataClass(DataType type) {
                 return InternalMap.class;
             case ROW:
                 return InternalRow.class;
+            case VECTOR:
+                // VECTOR values are stored as InternalArray of FLOAT32 elements.
+                return InternalArray.class;
             default:
                 throw new IllegalArgumentException("Illegal type: " + type);
         }
@@ -224,6 +227,10 @@ static FieldGetter createFieldGetter(DataType fieldType, int fieldPos) {
                 final int numFields = ((RowType) fieldType).getFieldCount();
                 fieldGetter = row -> row.getRow(fieldPos, numFields);
                 break;
+            case VECTOR:
+                // VECTOR values are InternalArray of FLOAT32 elements.
+                fieldGetter = row -> row.getArray(fieldPos);
+                break;
             default:
                 throw new IllegalArgumentException("Illegal type: " + fieldType);
         }
@@ -301,6 +308,13 @@ static FieldGetter createDeepFieldGetter(DataType fieldType, int fieldPos) {
                             return genericRow;
                         };
                 break;
+            case VECTOR:
+                fieldGetter =
+                        row -> {
+                            InternalArray array = row.getArray(fieldPos);
+                            return new GenericArray(array.toFloatArray());
+                        };
+                break;
             default:
                 // for primitive types, use the normal field getter
                 fieldGetter = createFieldGetter(fieldType, fieldPos);

fluss-common/src/main/java/org/apache/fluss/row/arrow/ArrowWriter.java

@@ -30,6 +30,7 @@
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.FieldVector;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorUnloader;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.complex.FixedSizeListVector;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.complex.ListVector;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.compression.CompressionCodec;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.compression.CompressionUtil;
@@ -267,6 +268,9 @@ public int serializeToOutputView(AbstractPagedOutputView outputView) throws IOEx
 
         // update row count only when we try to write records to the output.
         root.setRowCount(recordsCount);
+        for (ArrowFieldWriter fieldWriter : fieldWriters) {
+            fieldWriter.finish(recordsCount);
+        }
 
         // update the uncompressed body size.
         int uncompressedBodySizeInBytes = getBodyLength();
@@ -328,6 +332,15 @@ private void initFieldVector(FieldVector fieldVector) {
             ((BaseFixedWidthVector) fieldVector).allocateNew(INITIAL_CAPACITY);
         } else if (fieldVector instanceof BaseVariableWidthVector) {
             ((BaseVariableWidthVector) fieldVector).allocateNew(INITIAL_CAPACITY);
+        } else if (fieldVector instanceof FixedSizeListVector) {
+            // FixedSizeListVector: allocate the top-level validity bitmap and then
+            // recursively initialize the child (Float32) data vector.
+            FixedSizeListVector fslv = (FixedSizeListVector) fieldVector;
+            fslv.allocateNew();
+            FieldVector dataVector = fslv.getDataVector();
+            if (dataVector != null) {
+                initFieldVector(dataVector);
+            }
         } else if (fieldVector instanceof ListVector) {
             ListVector listVector = (ListVector) fieldVector;
             listVector.allocateNew();

fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowVectorColumnVector.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * 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
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.row.arrow.vectors;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.columnar.ArrayColumnVector;
+import org.apache.fluss.row.columnar.ColumnarArray;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.Float4Vector;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.complex.FixedSizeListVector;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * {@link org.apache.fluss.row.columnar.ColumnVector} backed by a shaded Arrow {@code
+ * FixedSizeListVector<Float32>} for reading {@link org.apache.fluss.types.VectorType} columns.
+ *
+ * <p>Each row {@code i} maps to child Float32 elements at indices {@code [i*dimension,
+ * (i+1)*dimension)} in the child {@code Float4Vector}. The returned {@link InternalArray} is a
+ * {@link ColumnarArray} view over an {@link ArrowFloatColumnVector} wrapping the child vector.
+ */
+@Internal
+public class ArrowVectorColumnVector implements ArrayColumnVector {
+
+    /** The FixedSizeListVector holding per-row validity and the stride-based child data. */
+    private final FixedSizeListVector vector;
+
+    /** The fixed number of float elements per row (equals the declared VECTOR dimension). */
+    private final int dimension;
+
+    /**
+     * A ColumnVector view over the child Float4Vector, shared across all rows for zero-copy {@link
+     * ColumnarArray} slicing.
+     */
+    private final ArrowFloatColumnVector elementVector;
+
+    /**
+     * Creates a new {@link ArrowVectorColumnVector}.
+     *
+     * @param vector the {@code FixedSizeListVector} to read from
+     * @param dimension the declared VECTOR dimension (must equal {@code vector.getListSize()})
+     */
+    public ArrowVectorColumnVector(FixedSizeListVector vector, int dimension) {
+        this.vector = checkNotNull(vector);
+        this.dimension = dimension;
+        this.elementVector =
+                new ArrowFloatColumnVector((Float4Vector) checkNotNull(vector.getDataVector()));
+    }
+
+    /**
+     * Returns the vector value at row {@code i} as an {@link InternalArray} of floats.
+     *
+     * <p>The returned array is a {@link ColumnarArray} window into the shared child vector,
+     * starting at element index {@code i * dimension} with length {@code dimension}.
+     *
+     * @param i row index (0-based)
+     * @return an {@link InternalArray} of {@code dimension} floats
+     */
+    @Override
+    public InternalArray getArray(int i) {
+        if (vector.getDataVector().getValueCount() == 0 && vector.getValueCount() > 0) {
+            vector.getDataVector().setValueCount(vector.getValueCount() * dimension);
+        }
+        int start = i * dimension;
+        return new ColumnarArray(elementVector, start, dimension);
+    }
+
+    @Override
+    public boolean isNullAt(int i) {
+        return vector.isNull(i);
+    }
+}

fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowFieldWriter.java

@@ -60,6 +60,14 @@ public void write(int rowIndex, DataGetters getters, int ordinal, boolean handle
         }
     }
 
+    /**
+     * Finishes writing the field vector for the given number of rows. Can be overridden by
+     * composite writers (e.g. {@link ArrowVectorWriter}) to finalize child vector state.
+     */
+    public void finish(int recordsCount) {
+        // default no-op
+    }
+
     /** Resets the state of the writer to write the next batch of fields. */
     public void reset() {
         fieldVector.reset();

fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowVectorWriter.java

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * 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
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.row.arrow.writers;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.row.DataGetters;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.arrow.ArrowWriter;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.complex.FixedSizeListVector;
+
+/**
+ * {@link ArrowFieldWriter} for {@link org.apache.fluss.types.VectorType}, writing {@link
+ * InternalArray} of float values into a shaded Arrow {@code FixedSizeListVector<Float32>}.
+ *
+ * <p>Unlike {@link ArrowArrayWriter} which uses {@code ListVector}'s {@code startNewValue/endValue}
+ * protocol, {@code FixedSizeListVector} uses stride-based child offsets: row {@code i} occupies
+ * child elements at indices {@code [i*listSize, (i+1)*listSize)}. Validity is set with {@code
+ * setNotNull(rowIndex)} and child elements are written via the delegated element writer.
+ */
+@Internal
+public class ArrowVectorWriter extends ArrowFieldWriter {
+
+    /** Writer for the Float32 child vector elements. */
+    private final ArrowFieldWriter elementWriter;
+
+    /**
+     * Running count of child element slots already written. Incremented by {@code listSize} for
+     * each row (including null rows, since FixedSizeListVector still allocates child slots for null
+     * rows).
+     */
+    private int offset;
+
+    /**
+     * Creates a new {@link ArrowVectorWriter}.
+     *
+     * @param vector the {@code FixedSizeListVector} to write into
+     * @param elementWriter writer for the Float32 child vector
+     */
+    public ArrowVectorWriter(FixedSizeListVector vector, ArrowFieldWriter elementWriter) {
+        super(vector);
+        this.elementWriter = elementWriter;
+        this.offset = 0;
+    }
+
+    @Override
+    public void doWrite(int rowIndex, DataGetters getters, int ordinal, boolean handleSafe) {
+        InternalArray array = getters.getArray(ordinal);
+        FixedSizeListVector listVector = (FixedSizeListVector) fieldVector;
+        int listSize = listVector.getListSize();
+        if (array.size() != listSize) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "VECTOR dimension mismatch: expected %d elements but got %d.",
+                            listSize, array.size()));
+        }
+        listVector.setNotNull(rowIndex);
+        for (int i = 0; i < listSize; i++) {
+            int elementIndex = offset + i;
+            // Use element-based index to determine handleSafe, not parent row count.
+            // When row count < INITIAL_CAPACITY but total elements > INITIAL_CAPACITY,
+            // we need safe mode for elements beyond the initial capacity.
+            boolean elementHandleSafe = elementIndex >= ArrowWriter.INITIAL_CAPACITY;
+            elementWriter.write(elementIndex, array, i, handleSafe || elementHandleSafe);
+        }
+        offset += listSize;
+    }
+
+    /**
+     * Overrides the base {@link ArrowFieldWriter#write} to always advance the {@code offset}
+     * counter by {@code listSize}, even for null rows.
+     *
+     * <p>This is required because {@code FixedSizeListVector} uses stride-based child indexing: row
+     * {@code i}'s child elements always occupy positions {@code [i*listSize, (i+1)*listSize)},
+     * regardless of whether the row is null. The base class short-circuits to {@code
+     * setNull(rowIndex)} without calling {@code doWrite}, so {@code offset} would never be
+     * incremented for null rows, causing subsequent non-null rows to write their child elements at
+     * the wrong positions.
+     */
+    @Override
+    public void write(int rowIndex, DataGetters getters, int ordinal, boolean handleSafe) {
+        if (getters.isNullAt(ordinal)) {
+            fieldVector.setNull(rowIndex);
+            offset += ((FixedSizeListVector) fieldVector).getListSize();
+        } else {
+            doWrite(rowIndex, getters, ordinal, handleSafe);
+        }
+    }
+
+    @Override
+    public void finish(int recordsCount) {
+        ((FixedSizeListVector) fieldVector).getDataVector().setValueCount(offset);
+    }
+
+    /**
+     * Resets the writer state for reuse (e.g. after batch serialization). The child element writer
+     * and offset counter are both reset to their initial state.
+     */
+    @Override
+    public void reset() {
+        super.reset();
+        elementWriter.reset();
+        offset = 0;
+    }
+}