[vortex] Upgrade vortex to 0.89.0 and rework integration for better performance

docs/docs/concepts/spec/fileformat.md

@@ -262,8 +262,23 @@ Limitations:
 
 Key features:
 - **Adaptive Encoding**: Automatically selects the best encoding per column based on data distribution
-- **Native Predicate Pushdown**: Supports filter expressions pushed down to the scan layer
+- **Native Predicate Pushdown**: Comparisons, `IS [NOT] NULL`, `IN`, `NOT IN`, `BETWEEN`, and prefix, suffix and substring matches are evaluated in the scan
 - **Column Projection**: Only reads requested columns from disk
+- **File Statistics**: Min, max and null counts come from the native writer, so scans can skip files without the writer collecting statistics row by row
+- **Vectorized Reads**: Batches are returned as columnar vectors to the engine
+
+Local files and OSS files are read and written by a native storage client. All other file systems
+are reached through the Paimon `FileIO` of the table.
+
+Format Options:
+
+| Option | Default | Type | Description |
+|:---|:---|:---|:---|
+| `vortex.worker-threads` | 4 | Integer | Minimum number of native threads that decode and encode Vortex data in the background. The pool is shared by the whole JVM and only grows. With 0 threads, only the thread that reads or writes a file drives the native work for it. |
+| `vortex.compact-encodings` | false | Boolean | PyPaimon only. Write with the compact Vortex encodings, which make files smaller and reads somewhat slower. |
+
+Java writers hand at least 8192 rows to Vortex in each batch, because every batch costs a native
+call and becomes a separate chunk. `write-batch-memory` still limits the memory of each batch.
 
 Limitations:
 1. Vortex does not support `MAP`, `MULTISET` or `VARIANT` types.

paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java

@@ -243,7 +243,8 @@ public boolean write(InternalRow currentRow) {
                 return false;
             }
         }
