diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md
index 18c1bb59d969..84bfcfba9496 100644
--- a/docs/docs/concepts/spec/fileformat.md
+++ b/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.
diff --git a/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java b/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
index f120b141cd84..cd43ee2964ec 100644
--- a/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
+++ b/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) {
diff --git a/paimon-arrow/src/main/java/org/apache/paimon/arrow/writer/ArrowFieldWriters.java b/paimon-arrow/src/main/java/org/apache/paimon/arrow/writer/ArrowFieldWriters.java
index 661c282bd87b..6077266d8a42 100644
--- a/paimon-arrow/src/main/java/org/apache/paimon/arrow/writer/ArrowFieldWriters.java
+++ b/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);
}
}
diff --git a/paimon-benchmark/paimon-micro-benchmarks/pom.xml b/paimon-benchmark/paimon-micro-benchmarks/pom.xml
index 7f99efb3b984..998191069fdf 100644
--- a/paimon-benchmark/paimon-micro-benchmarks/pom.xml
+++ b/paimon-benchmark/paimon-micro-benchmarks/pom.xml
@@ -152,6 +152,12 @@ under the License.
${project.version}
+
+ org.apache.paimon
+ paimon-vortex-format
+ ${project.version}
+
+
org.apache.paimonpaimon-arrow
diff --git a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableReadBenchmark.java b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableReadBenchmark.java
index 80a68563115f..b48ced7d980b 100644
--- a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableReadBenchmark.java
+++ b/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);
diff --git a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableWriterBenchmark.java b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableWriterBenchmark.java
index bc5e88ce8943..d056c546ca28 100644
--- a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/TableWriterBenchmark.java
+++ b/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();
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 77f40584ee5f..5bb753a1bbab 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/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)
diff --git a/paimon-python/pypaimon/execution/python/read/reader/format_vortex_reader.py b/paimon-python/pypaimon/execution/python/read/reader/format_vortex_reader.py
index 33ca6b474e5f..a931c9c0a235 100644
--- a/paimon-python/pypaimon/execution/python/read/reader/format_vortex_reader.py
+++ b/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
diff --git a/paimon-python/pypaimon/execution/python/read/reader/vortex_utils.py b/paimon-python/pypaimon/execution/python/read/reader/vortex_utils.py
index d66682898f9a..0578e1540eed 100644
--- a/paimon-python/pypaimon/execution/python/read/reader/vortex_utils.py
+++ b/paimon-python/pypaimon/execution/python/read/reader/vortex_utils.py
@@ -16,7 +16,9 @@
# under the License.
import os
-from typing import Dict, Optional, Tuple
+import threading
+from collections import OrderedDict
+from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlparse
from pypaimon.common.file_io import FileIO
@@ -67,3 +69,91 @@ def to_vortex_specified(file_io: FileIO, file_path: str) -> Tuple[str, Optional[
file_path_for_vortex = file_path_for_vortex.replace('oss://', 's3://')
return file_path_for_vortex, store_kwargs
+
+
+# Paimon never rewrites a data file in place, and every data file has a unique name. So a footer,
+# read once, stays valid for the path, and a later open of the same file needs no footer IO.
+_FOOTER_CACHE_SIZE = 1024
+_footer_cache: "OrderedDict[str, Any]" = OrderedDict()
+_store_cache: Dict[Tuple[str, Tuple[Tuple[str, str], ...]], Any] = {}
+_cache_lock = threading.Lock()
+
+
+def _store_for(url: str, store_kwargs: Dict[str, str]) -> Tuple[Any, str]:
+ """Return a shared store for the bucket of ``url``, and the key of ``url`` in it.
+
+ A store reaches every object in its bucket, so one client serves all files with the same
+ credentials. Creating a client per file costs a connection setup every time.
+ """
+ from vortex import store
+
+ parsed = urlparse(url)
+ root = f"{parsed.scheme}://{parsed.netloc}"
+ key = parsed.path.lstrip('/')
+ cache_key = (root, tuple(sorted((k, str(v)) for k, v in store_kwargs.items())))
+ with _cache_lock:
+ vortex_store = _store_cache.get(cache_key)
+ if vortex_store is None:
+ vortex_store = store.from_url(root, **store_kwargs)
+ _store_cache[cache_key] = vortex_store
+ return vortex_store, key
+
+
+def _cached_footer(cache_key: str) -> Any:
+ with _cache_lock:
+ footer = _footer_cache.get(cache_key)
+ if footer is not None:
+ _footer_cache.move_to_end(cache_key)
+ return footer
+
+
+def _cache_footer(cache_key: str, footer: Any) -> None:
+ with _cache_lock:
+ _footer_cache[cache_key] = footer
+ _footer_cache.move_to_end(cache_key)
+ while len(_footer_cache) > _FOOTER_CACHE_SIZE:
+ _footer_cache.popitem(last=False)
+
+
+def open_vortex_file(file_io: FileIO, file_path: str) -> Any:
+ """Open a Vortex data file of a Paimon table for reading.
+
+ A local file is memory-mapped: the page cache serves its segments, with no read threads and
+ no copies. A remote file is read through a shared store client. Either way, the footer of
+ a file that was opened before comes from a cache.
+ """
+ import vortex
+
+ file_path_for_vortex, store_kwargs = to_vortex_specified(file_io, file_path)
+ footer = _cached_footer(file_path_for_vortex)
+
+ if store_kwargs:
+ vortex_store, key = _store_for(file_path_for_vortex, store_kwargs)
+ vortex_file = vortex.open(key, store=vortex_store, footer=footer)
+ elif urlparse(file_path_for_vortex).scheme:
+ vortex_file = vortex.open(file_path_for_vortex, footer=footer)
+ else:
+ vortex_file = vortex.open(file_path_for_vortex, footer=footer, memory_map=True)
+
+ if footer is None:
+ _cache_footer(file_path_for_vortex, vortex_file.footer)
+ return vortex_file
+
+
+def write_vortex_file(file_io: FileIO, path: str, data: Any, compact: bool = False) -> None:
+ """Write a pyarrow table as a Vortex file, through a shared store client when remote.
+
+ ``compact`` selects the compact encodings, which make files smaller and reads slower.
+ """
+ import vortex
+ from vortex.io import VortexWriteOptions
+
+ options = VortexWriteOptions.compact() if compact else VortexWriteOptions.default()
+ file_path_for_vortex, store_kwargs = to_vortex_specified(file_io, path)
+ if store_kwargs:
+ vortex_store, key = _store_for(file_path_for_vortex, store_kwargs)
+ options.write(vortex.array(data), key, store=vortex_store)
+ else:
+ if not urlparse(file_path_for_vortex).scheme:
+ os.makedirs(os.path.dirname(file_path_for_vortex), exist_ok=True)
+ options.write(vortex.array(data), file_path_for_vortex)
diff --git a/paimon-python/pypaimon/execution/python/write/writer/data_writer.py b/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
index a46c5365377e..2387bc9a9b95 100644
--- a/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
@@ -266,7 +266,8 @@ def _write_file(self, file_path: str, data: pa.Table, file_format: str) -> Dict:
elif file_format == CoreOptions.FILE_FORMAT_LANCE:
self.file_io.write_lance(file_path, data)
elif file_format == CoreOptions.FILE_FORMAT_VORTEX:
- self.file_io.write_vortex(file_path, data)
+ self.file_io.write_vortex(file_path, data,
+ compact=self.options.vortex_compact_encodings())
elif file_format == CoreOptions.FILE_FORMAT_MOSAIC:
self.file_io.write_mosaic(file_path, data, options=self.mosaic_writer_options)
elif file_format == CoreOptions.FILE_FORMAT_ROW:
diff --git a/paimon-python/pypaimon/execution/python/write/writer/vector_writer.py b/paimon-python/pypaimon/execution/python/write/writer/vector_writer.py
index 7d4bce6c8a1a..5714091472f5 100644
--- a/paimon-python/pypaimon/execution/python/write/writer/vector_writer.py
+++ b/paimon-python/pypaimon/execution/python/write/writer/vector_writer.py
@@ -96,7 +96,8 @@ def _write_data_to_file(self, data: pa.Table):
zstd_level=self.zstd_level,
**self.parquet_writer_options)
elif self.vector_file_format == CoreOptions.FILE_FORMAT_VORTEX:
- self.file_io.write_vortex(file_path, data)
+ self.file_io.write_vortex(file_path, data,
+ compact=self.options.vortex_compact_encodings())
else:
raise ValueError(f"Unsupported vector file format: {self.vector_file_format}")
diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
index 9e9e14ac8a4f..0699620e437e 100644
--- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
+++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
@@ -696,18 +696,8 @@ def write_mosaic(self, path: str, data: pyarrow.Table, **kwargs):
def write_vortex(self, path: str, data: pyarrow.Table, **kwargs):
try:
- import vortex
- from vortex import store
-
- from pypaimon.execution.python.read.reader.vortex_utils import (to_vortex_specified)
- file_path_for_vortex, store_kwargs = to_vortex_specified(self, path)
-
- if store_kwargs:
- vortex_store = store.from_url(file_path_for_vortex, **store_kwargs)
- vortex_store.write(vortex.array(data))
- else:
- from vortex._lib.io import write as vortex_write
- vortex_write(vortex.array(data), file_path_for_vortex)
+ from pypaimon.execution.python.read.reader.vortex_utils import write_vortex_file
+ write_vortex_file(self, path, data, compact=kwargs.get('compact', False))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Vortex file {path}: {e}") from e
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py b/paimon-python/pypaimon/filesystem/local_file_io.py
index 2e53eb3123a2..8b7a56794218 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -440,19 +440,8 @@ def write_mosaic(self, path: str, data: pyarrow.Table, **kwargs):
def write_vortex(self, path: str, data: pyarrow.Table, **kwargs):
try:
- import vortex
- from vortex._lib.io import write as vortex_write
- os.makedirs(os.path.dirname(path), exist_ok=True)
-
- from pypaimon.execution.python.read.reader.vortex_utils import (to_vortex_specified)
- _, store_kwargs = to_vortex_specified(self, path)
-
- if store_kwargs:
- from vortex import store
- vortex_store = store.from_url(path, **store_kwargs)
- vortex_store.write(vortex.array(data))
- else:
- vortex_write(vortex.array(data), path)
+ from pypaimon.execution.python.read.reader.vortex_utils import write_vortex_file
+ write_vortex_file(self, path, data, compact=kwargs.get('compact', False))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Vortex file {path}: {e}") from e
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 60042cbf0c23..8793b4feb40a 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -1019,18 +1019,8 @@ def write_mosaic(self, path: str, data: pyarrow.Table, **kwargs):
def write_vortex(self, path: str, data: pyarrow.Table, **kwargs):
try:
- import vortex
- from vortex import store
-
- from pypaimon.execution.python.read.reader.vortex_utils import (to_vortex_specified)
- file_path_for_vortex, store_kwargs = to_vortex_specified(self, path)
-
- if store_kwargs:
- vortex_store = store.from_url(file_path_for_vortex, **store_kwargs)
- vortex_store.write(vortex.array(data))
- else:
- from vortex._lib.io import write as vortex_write
- vortex_write(vortex.array(data), file_path_for_vortex)
+ from pypaimon.execution.python.read.reader.vortex_utils import write_vortex_file
+ write_vortex_file(self, path, data, compact=kwargs.get('compact', False))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Vortex file {path}: {e}") from e
diff --git a/paimon-python/pypaimon/read/read_support.py b/paimon-python/pypaimon/read/read_support.py
index 65844f883ac0..4d33da47c17a 100644
--- a/paimon-python/pypaimon/read/read_support.py
+++ b/paimon-python/pypaimon/read/read_support.py
@@ -293,6 +293,17 @@ def _add_row_kind_column_to_batch(
return pyarrow.RecordBatch.from_arrays(columns, schema=new_schema)
+def _declared_output_schema(self):
+ # Resolving the read schema probes for the native runtime, which costs a failed import
+ # when it is absent. The result only depends on the read type and the projection.
+ key = (id(self.read_type), id(self.output_projection))
+ cached = getattr(self, '_declared_output_schema_cache', None)
+ if cached is None or cached[0] != key:
+ cached = (key, output_schema(read_arrow_schema(self.read_type), self.output_projection))
+ self._declared_output_schema_cache = cached
+ return cached[1]
+
+
def _project_batch_to_output(self, batch: pyarrow.RecordBatch) -> pyarrow.RecordBatch:
if self.output_projection is not None:
_, paths = self._reader_adapter()
@@ -301,7 +312,7 @@ def _project_batch_to_output(self, batch: pyarrow.RecordBatch) -> pyarrow.Record
# Project the already masked schema, retaining captured nullability.
adapted_projection = OutputProjection([(alias, path) for alias, path in outputs])
actual = output_schema(batch.schema, adapted_projection)
- declared = output_schema(read_arrow_schema(self.read_type), self.output_projection)
+ declared = _declared_output_schema(self)
fields = [field.with_nullable(field.nullable or actual[i].nullable) for i, field in enumerate(declared)]
if self.include_row_kind and self._row_kind_column in batch.schema.names:
arrays.insert(0, batch.column(self._row_kind_column))
diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py b/paimon-python/pypaimon/tests/hdfs_native_test.py
index ca27ace261d1..447595258b9a 100644
--- a/paimon-python/pypaimon/tests/hdfs_native_test.py
+++ b/paimon-python/pypaimon/tests/hdfs_native_test.py
@@ -905,20 +905,20 @@ def test_write_vortex_delegates_to_vortex_specified(self):
fake_vortex = types.ModuleType("vortex")
fake_vortex.array = MagicMock(return_value="varr")
fake_vortex.store = types.SimpleNamespace(from_url=MagicMock())
- fake_io = types.ModuleType("vortex._lib.io")
- fake_io.write = MagicMock()
- fake_lib = types.ModuleType("vortex._lib")
- fake_lib.io = fake_io
+ fake_io = types.ModuleType("vortex.io")
+ default_options = MagicMock()
+ fake_io.VortexWriteOptions = types.SimpleNamespace(
+ default=MagicMock(return_value=default_options), compact=MagicMock())
+ fake_vortex.io = fake_io
fake_modules = {
"vortex": fake_vortex,
- "vortex._lib": fake_lib,
- "vortex._lib.io": fake_io,
+ "vortex.io": fake_io,
}
with patch.dict(sys.modules, fake_modules), \
patch("pypaimon.execution.python.read.reader.vortex_utils.to_vortex_specified",
return_value=("hdfs://ns/x.vortex", None)):
self.fio.write_vortex("hdfs://ns/x.vortex", table)
- fake_io.write.assert_called_once_with("varr", "hdfs://ns/x.vortex")
+ default_options.write.assert_called_once_with("varr", "hdfs://ns/x.vortex")
if __name__ == "__main__":
diff --git a/paimon-python/pypaimon/tests/vortex_utils_test.py b/paimon-python/pypaimon/tests/vortex_utils_test.py
new file mode 100644
index 000000000000..f3caae2771d6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/vortex_utils_test.py
@@ -0,0 +1,132 @@
+# 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.
+
+
+import importlib.util
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon.common.file_io import FileIO
+from pypaimon.execution.python.read.reader import vortex_utils
+
+
+@unittest.skipIf(sys.version_info < (3, 11) or importlib.util.find_spec('vortex') is None,
+ "vortex-data requires Python >= 3.11")
+class VortexUtilsTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp()
+ self.file_io = FileIO.get(self.temp_dir, {})
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def test_local_file_round_trip_reuses_footer(self):
+ path = os.path.join(self.temp_dir, 'data', 'a.vortex')
+ table = pa.table({'id': [1, 2, 3], 'name': ['a', 'b', 'c']})
+ vortex_utils.write_vortex_file(self.file_io, path, table)
+
+ first = vortex_utils.open_vortex_file(self.file_io, path)
+ self.assertIn(path, vortex_utils._footer_cache)
+
+ # The second open takes the cached footer and reads the same rows.
+ second = vortex_utils.open_vortex_file(self.file_io, path)
+ self.assertEqual(len(first), len(second))
+ result = second.to_arrow(schema=pa.schema([('id', pa.int64()), ('name', pa.utf8())]))
+ self.assertEqual(result.read_all().to_pydict(), table.to_pydict())
+
+ def test_store_is_shared_and_keys_are_relative_to_the_bucket(self):
+ import vortex
+ from vortex.io import write
+
+ url = 'file://' + os.path.join(self.temp_dir, 'b.vortex')
+ store, key = vortex_utils._store_for(url, {})
+ same_store, _ = vortex_utils._store_for('file://' + os.path.join(self.temp_dir, 'c'), {})
+ self.assertIs(store, same_store)
+ self.assertEqual(key, os.path.join(self.temp_dir, 'b.vortex').lstrip('/'))
+
+ write(vortex.array(pa.table({'x': [1, 2]})), key, store=store)
+ self.assertEqual(len(vortex.open(key, store=store)), 2)
+
+ def test_compact_encodings_make_smaller_files_with_the_same_rows(self):
+ table = pa.table({'email': [f'user-{i * 7919 % 100003}@example.com' for i in range(200_000)]})
+ default_path = os.path.join(self.temp_dir, 'default.vortex')
+ compact_path = os.path.join(self.temp_dir, 'compact.vortex')
+ vortex_utils.write_vortex_file(self.file_io, default_path, table)
+ vortex_utils.write_vortex_file(self.file_io, compact_path, table, compact=True)
+
+ self.assertLess(os.path.getsize(compact_path), os.path.getsize(default_path))
+ compact = vortex_utils.open_vortex_file(self.file_io, compact_path)
+ rows = compact.to_arrow(schema=pa.schema([('email', pa.utf8())])).read_all()
+ self.assertEqual(rows.to_pydict(), table.to_pydict())
+
+ def test_table_option_selects_compact_encodings(self):
+ from unittest import mock
+
+ from pypaimon import CatalogFactory, Schema
+
+ catalog = CatalogFactory.create({'warehouse': os.path.join(self.temp_dir, 'wh')})
+ catalog.create_database('db', True)
+ pa_schema = pa.schema([('id', pa.int64())])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ pa_schema, options={'file.format': 'vortex', 'vortex.compact-encodings': 'true'}), False)
+ table = catalog.get_table('db.t')
+
+ with mock.patch.object(vortex_utils, 'write_vortex_file',
+ wraps=vortex_utils.write_vortex_file) as write:
+ builder = table.new_batch_write_builder()
+ table_write = builder.new_write()
+ commit = builder.new_commit()
+ table_write.write_arrow(pa.table({'id': [1, 2, 3]}, schema=pa_schema))
+ commit.commit(table_write.prepare_commit())
+ table_write.close()
+ commit.close()
+ self.assertTrue(write.call_args.kwargs['compact'])
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+ self.assertEqual(result.column('id').to_pylist(), [1, 2, 3])
+
+ def test_reader_slices_native_chunks_to_the_batch_size(self):
+ from pypaimon.execution.python.read.reader.format_vortex_reader import FormatVortexReader
+ from pypaimon.schema.data_types import AtomicType, DataField
+
+ path = os.path.join(self.temp_dir, 'd.vortex')
+ vortex_utils.write_vortex_file(self.file_io, path, pa.table({'id': list(range(5))}))
+
+ reader = FormatVortexReader(self.file_io, path, [DataField(0, 'id', AtomicType('BIGINT'))],
+ None, batch_size=2)
+ sizes = []
+ values = []
+ batch = reader.read_arrow_batch()
+ while batch is not None:
+ sizes.append(batch.num_rows)
+ values.extend(batch.column(0).to_pylist())
+ batch = reader.read_arrow_batch()
+ reader.close()
+
+ self.assertEqual(sizes, [2, 2, 1])
+ self.assertEqual(values, [0, 1, 2, 3, 4])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pyproject.toml b/paimon-python/pyproject.toml
index 624131379361..b7b1f65e2116 100644
--- a/paimon-python/pyproject.toml
+++ b/paimon-python/pyproject.toml
@@ -131,9 +131,10 @@ lance = [
"pylance>=0.20,<1; python_version >= '3.9'",
"pylance>=0.10,<1; python_version >= '3.8' and python_version < '3.9'",
]
-# Vortex 0.71 regresses native predicate pushdown on single-row files.
+# Keep in step with vortex.jni.version in paimon-vortex/paimon-vortex-jni/pom.xml, so that both
+# sides read the files that the other side writes.
vortex = [
- "vortex-data==0.70.0; python_version >= '3.11'",
+ "vortex-data==0.89.0; python_version >= '3.11'",
]
mosaic = [
"paimon-mosaic>=0.1.0; python_version >= '3.9'",
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/ViewBytesColumnVector.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/ViewBytesColumnVector.java
new file mode 100644
index 000000000000..bb396d5b4e22
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/ViewBytesColumnVector.java
@@ -0,0 +1,45 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.data.columnar.BytesColumnVector;
+
+import dev.vortex.arrow.ViewColumn;
+
+/** A string or binary column over a Vortex {@link ViewColumn}, copied on first use. */
+final class ViewBytesColumnVector implements BytesColumnVector {
+
+ private final ViewColumn column;
+
+ ViewBytesColumnVector(ViewColumn column) {
+ this.column = column;
+ }
+
+ @Override
+ public boolean isNullAt(int index) {
+ return column.isNull(index);
+ }
+
+ @Override
+ public Bytes getBytes(int index) {
+ int[] offsets = column.offsets();
+ int start = offsets[index];
+ return new Bytes(column.data(), start, offsets[index + 1] - start);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexArrowConverters.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexArrowConverters.java
new file mode 100644
index 000000000000..6f0238f0d3e4
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexArrowConverters.java
@@ -0,0 +1,267 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.arrow.converter.Arrow2PaimonVectorConverter;
+import org.apache.paimon.data.columnar.ByteColumnVector;
+import org.apache.paimon.data.columnar.BytesColumnVector;
+import org.apache.paimon.data.columnar.ColumnVector;
+import org.apache.paimon.data.columnar.DoubleColumnVector;
+import org.apache.paimon.data.columnar.FloatColumnVector;
+import org.apache.paimon.data.columnar.IntColumnVector;
+import org.apache.paimon.data.columnar.LongColumnVector;
+import org.apache.paimon.data.columnar.ShortColumnVector;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarBinaryType;
+import org.apache.paimon.types.VarCharType;
+
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.BaseFixedWidthVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.BitVector;
+
+import java.util.function.Function;
+
+/**
+ * Converts the Arrow vectors of Vortex scans to Paimon column vectors.
+ *
+ *
The default converters copy every string and binary value into a new array when it is read.
+ * Here, the first read of a column copies its whole data buffer to the heap once, and each value is
+ * a slice of that copy. Columns that are never read are never copied.
+ *
+ *
Fixed-width numbers are read straight from the data buffer, without boxing, and their null
+ * bits are only checked when the vector holds a null.
+ */
+final class VortexArrowConverters
+ extends Arrow2PaimonVectorConverter.Arrow2PaimonVectorConvertorVisitor {
+
+ static final VortexArrowConverters INSTANCE = new VortexArrowConverters();
+
+ private VortexArrowConverters() {}
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(CharType charType) {
+ return withHeapBytes(super.visit(charType));
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(VarCharType varCharType) {
+ return withHeapBytes(super.visit(varCharType));
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(BinaryType binaryType) {
+ return withHeapBytes(super.visit(binaryType));
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(VarBinaryType varBinaryType) {
+ return withHeapBytes(super.visit(varBinaryType));
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(TinyIntType tinyIntType) {
+ return fixedWidth(super.visit(tinyIntType), ByteColumn::new);
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(SmallIntType smallIntType) {
+ return fixedWidth(super.visit(smallIntType), ShortColumn::new);
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(IntType intType) {
+ return fixedWidth(super.visit(intType), IntColumn::new);
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(BigIntType bigIntType) {
+ return fixedWidth(super.visit(bigIntType), LongColumn::new);
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(FloatType floatType) {
+ return fixedWidth(super.visit(floatType), FloatColumn::new);
+ }
+
+ @Override
+ public Arrow2PaimonVectorConverter visit(DoubleType doubleType) {
+ return fixedWidth(super.visit(doubleType), DoubleColumn::new);
+ }
+
+ /**
+ * Uses {@code direct} for plain fixed-width vectors and {@code fallback} for anything else,
+ * such as dictionary-encoded or extension vectors.
+ */
+ private static Arrow2PaimonVectorConverter fixedWidth(
+ Arrow2PaimonVectorConverter fallback,
+ Function direct) {
+ return vector ->
+ vector instanceof BaseFixedWidthVector && !(vector instanceof BitVector)
+ ? direct.apply((BaseFixedWidthVector) vector)
+ : fallback.convertVector(vector);
+ }
+
+ /** Null checks of a fixed-width vector that skip the validity bits when there are no nulls. */
+ private abstract static class NullAware implements ColumnVector {
+
+ protected final ArrowBuf data;
+ private final BaseFixedWidthVector vector;
+ private final boolean hasNulls;
+
+ NullAware(BaseFixedWidthVector vector) {
+ this.vector = vector;
+ this.data = vector.getDataBuffer();
+ this.hasNulls = vector.getNullCount() > 0;
+ }
+
+ @Override
+ public boolean isNullAt(int index) {
+ return hasNulls && vector.isNull(index);
+ }
+ }
+
+ private static Arrow2PaimonVectorConverter withHeapBytes(Arrow2PaimonVectorConverter fallback) {
+ return vector ->
+ vector instanceof BaseVariableWidthVector
+ ? new HeapBytesColumn((BaseVariableWidthVector) vector)
+ : fallback.convertVector(vector);
+ }
+
+ /** A string or binary column whose values are slices of one heap copy of its data. */
+ private static final class HeapBytesColumn implements BytesColumnVector {
+
+ private final BaseVariableWidthVector vector;
+ private final ArrowBuf offsets;
+ private byte[] data;
+ private int base;
+
+ private HeapBytesColumn(BaseVariableWidthVector vector) {
+ this.vector = vector;
+ this.offsets = vector.getOffsetBuffer();
+ }
+
+ @Override
+ public boolean isNullAt(int index) {
+ return vector.isNull(index);
+ }
+
+ @Override
+ public Bytes getBytes(int index) {
+ if (data == null) {
+ copyData();
+ }
+ int start = offsets.getInt((long) index * BaseVariableWidthVector.OFFSET_WIDTH);
+ int end = offsets.getInt((long) (index + 1) * BaseVariableWidthVector.OFFSET_WIDTH);
+ return new Bytes(data, start - base, end - start);
+ }
+
+ private void copyData() {
+ int valueCount = vector.getValueCount();
+ if (valueCount == 0) {
+ data = new byte[0];
+ return;
+ }
+ base = offsets.getInt(0);
+ int end = offsets.getInt((long) valueCount * BaseVariableWidthVector.OFFSET_WIDTH);
+ byte[] copy = new byte[end - base];
+ vector.getDataBuffer().getBytes(base, copy);
+ data = copy;
+ }
+ }
+
+ private static final class ByteColumn extends NullAware implements ByteColumnVector {
+
+ private ByteColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public byte getByte(int index) {
+ return data.getByte(index);
+ }
+ }
+
+ private static final class ShortColumn extends NullAware implements ShortColumnVector {
+
+ private ShortColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public short getShort(int index) {
+ return data.getShort((long) index * Short.BYTES);
+ }
+ }
+
+ private static final class IntColumn extends NullAware implements IntColumnVector {
+
+ private IntColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public int getInt(int index) {
+ return data.getInt((long) index * Integer.BYTES);
+ }
+ }
+
+ private static final class LongColumn extends NullAware implements LongColumnVector {
+
+ private LongColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public long getLong(int index) {
+ return data.getLong((long) index * Long.BYTES);
+ }
+ }
+
+ private static final class FloatColumn extends NullAware implements FloatColumnVector {
+
+ private FloatColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public float getFloat(int index) {
+ return data.getFloat((long) index * Float.BYTES);
+ }
+ }
+
+ private static final class DoubleColumn extends NullAware implements DoubleColumnVector {
+
+ private DoubleColumn(BaseFixedWidthVector vector) {
+ super(vector);
+ }
+
+ @Override
+ public double getDouble(int index) {
+ return data.getDouble((long) index * Double.BYTES);
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
index d32c1669ad75..91b069dc6502 100644
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
@@ -23,7 +23,11 @@
import org.apache.paimon.format.FileFormatFactory;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.BinaryType;
@@ -53,10 +57,27 @@
import javax.annotation.Nullable;
import java.util.List;
+import java.util.Optional;
/** Vortex {@link FileFormat}. */
public class VortexFileFormat extends FileFormat {
+ public static final ConfigOption WORKER_THREADS =
+ ConfigOptions.key("vortex.worker-threads")
+ .intType()
+ .defaultValue(4)
+ .withDescription(
+ "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.");
+
+ /**
+ * Each written batch becomes a native chunk and costs a native call, so small batches make
+ * writes slow. The write batch memory still bounds every batch.
+ */
+ private static final int MIN_WRITE_BATCH_ROWS = 8192;
+
static {
System.setProperty("arrow.enable_unsafe_memory_access", "true");
}
@@ -73,7 +94,8 @@ public FormatReaderFactory createReaderFactory(
RowType dataSchemaRowType,
RowType projectedRowType,
@Nullable List predicates) {
- return new VortexReaderFactory(dataSchemaRowType, projectedRowType, predicates);
+ return new VortexReaderFactory(
+ dataSchemaRowType, projectedRowType, predicates, workerThreads());
}
@Override
@@ -82,10 +104,21 @@ public FormatWriterFactory createWriterFactory(RowType type) {
() ->
new ArrowFormatCWriter(
type,
- formatContext.writeBatchSize(),
+ Math.max(formatContext.writeBatchSize(), MIN_WRITE_BATCH_ROWS),
true,
formatContext.writeBatchMemory().getBytes(),
- null));
+ null),
+ workerThreads());
+ }
+
+ @Override
+ public Optional createStatsExtractor(
+ RowType type, SimpleColStatsCollector.Factory[] statsCollectors) {
+ return Optional.of(new VortexSimpleStatsExtractor(type, statsCollectors));
+ }
+
+ private int workerThreads() {
+ return formatContext.options().get(WORKER_THREADS);
}
@Override
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileIOReadable.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileIOReadable.java
new file mode 100644
index 000000000000..38dd9c2716aa
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileIOReadable.java
@@ -0,0 +1,155 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.VectoredReadable;
+
+import dev.vortex.io.NativeReadable;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Lets the native Vortex reader read a file through a Paimon {@link FileIO}.
+ *
+ *
Native worker threads read concurrently. A stream that supports positional reads is shared by
+ * all of them. Other streams are pooled, so each concurrent read seeks its own stream.
+ */
+final class VortexFileIOReadable implements NativeReadable {
+
+ private static final int COPY_CHUNK_SIZE = 1024 * 1024;
+
+ private final FileIO fileIO;
+ private final Path path;
+ private final long length;
+ private final ConcurrentLinkedQueue idleStreams =
+ new ConcurrentLinkedQueue<>();
+ private final List allStreams = new ArrayList<>();
+ private boolean closed;
+
+ VortexFileIOReadable(FileIO fileIO, Path path, long length) {
+ this.fileIO = fileIO;
+ this.path = path;
+ this.length = length;
+ }
+
+ @Override
+ public String name() {
+ return path.toString();
+ }
+
+ @Override
+ public long length() {
+ return length;
+ }
+
+ @Override
+ public void readFully(long position, ByteBuffer buffer) throws IOException {
+ if (position < 0 || position + buffer.remaining() > length) {
+ throw new EOFException(
+ String.format(
+ "Read of %d bytes at %d is past the end of %s (%d bytes)",
+ buffer.remaining(), position, path, length));
+ }
+
+ SeekableInputStream stream = borrow();
+ try {
+ byte[] chunk = new byte[Math.min(buffer.remaining(), COPY_CHUNK_SIZE)];
+ long offset = position;
+ while (buffer.hasRemaining()) {
+ int n = Math.min(buffer.remaining(), chunk.length);
+ read(stream, offset, chunk, n);
+ buffer.put(chunk, 0, n);
+ offset += n;
+ }
+ } finally {
+ idleStreams.offer(stream);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ List streams;
+ synchronized (this) {
+ closed = true;
+ streams = new ArrayList<>(allStreams);
+ allStreams.clear();
+ }
+ idleStreams.clear();
+
+ IOException failure = null;
+ for (SeekableInputStream stream : streams) {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private SeekableInputStream borrow() throws IOException {
+ SeekableInputStream stream = idleStreams.poll();
+ if (stream != null) {
+ return stream;
+ }
+
+ stream = fileIO.newInputStream(path);
+ synchronized (this) {
+ if (!closed) {
+ allStreams.add(stream);
+ return stream;
+ }
+ }
+ stream.close();
+ throw new IOException("Readable for " + path + " is closed");
+ }
+
+ private static void read(SeekableInputStream stream, long position, byte[] chunk, int n)
+ throws IOException {
+ if (stream instanceof VectoredReadable) {
+ ((VectoredReadable) stream).preadFully(position, chunk, 0, n);
+ return;
+ }
+
+ stream.seek(position);
+ int read = 0;
+ while (read < n) {
+ int r = stream.read(chunk, read, n - read);
+ if (r < 0) {
+ throw new EOFException("Unexpected end of stream while reading " + n + " bytes");
+ }
+ read += r;
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexInput.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexInput.java
new file mode 100644
index 000000000000..cc703bf9da25
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexInput.java
@@ -0,0 +1,66 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.Pair;
+
+import dev.vortex.api.DataSource;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Map;
+
+/** One Vortex file to read, through a native storage client or through a Paimon FileIO. */
+final class VortexInput {
+
+ @Nullable private final Pair> nativeLocation;
+ @Nullable private final VortexFileIOReadable readable;
+
+ private VortexInput(
+ @Nullable Pair> nativeLocation,
+ @Nullable VortexFileIOReadable readable) {
+ this.nativeLocation = nativeLocation;
+ this.readable = readable;
+ }
+
+ static VortexInput of(FileIO fileIO, Path path, long fileSize) {
+ Pair> location = VortexUtils.nativeLocation(fileIO, path, true);
+ if (location != null) {
+ return new VortexInput(location, null);
+ }
+ return new VortexInput(null, new VortexFileIOReadable(fileIO, path, fileSize));
+ }
+
+ /** The byte source that the reader must close after the data source, if any. */
+ @Nullable
+ Closeable readable() {
+ return readable;
+ }
+
+ DataSource open() {
+ if (nativeLocation != null) {
+ return DataSource.open(
+ VortexSessions.shared(), nativeLocation.getLeft(), nativeLocation.getRight());
+ }
+ return DataSource.open(VortexSessions.shared(), readable, 0);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutput.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutput.java
new file mode 100644
index 000000000000..4a7be7e0d8be
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutput.java
@@ -0,0 +1,90 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.utils.Pair;
+
+import dev.vortex.api.VortexWriter;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+
+/** One Vortex file to write, through a native storage client or through a Paimon stream. */
+final class VortexOutput {
+
+ private final String name;
+ @Nullable private final Pair> nativeLocation;
+ @Nullable private final VortexOutputStreamWritable writable;
+
+ private VortexOutput(
+ String name,
+ @Nullable Pair> nativeLocation,
+ @Nullable VortexOutputStreamWritable writable) {
+ this.name = name;
+ this.nativeLocation = nativeLocation;
+ this.writable = writable;
+ }
+
+ static VortexOutput of(FileIO fileIO, Path path) throws IOException {
+ Pair> location =
+ VortexUtils.nativeLocation(fileIO, path, false);
+ if (location != null) {
+ return new VortexOutput(path.toString(), location, null);
+ }
+ return new VortexOutput(
+ path.toString(),
+ null,
+ new VortexOutputStreamWritable(fileIO.newOutputStream(path, false), true));
+ }
+
+ /** A stream that its caller keeps and closes after the writer is closed. */
+ static VortexOutput of(PositionOutputStream out) {
+ return new VortexOutput("stream", null, new VortexOutputStreamWritable(out, false));
+ }
+
+ String name() {
+ return name;
+ }
+
+ /** The byte sink that the writer must close after the native writer is finished, if any. */
+ @Nullable
+ Closeable writable() {
+ return writable;
+ }
+
+ VortexWriter open(Schema arrowSchema, BufferAllocator allocator) throws IOException {
+ if (nativeLocation != null) {
+ return VortexWriter.create(
+ VortexSessions.shared(),
+ nativeLocation.getLeft(),
+ arrowSchema,
+ allocator,
+ nativeLocation.getRight());
+ }
+ return VortexWriter.create(VortexSessions.shared(), writable, arrowSchema, allocator);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutputStreamWritable.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutputStreamWritable.java
new file mode 100644
index 000000000000..659971ed642d
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexOutputStreamWritable.java
@@ -0,0 +1,59 @@
+/*
+ * 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.paimon.format.vortex;
+
+import org.apache.paimon.fs.PositionOutputStream;
+
+import dev.vortex.io.NativeWritable;
+
+import java.io.IOException;
+
+/**
+ * Lets the native Vortex writer write a file through a Paimon {@link PositionOutputStream}.
+ *
+ *
The native writer never closes it. Closing it closes the stream only when the writer owns the
+ * stream.
+ */
+final class VortexOutputStreamWritable implements NativeWritable {
+
+ private final PositionOutputStream out;
+ private final boolean ownsStream;
+
+ VortexOutputStreamWritable(PositionOutputStream out, boolean ownsStream) {
+ this.out = out;
+ this.ownsStream = ownsStream;
+ }
+
+ @Override
+ public void write(byte[] buffer, int offset, int length) throws IOException {
+ out.write(buffer, offset, length);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (ownsStream) {
+ out.close();
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
index 02326ba684af..034d2e6af645 100644
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
@@ -22,22 +22,34 @@
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.predicate.And;
+import org.apache.paimon.predicate.Between;
import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.Contains;
+import org.apache.paimon.predicate.EndsWith;
import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.GreaterOrEqual;
import org.apache.paimon.predicate.GreaterThan;
+import org.apache.paimon.predicate.In;
import org.apache.paimon.predicate.IsNotNull;
import org.apache.paimon.predicate.IsNull;
+import org.apache.paimon.predicate.LeafFunction;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LessOrEqual;
import org.apache.paimon.predicate.LessThan;
+import org.apache.paimon.predicate.NotContains;
+import org.apache.paimon.predicate.NotEndsWith;
import org.apache.paimon.predicate.NotEqual;
+import org.apache.paimon.predicate.NotIn;
+import org.apache.paimon.predicate.NotStartsWith;
import org.apache.paimon.predicate.Or;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.PredicateVisitor;
+import org.apache.paimon.predicate.StartsWith;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.DecimalType;
import org.apache.paimon.types.LocalZonedTimestampType;
import org.apache.paimon.types.TimestampType;
@@ -69,42 +81,44 @@ public Expression visit(LeafPredicate predicate) {
return null;
}
FieldRef fieldRef = fieldRefOpt.get();
+ LeafFunction function = predicate.function();
+ List