-        for (int i = 0; i < currentRow.getFieldCount(); i++) {
+        int fieldCount = currentRow.getFieldCount();
+        for (int i = 0; i < fieldCount; i++) {
             try {
                 fieldWriters[i].write(rowId, currentRow, i);
             } catch (OversizedAllocationException | IndexOutOfBoundsException e) {

paimon-arrow/src/main/java/org/apache/paimon/arrow/writer/ArrowFieldWriters.java

@@ -52,6 +52,8 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.IntArrayList;
 
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.BaseVariableWidthVector;
 import org.apache.arrow.vector.BigIntVector;
 import org.apache.arrow.vector.BitVector;
 import org.apache.arrow.vector.DateDayVector;
@@ -123,12 +125,12 @@ protected void doWrite(int rowIndex, DataGetters getters, int pos) {
             if (segments.length == 1) {
                 byte[] heapMemory = segments[0].getHeapMemory();
                 if (heapMemory != null) {
-                    ((VarCharVector) fieldVector)
-                            .setSafe(
-                                    rowIndex,
-                                    heapMemory,
-                                    binaryString.getOffset(),
-                                    binaryString.getSizeInBytes());
+                    setBytes(
+                            (VarCharVector) fieldVector,
+                            rowIndex,
+                            heapMemory,
+                            binaryString.getOffset(),
+                            binaryString.getSizeInBytes());
                     return;
                 }
             }
@@ -138,6 +140,30 @@ protected void doWrite(int rowIndex, DataGetters getters, int pos) {
         }
     }
 
+    /**
+     * Writes a value at {@code index}. When the value is the next one in sequence and fits in the
+     * current buffers, this skips the capacity loops and hole filling of {@code setSafe}, which
+     * cost more than the copy for short values.
+     */
+    private static void setBytes(
+            BaseVariableWidthVector vector, int index, byte[] value, int start, int length) {
+        int lastSet = vector.getLastSet();
+        if (index == lastSet + 1) {
+            ArrowBuf offsets = vector.getOffsetBuffer();
+            long nextOffsetPosition = (long) (index + 1) * BaseVariableWidthVector.OFFSET_WIDTH;
+            if (nextOffsetPosition + BaseVariableWidthVector.OFFSET_WIDTH <= offsets.capacity()
+                    && (index >> 3) < vector.getValidityBuffer().capacity()) {
+                long startOffset =
+                        offsets.getInt((long) index * BaseVariableWidthVector.OFFSET_WIDTH);
+                if (startOffset + length <= vector.getDataBuffer().capacity()) {
+                    vector.set(index, value, start, length);
+                    return;
+                }
+            }
+        }
+        vector.setSafe(index, value, start, length);
+    }
+
     /** Writer for BOOLEAN. */
     public static class BooleanWriter extends ArrowFieldWriter {
 
@@ -196,7 +222,8 @@ protected void doWrite(
 
         @Override
         protected void doWrite(int rowIndex, DataGetters getters, int pos) {
-            ((VarBinaryVector) fieldVector).setSafe(rowIndex, getters.getBinary(pos));
+            byte[] value = getters.getBinary(pos);
+            setBytes((VarBinaryVector) fieldVector, rowIndex, value, 0, value.length);
         }
     }
 

paimon-benchmark/paimon-micro-benchmarks/pom.xml

@@ -152,6 +152,12 @@ under the License.
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.paimon</groupId>
+            <artifactId>paimon-vortex-format</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.paimon</groupId>
             <artifactId>paimon-arrow</artifactId>

paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableReadBenchmark.java

@@ -80,6 +80,18 @@ public void testAvroRead() throws Exception {
          */
     }
 
+    @Test
+    public void testVortexRead() throws Exception {
+        innerTest(Collections.singletonMap("vortex", prepareData(vortex(), "vortex")));
+    }
+
+    @Test
+    public void testVortexReadProjection() throws Exception {
+        innerTestProjection(
+                Collections.singletonMap("vortex", prepareData(vortex(), "vortex")),
+                new int[] {0, 5, 10, 14});
+    }
+
     @Test
     public void testOrcReadProjection() throws Exception {
         innerTestProjection(
@@ -147,6 +159,12 @@ private Options parquet() {
         return options;
     }
 
+    private Options vortex() {
+        Options options = new Options();
+        options.set(CoreOptions.FILE_FORMAT, "vortex");
+        return options;
+    }
+
     private Options avro() {
         Options options = new Options();
         options.set(CoreOptions.FILE_FORMAT, CoreOptions.FILE_FORMAT_AVRO);

paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableWriterBenchmark.java

@@ -91,6 +91,13 @@ public void testParquet() throws Exception {
          */
     }
 
+    @Test
+    public void testVortex() throws Exception {
+        Options options = new Options();
+        options.set(CoreOptions.FILE_FORMAT, "vortex");
+        innerTest("vortex", options);
+    }
+
     @Test
     public void testOrc() throws Exception {
         Options options = new Options();

paimon-python/pypaimon/common/options/core_options.py

@@ -403,6 +403,15 @@ class CoreOptions:
         )
     )
 
+    VORTEX_COMPACT_ENCODINGS: ConfigOption[bool] = (
+        ConfigOptions.key("vortex.compact-encodings")
+        .boolean_type()
+        .default_value(False)
+        .with_description(
+            "Whether Python writers use the compact Vortex encodings, which make files smaller "
+            "and reads somewhat slower. Java writers always use the default encodings.")
+    )
+
     MOSAIC_NUM_BUCKETS: ConfigOption[int] = (
         ConfigOptions.key("mosaic.num-buckets")
         .int_type()
@@ -1395,6 +1404,9 @@ def mosaic_stats_columns(self, default=None):
             return [str(column).strip() for column in value if str(column).strip()]
         return []
 
+    def vortex_compact_encodings(self, default=None):
+        return self.options.get(CoreOptions.VORTEX_COMPACT_ENCODINGS, default)
+
     def mosaic_num_buckets(self, default=None):
         return self.options.get(CoreOptions.MOSAIC_NUM_BUCKETS, default)
 

paimon-python/pypaimon/execution/python/read/reader/format_vortex_reader.py

@@ -32,6 +32,11 @@ class FormatVortexReader(RecordBatchReader):
     and filters it based on the provided predicate and projection.
     """
 
+    # Vortex exports each chunk through the Arrow C interface. Small chunks make that per-chunk
+    # cost dominate, so Vortex produces chunks of up to this many rows, and the reader slices
+    # them to the requested batch size without a copy.
+    NATIVE_BATCH_SIZE = 65536
+
     # row_indices: from IndexedSplit (ANN vector search), discrete local row offsets within the file.
     # shard_range: from SlicedSplit (parallel shard scan), a contiguous [start, end) row range within the file.
     def __init__(self, file_io: FileIO, file_path: str, read_fields: List[DataField],
@@ -41,21 +46,15 @@ def __init__(self, file_io: FileIO, file_path: str, read_fields: List[DataField]
                  predicate_fields: Optional[Set[str]] = None):
         import vortex
 
-        from pypaimon.execution.python.read.reader.vortex_utils import to_vortex_specified
-        file_path_for_vortex, store_kwargs = to_vortex_specified(file_io, file_path)
-
-        if store_kwargs:
-            from vortex import store
-            vortex_store = store.from_url(file_path_for_vortex, **store_kwargs)
-            vortex_file = vortex_store.open()
-        else:
-            vortex_file = vortex.open(file_path_for_vortex)
+        from pypaimon.execution.python.read.reader.vortex_utils import open_vortex_file
+        vortex_file = open_vortex_file(file_io, file_path)
 
         self.read_fields = read_fields
         self._read_field_names = [f.name for f in read_fields]
 
         # Identify which fields exist in the file and which are missing
-        file_schema_names = set(vortex_file.dtype.to_arrow_schema().names)
+        file_schema = vortex_file.dtype.to_arrow_schema()
+        file_schema_names = set(file_schema.names)
         self.existing_fields = [f.name for f in read_fields if f.name in file_schema_names]
         self.missing_fields = [f.name for f in read_fields if f.name not in file_schema_names]
 
@@ -66,8 +65,7 @@ def __init__(self, file_io: FileIO, file_path: str, read_fields: List[DataField]
         if push_down_predicate is not None:
             try:
                 from vortex.arrow.expression import arrow_to_vortex
-                arrow_schema = vortex_file.dtype.to_arrow_schema()
-                vortex_expr = arrow_to_vortex(push_down_predicate, arrow_schema)
+                vortex_expr = arrow_to_vortex(push_down_predicate, file_schema)
             except Exception:
                 pass
 
@@ -79,13 +77,51 @@ def __init__(self, file_io: FileIO, file_path: str, read_fields: List[DataField]
             # index array. Acceptable trade-off vs reading the full file.
             indices = vortex.array(range(shard_range[0], shard_range[1]))
 
-        self.record_batch_reader = vortex_file.scan(
-            columns_for_vortex, expr=vortex_expr, indices=indices, batch_size=batch_size).to_arrow()
+        # Ask Vortex for offset-based strings and binaries. It exports views by default, and
+        # casting them in pyarrow decodes every value a second time.
+        export_schema = pa.schema([
+            file_schema.field(name).with_type(self._without_view_types(file_schema.field(name).type))
+            for name in self.existing_fields
+        ]) if self.existing_fields else None
+
+        self.record_batch_reader = vortex_file.to_arrow(
+            columns_for_vortex, expr=vortex_expr, indices=indices,
+            batch_size=max(batch_size, self.NATIVE_BATCH_SIZE), schema=export_schema)
+        self._batch_size = batch_size
+        self._pending: Optional[RecordBatch] = None
+        self._pending_offset = 0
 
         self._output_schema = (
             PyarrowFieldParser.from_paimon_schema(read_fields) if read_fields else None
         )
 
+    @staticmethod
+    def _without_view_types(data_type: pa.DataType) -> pa.DataType:
+        """Replace string_view/binary_view, also nested ones, with string/binary."""
+        if data_type == pa.string_view():
+            return pa.utf8()
+        if data_type == pa.binary_view():
+            return pa.binary()
+        if pa.types.is_struct(data_type):
+            return pa.struct([
+                data_type.field(i).with_type(
+                    FormatVortexReader._without_view_types(data_type.field(i).type))
+                for i in range(data_type.num_fields)
+            ])
+        if pa.types.is_list(data_type):
+            value_field = data_type.value_field
+            return pa.list_(value_field.with_type(
+                FormatVortexReader._without_view_types(value_field.type)))
+        if pa.types.is_large_list(data_type):
+            value_field = data_type.value_field
+            return pa.large_list(value_field.with_type(
+                FormatVortexReader._without_view_types(value_field.type)))
+        if pa.types.is_fixed_size_list(data_type):
+            value_field = data_type.value_field
+            return pa.list_(value_field.with_type(
+                FormatVortexReader._without_view_types(value_field.type)), data_type.list_size)
+        return data_type
+
     @staticmethod
     def _cast_view_types(batch: RecordBatch) -> RecordBatch:
         """Cast all string_view/binary_view columns to string/binary."""
@@ -109,10 +145,17 @@ def _cast_view_types(batch: RecordBatch) -> RecordBatch:
             return pa.RecordBatch.from_arrays(columns, schema=pa.schema(fields))
         return batch
 
+    def _next_slice(self) -> RecordBatch:
+        while self._pending is None or self._pending_offset >= self._pending.num_rows:
+            self._pending = self._cast_view_types(next(self.record_batch_reader))
+            self._pending_offset = 0
+        batch = self._pending.slice(self._pending_offset, self._batch_size)
+        self._pending_offset += self._batch_size
+        return batch
+
     def read_arrow_batch(self) -> Optional[RecordBatch]:
         try:
-            batch = next(self.record_batch_reader)
-            batch = self._cast_view_types(batch)
+            batch = self._next_slice()
 
             if not self.missing_fields:
                 return batch
@@ -150,3 +193,4 @@ def _type_for_missing(name: str) -> pa.DataType:
 
     def close(self):
         self.record_batch_reader = None
+        self._pending = None