You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ho...@apache.org on 2026/04/01 02:24:43 UTC
(paimon) 25/41: [vortex] Introduce vortex file format integration (#7543)
This is an automated email from the ASF dual-hosted git repository.
hope pushed a commit to branch release-1.4
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e1b930c800bd3388cfe516722753eb7f6555a817
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Mar 30 14:28:37 2026 +0800
[vortex] Introduce vortex file format integration (#7543)
See https://vortex.dev/
> 100x faster random access
> 10-20x faster scans5x faster writes
> Similar compression ratio
> (vs. Apache Parquet)
Vortex performs well in terms of vectors, and introducing it can
significantly help us gain significant advantages in vector storage.
(https://paimon.apache.org/docs/master/append-table/vector/)
---
.github/workflows/utitcase-vortex.yml | 72 ++
.../java/org/apache/paimon/arrow/ArrowUtils.java | 9 +-
paimon-vortex/paimon-vortex-format/pom.xml | 93 +++
.../org/apache/arrow/memory/BoundsChecking.java | 24 +
.../paimon/format/vortex/VortexFileFormat.java | 219 ++++++
.../format/vortex/VortexFileFormatFactory.java | 37 ++
.../format/vortex/VortexPredicateConverter.java | 211 ++++++
.../paimon/format/vortex/VortexReaderFactory.java | 79 +++
.../paimon/format/vortex/VortexRecordsReader.java | 154 +++++
.../paimon/format/vortex/VortexRecordsWriter.java | 118 ++++
.../paimon/format/vortex/VortexTypeUtils.java | 198 ++++++
.../apache/paimon/format/vortex/VortexUtils.java | 112 ++++
.../paimon/format/vortex/VortexWriterFactory.java | 70 ++
.../org.apache.paimon.format.FileFormatFactory | 16 +
.../vortex/VortexFileFormatReadWriteTest.java | 152 +++++
.../paimon/format/vortex/VortexFileFormatTest.java | 131 ++++
.../vortex/VortexPredicateConverterTest.java | 244 +++++++
.../format/vortex/VortexReaderWriterTest.java | 366 ++++++++++
.../paimon/format/vortex/VortexTypeUtilsTest.java | 132 ++++
paimon-vortex/paimon-vortex-jni/README.md | 89 +++
paimon-vortex/paimon-vortex-jni/pom.xml | 157 +++++
.../src/main/java/dev/vortex/api/Array.java | 68 ++
.../main/java/dev/vortex/api/ArrayIterator.java | 29 +
.../src/main/java/dev/vortex/api/DType.java | 249 +++++++
.../src/main/java/dev/vortex/api/Expression.java | 56 ++
.../src/main/java/dev/vortex/api/File.java | 31 +
.../src/main/java/dev/vortex/api/Files.java | 51 ++
.../src/main/java/dev/vortex/api/ScanOptions.java | 44 ++
.../src/main/java/dev/vortex/api/VortexWriter.java | 45 ++
.../java/dev/vortex/api/expressions/Binary.java | 228 +++++++
.../java/dev/vortex/api/expressions/GetItem.java | 95 +++
.../java/dev/vortex/api/expressions/Literal.java | 733 +++++++++++++++++++++
.../main/java/dev/vortex/api/expressions/Not.java | 91 +++
.../main/java/dev/vortex/api/expressions/Root.java | 64 ++
.../java/dev/vortex/api/expressions/Unknown.java | 52 ++
.../src/main/java/dev/vortex/api/proto/DTypes.java | 224 +++++++
.../java/dev/vortex/api/proto/EndianUtils.java | 74 +++
.../java/dev/vortex/api/proto/Expressions.java | 81 +++
.../main/java/dev/vortex/api/proto/Scalars.java | 368 +++++++++++
.../dev/vortex/api/proto/TemporalMetadatas.java | 93 +++
.../java/dev/vortex/arrow/ArrowAllocation.java | 33 +
.../src/main/java/dev/vortex/jni/JNIArray.java | 168 +++++
.../main/java/dev/vortex/jni/JNIArrayIterator.java | 76 +++
.../src/main/java/dev/vortex/jni/JNIDType.java | 131 ++++
.../src/main/java/dev/vortex/jni/JNIFile.java | 78 +++
.../src/main/java/dev/vortex/jni/JNIWriter.java | 63 ++
.../dev/vortex/jni/NativeArrayIteratorMethods.java | 34 +
.../java/dev/vortex/jni/NativeArrayMethods.java | 75 +++
.../java/dev/vortex/jni/NativeDTypeMethods.java | 92 +++
.../java/dev/vortex/jni/NativeFileMethods.java | 50 ++
.../src/main/java/dev/vortex/jni/NativeLoader.java | 83 +++
.../main/java/dev/vortex/jni/NativeLogging.java | 36 +
.../java/dev/vortex/jni/NativeWriterMethods.java | 37 ++
.../paimon-vortex-jni/src/main/proto/dtype.proto | 101 +++
.../paimon-vortex-jni/src/main/proto/expr.proto | 84 +++
.../paimon-vortex-jni/src/main/proto/scalar.proto | 37 ++
.../src/main/resources/META-INF/NOTICE | 9 +
paimon-vortex/pom.xml | 39 ++
pom.xml | 2 +
59 files changed, 6585 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/utitcase-vortex.yml b/.github/workflows/utitcase-vortex.yml
new file mode 100644
index 0000000000..fa83ab22d8
--- /dev/null
+++ b/.github/workflows/utitcase-vortex.yml
@@ -0,0 +1,72 @@
+################################################################################
+# 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.
+################################################################################
+
+name: UTCase Vortex
+
+on:
+ push:
+ paths:
+ - 'paimon-vortex/**'
+ pull_request:
+ paths:
+ - 'paimon-vortex/**'
+
+env:
+ JDK_VERSION: 8
+ MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
+ cancel-in-progress: true
+
+jobs:
+ vortex_test:
+ runs-on: ubuntu-latest
+
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v6
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v5
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Install Rust toolchain
+ uses: dtolnay/rust-toolchain@stable
+
+ - name: Clone and build Vortex native library
+ run: |
+ git clone --depth 1 https://github.com/spiraldb/vortex.git ${RUNNER_TEMP}/vortex
+ cd ${RUNNER_TEMP}/vortex
+ cargo build --package vortex-jni --release
+
+ - name: Copy native library to resources
+ run: |
+ RESOURCE_DIR=paimon-vortex/paimon-vortex-jni/src/main/resources/native/linux-amd64
+ mkdir -p ${RESOURCE_DIR}
+ cp ${RUNNER_TEMP}/vortex/target/release/libvortex_jni.so ${RESOURCE_DIR}/
+
+ - name: Build and test Vortex modules
+ timeout-minutes: 30
+ run: |
+ mvn -T 2C -B -ntp clean install -DskipTests
+ mvn -B -ntp verify -pl paimon-vortex/paimon-vortex-jni,paimon-vortex/paimon-vortex-format -Dcheckstyle.skip=true -Dspotless.check.skip=true
+ env:
+ MAVEN_OPTS: -Xmx4096m
diff --git a/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowUtils.java b/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowUtils.java
index 041927b7fb..83b6123219 100644
--- a/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowUtils.java
+++ b/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowUtils.java
@@ -48,6 +48,7 @@ import org.apache.arrow.vector.types.pojo.Schema;
import javax.annotation.Nullable;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.time.Instant;
@@ -276,11 +277,15 @@ public class ArrowUtils {
return ArrowCStruct.of(array, schema);
}
+ public static byte[] serializeToIpc(VectorSchemaRoot vsr) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ serializeToIpc(vsr, out);
+ return out.toByteArray();
+ }
+
public static void serializeToIpc(VectorSchemaRoot vsr, OutputStream out) {
try (ArrowStreamWriter writer = new ArrowStreamWriter(vsr, null, out)) {
- writer.start();
writer.writeBatch();
- writer.end();
} catch (IOException e) {
throw new RuntimeException("Failed to serialize VectorSchemaRoot to IPC", e);
}
diff --git a/paimon-vortex/paimon-vortex-format/pom.xml b/paimon-vortex/paimon-vortex-format/pom.xml
new file mode 100644
index 0000000000..dd8a724a60
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/pom.xml
@@ -0,0 +1,93 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>paimon-vortex</artifactId>
+ <groupId>org.apache.paimon</groupId>
+ <version>1.5-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-vortex-format</artifactId>
+ <name>Paimon : Vortex : Format</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-vortex-jni</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-arrow</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/arrow/memory/BoundsChecking.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/arrow/memory/BoundsChecking.java
new file mode 100644
index 0000000000..a654a03055
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/arrow/memory/BoundsChecking.java
@@ -0,0 +1,24 @@
+/*
+ * 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.arrow.memory;
+
+/** Overwrite BoundsChecking in arrow, disable bounds checkpoint always. */
+public class BoundsChecking {
+ public static final boolean BOUNDS_CHECKING_ENABLED = false;
+}
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
new file mode 100644
index 0000000000..eda8a3944c
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
@@ -0,0 +1,219 @@
+/*
+ * 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.vector.ArrowFormatWriter;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.BlobType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypeVisitor;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarBinaryType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.types.VariantType;
+import org.apache.paimon.types.VectorType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** Vortex {@link FileFormat}. */
+public class VortexFileFormat extends FileFormat {
+
+ static {
+ System.setProperty("arrow.enable_unsafe_memory_access", "true");
+ }
+
+ private final FileFormatFactory.FormatContext formatContext;
+
+ public VortexFileFormat(FileFormatFactory.FormatContext formatContext) {
+ super("vortex");
+ this.formatContext = formatContext;
+ }
+
+ @Override
+ public FormatReaderFactory createReaderFactory(
+ RowType dataSchemaRowType,
+ RowType projectedRowType,
+ @Nullable List<Predicate> predicates) {
+ return new VortexReaderFactory(projectedRowType, predicates);
+ }
+
+ @Override
+ public FormatWriterFactory createWriterFactory(RowType type) {
+ return new VortexWriterFactory(
+ type,
+ () ->
+ new ArrowFormatWriter(
+ type,
+ formatContext.writeBatchSize(),
+ true,
+ formatContext.writeBatchMemory().getBytes()));
+ }
+
+ @Override
+ public void validateDataFields(RowType rowType) {
+ rowType.accept(new VortexRowTypeVisitor());
+ }
+
+ static class VortexRowTypeVisitor implements DataTypeVisitor<Void> {
+
+ @Override
+ public Void visit(CharType charType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(VarCharType varCharType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(BooleanType booleanType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(BinaryType binaryType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(VarBinaryType varBinaryType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(DecimalType decimalType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(TinyIntType tinyIntType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(SmallIntType smallIntType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(IntType intType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(BigIntType bigIntType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(FloatType floatType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(DoubleType doubleType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(DateType dateType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(TimeType timeType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(TimestampType timestampType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(LocalZonedTimestampType localZonedTimestampType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(VariantType variantType) {
+ throw new UnsupportedOperationException(
+ "Vortex file format does not support type VARIANT");
+ }
+
+ @Override
+ public Void visit(BlobType blobType) {
+ throw new UnsupportedOperationException(
+ "Vortex file format does not support type BLOB");
+ }
+
+ @Override
+ public Void visit(ArrayType arrayType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(VectorType vectorType) {
+ return null;
+ }
+
+ @Override
+ public Void visit(MultisetType multisetType) {
+ throw new UnsupportedOperationException(
+ "Vortex file format does not support type MULTISET");
+ }
+
+ @Override
+ public Void visit(MapType mapType) {
+ throw new UnsupportedOperationException("Vortex file format does not support type MAP");
+ }
+
+ @Override
+ public Void visit(RowType rowType) {
+ for (DataField field : rowType.getFields()) {
+ field.type().accept(this);
+ }
+ return null;
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormatFactory.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormatFactory.java
new file mode 100644
index 0000000000..54680635c0
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormatFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.format.FileFormatFactory;
+
+/** Factory to create {@link VortexFileFormat}. */
+public class VortexFileFormatFactory implements FileFormatFactory {
+
+ public static final String IDENTIFIER = "vortex";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public VortexFileFormat create(FormatContext formatContext) {
+ return new VortexFileFormat(formatContext);
+ }
+}
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
new file mode 100644
index 0000000000..ff1e1607a2
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
@@ -0,0 +1,211 @@
+/*
+ * 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.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.predicate.And;
+import org.apache.paimon.predicate.CompoundPredicate;
+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.IsNotNull;
+import org.apache.paimon.predicate.IsNull;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LessOrEqual;
+import org.apache.paimon.predicate.LessThan;
+import org.apache.paimon.predicate.NotEqual;
+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.types.DataType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
+
+import dev.vortex.api.Expression;
+import dev.vortex.api.expressions.Binary;
+import dev.vortex.api.expressions.GetItem;
+import dev.vortex.api.expressions.Literal;
+import dev.vortex.api.expressions.Not;
+import dev.vortex.api.expressions.Root;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** Converts Paimon {@link Predicate} to Vortex {@link Expression}. */
+public class VortexPredicateConverter implements PredicateVisitor<Expression> {
+
+ public static final VortexPredicateConverter INSTANCE = new VortexPredicateConverter();
+
+ @Nullable
+ public static Expression toVortexExpression(@Nullable List<Predicate> predicates) {
+ if (predicates == null || predicates.isEmpty()) {
+ return null;
+ }
+ return PredicateBuilder.and(predicates).visit(INSTANCE);
+ }
+
+ @Override
+ public Expression visit(LeafPredicate predicate) {
+ Optional<FieldRef> fieldRefOpt = predicate.fieldRefOptional();
+ if (!fieldRefOpt.isPresent()) {
+ return null;
+ }
+ FieldRef fieldRef = fieldRefOpt.get();
+ Expression field = GetItem.of(Root.INSTANCE, fieldRef.name());
+
+ if (predicate.function() instanceof IsNull) {
+ return Not.of(Binary.notEq(field, Literal.nullLit()));
+ }
+ if (predicate.function() instanceof IsNotNull) {
+ return Binary.notEq(field, Literal.nullLit());
+ }
+
+ List<Object> literals = predicate.literals();
+ if (literals == null || literals.isEmpty()) {
+ return null;
+ }
+
+ Literal<?> vortexLiteral = toLiteral(fieldRef.type(), literals.get(0));
+ if (vortexLiteral == null) {
+ return null;
+ }
+
+ if (predicate.function() instanceof Equal) {
+ return Binary.eq(field, vortexLiteral);
+ } else if (predicate.function() instanceof NotEqual) {
+ return Binary.notEq(field, vortexLiteral);
+ } else if (predicate.function() instanceof GreaterThan) {
+ return Binary.gt(field, vortexLiteral);
+ } else if (predicate.function() instanceof GreaterOrEqual) {
+ return Binary.gtEq(field, vortexLiteral);
+ } else if (predicate.function() instanceof LessThan) {
+ return Binary.lt(field, vortexLiteral);
+ } else if (predicate.function() instanceof LessOrEqual) {
+ return Binary.ltEq(field, vortexLiteral);
+ }
+
+ // unsupported function (e.g. In, Between)
+ return null;
+ }
+
+ @Override
+ public Expression visit(CompoundPredicate predicate) {
+ if (predicate.function() instanceof And) {
+ List<Expression> children = new ArrayList<>();
+ for (Predicate child : predicate.children()) {
+ Expression expr = child.visit(this);
+ if (expr != null) {
+ children.add(expr);
+ }
+ }
+ if (children.isEmpty()) {
+ return null;
+ }
+ Expression result = children.get(0);
+ for (int i = 1; i < children.size(); i++) {
+ result = Binary.and(result, children.get(i));
+ }
+ return result;
+ } else if (predicate.function() instanceof Or) {
+ List<Expression> children = new ArrayList<>();
+ for (Predicate child : predicate.children()) {
+ Expression expr = child.visit(this);
+ if (expr == null) {
+ return null;
+ }
+ children.add(expr);
+ }
+ Expression result = children.get(0);
+ for (int i = 1; i < children.size(); i++) {
+ result = Binary.or(result, children.get(i));
+ }
+ return result;
+ }
+
+ return null;
+ }
+
+ @Nullable
+ private static Literal<?> toLiteral(DataType type, Object value) {
+ if (value == null) {
+ return Literal.nullLit();
+ }
+ switch (type.getTypeRoot()) {
+ case BOOLEAN:
+ return Literal.bool((Boolean) value);
+ case TINYINT:
+ return Literal.int8((Byte) value);
+ case SMALLINT:
+ return Literal.int16((Short) value);
+ case INTEGER:
+ case DATE:
+ return Literal.int32((Integer) value);
+ case BIGINT:
+ return Literal.int64((Long) value);
+ case FLOAT:
+ return Literal.float32((Float) value);
+ case DOUBLE:
+ return Literal.float64((Double) value);
+ case CHAR:
+ case VARCHAR:
+ return Literal.string(
+ value instanceof BinaryString ? value.toString() : (String) value);
+ case DECIMAL:
+ Decimal decimal = (Decimal) value;
+ return Literal.decimal(
+ decimal.toBigDecimal(), decimal.precision(), decimal.scale());
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ Timestamp ts = (Timestamp) value;
+ return toTimestampLiteral(
+ ts, ((TimestampType) type).getPrecision(), Optional.empty());
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ Timestamp lzTs = (Timestamp) value;
+ return toTimestampLiteral(
+ lzTs, ((LocalZonedTimestampType) type).getPrecision(), Optional.of("UTC"));
+ default:
+ return null;
+ }
+ }
+
+ private static Literal<Long> toTimestampLiteral(
+ Timestamp ts, int precision, Optional<String> timeZone) {
+ if (precision <= 0) {
+ // seconds
+ return Literal.timestampMillis(ts.getMillisecond(), timeZone);
+ } else if (precision <= 3) {
+ // millis
+ return Literal.timestampMillis(ts.getMillisecond(), timeZone);
+ } else if (precision <= 6) {
+ // micros
+ return Literal.timestampMicros(
+ ts.getMillisecond() * 1000 + ts.getNanoOfMillisecond() / 1000, timeZone);
+ } else {
+ // nanos
+ return Literal.timestampNanos(
+ ts.getMillisecond() * 1_000_000 + ts.getNanoOfMillisecond(), timeZone);
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexReaderFactory.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexReaderFactory.java
new file mode 100644
index 0000000000..036f716a16
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexReaderFactory.java
@@ -0,0 +1,79 @@
+/*
+ * 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.InternalRow;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.RoaringBitmap32;
+
+import dev.vortex.api.Expression;
+
+import javax.annotation.Nullable;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.format.vortex.VortexUtils.toVortexSpecified;
+
+/** A factory to create Vortex Reader. */
+public class VortexReaderFactory implements FormatReaderFactory {
+
+ private final RowType projectedRowType;
+ @Nullable private final List<Predicate> predicates;
+
+ public VortexReaderFactory(RowType projectedRowType, @Nullable List<Predicate> predicates) {
+ this.projectedRowType = projectedRowType;
+ this.predicates = predicates;
+ }
+
+ @Override
+ public FileRecordReader<InternalRow> createReader(Context context) {
+ long[] rowIndices = toRowIndices(context.selection());
+ Expression predicate = VortexPredicateConverter.toVortexExpression(predicates);
+ Pair<Path, Map<String, String>> vortexSpecified =
+ toVortexSpecified(context.fileIO(), context.filePath());
+ return new VortexRecordsReader(
+ vortexSpecified.getLeft(),
+ projectedRowType,
+ rowIndices,
+ predicate,
+ vortexSpecified.getRight());
+ }
+
+ @Nullable
+ private static long[] toRowIndices(@Nullable RoaringBitmap32 selection) {
+ if (selection == null) {
+ return null;
+ }
+ long cardinality = selection.getCardinality();
+ long[] indices = new long[(int) cardinality];
+ Iterator<Integer> iter = selection.iterator();
+ int i = 0;
+ while (iter.hasNext()) {
+ indices[i++] = iter.next();
+ }
+ return indices;
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java
new file mode 100644
index 0000000000..f7cfb25e29
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java
@@ -0,0 +1,154 @@
+/*
+ * 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.reader.ArrowBatchReader;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.LongIterator;
+
+import dev.vortex.api.Array;
+import dev.vortex.api.ArrayIterator;
+import dev.vortex.api.Expression;
+import dev.vortex.api.File;
+import dev.vortex.api.Files;
+import dev.vortex.api.ImmutableScanOptions;
+import dev.vortex.arrow.ArrowAllocation;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+/** File reader for Vortex format. */
+public class VortexRecordsReader implements FileRecordReader<InternalRow> {
+
+ private final ArrowBatchReader arrowBatchReader;
+ private final Path filePath;
+ private final BufferAllocator allocator;
+ private final ArrayIterator arrayIterator;
+ private final File vortexFile;
+ private final LongIterator positionIterator;
+ private VectorSchemaRoot reuse;
+ private Array currentArray;
+ private long returnedPosition = -1;
+
+ public VortexRecordsReader(
+ Path path,
+ RowType projectedRowType,
+ @Nullable long[] rowIndices,
+ @Nullable Expression predicate,
+ Map<String, String> storageOptions) {
+ this.filePath = path;
+ this.allocator =
+ ArrowAllocation.rootAllocator()
+ .newChildAllocator("vortex-reader", 0, Long.MAX_VALUE);
+ try {
+ this.vortexFile = Files.open(path.toUri().toString(), storageOptions);
+ try {
+ ImmutableScanOptions.Builder scanBuilder = ImmutableScanOptions.builder();
+ scanBuilder.addAllColumns(projectedRowType.getFieldNames());
+ if (rowIndices != null) {
+ scanBuilder.rowIndices(rowIndices);
+ }
+ if (predicate != null) {
+ scanBuilder.predicate(predicate);
+ }
+ this.arrayIterator = vortexFile.newScan(scanBuilder.build());
+ } catch (Exception e) {
+ vortexFile.close();
+ throw e;
+ }
+ } catch (Exception e) {
+ allocator.close();
+ throw e;
+ }
+ this.arrowBatchReader = new ArrowBatchReader(projectedRowType, true);
+ this.positionIterator =
+ rowIndices != null
+ ? LongIterator.fromArray(rowIndices)
+ : LongIterator.fromRange(0, vortexFile.rowCount());
+ }
+
+ @Nullable
+ @Override
+ public FileRecordIterator<InternalRow> readBatch() throws IOException {
+ if (!arrayIterator.hasNext()) {
+ return null;
+ }
+
+ releaseCurrentArray();
+ Array array = arrayIterator.next();
+ this.currentArray = array;
+ VectorSchemaRoot vsr = array.exportToArrow(allocator, reuse);
+ this.reuse = vsr;
+ Iterator<InternalRow> rows = arrowBatchReader.readBatch(vsr).iterator();
+
+ return new FileRecordIterator<InternalRow>() {
+ @Override
+ public long returnedPosition() {
+ return returnedPosition;
+ }
+
+ @Override
+ public Path filePath() {
+ return filePath;
+ }
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ if (!rows.hasNext()) {
+ return null;
+ }
+ returnedPosition = positionIterator.next();
+ return rows.next();
+ }
+
+ @Override
+ public void releaseBatch() {
+ releaseCurrentArray();
+ }
+ };
+ }
+
+ private void releaseCurrentArray() {
+ if (currentArray != null) {
+ currentArray.close();
+ currentArray = null;
+ }
+ }
+
+ @Override
+ public void close() {
+ releaseCurrentArray();
+ if (reuse != null) {
+ reuse.close();
+ }
+ arrayIterator.close();
+ vortexFile.close();
+ allocator.close();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsWriter.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsWriter.java
new file mode 100644
index 0000000000..9691f45670
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsWriter.java
@@ -0,0 +1,118 @@
+/*
+ * 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.ArrowBundleRecords;
+import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.BundleFormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.BundleRecords;
+import org.apache.paimon.types.RowType;
+
+import dev.vortex.api.DType;
+import dev.vortex.api.VortexWriter;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.apache.paimon.arrow.ArrowUtils.serializeToIpc;
+
+/** Vortex records writer. */
+public class VortexRecordsWriter implements BundleFormatWriter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(VortexRecordsWriter.class);
+
+ private final ArrowFormatWriter arrowFormatWriter;
+ private final VortexWriter nativeWriter;
+ private final String path;
+ private long bytesWritten = 0;
+ private long jniCost = 0;
+
+ public VortexRecordsWriter(
+ RowType rowType,
+ ArrowFormatWriter arrowFormatWriter,
+ Path path,
+ Map<String, String> storageOptions)
+ throws IOException {
+ this.arrowFormatWriter = arrowFormatWriter;
+ this.path = path.toUri().toString();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ this.nativeWriter = VortexWriter.create(this.path, dtype, storageOptions);
+ }
+
+ @Override
+ public void addElement(InternalRow internalRow) throws IOException {
+ if (!arrowFormatWriter.write(internalRow)) {
+ flush();
+ if (!arrowFormatWriter.write(internalRow)) {
+ throw new RuntimeException("Exception happens while write to vortex file");
+ }
+ }
+ }
+
+ @Override
+ public void writeBundle(BundleRecords bundleRecords) throws IOException {
+ if (bundleRecords instanceof ArrowBundleRecords) {
+ writeVsr(((ArrowBundleRecords) bundleRecords).getVectorSchemaRoot());
+ } else {
+ for (InternalRow row : bundleRecords) {
+ addElement(row);
+ }
+ }
+ }
+
+ @Override
+ public boolean reachTargetSize(boolean suggestedCheck, long targetSize) {
+ // Note: bytesWritten tracks Arrow IPC serialized bytes, not the actual Vortex file size
+ // (which may differ due to Vortex's own compression/encoding).
+ return suggestedCheck && (bytesWritten > targetSize);
+ }
+
+ @Override
+ public void close() throws IOException {
+ flush();
+ LOG.info("Jni cost: {}ms for file: {}", jniCost, path);
+ long t1 = System.currentTimeMillis();
+ nativeWriter.close();
+ arrowFormatWriter.close();
+ long closeCost = (System.currentTimeMillis() - t1);
+ LOG.info("Close cost: {}ms for file: {}", closeCost, path);
+ }
+
+ private void flush() throws IOException {
+ arrowFormatWriter.flush();
+ if (!arrowFormatWriter.empty()) {
+ writeVsr(arrowFormatWriter.getVectorSchemaRoot());
+ }
+ arrowFormatWriter.reset();
+ }
+
+ private void writeVsr(VectorSchemaRoot vsr) throws IOException {
+ byte[] arrowData = serializeToIpc(vsr);
+ bytesWritten += arrowData.length;
+ long t1 = System.currentTimeMillis();
+ nativeWriter.writeBatch(arrowData);
+ jniCost += (System.currentTimeMillis() - t1);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexTypeUtils.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexTypeUtils.java
new file mode 100644
index 0000000000..271c14d196
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexTypeUtils.java
@@ -0,0 +1,198 @@
+/*
+ * 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.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeDefaultVisitor;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarBinaryType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.types.VectorType;
+
+import dev.vortex.api.DType;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Utilities for converting Paimon types to Vortex DType. */
+public class VortexTypeUtils {
+
+ public static DType toDType(RowType rowType) {
+ // Vortex does not support nullable top-level structs
+ return toStructDType(rowType, false);
+ }
+
+ private static DType toStructDType(RowType rowType, boolean isNullable) {
+ List<DataField> fields = rowType.getFields();
+ String[] fieldNames = new String[fields.size()];
+ DType[] fieldTypes = new DType[fields.size()];
+ for (int i = 0; i < fields.size(); i++) {
+ DataField field = fields.get(i);
+ fieldNames[i] = field.name();
+ fieldTypes[i] = field.type().accept(new VortexTypeVisitor());
+ }
+ return DType.newStruct(fieldNames, fieldTypes, isNullable);
+ }
+
+ private static class VortexTypeVisitor extends DataTypeDefaultVisitor<DType> {
+
+ @Override
+ public DType visit(TinyIntType tinyIntType) {
+ return DType.newByte(tinyIntType.isNullable());
+ }
+
+ @Override
+ public DType visit(SmallIntType smallIntType) {
+ return DType.newShort(smallIntType.isNullable());
+ }
+
+ @Override
+ public DType visit(IntType intType) {
+ return DType.newInt(intType.isNullable());
+ }
+
+ @Override
+ public DType visit(BigIntType bigIntType) {
+ return DType.newLong(bigIntType.isNullable());
+ }
+
+ @Override
+ public DType visit(FloatType floatType) {
+ return DType.newFloat(floatType.isNullable());
+ }
+
+ @Override
+ public DType visit(DoubleType doubleType) {
+ return DType.newDouble(doubleType.isNullable());
+ }
+
+ @Override
+ public DType visit(CharType charType) {
+ return DType.newUtf8(charType.isNullable());
+ }
+
+ @Override
+ public DType visit(VarCharType varCharType) {
+ return DType.newUtf8(varCharType.isNullable());
+ }
+
+ @Override
+ public DType visit(BooleanType booleanType) {
+ return DType.newBool(booleanType.isNullable());
+ }
+
+ @Override
+ public DType visit(BinaryType binaryType) {
+ return DType.newBinary(binaryType.isNullable());
+ }
+
+ @Override
+ public DType visit(VarBinaryType varBinaryType) {
+ return DType.newBinary(varBinaryType.isNullable());
+ }
+
+ @Override
+ public DType visit(DecimalType decimalType) {
+ return DType.newDecimal(
+ decimalType.getPrecision(), decimalType.getScale(), decimalType.isNullable());
+ }
+
+ @Override
+ public DType visit(DateType dateType) {
+ return DType.newDate(DType.TimeUnit.DAYS, dateType.isNullable());
+ }
+
+ @Override
+ public DType visit(TimeType timeType) {
+ return DType.newTime(DType.TimeUnit.MILLISECONDS, timeType.isNullable());
+ }
+
+ @Override
+ public DType visit(TimestampType timestampType) {
+ DType.TimeUnit unit;
+ int precision = timestampType.getPrecision();
+ if (precision <= 0) {
+ unit = DType.TimeUnit.SECONDS;
+ } else if (precision <= 3) {
+ unit = DType.TimeUnit.MILLISECONDS;
+ } else if (precision <= 6) {
+ unit = DType.TimeUnit.MICROSECONDS;
+ } else {
+ unit = DType.TimeUnit.NANOSECONDS;
+ }
+ return DType.newTimestamp(unit, Optional.empty(), timestampType.isNullable());
+ }
+
+ @Override
+ public DType visit(LocalZonedTimestampType lzTimestampType) {
+ DType.TimeUnit unit;
+ int precision = lzTimestampType.getPrecision();
+ if (precision <= 0) {
+ unit = DType.TimeUnit.SECONDS;
+ } else if (precision <= 3) {
+ unit = DType.TimeUnit.MILLISECONDS;
+ } else if (precision <= 6) {
+ unit = DType.TimeUnit.MICROSECONDS;
+ } else {
+ unit = DType.TimeUnit.NANOSECONDS;
+ }
+ return DType.newTimestamp(unit, Optional.of("UTC"), lzTimestampType.isNullable());
+ }
+
+ @Override
+ public DType visit(ArrayType arrayType) {
+ DType elementType = arrayType.getElementType().accept(this);
+ return DType.newList(elementType, arrayType.isNullable());
+ }
+
+ @Override
+ public DType visit(VectorType vectorType) {
+ DType elementType = vectorType.getElementType().accept(this);
+ return DType.newFixedSizeList(
+ elementType, vectorType.getLength(), vectorType.isNullable());
+ }
+
+ @Override
+ public DType visit(RowType rowType) {
+ return VortexTypeUtils.toStructDType(rowType, rowType.isNullable());
+ }
+
+ @Override
+ protected DType defaultMethod(DataType dataType) {
+ throw new UnsupportedOperationException(
+ "Vortex does not support type: " + dataType.asSQLString());
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java
new file mode 100644
index 0000000000..a228f5d9b9
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java
@@ -0,0 +1,112 @@
+/*
+ * 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.PluginFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.RESTTokenFileIO;
+import org.apache.paimon.utils.Pair;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Utils for Vortex storage options. */
+public class VortexUtils {
+
+ private static final Class<?> ossFileIOKlass;
+ private static final Class<?> pluginFileIO;
+ private static final Class<?> jindoFileIOKlass;
+
+ static {
+ Class<?> klass;
+ try {
+ klass = Class.forName("org.apache.paimon.oss.OSSFileIO");
+ } catch (ClassNotFoundException | NoClassDefFoundError e) {
+ klass = null;
+ }
+ ossFileIOKlass = klass;
+
+ try {
+ klass = Class.forName("org.apache.paimon.jindo.JindoFileIO");
+ } catch (ClassNotFoundException | NoClassDefFoundError e) {
+ klass = null;
+ }
+ jindoFileIOKlass = klass;
+
+ try {
+ klass = Class.forName("org.apache.paimon.fs.PluginFileIO");
+ } catch (ClassNotFoundException | NoClassDefFoundError e) {
+ klass = null;
+ }
+ pluginFileIO = klass;
+ }
+
+ public static Pair<Path, Map<String, String>> toVortexSpecified(FileIO fileIO, Path path) {
+ URI uri = path.toUri();
+ String schema = uri.getScheme();
+
+ if (fileIO instanceof RESTTokenFileIO) {
+ try {
+ fileIO = ((RESTTokenFileIO) fileIO).fileIO();
+ } catch (IOException e) {
+ throw new RuntimeException("Can't get fileIO from RESTTokenFileIO", e);
+ }
+ }
+
+ Options originOptions;
+ if (ossFileIOKlass != null && ossFileIOKlass.isInstance(fileIO)) {
+ originOptions = invokeHadoopOptions(fileIO);
+ } else if (jindoFileIOKlass != null && jindoFileIOKlass.isInstance(fileIO)) {
+ originOptions = invokeHadoopOptions(fileIO);
+ } else if (pluginFileIO != null && pluginFileIO.isInstance(fileIO)) {
+ originOptions = ((PluginFileIO) fileIO).options();
+ } else {
+ originOptions = new Options();
+ }
+
+ Path converted = path;
+ Map<String, String> storageOptions = new HashMap<>();
+ if ("oss".equals(schema)) {
+ storageOptions.put(
+ "endpoint",
+ "https://" + uri.getHost() + "." + originOptions.get("fs.oss.endpoint"));
+ storageOptions.put("access_key_id", originOptions.get("fs.oss.accessKeyId"));
+ storageOptions.put("secret_access_key", originOptions.get("fs.oss.accessKeySecret"));
+ storageOptions.put("virtual_hosted_style_request", "true");
+ if (originOptions.containsKey("fs.oss.securityToken")) {
+ storageOptions.put("session_token", originOptions.get("fs.oss.securityToken"));
+ }
+ converted = new Path(uri.toString().replace("oss://", "s3://"));
+ }
+
+ return Pair.of(converted, storageOptions);
+ }
+
+ private static Options invokeHadoopOptions(Object fileIO) {
+ try {
+ return (Options) fileIO.getClass().getMethod("hadoopOptions").invoke(fileIO);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to invoke hadoopOptions", e);
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
new file mode 100644
index 0000000000..fa08a80ce8
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
@@ -0,0 +1,70 @@
+/*
+ * 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.vector.ArrowFormatWriter;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.format.SupportsDirectWrite;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import static org.apache.paimon.format.vortex.VortexUtils.toVortexSpecified;
+
+/** A factory to create Vortex {@link FormatWriter}. */
+public class VortexWriterFactory implements FormatWriterFactory, SupportsDirectWrite {
+
+ private static final Logger LOG = LoggerFactory.getLogger(VortexWriterFactory.class);
+
+ private final RowType rowType;
+ private final Supplier<ArrowFormatWriter> arrowFormatWriterSupplier;
+
+ public VortexWriterFactory(
+ RowType rowType, Supplier<ArrowFormatWriter> arrowFormatWriterSupplier) {
+ this.rowType = rowType;
+ this.arrowFormatWriterSupplier = arrowFormatWriterSupplier;
+ }
+
+ @Override
+ public FormatWriter create(PositionOutputStream positionOutputStream, String compression) {
+ throw new UnsupportedOperationException(
+ "VortexWriterFactory does not support PositionOutputStream, "
+ + "please use create(FileIO fileIO, Path path, String compression) instead.");
+ }
+
+ @Override
+ public FormatWriter create(FileIO fileIO, Path path, String compression) throws IOException {
+ Pair<Path, Map<String, String>> vortexSpecified = toVortexSpecified(fileIO, path);
+ return new VortexRecordsWriter(
+ rowType,
+ arrowFormatWriterSupplier.get(),
+ vortexSpecified.getLeft(),
+ vortexSpecified.getRight());
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory b/paimon-vortex/paimon-vortex-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory
new file mode 100644
index 0000000000..554e95a7bd
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.format.vortex.VortexFileFormatFactory
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatReadWriteTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatReadWriteTest.java
new file mode 100644
index 0000000000..a270d2e92f
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatReadWriteTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.BinaryVector;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatReadWriteTest;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import dev.vortex.api.DType;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.math.BigDecimal;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.data.BinaryString.fromString;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Test for Vortex file format read/write using the base test framework. */
+public class VortexFileFormatReadWriteTest extends FormatReadWriteTest {
+
+ protected VortexFileFormatReadWriteTest() {
+ super("vortex");
+ }
+
+ @BeforeAll
+ static void checkNativeLibrary() {
+ assumeTrue(isNativeAvailable(), "Vortex native library not available, skipping tests");
+ }
+
+ private static boolean isNativeAvailable() {
+ try {
+ dev.vortex.jni.NativeLoader.loadJni();
+ return true;
+ } catch (Throwable t) {
+ return false;
+ }
+ }
+
+ private static boolean isFixedSizeListSupported() {
+ try {
+ DType.newFixedSizeList(DType.newInt(false), 2, false);
+ return true;
+ } catch (Throwable t) {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean supportNestedReadPruning() {
+ return false;
+ }
+
+ @Override
+ public String compression() {
+ return "";
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return new VortexFileFormat(new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ }
+
+ @Override
+ protected RowType rowTypeForFullTypesTest() {
+ RowType.Builder builder =
+ RowType.builder()
+ .field("id", DataTypes.INT().notNull())
+ .field("name", DataTypes.STRING())
+ .field("salary", DataTypes.DOUBLE().notNull())
+ .field("strArray", DataTypes.ARRAY(DataTypes.STRING()).nullable())
+ .field("intArray", DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .field("boolean", DataTypes.BOOLEAN().nullable())
+ .field("tinyint", DataTypes.TINYINT())
+ .field("smallint", DataTypes.SMALLINT())
+ .field("bigint", DataTypes.BIGINT())
+ .field("bytes", DataTypes.BYTES())
+ .field("timestamp", DataTypes.TIMESTAMP())
+ .field("timestamp_3", DataTypes.TIMESTAMP(3))
+ .field("date", DataTypes.DATE())
+ .field("decimal", DataTypes.DECIMAL(2, 2))
+ .field("decimal2", DataTypes.DECIMAL(38, 2))
+ .field("decimal3", DataTypes.DECIMAL(10, 1))
+ .field(
+ "rowArray",
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 0,
+ "int0",
+ DataTypes.INT().notNull(),
+ "nested row int field 0"),
+ DataTypes.FIELD(
+ 1,
+ "double1",
+ DataTypes.DOUBLE().notNull(),
+ "nested row double field 1"))))
+ .field("floatVector", DataTypes.VECTOR(3, DataTypes.FLOAT()));
+
+ RowType rowType = builder.build();
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ rowType = rowType.notNull();
+ }
+ return rowType;
+ }
+
+ @Override
+ protected GenericRow expectedRowForFullTypesTest() {
+ return GenericRow.of(
+ 1,
+ fromString("name"),
+ 5.26D,
+ new GenericArray(new Object[] {fromString("123"), fromString("456")}),
+ new GenericArray(new Object[] {123, 456}),
+ true,
+ (byte) 3,
+ (short) 6,
+ 12304L,
+ new byte[] {1, 5, 2},
+ Timestamp.fromMicros(123123123),
+ Timestamp.fromEpochMillis(123123123),
+ 2456,
+ Decimal.fromBigDecimal(new BigDecimal("0.22"), 2, 2),
+ Decimal.fromBigDecimal(new BigDecimal("12312455.22"), 38, 2),
+ Decimal.fromBigDecimal(new BigDecimal("12455.1"), 10, 1),
+ new GenericArray(new Object[] {GenericRow.of(1, 0.1D), GenericRow.of(2, 0.2D)}),
+ BinaryVector.fromPrimitiveArray(new float[] {1.0f, 2.0f, 3.0f}));
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
new file mode 100644
index 0000000000..e3925ee361
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.format.FileFormatFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Test for Vortex file format. */
+public class VortexFileFormatTest {
+
+ @Test
+ public void testIdentifier() {
+ VortexFileFormatFactory factory = new VortexFileFormatFactory();
+ assertEquals("vortex", factory.identifier());
+ }
+
+ @Test
+ public void testCreateFormat() {
+ VortexFileFormatFactory factory = new VortexFileFormatFactory();
+ VortexFileFormat format =
+ factory.create(new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ assertEquals("vortex", format.getFormatIdentifier());
+ }
+
+ @Test
+ public void testCreateReaderFactory() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ assertDoesNotThrow(() -> format.createReaderFactory(rowType, rowType, null));
+ }
+
+ @Test
+ public void testCreateWriterFactory() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ assertDoesNotThrow(() -> format.createWriterFactory(rowType));
+ }
+
+ @Test
+ public void testValidateDataFields_UnsupportedMapType() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()));
+ assertThrows(UnsupportedOperationException.class, () -> format.validateDataFields(rowType));
+ }
+
+ @Test
+ public void testValidateDataFields_UnsupportedMultisetType() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.MULTISET(DataTypes.STRING()));
+ assertThrows(UnsupportedOperationException.class, () -> format.validateDataFields(rowType));
+ }
+
+ @Test
+ public void testValidateDataFields_UnsupportedVariantType() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.VARIANT());
+ assertThrows(UnsupportedOperationException.class, () -> format.validateDataFields(rowType));
+ }
+
+ @Test
+ public void testValidateDataFields_UnsupportedBlobType() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ assertThrows(UnsupportedOperationException.class, () -> format.validateDataFields(rowType));
+ }
+
+ @Test
+ public void testValidateDataFields_SupportedTypes() {
+ VortexFileFormat format =
+ new VortexFileFormat(
+ new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .field("salary", DataTypes.DOUBLE())
+ .field("isActive", DataTypes.BOOLEAN())
+ .field("bytes", DataTypes.BYTES())
+ .field("timestamp", DataTypes.TIMESTAMP())
+ .field("date", DataTypes.DATE())
+ .field("decimal", DataTypes.DECIMAL(10, 2))
+ .field("array", DataTypes.ARRAY(DataTypes.INT()))
+ .field("tinyint", DataTypes.TINYINT())
+ .field("smallint", DataTypes.SMALLINT())
+ .field("bigint", DataTypes.BIGINT())
+ .field("float", DataTypes.FLOAT())
+ .field("time", DataTypes.TIME())
+ .field("ltz", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .field("vector", DataTypes.VECTOR(3, DataTypes.FLOAT()))
+ .build();
+
+ assertDoesNotThrow(() -> format.validateDataFields(rowType));
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java
new file mode 100644
index 0000000000..1b89659110
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java
@@ -0,0 +1,244 @@
+/*
+ * 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.Timestamp;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import dev.vortex.api.Expression;
+import dev.vortex.api.expressions.Binary;
+import dev.vortex.api.expressions.GetItem;
+import dev.vortex.api.expressions.Literal;
+import dev.vortex.api.expressions.Not;
+import dev.vortex.api.expressions.Root;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/** Tests for {@link VortexPredicateConverter}. */
+public class VortexPredicateConverterTest {
+
+ private static final RowType ROW_TYPE =
+ RowType.builder()
+ .field("f_int", DataTypes.INT())
+ .field("f_bigint", DataTypes.BIGINT())
+ .field("f_string", DataTypes.STRING())
+ .build();
+
+ private static final PredicateBuilder BUILDER = new PredicateBuilder(ROW_TYPE);
+
+ private static Expression field(String name) {
+ return GetItem.of(Root.INSTANCE, name);
+ }
+
+ @Test
+ public void testEqual() {
+ Predicate predicate = BUILDER.equal(0, 42);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.eq(field("f_int"), Literal.int32(42)), result);
+ }
+
+ @Test
+ public void testNotEqual() {
+ Predicate predicate = BUILDER.notEqual(1, 100L);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.notEq(field("f_bigint"), Literal.int64(100L)), result);
+ }
+
+ @Test
+ public void testGreaterThan() {
+ Predicate predicate = BUILDER.greaterThan(0, 10);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.gt(field("f_int"), Literal.int32(10)), result);
+ }
+
+ @Test
+ public void testGreaterOrEqual() {
+ Predicate predicate = BUILDER.greaterOrEqual(0, 10);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.gtEq(field("f_int"), Literal.int32(10)), result);
+ }
+
+ @Test
+ public void testLessThan() {
+ Predicate predicate = BUILDER.lessThan(1, 50L);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.lt(field("f_bigint"), Literal.int64(50L)), result);
+ }
+
+ @Test
+ public void testLessOrEqual() {
+ Predicate predicate = BUILDER.lessOrEqual(1, 50L);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.ltEq(field("f_bigint"), Literal.int64(50L)), result);
+ }
+
+ @Test
+ public void testIsNull() {
+ Predicate predicate = BUILDER.isNull(0);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Not.of(Binary.notEq(field("f_int"), Literal.nullLit())), result);
+ }
+
+ @Test
+ public void testIsNotNull() {
+ Predicate predicate = BUILDER.isNotNull(0);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.notEq(field("f_int"), Literal.nullLit()), result);
+ }
+
+ @Test
+ public void testAnd() {
+ Predicate p1 = BUILDER.greaterThan(0, 10);
+ Predicate p2 = BUILDER.lessThan(0, 100);
+ Predicate and = PredicateBuilder.and(p1, p2);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(and));
+ assertEquals(
+ Binary.and(
+ Binary.gt(field("f_int"), Literal.int32(10)),
+ Binary.lt(field("f_int"), Literal.int32(100))),
+ result);
+ }
+
+ @Test
+ public void testOr() {
+ Predicate p1 = BUILDER.equal(0, 1);
+ Predicate p2 = BUILDER.equal(0, 2);
+ Predicate or = PredicateBuilder.or(p1, p2);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(or));
+ assertEquals(
+ Binary.or(
+ Binary.eq(field("f_int"), Literal.int32(1)),
+ Binary.eq(field("f_int"), Literal.int32(2))),
+ result);
+ }
+
+ @Test
+ public void testMultiplePredicatesAsAnd() {
+ Predicate p1 = BUILDER.greaterThan(0, 5);
+ Predicate p2 = BUILDER.lessThan(1, 200L);
+ Expression result = VortexPredicateConverter.toVortexExpression(Arrays.asList(p1, p2));
+ assertEquals(
+ Binary.and(
+ Binary.gt(field("f_int"), Literal.int32(5)),
+ Binary.lt(field("f_bigint"), Literal.int64(200L))),
+ result);
+ }
+
+ @Test
+ public void testNullPredicates() {
+ assertNull(VortexPredicateConverter.toVortexExpression(null));
+ }
+
+ @Test
+ public void testEmptyPredicates() {
+ assertNull(VortexPredicateConverter.toVortexExpression(Collections.emptyList()));
+ }
+
+ @Test
+ public void testStringLiteral() {
+ Predicate predicate =
+ BUILDER.equal(2, org.apache.paimon.data.BinaryString.fromString("hello"));
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(Binary.eq(field("f_string"), Literal.string("hello")), result);
+ }
+
+ @Test
+ public void testTimestampMillisPrecision() {
+ // TIMESTAMP(3) should produce timestampMillis
+ RowType tsRowType = RowType.builder().field("f_ts", DataTypes.TIMESTAMP(3)).build();
+ PredicateBuilder tsBuilder = new PredicateBuilder(tsRowType);
+ Timestamp ts = Timestamp.fromEpochMillis(123456789L);
+ Predicate predicate = tsBuilder.equal(0, ts);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(
+ Binary.eq(field("f_ts"), Literal.timestampMillis(123456789L, Optional.empty())),
+ result);
+ }
+
+ @Test
+ public void testTimestampMicrosPrecision() {
+ // TIMESTAMP(6) should produce timestampMicros
+ RowType tsRowType = RowType.builder().field("f_ts", DataTypes.TIMESTAMP(6)).build();
+ PredicateBuilder tsBuilder = new PredicateBuilder(tsRowType);
+ // 123456789 millis + 123000 nanos = 123456789_123 micros
+ Timestamp ts = Timestamp.fromMicros(123456789123L);
+ Predicate predicate = tsBuilder.equal(0, ts);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(
+ Binary.eq(field("f_ts"), Literal.timestampMicros(123456789123L, Optional.empty())),
+ result);
+ }
+
+ @Test
+ public void testTimestampNanosPrecision() {
+ // TIMESTAMP(9) should produce timestampNanos
+ RowType tsRowType = RowType.builder().field("f_ts", DataTypes.TIMESTAMP(9)).build();
+ PredicateBuilder tsBuilder = new PredicateBuilder(tsRowType);
+ Timestamp ts = Timestamp.fromEpochMillis(123456L, 789012);
+ Predicate predicate = tsBuilder.equal(0, ts);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ // 123456 ms * 1_000_000 + 789012 nanos = 123456000789012 nanos
+ assertEquals(
+ Binary.eq(
+ field("f_ts"),
+ Literal.timestampNanos(123456L * 1_000_000 + 789012, Optional.empty())),
+ result);
+ }
+
+ @Test
+ public void testTimestampWithLocalTimeZone() {
+ // TIMESTAMP_LTZ(3) should produce timestampMillis with UTC
+ RowType tsRowType =
+ RowType.builder()
+ .field("f_ts_ltz", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3))
+ .build();
+ PredicateBuilder tsBuilder = new PredicateBuilder(tsRowType);
+ Timestamp ts = Timestamp.fromEpochMillis(123456789L);
+ Predicate predicate = tsBuilder.equal(0, ts);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(Collections.singletonList(predicate));
+ assertEquals(
+ Binary.eq(
+ field("f_ts_ltz"), Literal.timestampMillis(123456789L, Optional.of("UTC"))),
+ result);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java
new file mode 100644
index 0000000000..da9b3c3266
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java
@@ -0,0 +1,366 @@
+/*
+ * 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.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.SupportsDirectWrite;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.RoaringBitmap32;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Test read write for Vortex file format. */
+public class VortexReaderWriterTest {
+
+ @BeforeAll
+ static void checkNativeLibrary() {
+ assumeTrue(isNativeAvailable(), "Vortex native library not available, skipping tests");
+ }
+
+ private static boolean isNativeAvailable() {
+ try {
+ dev.vortex.jni.NativeLoader.loadJni();
+ return true;
+ } catch (Throwable t) {
+ return false;
+ }
+ }
+
+ @Test
+ public void testWriteAndRead(@TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_data_" + UUID.randomUUID());
+
+ // Write data
+ List<InternalRow> expectedRows = new ArrayList<>();
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ expectedRows.add(GenericRow.of(1, BinaryString.fromString("hello")));
+ writer.addElement(expectedRows.get(0));
+ expectedRows.add(GenericRow.of(2, BinaryString.fromString("world")));
+ writer.addElement(expectedRows.get(1));
+ }
+
+ InternalRowSerializer internalRowSerializer = new InternalRowSerializer(rowType);
+ // Read data and check
+ FormatReaderFactory readerFactory = format.createReaderFactory(rowType, rowType, null);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), null));
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(reader)) {
+ assertNotNull(reader);
+
+ List<InternalRow> actualRows = new ArrayList<>();
+ while (iterator.hasNext()) {
+ actualRows.add(internalRowSerializer.copy(iterator.next()));
+ }
+
+ assertEquals(expectedRows.size(), actualRows.size());
+ for (int i = 0; i < expectedRows.size(); i++) {
+ assertEquals(expectedRows.get(i).getInt(0), actualRows.get(i).getInt(0));
+ assertEquals(expectedRows.get(i).getString(1), actualRows.get(i).getString(1));
+ }
+ }
+ }
+
+ @Test
+ public void testWriteAndReadMultipleTypes(@TempDir java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("f_int", DataTypes.INT())
+ .field("f_bigint", DataTypes.BIGINT())
+ .field("f_double", DataTypes.DOUBLE())
+ .field("f_boolean", DataTypes.BOOLEAN())
+ .field("f_string", DataTypes.STRING())
+ .build();
+
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_multi_" + UUID.randomUUID());
+
+ // Write data
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ writer.addElement(GenericRow.of(1, 100L, 1.5D, true, BinaryString.fromString("row1")));
+ writer.addElement(GenericRow.of(2, 200L, 2.5D, false, BinaryString.fromString("row2")));
+ writer.addElement(GenericRow.of(3, 300L, 3.5D, true, BinaryString.fromString("row3")));
+ }
+
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+ // Read data and check
+ FormatReaderFactory readerFactory = format.createReaderFactory(rowType, rowType, null);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), null));
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(reader)) {
+
+ List<InternalRow> actualRows = new ArrayList<>();
+ while (iterator.hasNext()) {
+ actualRows.add(serializer.copy(iterator.next()));
+ }
+
+ assertEquals(3, actualRows.size());
+ assertEquals(1, actualRows.get(0).getInt(0));
+ assertEquals(100L, actualRows.get(0).getLong(1));
+ assertEquals(1.5D, actualRows.get(0).getDouble(2));
+ assertEquals(true, actualRows.get(0).getBoolean(3));
+ assertEquals(BinaryString.fromString("row1"), actualRows.get(0).getString(4));
+
+ assertEquals(3, actualRows.get(2).getInt(0));
+ assertEquals(300L, actualRows.get(2).getLong(1));
+ }
+ }
+
+ @Test
+ public void testReadWithSelection(@TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_selection_" + UUID.randomUUID());
+
+ // Write 5 rows
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i < 5; i++) {
+ writer.addElement(GenericRow.of(i, BinaryString.fromString("row" + i)));
+ }
+ }
+
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+
+ // Read with selection: only rows 1 and 3
+ RoaringBitmap32 selection = RoaringBitmap32.bitmapOf(1, 3);
+ FormatReaderFactory readerFactory = format.createReaderFactory(rowType, rowType, null);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO,
+ testFile,
+ fileIO.getFileSize(testFile),
+ selection));
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(reader)) {
+
+ List<InternalRow> actualRows = new ArrayList<>();
+ while (iterator.hasNext()) {
+ actualRows.add(serializer.copy(iterator.next()));
+ }
+
+ assertEquals(2, actualRows.size());
+ assertEquals(1, actualRows.get(0).getInt(0));
+ assertEquals(BinaryString.fromString("row1"), actualRows.get(0).getString(1));
+ assertEquals(3, actualRows.get(1).getInt(0));
+ assertEquals(BinaryString.fromString("row3"), actualRows.get(1).getString(1));
+ }
+ }
+
+ @Test
+ public void testReadWithPredicate(@TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("f_int", DataTypes.INT())
+ .field("f_string", DataTypes.STRING())
+ .build();
+
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_predicate_" + UUID.randomUUID());
+
+ // Write 5 rows: (0, "row0"), (1, "row1"), ..., (4, "row4")
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i < 5; i++) {
+ writer.addElement(GenericRow.of(i, BinaryString.fromString("row" + i)));
+ }
+ }
+
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+
+ // Read with predicate: f_int > 2 (should return rows 3 and 4)
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Predicate predicate = builder.greaterThan(0, 2);
+ FormatReaderFactory readerFactory =
+ format.createReaderFactory(rowType, rowType, Collections.singletonList(predicate));
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), null));
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(reader)) {
+
+ List<InternalRow> actualRows = new ArrayList<>();
+ while (iterator.hasNext()) {
+ actualRows.add(serializer.copy(iterator.next()));
+ }
+
+ assertEquals(2, actualRows.size());
+ assertEquals(3, actualRows.get(0).getInt(0));
+ assertEquals(BinaryString.fromString("row3"), actualRows.get(0).getString(1));
+ assertEquals(4, actualRows.get(1).getInt(0));
+ assertEquals(BinaryString.fromString("row4"), actualRows.get(1).getString(1));
+ }
+ }
+
+ @Test
+ public void testReturnedPositionWithRowIndices(@TempDir java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_position_" + UUID.randomUUID());
+
+ // Write 10 rows: (0, "row0"), ..., (9, "row9")
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i < 10; i++) {
+ writer.addElement(GenericRow.of(i, BinaryString.fromString("row" + i)));
+ }
+ }
+
+ // Read with selection: rows at indices 2, 5, 8
+ RoaringBitmap32 selection = RoaringBitmap32.bitmapOf(2, 5, 8);
+ FormatReaderFactory readerFactory = format.createReaderFactory(rowType, rowType, null);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), selection))) {
+
+ long[] expectedPositions = new long[] {2, 5, 8};
+ int idx = 0;
+ RecordReader.RecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ FileRecordIterator<InternalRow> fileIter = (FileRecordIterator<InternalRow>) batch;
+ InternalRow row;
+ while ((row = fileIter.next()) != null) {
+ assertEquals(
+ expectedPositions[idx],
+ fileIter.returnedPosition(),
+ "returnedPosition mismatch at index " + idx);
+ assertEquals((int) expectedPositions[idx], row.getInt(0));
+ idx++;
+ }
+ batch.releaseBatch();
+ }
+ assertEquals(3, idx, "Should have read exactly 3 rows");
+ }
+ }
+
+ @Test
+ public void testReturnedPositionSequential(@TempDir java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+ Options options = new Options();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile = new Path(new Path(tempDir.toUri()), "test_seq_pos_" + UUID.randomUUID());
+
+ // Write 5 rows
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i < 5; i++) {
+ writer.addElement(GenericRow.of(i, BinaryString.fromString("row" + i)));
+ }
+ }
+
+ // Read without selection — positions should be 0, 1, 2, 3, 4
+ FormatReaderFactory readerFactory = format.createReaderFactory(rowType, rowType, null);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), null))) {
+
+ long expectedPos = 0;
+ RecordReader.RecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ FileRecordIterator<InternalRow> fileIter = (FileRecordIterator<InternalRow>) batch;
+ InternalRow row;
+ while ((row = fileIter.next()) != null) {
+ assertEquals(
+ expectedPos,
+ fileIter.returnedPosition(),
+ "Sequential position mismatch");
+ assertEquals((int) expectedPos, row.getInt(0));
+ expectedPos++;
+ }
+ batch.releaseBatch();
+ }
+ assertEquals(5, expectedPos, "Should have read exactly 5 rows");
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexTypeUtilsTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexTypeUtilsTest.java
new file mode 100644
index 0000000000..8a06c273e5
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexTypeUtilsTest.java
@@ -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.
+ */
+
+package org.apache.paimon.format.vortex;
+
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import dev.vortex.api.DType;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Test for {@link VortexTypeUtils}. */
+public class VortexTypeUtilsTest {
+
+ @BeforeAll
+ static void checkNativeLibrary() {
+ assumeTrue(isNativeAvailable(), "Vortex native library not available, skipping tests");
+ }
+
+ private static boolean isNativeAvailable() {
+ try {
+ dev.vortex.jni.NativeLoader.loadJni();
+ return true;
+ } catch (Throwable t) {
+ return false;
+ }
+ }
+
+ @Test
+ public void testSimpleTypes() {
+ RowType rowType =
+ RowType.builder()
+ .field("f_int", DataTypes.INT())
+ .field("f_bigint", DataTypes.BIGINT())
+ .field("f_string", DataTypes.STRING())
+ .field("f_boolean", DataTypes.BOOLEAN())
+ .field("f_double", DataTypes.DOUBLE())
+ .field("f_float", DataTypes.FLOAT())
+ .build();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ assertEquals(DType.Variant.STRUCT, dtype.getVariant());
+ assertEquals(6, dtype.getFieldNames().size());
+ assertEquals("f_int", dtype.getFieldNames().get(0));
+ assertEquals("f_bigint", dtype.getFieldNames().get(1));
+ assertEquals("f_string", dtype.getFieldNames().get(2));
+ assertEquals("f_boolean", dtype.getFieldNames().get(3));
+ assertEquals("f_double", dtype.getFieldNames().get(4));
+ assertEquals("f_float", dtype.getFieldNames().get(5));
+ }
+
+ @Test
+ public void testDecimalType() {
+ RowType rowType = RowType.builder().field("f_decimal", DataTypes.DECIMAL(10, 2)).build();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ DType decimalField = dtype.getFieldTypes().get(0);
+ assertEquals(DType.Variant.DECIMAL, decimalField.getVariant());
+ assertEquals(10, decimalField.getPrecision());
+ assertEquals(2, decimalField.getScale());
+ }
+
+ @Test
+ public void testArrayType() {
+ RowType rowType =
+ RowType.builder().field("f_array", DataTypes.ARRAY(DataTypes.INT())).build();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ DType arrayField = dtype.getFieldTypes().get(0);
+ assertEquals(DType.Variant.LIST, arrayField.getVariant());
+ }
+
+ @Test
+ public void testNestedRowType() {
+ RowType innerType =
+ RowType.builder()
+ .field("inner_int", DataTypes.INT())
+ .field("inner_str", DataTypes.STRING())
+ .build();
+ RowType rowType = RowType.builder().field("f_row", innerType).build();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ DType rowField = dtype.getFieldTypes().get(0);
+ assertEquals(DType.Variant.STRUCT, rowField.getVariant());
+ assertEquals(2, rowField.getFieldNames().size());
+ assertEquals("inner_int", rowField.getFieldNames().get(0));
+ assertEquals("inner_str", rowField.getFieldNames().get(1));
+ }
+
+ @Test
+ public void testUnsupportedMapType() {
+ RowType rowType =
+ RowType.builder()
+ .field("f_map", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()))
+ .build();
+
+ assertThrows(UnsupportedOperationException.class, () -> VortexTypeUtils.toDType(rowType));
+ }
+
+ @Test
+ public void testNullability() {
+ RowType rowType =
+ RowType.builder()
+ .field("f_nullable", DataTypes.INT().nullable())
+ .field("f_not_null", DataTypes.INT().notNull())
+ .build();
+
+ DType dtype = VortexTypeUtils.toDType(rowType);
+ assertEquals(true, dtype.getFieldTypes().get(0).isNullable());
+ assertEquals(false, dtype.getFieldTypes().get(1).isNullable());
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/README.md b/paimon-vortex/paimon-vortex-jni/README.md
new file mode 100644
index 0000000000..a8ba77a471
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/README.md
@@ -0,0 +1,89 @@
+# Paimon Vortex JNI
+
+This module provides Java JNI bindings for the [Vortex](https://github.com/spiraldb/vortex) columnar file format.
+
+## Source Code Origin
+
+The Java source code under `src/main/java/dev/vortex/` is copied from the upstream
+[vortex-jni](https://github.com/spiraldb/vortex/tree/develop/java/vortex-jni) module
+(Apache License 2.0). The `dev.vortex.jni` package name is preserved to match the JNI
+native method signatures in the pre-compiled Rust library.
+
+Key adaptations from upstream:
+- Build system changed from Gradle to Maven.
+- Protobuf `.proto` files copied to `src/main/proto/` and compiled via `protobuf-maven-plugin`.
+- Target Java version set to 1.8 for Paimon compatibility.
+
+## Building the Native Library
+
+The module requires a platform-specific native library (`libvortex_jni`) built from the
+Vortex Rust project. This library is **not** published to Maven Central and must be built
+from source.
+
+### Prerequisites
+
+- **Rust toolchain**: Install via [rustup](https://rustup.rs/)
+ ```bash
+ curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
+ ```
+- **C compiler**: `gcc` (Linux) or Xcode Command Line Tools (macOS).
+
+### Build Steps
+
+```bash
+# 1. Clone the Vortex repository
+git clone https://github.com/spiraldb/vortex.git
+cd vortex
+
+# 2. Build the native library
+cargo build --package vortex-jni # debug build
+cargo build --package vortex-jni --release # release build (optimized)
+
+# 3. Copy to resources (example for macOS Apple Silicon)
+mkdir -p /path/to/paimon-vortex-jni/src/main/resources/native/darwin-aarch64
+cp target/debug/libvortex_jni.dylib \
+ /path/to/paimon-vortex-jni/src/main/resources/native/darwin-aarch64/
+```
+
+Platform-specific paths:
+
+| Platform | Directory | Library file |
+|-----------------------|----------------------|-------------------------|
+| macOS Apple Silicon | `darwin-aarch64` | `libvortex_jni.dylib` |
+| macOS Intel | `darwin-x86_64` | `libvortex_jni.dylib` |
+| Linux x86_64 | `linux-amd64` | `libvortex_jni.so` |
+| Linux aarch64 | `linux-aarch64` | `libvortex_jni.so` |
+
+You only need to provide the library for your current platform.
+
+### Expected Resources Layout
+
+```
+src/main/resources/native/
+├── darwin-aarch64/
+│ └── libvortex_jni.dylib
+├── darwin-x86_64/
+│ └── libvortex_jni.dylib
+├── linux-amd64/
+│ └── libvortex_jni.so
+└── linux-aarch64/
+ └── libvortex_jni.so
+```
+
+## Verification
+
+```bash
+mvn test -pl paimon-vortex/paimon-vortex-jni -Dcheckstyle.skip=true -Dspotless.check.skip=true
+```
+
+Tests that require the native library use `assumeTrue` to check availability at runtime.
+If the library is not found, those tests will be skipped rather than fail.
+
+## Troubleshooting
+
+| Problem | Solution |
+|---------|----------|
+| `UnsatisfiedLinkError` | Ensure the `.dylib`/`.so` is in the correct `native/{os}-{arch}/` directory. |
+| `FileNotFoundException: Library not found` | Check `os.name` and `os.arch` system properties match the directory name. |
+| Rust compilation fails | Ensure Rust toolchain is up-to-date: `rustup update`. |
+| Cross-compilation needed | Use `cargo build --target <triple>`, e.g. `x86_64-unknown-linux-gnu`. |
diff --git a/paimon-vortex/paimon-vortex-jni/pom.xml b/paimon-vortex/paimon-vortex-jni/pom.xml
new file mode 100644
index 0000000000..27b6a5e483
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/pom.xml
@@ -0,0 +1,157 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>paimon-vortex</artifactId>
+ <groupId>org.apache.paimon</groupId>
+ <version>1.5-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-vortex-jni</artifactId>
+ <name>Paimon : Vortex JNI</name>
+
+ <properties>
+ <target.java.version>1.8</target.java.version>
+ <vortex.jni.version>0.24.0</vortex.jni.version>
+ <protobuf.version>3.25.5</protobuf.version>
+ <immutables.version>2.10.1</immutables.version>
+ <!-- vortex-jni uses shaded Guava, skip spotless/checkstyle rewriting -->
+ <spotless.check.skip>true</spotless.check.skip>
+ <spotless.apply.skip>true</spotless.apply.skip>
+ <checkstyle.skip>true</checkstyle.skip>
+ </properties>
+
+ <dependencies>
+ <!-- Arrow C Data Interface -->
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-c-data</artifactId>
+ <version>${arrow.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-core</artifactId>
+ <version>${arrow.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-netty</artifactId>
+ <version>${arrow.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-vector</artifactId>
+ <version>${arrow.version}</version>
+ </dependency>
+
+ <!-- Protobuf -->
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <version>${protobuf.version}</version>
+ </dependency>
+
+ <!-- Guava (shaded) -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-shade-guava-30</artifactId>
+ <version>${paimon.shade.guava.version}-${paimon.shade.version}</version>
+ </dependency>
+
+ <!-- Immutables (compile-only annotation processor) -->
+ <dependency>
+ <groupId>org.immutables</groupId>
+ <artifactId>value</artifactId>
+ <version>${immutables.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit5.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Protobuf compilation -->
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <version>0.6.1</version>
+ <configuration>
+ <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
+ <protoSourceRoot>${project.basedir}/src/main/proto</protoSourceRoot>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- OS detection for protoc -->
+ <plugin>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>1.7.1</version>
+ <executions>
+ <execution>
+ <phase>initialize</phase>
+ <goals>
+ <goal>detect</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Surefire config for JNI (following paimon-lumina pattern) -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <forkCount>1</forkCount>
+ <redirectTestOutputToFile>true</redirectTestOutputToFile>
+ <parallel>none</parallel>
+ </configuration>
+ </plugin>
+ </plugins>
+
+ <extensions>
+ <extension>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>1.7.1</version>
+ </extension>
+ </extensions>
+ </build>
+</project>
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Array.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Array.java
new file mode 100644
index 0000000000..e05f4d9b3c
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Array.java
@@ -0,0 +1,68 @@
+/*
+ * 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 dev.vortex.api;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+
+import java.math.BigDecimal;
+
+/** Vortex array interface for accessing columnar data. */
+public interface Array extends AutoCloseable {
+ long getLen();
+
+ long nbytes();
+
+ VectorSchemaRoot exportToArrow(BufferAllocator allocator, VectorSchemaRoot reuse);
+
+ DType getDataType();
+
+ Array getField(int index);
+
+ Array slice(int start, int stop);
+
+ boolean getNull(int index);
+
+ int getNullCount();
+
+ byte getByte(int index);
+
+ short getShort(int index);
+
+ int getInt(int index);
+
+ long getLong(int index);
+
+ boolean getBool(int index);
+
+ float getFloat(int index);
+
+ double getDouble(int index);
+
+ BigDecimal getBigDecimal(int index);
+
+ String getUTF8(int index);
+
+ void getUTF8_ptr_len(int index, long[] ptr, int[] len);
+
+ byte[] getBinary(int index);
+
+ @Override
+ void close();
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ArrayIterator.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ArrayIterator.java
new file mode 100644
index 0000000000..41df75e53a
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ArrayIterator.java
@@ -0,0 +1,29 @@
+/*
+ * 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 dev.vortex.api;
+
+import java.util.Iterator;
+
+/** An iterator over Vortex arrays with type information and resource management. */
+public interface ArrayIterator extends AutoCloseable, Iterator<Array> {
+ DType getDataType();
+
+ @Override
+ void close();
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DType.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DType.java
new file mode 100644
index 0000000000..7f834c7d36
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DType.java
@@ -0,0 +1,249 @@
+/*
+ * 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 dev.vortex.api;
+
+import dev.vortex.jni.JNIDType;
+import dev.vortex.jni.NativeDTypeMethods;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Vortex logical type interface representing the schema and metadata for array data. */
+public interface DType extends AutoCloseable {
+
+ Variant getVariant();
+
+ boolean isNullable();
+
+ List<String> getFieldNames();
+
+ List<DType> getFieldTypes();
+
+ DType getElementType();
+
+ int getFixedSizeListSize();
+
+ boolean isDate();
+
+ boolean isTime();
+
+ boolean isTimestamp();
+
+ TimeUnit getTimeUnit();
+
+ Optional<String> getTimeZone();
+
+ boolean isDecimal();
+
+ int getPrecision();
+
+ byte getScale();
+
+ @Override
+ void close();
+
+ static DType newByte(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newByte(isNullable), true);
+ }
+
+ static DType newShort(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newShort(isNullable), true);
+ }
+
+ static DType newInt(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newInt(isNullable), true);
+ }
+
+ static DType newLong(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newLong(isNullable), true);
+ }
+
+ static DType newFloat(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newFloat(isNullable), true);
+ }
+
+ static DType newDouble(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newDouble(isNullable), true);
+ }
+
+ static DType newDecimal(int precision, int scale, boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newDecimal(precision, scale, isNullable), true);
+ }
+
+ static DType newUtf8(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newUtf8(isNullable), true);
+ }
+
+ static DType newBinary(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newBinary(isNullable), true);
+ }
+
+ static DType newBool(boolean isNullable) {
+ return new JNIDType(NativeDTypeMethods.newBool(isNullable), true);
+ }
+
+ static DType newList(DType element, boolean isNullable) {
+ JNIDType jniType = (JNIDType) element;
+ return new JNIDType(NativeDTypeMethods.newList(jniType.getPointer(), isNullable), true);
+ }
+
+ static DType newFixedSizeList(DType element, int size, boolean isNullable) {
+ JNIDType jniType = (JNIDType) element;
+ return new JNIDType(
+ NativeDTypeMethods.newFixedSizeList(jniType.getPointer(), size, isNullable), true);
+ }
+
+ static DType newStruct(String[] fieldNames, DType[] fieldTypes, boolean isNullable) {
+ long[] ptrs = new long[fieldTypes.length];
+ for (int i = 0; i < fieldTypes.length; i++) {
+ ptrs[i] = ((JNIDType) fieldTypes[i]).getPointer();
+ }
+ return new JNIDType(NativeDTypeMethods.newStruct(fieldNames, ptrs, isNullable), true);
+ }
+
+ static DType newTimestamp(TimeUnit unit, Optional<String> timeZone, boolean isNullable) {
+ byte timeUnit = unit.asByte();
+ return new JNIDType(
+ NativeDTypeMethods.newTimestamp(timeUnit, timeZone.orElse(null), isNullable), true);
+ }
+
+ static DType newDate(TimeUnit unit, boolean isNullable) {
+ byte timeUnit = unit.asByte();
+ return new JNIDType(NativeDTypeMethods.newDate(timeUnit, isNullable), true);
+ }
+
+ static DType newTime(TimeUnit unit, boolean isNullable) {
+ byte timeUnit = unit.asByte();
+ return new JNIDType(NativeDTypeMethods.newTime(timeUnit, isNullable), true);
+ }
+
+ /** Time units supported by Vortex temporal data types. */
+ enum TimeUnit {
+ NANOSECONDS,
+ MICROSECONDS,
+ MILLISECONDS,
+ SECONDS,
+ DAYS;
+
+ public static TimeUnit from(byte unit) {
+ switch (unit) {
+ case 0:
+ return NANOSECONDS;
+ case 1:
+ return MICROSECONDS;
+ case 2:
+ return MILLISECONDS;
+ case 3:
+ return SECONDS;
+ case 4:
+ return DAYS;
+ default:
+ throw new IllegalArgumentException("Unknown TimeUnit: " + unit);
+ }
+ }
+
+ public byte asByte() {
+ switch (this) {
+ case NANOSECONDS:
+ return 0;
+ case MICROSECONDS:
+ return 1;
+ case MILLISECONDS:
+ return 2;
+ case SECONDS:
+ return 3;
+ case DAYS:
+ return 4;
+ default:
+ throw new IllegalArgumentException("Unknown TimeUnit: " + this);
+ }
+ }
+ }
+
+ /** All supported data type variants in Vortex. */
+ enum Variant {
+ NULL,
+ BOOL,
+ PRIMITIVE_U8,
+ PRIMITIVE_U16,
+ PRIMITIVE_U32,
+ PRIMITIVE_U64,
+ PRIMITIVE_I8,
+ PRIMITIVE_I16,
+ PRIMITIVE_I32,
+ PRIMITIVE_I64,
+ PRIMITIVE_F16,
+ PRIMITIVE_F32,
+ PRIMITIVE_F64,
+ UTF8,
+ BINARY,
+ STRUCT,
+ LIST,
+ EXTENSION,
+ DECIMAL,
+ FIXED_SIZE_LIST;
+
+ public static Variant from(byte variant) {
+ switch (variant) {
+ case 0:
+ return NULL;
+ case 1:
+ return BOOL;
+ case 2:
+ return PRIMITIVE_U8;
+ case 3:
+ return PRIMITIVE_U16;
+ case 4:
+ return PRIMITIVE_U32;
+ case 5:
+ return PRIMITIVE_U64;
+ case 6:
+ return PRIMITIVE_I8;
+ case 7:
+ return PRIMITIVE_I16;
+ case 8:
+ return PRIMITIVE_I32;
+ case 9:
+ return PRIMITIVE_I64;
+ case 10:
+ return PRIMITIVE_F16;
+ case 11:
+ return PRIMITIVE_F32;
+ case 12:
+ return PRIMITIVE_F64;
+ case 13:
+ return UTF8;
+ case 14:
+ return BINARY;
+ case 15:
+ return STRUCT;
+ case 16:
+ return LIST;
+ case 17:
+ return EXTENSION;
+ case 18:
+ return DECIMAL;
+ case 19:
+ return FIXED_SIZE_LIST;
+ default:
+ throw new IllegalArgumentException("Unknown DType variant: " + variant);
+ }
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Expression.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Expression.java
new file mode 100644
index 0000000000..4211611425
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Expression.java
@@ -0,0 +1,56 @@
+/*
+ * 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 dev.vortex.api;
+
+import dev.vortex.api.expressions.Binary;
+import dev.vortex.api.expressions.GetItem;
+import dev.vortex.api.expressions.Literal;
+import dev.vortex.api.expressions.Not;
+import dev.vortex.api.expressions.Root;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Vortex expression language. */
+public interface Expression {
+ String id();
+
+ List<Expression> children();
+
+ Optional<byte[]> metadata();
+
+ default <T> T accept(Visitor<T> visitor) {
+ return visitor.visitOther(this);
+ }
+
+ /** Visitor interface for expression tree traversal. */
+ interface Visitor<T> {
+ T visitLiteral(Literal<?> literal);
+
+ T visitRoot(Root root);
+
+ T visitBinary(Binary binary);
+
+ T visitNot(Not not);
+
+ T visitGetItem(GetItem getItem);
+
+ T visitOther(Expression expression);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/File.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/File.java
new file mode 100644
index 0000000000..beeb8f7548
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/File.java
@@ -0,0 +1,31 @@
+/*
+ * 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 dev.vortex.api;
+
+/** Interface for reading Vortex format files. */
+public interface File extends AutoCloseable {
+ DType getDType();
+
+ long rowCount();
+
+ ArrayIterator newScan(ScanOptions options);
+
+ @Override
+ void close();
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Files.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Files.java
new file mode 100644
index 0000000000..4fb7ed9b31
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Files.java
@@ -0,0 +1,51 @@
+/*
+ * 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 dev.vortex.api;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+
+import dev.vortex.jni.JNIFile;
+import dev.vortex.jni.NativeFileMethods;
+
+import java.net.URI;
+import java.nio.file.Paths;
+import java.util.Map;
+
+/** Utility class for opening Vortex files. */
+public final class Files {
+
+ private Files() {}
+
+ public static File open(String path) {
+ return open(path, java.util.Collections.emptyMap());
+ }
+
+ public static File open(String path, Map<String, String> properties) {
+ if (path.startsWith("/")) {
+ return open(Paths.get(path).toUri(), properties);
+ }
+ return open(URI.create(path), properties);
+ }
+
+ public static File open(URI uri, Map<String, String> properties) {
+ long ptr = NativeFileMethods.open(uri.toString(), properties);
+ Preconditions.checkArgument(ptr > 0, "Failed to open file: %s", uri);
+ return new JNIFile(ptr);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ScanOptions.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ScanOptions.java
new file mode 100644
index 0000000000..4afd2d4ab4
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ScanOptions.java
@@ -0,0 +1,44 @@
+/*
+ * 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 dev.vortex.api;
+
+import org.immutables.value.Value;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Options for configuring a Vortex file scan. */
+@Value.Immutable
+public interface ScanOptions {
+ List<String> columns();
+
+ Optional<Expression> predicate();
+
+ Optional<long[]> rowRange();
+
+ Optional<long[]> rowIndices();
+
+ static ScanOptions of() {
+ return ImmutableScanOptions.builder().build();
+ }
+
+ static ImmutableScanOptions.Builder builder() {
+ return ImmutableScanOptions.builder();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java
new file mode 100644
index 0000000000..4b25837821
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/VortexWriter.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 dev.vortex.api;
+
+import dev.vortex.jni.JNIDType;
+import dev.vortex.jni.JNIWriter;
+import dev.vortex.jni.NativeWriterMethods;
+
+import java.io.IOException;
+import java.util.Map;
+
+/** Writer for creating Vortex files from Arrow data. */
+public interface VortexWriter extends AutoCloseable {
+
+ static VortexWriter create(String uri, DType dtype, Map<String, String> options)
+ throws IOException {
+ long ptr = NativeWriterMethods.create(uri, ((JNIDType) dtype).getPointer(), options);
+ if (ptr <= 0) {
+ throw new IOException(
+ "Failed to create Vortex writer for: " + uri + " (got ptr=" + ptr + ")");
+ }
+ return new JNIWriter(ptr);
+ }
+
+ void writeBatch(byte[] arrowData) throws IOException;
+
+ @Override
+ void close() throws IOException;
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Binary.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Binary.java
new file mode 100644
index 0000000000..0761fddfa5
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Binary.java
@@ -0,0 +1,228 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import dev.vortex.api.Expression;
+import dev.vortex.proto.ExprProtos;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+/** Binary expression operating on two child expressions with a binary operator. */
+public final class Binary implements Expression {
+ private final BinaryOp operator;
+ private final Expression left;
+ private final Expression right;
+
+ private Binary(BinaryOp operator, Expression left, Expression right) {
+ this.operator = operator;
+ this.left = left;
+ this.right = right;
+ }
+
+ public static Binary parse(byte[] metadata, List<Expression> children) {
+ if (children.size() != 2) {
+ throw new IllegalArgumentException(
+ "Binary expression must have exactly two children, found: " + children.size());
+ }
+ try {
+ ExprProtos.BinaryOpts opts = ExprProtos.BinaryOpts.parseFrom(metadata);
+ BinaryOp operator = BinaryOp.fromProto(opts.getOp());
+ return new Binary(operator, children.get(0), children.get(1));
+ } catch (InvalidProtocolBufferException e) {
+ throw new IllegalArgumentException("Failed to parse Binary metadata", e);
+ }
+ }
+
+ public static Binary of(BinaryOp operator, Expression left, Expression right) {
+ return new Binary(operator, left, right);
+ }
+
+ public static Binary and(Expression first, Expression... rest) {
+ Expression rhs = Stream.of(rest).reduce(Binary::and).orElse(Literal.bool(true));
+ return new Binary(BinaryOp.AND, first, rhs);
+ }
+
+ public static Binary or(Expression first, Expression... rest) {
+ Expression rhs = Stream.of(rest).reduce(Binary::or).orElse(Literal.bool(false));
+ return new Binary(BinaryOp.OR, first, rhs);
+ }
+
+ public static Binary eq(Expression left, Expression right) {
+ return new Binary(BinaryOp.EQ, left, right);
+ }
+
+ public static Binary notEq(Expression left, Expression right) {
+ return new Binary(BinaryOp.NOT_EQ, left, right);
+ }
+
+ public static Binary gt(Expression left, Expression right) {
+ return new Binary(BinaryOp.GT, left, right);
+ }
+
+ public static Binary gtEq(Expression left, Expression right) {
+ return new Binary(BinaryOp.GT_EQ, left, right);
+ }
+
+ public static Binary lt(Expression left, Expression right) {
+ return new Binary(BinaryOp.LT, left, right);
+ }
+
+ public static Binary ltEq(Expression left, Expression right) {
+ return new Binary(BinaryOp.LT_EQ, left, right);
+ }
+
+ @Override
+ public String id() {
+ return "vortex.binary";
+ }
+
+ @Override
+ public List<Expression> children() {
+ return java.util.Arrays.asList(left, right);
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(
+ ExprProtos.BinaryOpts.newBuilder().setOp(operator.toProto()).build().toByteArray());
+ }
+
+ @Override
+ public String toString() {
+ return "(" + left + " " + operator + " " + right + ")";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) return false;
+ Binary binary = (Binary) o;
+ return operator == binary.operator
+ && Objects.equals(left, binary.left)
+ && Objects.equals(right, binary.right);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(operator, left, right);
+ }
+
+ @Override
+ public <T> T accept(Visitor<T> visitor) {
+ return visitor.visitBinary(this);
+ }
+
+ public BinaryOp getOperator() {
+ return operator;
+ }
+
+ public Expression getLeft() {
+ return left;
+ }
+
+ public Expression getRight() {
+ return right;
+ }
+
+ /** Binary operators supported by binary expressions. */
+ public enum BinaryOp {
+ EQ,
+ NOT_EQ,
+ GT,
+ GT_EQ,
+ LT,
+ LT_EQ,
+ AND,
+ OR;
+
+ @Override
+ public String toString() {
+ switch (this) {
+ case EQ:
+ return "==";
+ case NOT_EQ:
+ return "!=";
+ case GT:
+ return ">";
+ case GT_EQ:
+ return ">=";
+ case LT:
+ return "<";
+ case LT_EQ:
+ return "<=";
+ case AND:
+ return "&&";
+ case OR:
+ return "||";
+ default:
+ throw new IllegalStateException("Unknown Operator: " + this);
+ }
+ }
+
+ static BinaryOp fromProto(ExprProtos.BinaryOpts.BinaryOp proto) {
+ switch (proto) {
+ case Eq:
+ return EQ;
+ case NotEq:
+ return NOT_EQ;
+ case Gt:
+ return GT;
+ case Gte:
+ return GT_EQ;
+ case Lt:
+ return LT;
+ case Lte:
+ return LT_EQ;
+ case And:
+ return AND;
+ case Or:
+ return OR;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported binary operator proto: " + proto);
+ }
+ }
+
+ ExprProtos.BinaryOpts.BinaryOp toProto() {
+ switch (this) {
+ case EQ:
+ return ExprProtos.BinaryOpts.BinaryOp.Eq;
+ case NOT_EQ:
+ return ExprProtos.BinaryOpts.BinaryOp.NotEq;
+ case GT:
+ return ExprProtos.BinaryOpts.BinaryOp.Gt;
+ case GT_EQ:
+ return ExprProtos.BinaryOpts.BinaryOp.Gte;
+ case LT:
+ return ExprProtos.BinaryOpts.BinaryOp.Lt;
+ case LT_EQ:
+ return ExprProtos.BinaryOpts.BinaryOp.Lte;
+ case AND:
+ return ExprProtos.BinaryOpts.BinaryOp.And;
+ case OR:
+ return ExprProtos.BinaryOpts.BinaryOp.Or;
+ default:
+ throw new IllegalArgumentException("Unsupported binary operator: " + this);
+ }
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/GetItem.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/GetItem.java
new file mode 100644
index 0000000000..4cfcb8561b
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/GetItem.java
@@ -0,0 +1,95 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import dev.vortex.api.Expression;
+import dev.vortex.proto.ExprProtos;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Expression that extracts a field from a parent expression. */
+public final class GetItem implements Expression {
+ private final String path;
+ private final Expression child;
+
+ private GetItem(Expression child, String path) {
+ this.child = child;
+ this.path = path;
+ }
+
+ public static GetItem of(Expression child, String path) {
+ return new GetItem(child, path);
+ }
+
+ public static GetItem parse(byte[] metadata, List<Expression> children) {
+ if (children.size() != 1) {
+ throw new IllegalArgumentException(
+ "GetItem expression must have exactly one child, found: " + children.size());
+ }
+ try {
+ ExprProtos.GetItemOpts opts = ExprProtos.GetItemOpts.parseFrom(metadata);
+ return new GetItem(children.get(0), opts.getPath());
+ } catch (InvalidProtocolBufferException e) {
+ throw new IllegalArgumentException("Failed to parse GetItem metadata", e);
+ }
+ }
+
+ public Expression getChild() {
+ return child;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ @Override
+ public String id() {
+ return "vortex.get_item";
+ }
+
+ @Override
+ public List<Expression> children() {
+ return java.util.Collections.singletonList(child);
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(ExprProtos.GetItemOpts.newBuilder().setPath(path).build().toByteArray());
+ }
+
+ @Override
+ public <T> T accept(Visitor<T> visitor) {
+ return visitor.visitGetItem(this);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof GetItem)) return false;
+ GetItem getItem = (GetItem) o;
+ return Objects.equals(path, getItem.path);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(path);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Literal.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Literal.java
new file mode 100644
index 0000000000..bd8ca9db75
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Literal.java
@@ -0,0 +1,733 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import dev.vortex.api.Expression;
+import dev.vortex.api.proto.EndianUtils;
+import dev.vortex.api.proto.Scalars;
+import dev.vortex.api.proto.TemporalMetadatas;
+import dev.vortex.proto.DTypeProtos;
+import dev.vortex.proto.ExprProtos;
+import dev.vortex.proto.ScalarProtos;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Literal value expression in the Vortex query system. */
+public abstract class Literal<T> implements Expression {
+ private final T value;
+
+ private Literal(T value) {
+ this.value = value;
+ }
+
+ public static Literal<?> parse(byte[] metadata, List<Expression> children) {
+ if (!children.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Literal expression must have no children, found: " + children.size());
+ }
+ try {
+ ExprProtos.LiteralOpts opts = ExprProtos.LiteralOpts.parseFrom(metadata);
+ return deserializeLiteral(opts, children);
+ } catch (InvalidProtocolBufferException e) {
+ throw new IllegalArgumentException("Failed to parse literal metadata", e);
+ }
+ }
+
+ public T getValue() {
+ return this.value;
+ }
+
+ @Override
+ public String id() {
+ return "vortex.literal";
+ }
+
+ @Override
+ public List<Expression> children() {
+ return java.util.Collections.emptyList();
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(
+ ExprProtos.LiteralOpts.newBuilder()
+ .setValue(this.acceptLiteralVisitor(LiteralToScalar.INSTANCE))
+ .build()
+ .toByteArray());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(getValue());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Literal)) return false;
+ Literal<?> literal = (Literal<?>) o;
+ return Objects.equals(value, literal.value);
+ }
+
+ public static Literal<Void> nullLit() {
+ return NullLiteral.INSTANCE;
+ }
+
+ public static Literal<Boolean> bool(Boolean value) {
+ return new BooleanLiteral(value);
+ }
+
+ public static Literal<Byte> int8(Byte value) {
+ return new Int8Literal(value);
+ }
+
+ public static Literal<Short> int16(Short value) {
+ return new Int16Literal(value);
+ }
+
+ public static Literal<Integer> int32(Integer value) {
+ return new Int32Literal(value);
+ }
+
+ public static Literal<Long> int64(Long value) {
+ return new Int64Literal(value);
+ }
+
+ public static Literal<Float> float32(Float value) {
+ return new Float32Literal(value);
+ }
+
+ public static Literal<Double> float64(Double value) {
+ return new Float64Literal(value);
+ }
+
+ public static Literal<BigDecimal> decimal(BigDecimal value, int precision, int scale) {
+ return new DecimalLiteral(value, precision, scale);
+ }
+
+ public static Literal<String> string(String value) {
+ return new StringLiteral(value);
+ }
+
+ public static Literal<byte[]> bytes(byte[] value) {
+ return new BytesLiteral(value);
+ }
+
+ public static Literal<Integer> timeSeconds(Integer value) {
+ return new TimeSeconds(value);
+ }
+
+ public static Literal<Integer> timeMillis(Integer value) {
+ return new TimeMillis(value);
+ }
+
+ public static Literal<Long> timeMicros(Long value) {
+ return new TimeMicros(value);
+ }
+
+ public static Literal<Long> timeNanos(Long value) {
+ return new TimeNanos(value);
+ }
+
+ public static Literal<Integer> dateDays(Integer value) {
+ return new DateDays(value);
+ }
+
+ public static Literal<Long> dateMillis(Long value) {
+ return new DateMillis(value);
+ }
+
+ public static Literal<Long> timestampMillis(Long value, Optional<String> timeZone) {
+ return new TimestampMillis(value, timeZone);
+ }
+
+ public static Literal<Long> timestampMicros(Long value, Optional<String> timeZone) {
+ return new TimestampMicros(value, timeZone);
+ }
+
+ public static Literal<Long> timestampNanos(Long value, Optional<String> timeZone) {
+ return new TimestampNanos(value, timeZone);
+ }
+
+ @Override
+ public <R> R accept(Expression.Visitor<R> visitor) {
+ return visitor.visitLiteral(this);
+ }
+
+ public abstract <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor);
+
+ /** Visitor interface for processing literals in a type-safe manner. */
+ public interface LiteralVisitor<U> {
+ U visitNull();
+
+ U visitBoolean(Boolean literal);
+
+ U visitInt8(Byte literal);
+
+ U visitInt16(Short literal);
+
+ U visitInt32(Integer literal);
+
+ U visitInt64(Long literal);
+
+ U visitDateDays(Integer days);
+
+ U visitDateMillis(Long millis);
+
+ U visitTimeSeconds(Integer seconds);
+
+ U visitTimeMillis(Integer seconds);
+
+ U visitTimeMicros(Long seconds);
+
+ U visitTimeNanos(Long seconds);
+
+ U visitTimestampMillis(Long epochMillis, Optional<String> timeZone);
+
+ U visitTimestampMicros(Long epochMicros, Optional<String> timeZone);
+
+ U visitTimestampNanos(Long epochNanos, Optional<String> timeZone);
+
+ U visitFloat32(Float literal);
+
+ U visitFloat64(Double literal);
+
+ U visitDecimal(BigDecimal decimal, int precision, int scale);
+
+ U visitString(String literal);
+
+ U visitBytes(byte[] literal);
+ }
+
+ static final class NullLiteral extends Literal<Void> {
+ static final NullLiteral INSTANCE = new NullLiteral();
+
+ private NullLiteral() {
+ super(null);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitNull();
+ }
+ }
+
+ static final class BooleanLiteral extends Literal<Boolean> {
+ BooleanLiteral(Boolean value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitBoolean(getValue());
+ }
+ }
+
+ static final class Int8Literal extends Literal<Byte> {
+ Int8Literal(Byte value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitInt8(getValue());
+ }
+ }
+
+ static final class Int16Literal extends Literal<Short> {
+ Int16Literal(Short value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitInt16(getValue());
+ }
+ }
+
+ static final class Int32Literal extends Literal<Integer> {
+ Int32Literal(Integer value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitInt32(getValue());
+ }
+ }
+
+ static final class Int64Literal extends Literal<Long> {
+ Int64Literal(Long value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitInt64(getValue());
+ }
+ }
+
+ static final class Float32Literal extends Literal<Float> {
+ Float32Literal(Float value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitFloat32(getValue());
+ }
+ }
+
+ static final class Float64Literal extends Literal<Double> {
+ Float64Literal(Double value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitFloat64(getValue());
+ }
+ }
+
+ static final class DecimalLiteral extends Literal<BigDecimal> {
+ private final int precision;
+ private final int scale;
+
+ DecimalLiteral(BigDecimal value, int precision, int scale) {
+ super(value);
+ if (!Objects.isNull(value)) {
+ Preconditions.checkArgument(
+ scale == value.scale(), "scale %s ≠value scale %s", scale, value.scale());
+ }
+ this.precision = precision;
+ this.scale = scale;
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitDecimal(getValue(), precision, scale);
+ }
+ }
+
+ static final class StringLiteral extends Literal<String> {
+ StringLiteral(String value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitString(getValue());
+ }
+ }
+
+ static final class BytesLiteral extends Literal<byte[]> {
+ BytesLiteral(byte[] value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitBytes(getValue());
+ }
+ }
+
+ static final class TimeSeconds extends Literal<Integer> {
+ TimeSeconds(Integer value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimeSeconds(getValue());
+ }
+ }
+
+ static final class TimeMillis extends Literal<Integer> {
+ TimeMillis(Integer value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimeMillis(getValue());
+ }
+ }
+
+ static final class TimeMicros extends Literal<Long> {
+ TimeMicros(Long value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimeMicros(getValue());
+ }
+ }
+
+ static final class TimeNanos extends Literal<Long> {
+ TimeNanos(Long value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimeNanos(getValue());
+ }
+ }
+
+ static final class DateDays extends Literal<Integer> {
+ DateDays(Integer value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitDateDays(getValue());
+ }
+ }
+
+ static final class DateMillis extends Literal<Long> {
+ DateMillis(Long value) {
+ super(value);
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitDateMillis(getValue());
+ }
+ }
+
+ static final class TimestampMillis extends Literal<Long> {
+ private final Optional<String> timeZone;
+
+ TimestampMillis(Long value, Optional<String> timeZone) {
+ super(value);
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimestampMillis(getValue(), timeZone);
+ }
+ }
+
+ static final class TimestampMicros extends Literal<Long> {
+ private final Optional<String> timeZone;
+
+ TimestampMicros(Long value, Optional<String> timeZone) {
+ super(value);
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimestampMicros(getValue(), timeZone);
+ }
+ }
+
+ static final class TimestampNanos extends Literal<Long> {
+ private final Optional<String> timeZone;
+
+ TimestampNanos(Long value, Optional<String> timeZone) {
+ super(value);
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+ return visitor.visitTimestampNanos(getValue(), timeZone);
+ }
+ }
+
+ static final class LiteralToScalar implements LiteralVisitor<ScalarProtos.Scalar> {
+ static final LiteralToScalar INSTANCE = new LiteralToScalar();
+
+ private LiteralToScalar() {}
+
+ @Override
+ public ScalarProtos.Scalar visitNull() {
+ return Scalars.nullNull();
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitBoolean(Boolean literal) {
+ return Objects.isNull(literal) ? Scalars.nullBool() : Scalars.bool(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitInt8(Byte literal) {
+ return Objects.isNull(literal) ? Scalars.nullInt8() : Scalars.int8(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitInt16(Short literal) {
+ return Objects.isNull(literal) ? Scalars.nullInt16() : Scalars.int16(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitInt32(Integer literal) {
+ return Objects.isNull(literal) ? Scalars.nullInt32() : Scalars.int32(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitInt64(Long literal) {
+ return Objects.isNull(literal) ? Scalars.nullInt64() : Scalars.int64(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitDateDays(Integer days) {
+ return Objects.isNull(days) ? Scalars.nullDateDays() : Scalars.dateDays(days);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitDateMillis(Long millis) {
+ return Objects.isNull(millis) ? Scalars.nullDateMillis() : Scalars.dateMillis(millis);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimeSeconds(Integer seconds) {
+ return Objects.isNull(seconds)
+ ? Scalars.nullTimeSeconds()
+ : Scalars.timeSeconds(seconds);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimeMillis(Integer seconds) {
+ return Objects.isNull(seconds) ? Scalars.nullTimeMillis() : Scalars.timeMillis(seconds);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimeMicros(Long seconds) {
+ return Objects.isNull(seconds) ? Scalars.nullTimeMicros() : Scalars.timeMicros(seconds);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimeNanos(Long seconds) {
+ return Objects.isNull(seconds) ? Scalars.nullTimeNanos() : Scalars.timeNanos(seconds);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimestampMillis(
+ Long epochMillis, Optional<String> timeZone) {
+ return Objects.isNull(epochMillis)
+ ? Scalars.nullTimestampMillis(timeZone)
+ : Scalars.timestampMillis(epochMillis, timeZone);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimestampMicros(
+ Long epochMicros, Optional<String> timeZone) {
+ return Objects.isNull(epochMicros)
+ ? Scalars.nullTimestampMicros(timeZone)
+ : Scalars.timestampMicros(epochMicros, timeZone);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitTimestampNanos(Long epochNanos, Optional<String> timeZone) {
+ return Objects.isNull(epochNanos)
+ ? Scalars.nullTimestampNanos(timeZone)
+ : Scalars.timestampNanos(epochNanos, timeZone);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitFloat32(Float literal) {
+ return Objects.isNull(literal) ? Scalars.nullFloat32() : Scalars.float32(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitFloat64(Double literal) {
+ return Objects.isNull(literal) ? Scalars.nullFloat64() : Scalars.float64(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitDecimal(BigDecimal decimal, int precision, int scale) {
+ return Objects.isNull(decimal)
+ ? Scalars.nullDecimal(precision, scale)
+ : Scalars.decimal(decimal, precision, scale);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitString(String literal) {
+ return Objects.isNull(literal) ? Scalars.nullString() : Scalars.string(literal);
+ }
+
+ @Override
+ public ScalarProtos.Scalar visitBytes(byte[] literal) {
+ return Objects.isNull(literal) ? Scalars.nullBytes() : Scalars.bytes(literal);
+ }
+ }
+
+ private static Literal<?> deserializeLiteral(
+ ExprProtos.LiteralOpts literal, List<Expression> children) {
+ ScalarProtos.Scalar literalScalar = literal.getValue();
+ DTypeProtos.DType dtype = literalScalar.getDtype();
+
+ if (dtype.hasExtension()) {
+ return deserializeExtensionLiteral(literal);
+ }
+
+ ScalarProtos.ScalarValue scalarValue = literalScalar.getValue();
+
+ switch (scalarValue.getKindCase()) {
+ case NULL_VALUE:
+ return nullLiteral(dtype);
+ case BOOL_VALUE:
+ return Literal.bool(scalarValue.getBoolValue());
+ case INT64_VALUE:
+ return Literal.int64(scalarValue.getInt64Value());
+ case UINT64_VALUE:
+ return Literal.int64(scalarValue.getUint64Value());
+ case F32_VALUE:
+ return Literal.float32(scalarValue.getF32Value());
+ case F64_VALUE:
+ return Literal.float64(scalarValue.getF64Value());
+ case STRING_VALUE:
+ return Literal.string(scalarValue.getStringValue());
+ case BYTES_VALUE:
+ if (dtype.hasDecimal()) {
+ ByteString littleEndian = scalarValue.getBytesValue();
+ byte[] bigEndian = EndianUtils.reverse(littleEndian);
+ BigDecimal value =
+ new BigDecimal(
+ new BigInteger(bigEndian), dtype.getDecimal().getScale());
+ return Literal.decimal(
+ value,
+ dtype.getDecimal().getPrecision(),
+ dtype.getDecimal().getScale());
+ } else {
+ return Literal.bytes(scalarValue.getBytesValue().toByteArray());
+ }
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported ScalarValue type encountered: " + scalarValue);
+ }
+ }
+
+ private static Literal<?> deserializeExtensionLiteral(ExprProtos.LiteralOpts literal) {
+ ScalarProtos.Scalar scalar = literal.getValue();
+ DTypeProtos.DType scalarType = scalar.getDtype();
+
+ Preconditions.checkArgument(scalarType.hasExtension());
+
+ DTypeProtos.Extension extType = scalarType.getExtension();
+ String extId = scalarType.getExtension().getId();
+
+ switch (extId) {
+ case "vortex.time":
+ {
+ byte timeUnit =
+ TemporalMetadatas.getTimeUnit(extType.getMetadata().toByteArray());
+ if (timeUnit == TemporalMetadatas.TIME_UNIT_SECONDS) {
+ return Literal.timeSeconds(
+ Math.toIntExact(scalar.getValue().getInt64Value()));
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_MILLIS) {
+ return Literal.timeMillis(
+ Math.toIntExact(scalar.getValue().getInt64Value()));
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_MICROS) {
+ return Literal.timeMicros(scalar.getValue().getInt64Value());
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_NANOS) {
+ return Literal.timeNanos(scalar.getValue().getInt64Value());
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported TIME time unit: " + timeUnit);
+ }
+ }
+ case "vortex.date":
+ {
+ byte timeUnit =
+ TemporalMetadatas.getTimeUnit(extType.getMetadata().toByteArray());
+ if (timeUnit == TemporalMetadatas.TIME_UNIT_DAYS) {
+ return Literal.dateDays(Math.toIntExact(scalar.getValue().getInt64Value()));
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_MILLIS) {
+ return Literal.dateMillis(scalar.getValue().getInt64Value());
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported DATE time unit: " + timeUnit);
+ }
+ }
+ case "vortex.timestamp":
+ {
+ byte timeUnit =
+ TemporalMetadatas.getTimeUnit(extType.getMetadata().toByteArray());
+ Optional<String> timeZone =
+ TemporalMetadatas.getTimeZone(extType.getMetadata().toByteArray());
+ if (timeUnit == TemporalMetadatas.TIME_UNIT_MILLIS) {
+ return Literal.timestampMillis(scalar.getValue().getInt64Value(), timeZone);
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_MICROS) {
+ return Literal.timestampMicros(scalar.getValue().getInt64Value(), timeZone);
+ } else if (timeUnit == TemporalMetadatas.TIME_UNIT_NANOS) {
+ return Literal.timestampNanos(scalar.getValue().getInt64Value(), timeZone);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported TIMESTAMP time unit: " + timeUnit);
+ }
+ }
+ default:
+ throw new UnsupportedOperationException("Unsupported extension type: " + extId);
+ }
+ }
+
+ private static Literal<?> nullLiteral(DTypeProtos.DType type) {
+ switch (type.getDtypeTypeCase()) {
+ case NULL:
+ return Literal.nullLit();
+ case BOOL:
+ return Literal.bool(null);
+ case PRIMITIVE:
+ switch (type.getPrimitive().getType()) {
+ case U8:
+ case I8:
+ return Literal.int8(null);
+ case U16:
+ case I16:
+ return Literal.int16(null);
+ case U32:
+ case I32:
+ return Literal.int32(null);
+ case U64:
+ case I64:
+ return Literal.int64(null);
+ case F32:
+ return Literal.float32(null);
+ case F64:
+ return Literal.float64(null);
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported ScalarValue type encountered: " + type);
+ }
+ case DECIMAL:
+ return Literal.decimal(
+ null, type.getDecimal().getPrecision(), type.getDecimal().getScale());
+ case UTF8:
+ return Literal.string(null);
+ case BINARY:
+ return Literal.bytes(null);
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported ScalarValue type encountered: " + type);
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Not.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Not.java
new file mode 100644
index 0000000000..4c65a50aaf
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Not.java
@@ -0,0 +1,91 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import dev.vortex.api.Expression;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Logical NOT expression that negates its child expression. */
+public final class Not implements Expression {
+ private final Expression child;
+
+ private Not(Expression child) {
+ this.child = child;
+ }
+
+ public static Not parse(byte[] metadata, List<Expression> children) {
+ if (children.size() != 1) {
+ throw new IllegalArgumentException(
+ "Not expression must have exactly one child, found: " + children.size());
+ }
+ if (metadata.length > 0) {
+ throw new IllegalArgumentException(
+ "Not expression must not have metadata, found: " + metadata.length);
+ }
+ return new Not(children.get(0));
+ }
+
+ public static Not of(Expression child) {
+ return new Not(child);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) return false;
+ Not other = (Not) o;
+ return Objects.equals(child, other.child);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(child);
+ }
+
+ @Override
+ public String id() {
+ return "vortex.not";
+ }
+
+ @Override
+ public List<Expression> children() {
+ return java.util.Collections.singletonList(child);
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(new byte[] {});
+ }
+
+ @Override
+ public String toString() {
+ return "not(" + child + ")";
+ }
+
+ public Expression getChild() {
+ return child;
+ }
+
+ @Override
+ public <T> T accept(Visitor<T> visitor) {
+ return visitor.visitNot(this);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Root.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Root.java
new file mode 100644
index 0000000000..fb9aa99869
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Root.java
@@ -0,0 +1,64 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import dev.vortex.api.Expression;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Root expression in a Vortex expression tree (singleton). */
+public final class Root implements Expression {
+ public static final Root INSTANCE = new Root();
+
+ private Root() {}
+
+ public static Root parse(byte[] _metadata, List<Expression> children) {
+ if (!children.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Root expression must have no children, found: " + children.size());
+ }
+ return INSTANCE;
+ }
+
+ @Override
+ public String id() {
+ return "vortex.root";
+ }
+
+ @Override
+ public List<Expression> children() {
+ return java.util.Collections.emptyList();
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(new byte[] {});
+ }
+
+ @Override
+ public String toString() {
+ return "$";
+ }
+
+ @Override
+ public <T> T accept(Visitor<T> visitor) {
+ return visitor.visitRoot(this);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Unknown.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Unknown.java
new file mode 100644
index 0000000000..4256f4cf7b
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Unknown.java
@@ -0,0 +1,52 @@
+/*
+ * 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 dev.vortex.api.expressions;
+
+import dev.vortex.api.Expression;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Generic expression deserialized from Vortex without a concrete Java type. */
+public final class Unknown implements Expression {
+ private final String id;
+ private final List<Expression> children;
+ private final byte[] metadata;
+
+ public Unknown(String id, List<Expression> children, byte[] metadata) {
+ this.id = id;
+ this.children = children;
+ this.metadata = metadata;
+ }
+
+ @Override
+ public String id() {
+ return id;
+ }
+
+ @Override
+ public List<Expression> children() {
+ return children;
+ }
+
+ @Override
+ public Optional<byte[]> metadata() {
+ return Optional.of(metadata);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/DTypes.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/DTypes.java
new file mode 100644
index 0000000000..d0cf7f9f35
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/DTypes.java
@@ -0,0 +1,224 @@
+/*
+ * 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 dev.vortex.api.proto;
+
+import com.google.protobuf.ByteString;
+import dev.vortex.proto.DTypeProtos;
+
+import java.util.Optional;
+
+import static dev.vortex.api.proto.TemporalMetadatas.TIME_UNIT_MICROS;
+import static dev.vortex.api.proto.TemporalMetadatas.TIME_UNIT_NANOS;
+
+/** Factory class for creating Vortex data type definitions. */
+public final class DTypes {
+ private DTypes() {}
+
+ static DTypeProtos.DType nullType() {
+ return DTypeProtos.DType.newBuilder()
+ .setNull(DTypeProtos.Null.newBuilder().build())
+ .build();
+ }
+
+ static DTypeProtos.DType bool(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setBool(DTypeProtos.Bool.newBuilder().setNullable(nullable).build())
+ .build();
+ }
+
+ static DTypeProtos.DType int8(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.I8)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType int16(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.I16)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType int32(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.I32)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType int64(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.I64)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType float32(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.F32)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType float64(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setPrimitive(
+ DTypeProtos.Primitive.newBuilder()
+ .setType(DTypeProtos.PType.F64)
+ .setNullable(nullable)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType decimal(boolean nullable, int precision, int scale) {
+ return DTypeProtos.DType.newBuilder()
+ .setDecimal(
+ DTypeProtos.Decimal.newBuilder()
+ .setNullable(nullable)
+ .setPrecision(precision)
+ .setScale(scale)
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType string(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setUtf8(DTypeProtos.Utf8.newBuilder().setNullable(nullable).build())
+ .build();
+ }
+
+ static DTypeProtos.DType binary(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setBinary(DTypeProtos.Binary.newBuilder().setNullable(nullable).build())
+ .build();
+ }
+
+ static DTypeProtos.DType dateDays(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.date")
+ .setStorageDtype(DTypes.int32(nullable))
+ .setMetadata(ByteString.copyFrom(TemporalMetadatas.DATE_DAYS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType dateMillis(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.date")
+ .setStorageDtype(DTypes.int64(nullable))
+ .setMetadata(
+ ByteString.copyFrom(TemporalMetadatas.DATE_MILLIS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType timeSeconds(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.time")
+ .setStorageDtype(DTypes.int32(nullable))
+ .setMetadata(
+ ByteString.copyFrom(TemporalMetadatas.TIME_SECONDS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType timeMillis(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.time")
+ .setStorageDtype(DTypes.int32(nullable))
+ .setMetadata(
+ ByteString.copyFrom(TemporalMetadatas.TIME_MILLIS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType timeMicros(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.time")
+ .setStorageDtype(DTypes.int64(nullable))
+ .setMetadata(
+ ByteString.copyFrom(TemporalMetadatas.TIME_MICROS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType timeNanos(boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.time")
+ .setStorageDtype(DTypes.int64(nullable))
+ .setMetadata(
+ ByteString.copyFrom(TemporalMetadatas.TIME_NANOS.get()))
+ .build())
+ .build();
+ }
+
+ static DTypeProtos.DType timestampMillis(Optional<String> timeZone, boolean nullable) {
+ return timestamp(TemporalMetadatas.TIME_UNIT_MILLIS, timeZone, nullable);
+ }
+
+ static DTypeProtos.DType timestampMicros(Optional<String> timeZone, boolean nullable) {
+ return timestamp(TIME_UNIT_MICROS, timeZone, nullable);
+ }
+
+ static DTypeProtos.DType timestampNanos(Optional<String> timeZone, boolean nullable) {
+ return timestamp(TIME_UNIT_NANOS, timeZone, nullable);
+ }
+
+ private static DTypeProtos.DType timestamp(
+ byte timeUnit, Optional<String> timeZone, boolean nullable) {
+ return DTypeProtos.DType.newBuilder()
+ .setExtension(
+ DTypeProtos.Extension.newBuilder()
+ .setId("vortex.timestamp")
+ .setStorageDtype(DTypes.int64(nullable))
+ .setMetadata(
+ ByteString.copyFrom(
+ TemporalMetadatas.timestamp(timeUnit, timeZone)))
+ .build())
+ .build();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/EndianUtils.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/EndianUtils.java
new file mode 100644
index 0000000000..ff28ed2d2d
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/EndianUtils.java
@@ -0,0 +1,74 @@
+/*
+ * 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 dev.vortex.api.proto;
+
+import com.google.protobuf.ByteString;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+/** Utility class for endianness conversions in Vortex protocol buffers. */
+public final class EndianUtils {
+ public static byte[] reverse(ByteString src) {
+ byte[] dst = new byte[src.size()];
+ for (int i = 0; i < dst.length; i++) {
+ dst[i] = src.byteAt(dst.length - 1 - i);
+ }
+ return dst;
+ }
+
+ public static byte[] littleEndianDecimal(BigDecimal decimal) {
+ BigInteger unscaled = decimal.unscaledValue();
+ byte[] bigEndianBytes = unscaled.toByteArray();
+
+ int targetSize;
+ if (bigEndianBytes.length <= 1) {
+ targetSize = 1;
+ } else if (bigEndianBytes.length <= 2) {
+ targetSize = 2;
+ } else if (bigEndianBytes.length <= 4) {
+ targetSize = 4;
+ } else if (bigEndianBytes.length <= 8) {
+ targetSize = 8;
+ } else if (bigEndianBytes.length <= 16) {
+ targetSize = 16;
+ } else if (bigEndianBytes.length <= 32) {
+ targetSize = 32;
+ } else {
+ throw new IllegalArgumentException(
+ "BigDecimal with "
+ + bigEndianBytes.length
+ + " bytes overflows maximum Vortex decimal size");
+ }
+
+ byte[] result = new byte[targetSize];
+
+ for (int i = 0; i < bigEndianBytes.length; i++) {
+ result[i] = bigEndianBytes[bigEndianBytes.length - 1 - i];
+ }
+
+ if (unscaled.signum() < 0) {
+ for (int i = bigEndianBytes.length; i < targetSize; i++) {
+ result[i] = (byte) 0xFF;
+ }
+ }
+
+ return result;
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Expressions.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Expressions.java
new file mode 100644
index 0000000000..2d95ecfa05
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Expressions.java
@@ -0,0 +1,81 @@
+/*
+ * 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 dev.vortex.api.proto;
+
+import com.google.protobuf.ByteString;
+import dev.vortex.api.Expression;
+import dev.vortex.api.expressions.Binary;
+import dev.vortex.api.expressions.GetItem;
+import dev.vortex.api.expressions.Literal;
+import dev.vortex.api.expressions.Not;
+import dev.vortex.api.expressions.Root;
+import dev.vortex.api.expressions.Unknown;
+import dev.vortex.proto.ExprProtos;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Serialize/deserialize Vortex expressions to/from protocol buffers. */
+public final class Expressions {
+ public static ExprProtos.Expr serialize(Expression expression) {
+ ByteString metadata =
+ ByteString.copyFrom(
+ expression
+ .metadata()
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ "Expression is not serializable: "
+ + expression.id())));
+
+ return ExprProtos.Expr.newBuilder()
+ .setId(expression.id())
+ .addAllChildren(
+ expression.children().stream()
+ .map(Expressions::serialize)
+ .collect(Collectors.toList()))
+ .setMetadata(metadata)
+ .build();
+ }
+
+ public static Expression deserialize(ExprProtos.Expr expr) {
+ byte[] metadata = expr.getMetadata().toByteArray();
+ List<Expression> children =
+ expr.getChildrenList().stream()
+ .map(Expressions::deserialize)
+ .collect(Collectors.toList());
+
+ switch (expr.getId()) {
+ case "vortex.binary":
+ return Binary.parse(metadata, children);
+ case "vortex.get_item":
+ return GetItem.parse(metadata, children);
+ case "vortex.root":
+ return Root.parse(metadata, children);
+ case "vortex.literal":
+ return Literal.parse(metadata, children);
+ case "vortex.not":
+ return Not.parse(metadata, children);
+ default:
+ return new Unknown(expr.getId(), children, expr.getMetadata().toByteArray());
+ }
+ }
+
+ private Expressions() {}
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Scalars.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Scalars.java
new file mode 100644
index 0000000000..6402bb893d
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Scalars.java
@@ -0,0 +1,368 @@
+/*
+ * 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 dev.vortex.api.proto;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.NullValue;
+import dev.vortex.proto.ScalarProtos;
+
+import java.math.BigDecimal;
+import java.util.Optional;
+
+/** Factory class for creating Vortex scalar values with their associated data types. */
+public final class Scalars {
+ private Scalars() {}
+
+ public static ScalarProtos.Scalar nullNull() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.nullType())
+ .build();
+ }
+
+ public static ScalarProtos.Scalar bool(boolean value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setBoolValue(value).build())
+ .setDtype(DTypes.bool(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullBool() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.bool(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar int8(byte value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.int8(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullInt8() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.int8(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar int16(short value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.int16(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullInt16() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.int16(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar int32(int value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.int32(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullInt32() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.int32(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar int64(long value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.int64(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullInt64() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.int64(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar float32(float value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setF32Value(value).build())
+ .setDtype(DTypes.float32(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullFloat32() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.float32(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar float64(double value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setF64Value(value).build())
+ .setDtype(DTypes.float64(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullFloat64() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.float64(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar string(String value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setStringValue(value).build())
+ .setDtype(DTypes.string(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullString() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.string(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar decimal(BigDecimal decimal, int precision, int scale) {
+ byte[] littleEndian = EndianUtils.littleEndianDecimal(decimal);
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setBytesValue(ByteString.copyFrom(littleEndian))
+ .build())
+ .setDtype(DTypes.decimal(false, precision, scale))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullDecimal(int precision, int scale) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setNullValue(NullValue.NULL_VALUE))
+ .setDtype(DTypes.decimal(true, precision, scale))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar bytes(byte[] value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setBytesValue(ByteString.copyFrom(value))
+ .build())
+ .setDtype(DTypes.binary(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullBytes() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.binary(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar dateDays(int value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.dateDays(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullDateDays() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.dateDays(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar dateMillis(long value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.dateMillis(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullDateMillis() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.dateMillis(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timeSeconds(int value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timeSeconds(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimeSeconds() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timeSeconds(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timeMillis(int value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timeMillis(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimeMillis() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timeMillis(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timeMicros(long value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timeMicros(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimeMicros() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timeMicros(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timeNanos(long value) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timeNanos(false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimeNanos() {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timeNanos(true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timestampMillis(long value, Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timestampMillis(timeZone, false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimestampMillis(Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timestampMillis(timeZone, true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timestampMicros(long value, Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timestampMicros(timeZone, false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimestampMicros(Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timestampMicros(timeZone, true))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar timestampNanos(long value, Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(ScalarProtos.ScalarValue.newBuilder().setInt64Value(value).build())
+ .setDtype(DTypes.timestampNanos(timeZone, false))
+ .build();
+ }
+
+ public static ScalarProtos.Scalar nullTimestampNanos(Optional<String> timeZone) {
+ return ScalarProtos.Scalar.newBuilder()
+ .setValue(
+ ScalarProtos.ScalarValue.newBuilder()
+ .setNullValue(NullValue.NULL_VALUE)
+ .build())
+ .setDtype(DTypes.timestampNanos(timeZone, true))
+ .build();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/TemporalMetadatas.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/TemporalMetadatas.java
new file mode 100644
index 0000000000..bbdcc60ea2
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/TemporalMetadatas.java
@@ -0,0 +1,93 @@
+/*
+ * 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 dev.vortex.api.proto;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+import org.apache.paimon.shade.guava30.com.google.common.base.Supplier;
+import org.apache.paimon.shade.guava30.com.google.common.base.Suppliers;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+
+/** Utility class for creating and parsing temporal metadata in Vortex protocol buffers. */
+public final class TemporalMetadatas {
+ private TemporalMetadatas() {}
+
+ public static byte TIME_UNIT_NANOS = 0;
+ public static byte TIME_UNIT_MICROS = 1;
+ public static byte TIME_UNIT_MILLIS = 2;
+ public static byte TIME_UNIT_SECONDS = 3;
+ public static byte TIME_UNIT_DAYS = 4;
+
+ public static final Supplier<byte[]> DATE_DAYS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_DAYS});
+ public static final Supplier<byte[]> DATE_MILLIS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_MILLIS});
+ public static final Supplier<byte[]> TIME_SECONDS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_SECONDS});
+ public static final Supplier<byte[]> TIME_MILLIS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_MILLIS});
+ public static final Supplier<byte[]> TIME_MICROS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_MICROS});
+ public static final Supplier<byte[]> TIME_NANOS =
+ Suppliers.memoize(() -> new byte[] {TIME_UNIT_NANOS});
+
+ public static byte[] timestamp(byte timeUnit, Optional<String> timeZone) {
+ Preconditions.checkArgument(
+ timeUnit >= TIME_UNIT_NANOS && timeUnit < TIME_UNIT_DAYS,
+ "invalid timeUnit for Timestamp:" + timeUnit);
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ baos.write(timeUnit);
+ if (timeZone.isPresent()) {
+ byte[] timeZoneBytes = timeZone.get().getBytes(StandardCharsets.UTF_8);
+ int lenLow = timeZoneBytes.length & 0xFF;
+ int lenHigh = (timeZoneBytes.length >> 8) & 0xFF;
+ baos.write(lenLow);
+ baos.write(lenHigh);
+ baos.write(timeZoneBytes, 0, timeZoneBytes.length);
+ } else {
+ baos.write(0);
+ baos.write(0);
+ }
+ return baos.toByteArray();
+ }
+
+ public static byte getTimeUnit(byte[] serializedMetadata) {
+ byte timeUnit = serializedMetadata[0];
+ Preconditions.checkArgument(
+ timeUnit >= TIME_UNIT_NANOS && timeUnit <= TIME_UNIT_DAYS,
+ "invalid timeUnit byte: " + timeUnit);
+ return timeUnit;
+ }
+
+ public static Optional<String> getTimeZone(byte[] serializedMetadata) {
+ byte lenLow = serializedMetadata[1];
+ byte lenHigh = serializedMetadata[2];
+ int len = ((lenHigh & 0xFF) << 8) | (lenLow & 0xFF);
+ if (len == 0) {
+ return Optional.empty();
+ } else {
+ byte[] timeZoneBytes = new byte[len];
+ System.arraycopy(serializedMetadata, 3, timeZoneBytes, 0, len);
+ return Optional.of(new String(timeZoneBytes, StandardCharsets.UTF_8));
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/arrow/ArrowAllocation.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/arrow/ArrowAllocation.java
new file mode 100644
index 0000000000..6227142a9e
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/arrow/ArrowAllocation.java
@@ -0,0 +1,33 @@
+/*
+ * 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 dev.vortex.arrow;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+
+/** Shared root allocator for Apache Arrow memory operations. */
+public final class ArrowAllocation {
+ private static final RootAllocator ROOT_ALLOCATOR = new RootAllocator(Long.MAX_VALUE);
+
+ private ArrowAllocation() {}
+
+ public static BufferAllocator rootAllocator() {
+ return ROOT_ALLOCATOR;
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArray.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArray.java
new file mode 100644
index 0000000000..b2ea877425
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArray.java
@@ -0,0 +1,168 @@
+/*
+ * 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 dev.vortex.jni;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+
+import dev.vortex.api.Array;
+import dev.vortex.api.DType;
+import org.apache.arrow.c.ArrowArray;
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.CDataDictionaryProvider;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+
+import java.math.BigDecimal;
+import java.util.OptionalLong;
+
+/** JNI implementation of the Array interface. */
+public final class JNIArray implements Array {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private final ThreadLocal<long[]> schemaPtr = ThreadLocal.withInitial(() -> new long[1]);
+ private final ThreadLocal<long[]> arrayPtr = ThreadLocal.withInitial(() -> new long[1]);
+
+ private OptionalLong pointer;
+
+ public JNIArray(long pointer) {
+ Preconditions.checkArgument(pointer > 0, "Invalid pointer address: " + pointer);
+ this.pointer = OptionalLong.of(pointer);
+ }
+
+ @Override
+ public long getLen() {
+ return NativeArrayMethods.getLen(pointer.getAsLong());
+ }
+
+ @Override
+ public long nbytes() {
+ return NativeArrayMethods.nbytes(pointer.getAsLong());
+ }
+
+ @Override
+ public VectorSchemaRoot exportToArrow(BufferAllocator allocator, VectorSchemaRoot reuse) {
+ NativeArrayMethods.exportToArrow(pointer.getAsLong(), schemaPtr.get(), arrayPtr.get());
+ try (ArrowSchema arrowSchema = ArrowSchema.wrap(schemaPtr.get()[0]);
+ ArrowArray arrowArray = ArrowArray.wrap(arrayPtr.get()[0]);
+ CDataDictionaryProvider provider = new CDataDictionaryProvider()) {
+ if (reuse != null) {
+ Data.importIntoVectorSchemaRoot(allocator, arrowArray, reuse, provider);
+ return reuse;
+ } else {
+ return Data.importVectorSchemaRoot(
+ allocator, arrowArray, arrowSchema, new CDataDictionaryProvider());
+ }
+ } finally {
+ NativeArrayMethods.dropArrowSchema(schemaPtr.get()[0]);
+ NativeArrayMethods.dropArrowArray(arrayPtr.get()[0]);
+ }
+ }
+
+ @Override
+ public DType getDataType() {
+ return new JNIDType(NativeArrayMethods.getDataType(pointer.getAsLong()));
+ }
+
+ @Override
+ public Array getField(int index) {
+ return new JNIArray(NativeArrayMethods.getField(pointer.getAsLong(), index));
+ }
+
+ @Override
+ public Array slice(int start, int stop) {
+ return new JNIArray(NativeArrayMethods.slice(pointer.getAsLong(), start, stop));
+ }
+
+ @Override
+ public boolean getNull(int index) {
+ return NativeArrayMethods.getNull(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public int getNullCount() {
+ return NativeArrayMethods.getNullCount(pointer.getAsLong());
+ }
+
+ @Override
+ public byte getByte(int index) {
+ return NativeArrayMethods.getByte(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public short getShort(int index) {
+ return NativeArrayMethods.getShort(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public int getInt(int index) {
+ return NativeArrayMethods.getInt(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public long getLong(int index) {
+ return NativeArrayMethods.getLong(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public boolean getBool(int index) {
+ return NativeArrayMethods.getBool(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public float getFloat(int index) {
+ return NativeArrayMethods.getFloat(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public double getDouble(int index) {
+ return NativeArrayMethods.getDouble(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int index) {
+ return NativeArrayMethods.getBigDecimal(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public String getUTF8(int index) {
+ return NativeArrayMethods.getUTF8(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public void getUTF8_ptr_len(int index, long[] ptr, int[] len) {
+ NativeArrayMethods.getUTF8_ptr_len(pointer.getAsLong(), index, ptr, len);
+ }
+
+ @Override
+ public byte[] getBinary(int index) {
+ return NativeArrayMethods.getBinary(pointer.getAsLong(), index);
+ }
+
+ @Override
+ public void close() {
+ if (!pointer.isPresent()) {
+ return;
+ }
+ NativeArrayMethods.free(pointer.getAsLong());
+ pointer = OptionalLong.empty();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArrayIterator.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArrayIterator.java
new file mode 100644
index 0000000000..9fde302c26
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArrayIterator.java
@@ -0,0 +1,76 @@
+/*
+ * 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 dev.vortex.jni;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+
+import dev.vortex.api.Array;
+import dev.vortex.api.ArrayIterator;
+import dev.vortex.api.DType;
+
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/** JNI implementation of the ArrayIterator interface. */
+public final class JNIArrayIterator implements ArrayIterator {
+ private OptionalLong pointer;
+ private Optional<Array> next;
+
+ public JNIArrayIterator(long pointer) {
+ Preconditions.checkArgument(pointer > 0, "Invalid pointer address: " + pointer);
+ this.pointer = OptionalLong.of(pointer);
+ advance();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return next.isPresent();
+ }
+
+ @Override
+ public Array next() {
+ Array array = this.next.get();
+ advance();
+ return array;
+ }
+
+ @Override
+ public DType getDataType() {
+ return new JNIDType(NativeArrayIteratorMethods.getDType(pointer.getAsLong()));
+ }
+
+ @Override
+ public void close() {
+ if (!pointer.isPresent()) {
+ return;
+ }
+ NativeArrayIteratorMethods.free(pointer.getAsLong());
+ pointer = OptionalLong.empty();
+ next = Optional.empty();
+ }
+
+ private void advance() {
+ long next = NativeArrayIteratorMethods.take(pointer.getAsLong());
+ if (next <= 0) {
+ this.next = Optional.empty();
+ } else {
+ this.next = Optional.of(new JNIArray(next));
+ }
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIDType.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIDType.java
new file mode 100644
index 0000000000..6f739597cc
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIDType.java
@@ -0,0 +1,131 @@
+/*
+ * 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 dev.vortex.jni;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+
+import dev.vortex.api.DType;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/** JNI implementation of the DType interface. */
+public final class JNIDType implements DType {
+ OptionalLong pointer;
+ final boolean isOwned;
+
+ public JNIDType(long pointer) {
+ this(pointer, false);
+ }
+
+ public long getPointer() {
+ return pointer.getAsLong();
+ }
+
+ public JNIDType(long pointer, boolean isOwned) {
+ Preconditions.checkArgument(pointer > 0, "Invalid pointer address: " + pointer);
+ this.pointer = OptionalLong.of(pointer);
+ this.isOwned = isOwned;
+ }
+
+ @Override
+ public Variant getVariant() {
+ return Variant.from(NativeDTypeMethods.getVariant(pointer.getAsLong()));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return NativeDTypeMethods.isNullable(pointer.getAsLong());
+ }
+
+ @Override
+ public List<String> getFieldNames() {
+ return NativeDTypeMethods.getFieldNames(pointer.getAsLong());
+ }
+
+ @Override
+ public List<DType> getFieldTypes() {
+ return Lists.transform(
+ NativeDTypeMethods.getFieldTypes(pointer.getAsLong()), JNIDType::new);
+ }
+
+ @Override
+ public DType getElementType() {
+ return new JNIDType(NativeDTypeMethods.getElementType(pointer.getAsLong()));
+ }
+
+ @Override
+ public int getFixedSizeListSize() {
+ return NativeDTypeMethods.getFixedSizeListSize(pointer.getAsLong());
+ }
+
+ @Override
+ public boolean isDate() {
+ return NativeDTypeMethods.isDate(pointer.getAsLong());
+ }
+
+ @Override
+ public boolean isTime() {
+ return NativeDTypeMethods.isTime(pointer.getAsLong());
+ }
+
+ @Override
+ public boolean isTimestamp() {
+ return NativeDTypeMethods.isTimestamp(pointer.getAsLong());
+ }
+
+ @Override
+ public TimeUnit getTimeUnit() {
+ return TimeUnit.from(NativeDTypeMethods.getTimeUnit(pointer.getAsLong()));
+ }
+
+ @Override
+ public Optional<String> getTimeZone() {
+ return Optional.ofNullable(NativeDTypeMethods.getTimeZone(pointer.getAsLong()));
+ }
+
+ @Override
+ public boolean isDecimal() {
+ return NativeDTypeMethods.isDecimal(pointer.getAsLong());
+ }
+
+ @Override
+ public int getPrecision() {
+ return NativeDTypeMethods.getDecimalPrecision(pointer.getAsLong());
+ }
+
+ @Override
+ public byte getScale() {
+ return NativeDTypeMethods.getDecimalScale(pointer.getAsLong());
+ }
+
+ @Override
+ public void close() {
+ if (isOwned && pointer.isPresent()) {
+ NativeDTypeMethods.free(pointer.getAsLong());
+ pointer = OptionalLong.empty();
+ }
+ }
+
+ public static JNIDType ownedDType(long pointer) {
+ return new JNIDType(pointer, true);
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIFile.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIFile.java
new file mode 100644
index 0000000000..a640e79181
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIFile.java
@@ -0,0 +1,78 @@
+/*
+ * 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 dev.vortex.jni;
+
+import org.apache.paimon.shade.guava30.com.google.common.base.Preconditions;
+
+import dev.vortex.api.ArrayIterator;
+import dev.vortex.api.DType;
+import dev.vortex.api.File;
+import dev.vortex.api.ScanOptions;
+import dev.vortex.api.proto.Expressions;
+
+import java.util.OptionalLong;
+
+/** JNI implementation of the File interface. */
+public final class JNIFile implements File {
+ private OptionalLong pointer;
+
+ public JNIFile(long pointer) {
+ Preconditions.checkArgument(pointer > 0, "Invalid pointer address: " + pointer);
+ this.pointer = OptionalLong.of(pointer);
+ }
+
+ @Override
+ public DType getDType() {
+ return new JNIDType(NativeFileMethods.dtype(pointer.getAsLong()));
+ }
+
+ @Override
+ public long rowCount() {
+ return NativeFileMethods.rowCount(pointer.getAsLong());
+ }
+
+ @Override
+ public ArrayIterator newScan(ScanOptions options) {
+ byte[] predicateProto = null;
+
+ if (options.predicate().isPresent()) {
+ predicateProto = Expressions.serialize(options.predicate().get()).toByteArray();
+ }
+
+ long[] rowIndices = options.rowIndices().orElse(null);
+ long[] rowRange = options.rowRange().orElse(null);
+
+ return new JNIArrayIterator(
+ NativeFileMethods.scan(
+ pointer.getAsLong(),
+ options.columns(),
+ predicateProto,
+ rowRange,
+ rowIndices));
+ }
+
+ @Override
+ public void close() {
+ if (!pointer.isPresent()) {
+ return;
+ }
+ NativeFileMethods.close(pointer.getAsLong());
+ pointer = OptionalLong.empty();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIWriter.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIWriter.java
new file mode 100644
index 0000000000..260cc82713
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIWriter.java
@@ -0,0 +1,63 @@
+/*
+ * 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 dev.vortex.jni;
+
+import dev.vortex.api.VortexWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.OptionalLong;
+
+/** JNI implementation of VortexWriter. */
+public final class JNIWriter implements VortexWriter, AutoCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(JNIWriter.class);
+
+ private OptionalLong ptr;
+
+ public JNIWriter(long ptr) {
+ this.ptr = OptionalLong.of(ptr);
+ logger.debug("Created JNIWriter with ptr={}", ptr);
+ }
+
+ @Override
+ public void writeBatch(byte[] arrowData) throws IOException {
+ logger.trace("Writing batch with {} bytes", arrowData.length);
+
+ boolean success = NativeWriterMethods.writeBatch(ptr.getAsLong(), arrowData);
+ if (!success) {
+ logger.error("Failed to write batch to Vortex file");
+ throw new IOException("Failed to write batch to Vortex file");
+ }
+ }
+
+ @Override
+ public void close() {
+ if (!this.ptr.isPresent()) {
+ logger.debug("Attempted to close already closed JNIWriter, skipping");
+ return;
+ }
+
+ long ptr = this.ptr.getAsLong();
+
+ logger.debug("Closing JNIWriter with ptr={}", ptr);
+ NativeWriterMethods.close(ptr);
+ this.ptr = OptionalLong.empty();
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java
new file mode 100644
index 0000000000..08c6fc7dd2
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java
@@ -0,0 +1,34 @@
+/*
+ * 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 dev.vortex.jni;
+
+/** Native JNI methods for array iterator operations. */
+public final class NativeArrayIteratorMethods {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeArrayIteratorMethods() {}
+
+ public static native void free(long pointer);
+
+ public static native long take(long pointer);
+
+ public static native long getDType(long pointer);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayMethods.java
new file mode 100644
index 0000000000..d51dc96c70
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayMethods.java
@@ -0,0 +1,75 @@
+/*
+ * 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 dev.vortex.jni;
+
+import java.math.BigDecimal;
+
+/** Native JNI methods for array operations. */
+public final class NativeArrayMethods {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeArrayMethods() {}
+
+ public static native long nbytes(long pointer);
+
+ public static native void exportToArrow(
+ long pointer, long[] schemaPointer, long[] arrayPointer);
+
+ public static native void dropArrowSchema(long arrowSchemaPtr);
+
+ public static native void dropArrowArray(long arrowArrayPtr);
+
+ public static native void free(long pointer);
+
+ public static native long getLen(long pointer);
+
+ public static native long getDataType(long pointer);
+
+ public static native long getField(long pointer, int index);
+
+ public static native long slice(long pointer, int start, int stop);
+
+ public static native boolean getNull(long pointer, int index);
+
+ public static native int getNullCount(long pointer);
+
+ public static native byte getByte(long pointer, int index);
+
+ public static native short getShort(long pointer, int index);
+
+ public static native int getInt(long pointer, int index);
+
+ public static native long getLong(long pointer, int index);
+
+ public static native boolean getBool(long pointer, int index);
+
+ public static native float getFloat(long pointer, int index);
+
+ public static native double getDouble(long pointer, int index);
+
+ public static native BigDecimal getBigDecimal(long pointer, int index);
+
+ public static native String getUTF8(long pointer, int index);
+
+ public static native void getUTF8_ptr_len(long pointer, int index, long[] outPtr, int[] outLen);
+
+ public static native byte[] getBinary(long pointer, int index);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeDTypeMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeDTypeMethods.java
new file mode 100644
index 0000000000..ee64e0eb17
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeDTypeMethods.java
@@ -0,0 +1,92 @@
+/*
+ * 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 dev.vortex.jni;
+
+import java.util.List;
+
+/** Native JNI methods for DType operations. */
+public final class NativeDTypeMethods {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeDTypeMethods() {}
+
+ public static native long newByte(boolean isNullable);
+
+ public static native long newShort(boolean isNullable);
+
+ public static native long newInt(boolean isNullable);
+
+ public static native long newLong(boolean isNullable);
+
+ public static native long newFloat(boolean isNullable);
+
+ public static native long newDouble(boolean isNullable);
+
+ public static native long newDecimal(int precision, int scale, boolean isNullable);
+
+ public static native long newUtf8(boolean isNullable);
+
+ public static native long newBinary(boolean isNullable);
+
+ public static native long newBool(boolean isNullable);
+
+ public static native long newList(long elementTypePtr, boolean isNullable);
+
+ public static native long newFixedSizeList(long elementTypePtr, int size, boolean isNullable);
+
+ public static native long newStruct(String[] fieldNames, long[] fieldTypes, boolean isNullable);
+
+ public static native long newTimestamp(byte timeUnit, String zone, boolean isNullable);
+
+ public static native long newDate(byte timeUnit, boolean isNullable);
+
+ public static native long newTime(byte timeUnit, boolean isNullable);
+
+ public static native void free(long pointer);
+
+ public static native byte getVariant(long pointer);
+
+ public static native boolean isNullable(long pointer);
+
+ public static native List<String> getFieldNames(long pointer);
+
+ public static native List<Long> getFieldTypes(long pointer);
+
+ public static native long getElementType(long pointer);
+
+ public static native int getFixedSizeListSize(long pointer);
+
+ public static native boolean isDate(long pointer);
+
+ public static native boolean isTime(long pointer);
+
+ public static native boolean isTimestamp(long pointer);
+
+ public static native byte getTimeUnit(long pointer);
+
+ public static native String getTimeZone(long pointer);
+
+ public static native boolean isDecimal(long pointer);
+
+ public static native int getDecimalPrecision(long pointer);
+
+ public static native byte getDecimalScale(long pointer);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeFileMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeFileMethods.java
new file mode 100644
index 0000000000..deddfe4a51
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeFileMethods.java
@@ -0,0 +1,50 @@
+/*
+ * 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 dev.vortex.jni;
+
+import java.util.List;
+import java.util.Map;
+
+/** Native JNI methods for file operations. */
+public final class NativeFileMethods {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeFileMethods() {}
+
+ public static native List<String> listVortexFiles(String uri, Map<String, String> options);
+
+ public static native void delete(String[] uris, Map<String, String> options);
+
+ public static native long open(String uri, Map<String, String> options);
+
+ public static native long rowCount(long pointer);
+
+ public static native long dtype(long pointer);
+
+ public static native void close(long pointer);
+
+ public static native long scan(
+ long pointer,
+ List<String> columns,
+ byte[] predicateProto,
+ long[] rowRange,
+ long[] rowIndices);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLoader.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLoader.java
new file mode 100644
index 0000000000..b6fef957e8
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLoader.java
@@ -0,0 +1,83 @@
+/*
+ * 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 dev.vortex.jni;
+
+import org.apache.paimon.shade.guava30.com.google.common.io.ByteStreams;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Locale;
+
+/** Utility class for loading the native Vortex JNI library. */
+public final class NativeLoader {
+ private static boolean loaded = false;
+
+ private NativeLoader() {}
+
+ public static synchronized void loadJni() {
+ if (loaded) {
+ return;
+ }
+
+ String osName = System.getProperty("os.name").toLowerCase(Locale.ROOT);
+ String osArch = System.getProperty("os.arch").toLowerCase(Locale.ROOT);
+ String libName = "libvortex_jni";
+
+ String osShortName;
+ String libExt;
+ if (osName.contains("win")) {
+ osShortName = "win";
+ libExt = ".dll";
+ libName += libExt;
+ } else if (osName.contains("mac")) {
+ osShortName = "darwin";
+ libExt = ".dylib";
+ libName += libExt;
+ } else if (osName.contains("nix") || osName.contains("nux")) {
+ osShortName = "linux";
+ libExt = ".so";
+ libName += libExt;
+ } else {
+ throw new UnsupportedOperationException("Unsupported OS: " + osName);
+ }
+
+ String libPath = "/native/" + osShortName + "-" + osArch + "/" + libName;
+ try (InputStream in = NativeLoader.class.getResourceAsStream(libPath)) {
+ if (in == null) {
+ throw new FileNotFoundException("Library not found: " + libPath);
+ }
+ File tempFile = File.createTempFile("libvortex_jni", libExt);
+ tempFile.deleteOnExit();
+
+ try (OutputStream out = new FileOutputStream(tempFile)) {
+ ByteStreams.copy(in, out);
+ }
+ libName = tempFile.getAbsolutePath();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to load library: " + e.getMessage(), e);
+ }
+
+ System.load(libName);
+ loaded = true;
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLogging.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLogging.java
new file mode 100644
index 0000000000..53a42af72b
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeLogging.java
@@ -0,0 +1,36 @@
+/*
+ * 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 dev.vortex.jni;
+
+/** Utility class for configuring native logging levels in the Vortex JNI layer. */
+public final class NativeLogging {
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeLogging() {}
+
+ public static final int ERROR = 0;
+ public static final int WARN = 1;
+ public static final int INFO = 2;
+ public static final int DEBUG = 3;
+ public static final int TRACE = 4;
+
+ public static native void initLogging(int level);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriterMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriterMethods.java
new file mode 100644
index 0000000000..50a1bd3ed4
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriterMethods.java
@@ -0,0 +1,37 @@
+/*
+ * 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 dev.vortex.jni;
+
+import java.util.Map;
+
+/** Native JNI methods for writing Vortex files. */
+public final class NativeWriterMethods {
+
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeWriterMethods() {}
+
+ public static native long create(String uri, long dtype, Map<String, String> options);
+
+ public static native boolean writeBatch(long writerPtr, byte[] arrowData);
+
+ public static native void close(long writerPtr);
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/proto/dtype.proto b/paimon-vortex/paimon-vortex-jni/src/main/proto/dtype.proto
new file mode 100644
index 0000000000..77af16f806
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/proto/dtype.proto
@@ -0,0 +1,101 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+syntax = "proto3";
+
+package vortex.dtype;
+
+option java_package = "dev.vortex.proto";
+option java_outer_classname = "DTypeProtos";
+
+enum PType {
+ U8 = 0;
+ U16 = 1;
+ U32 = 2;
+ U64 = 3;
+ I8 = 4;
+ I16 = 5;
+ I32 = 6;
+ I64 = 7;
+ F16 = 8;
+ F32 = 9;
+ F64 = 10;
+}
+
+message Null {}
+
+message Bool {
+ bool nullable = 1;
+}
+
+message Primitive {
+ PType type = 1;
+ bool nullable = 2;
+}
+
+message Decimal {
+ uint32 precision = 1;
+ int32 scale = 2;
+ bool nullable = 3;
+}
+
+message Utf8 {
+ bool nullable = 1;
+}
+
+message Binary {
+ bool nullable = 1;
+}
+
+message Struct {
+ repeated string names = 1;
+ repeated DType dtypes = 2;
+ bool nullable = 3;
+}
+
+message List {
+ DType element_type = 1;
+ bool nullable = 2;
+}
+
+message FixedSizeList {
+ DType element_type = 1;
+ uint32 size = 2;
+ bool nullable = 3;
+}
+
+message Extension {
+ string id = 1;
+ DType storage_dtype = 2;
+ optional bytes metadata = 3;
+}
+
+message Variant {
+ bool nullable = 1;
+}
+
+message DType {
+ oneof dtype_type {
+ Null null = 1;
+ Bool bool = 2;
+ Primitive primitive = 3;
+ Decimal decimal = 4;
+ Utf8 utf8 = 5;
+ Binary binary = 6;
+ Struct struct = 7;
+ List list = 8;
+ Extension extension = 9;
+ FixedSizeList fixed_size_list = 10;
+ Variant variant = 11;
+ }
+}
+
+message Field {
+ oneof field_type {
+ string name = 1;
+ }
+}
+
+message FieldPath {
+ repeated Field path = 1;
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/proto/expr.proto b/paimon-vortex/paimon-vortex-jni/src/main/proto/expr.proto
new file mode 100644
index 0000000000..66b778c75c
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/proto/expr.proto
@@ -0,0 +1,84 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+syntax = "proto3";
+
+package vortex.expr;
+
+import "scalar.proto";
+import "dtype.proto";
+
+option java_package = "dev.vortex.proto";
+option java_outer_classname = "ExprProtos";
+
+message Expr {
+ string id = 1;
+ repeated Expr children = 2;
+ optional bytes metadata = 3;
+}
+
+message AggregateFn {
+ string id = 1;
+ optional bytes metadata = 2;
+}
+
+message LiteralOpts {
+ vortex.scalar.Scalar value = 1;
+}
+
+message PackOpts {
+ repeated string paths = 1;
+ bool nullable = 2;
+}
+
+message GetItemOpts {
+ string path = 1;
+}
+
+message BinaryOpts {
+ BinaryOp op = 1;
+
+ enum BinaryOp {
+ Eq = 0;
+ NotEq = 1;
+ Gt = 2;
+ Gte = 3;
+ Lt = 4;
+ Lte = 5;
+ And = 6;
+ Or = 7;
+ Add = 8;
+ Sub = 9;
+ Mul = 10;
+ Div = 11;
+ }
+}
+
+message BetweenOpts {
+ bool lower_strict = 1;
+ bool upper_strict = 2;
+}
+
+message LikeOpts {
+ bool negated = 1;
+ bool case_insensitive = 2;
+}
+
+message CastOpts {
+ vortex.dtype.DType target = 1;
+}
+
+message FieldNames {
+ repeated string names = 1;
+}
+
+message SelectOpts {
+ oneof opts {
+ FieldNames include = 1;
+ FieldNames exclude = 2;
+ }
+}
+
+message CaseWhenOpts {
+ uint32 num_children = 1;
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/proto/scalar.proto b/paimon-vortex/paimon-vortex-jni/src/main/proto/scalar.proto
new file mode 100644
index 0000000000..4e5b64aa34
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/proto/scalar.proto
@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+syntax = "proto3";
+
+package vortex.scalar;
+
+option java_package = "dev.vortex.proto";
+option java_outer_classname = "ScalarProtos";
+
+import "dtype.proto";
+import "google/protobuf/struct.proto";
+
+message Scalar {
+ vortex.dtype.DType dtype = 1;
+ ScalarValue value = 2;
+}
+
+message ScalarValue {
+ oneof kind {
+ google.protobuf.NullValue null_value = 1;
+ bool bool_value = 2;
+ sint64 int64_value = 3;
+ uint64 uint64_value = 4;
+ float f32_value = 5;
+ double f64_value = 6;
+ string string_value = 7;
+ bytes bytes_value = 8;
+ ListValue list_value = 9;
+ uint64 f16_value = 10;
+ Scalar variant_value = 11;
+ }
+}
+
+message ListValue {
+ repeated ScalarValue values = 1;
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/resources/META-INF/NOTICE b/paimon-vortex/paimon-vortex-jni/src/main/resources/META-INF/NOTICE
new file mode 100644
index 0000000000..d0d194b113
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,9 @@
+Paimon : Vortex JNI
+Copyright 2023-2025 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This product includes software from the Vortex project
+(https://github.com/vortex-data/vortex), licensed under the
+Apache License, Version 2.0.
diff --git a/paimon-vortex/pom.xml b/paimon-vortex/pom.xml
new file mode 100644
index 0000000000..c577b74e52
--- /dev/null
+++ b/paimon-vortex/pom.xml
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>paimon-parent</artifactId>
+ <groupId>org.apache.paimon</groupId>
+ <version>1.5-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-vortex</artifactId>
+ <name>Paimon : Vortex : </name>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>paimon-vortex-jni</module>
+ <module>paimon-vortex-format</module>
+ </modules>
+</project>
diff --git a/pom.xml b/pom.xml
index 5ed861b5e9..68e0c26844 100644
--- a/pom.xml
+++ b/pom.xml
@@ -73,6 +73,7 @@ under the License.
<module>tools/ci/paimon-ci-tools</module>
<module>paimon-api</module>
<module>paimon-lumina</module>
+ <module>paimon-vortex</module>
</modules>
<properties>
@@ -626,6 +627,7 @@ under the License.
<exclude>paimon-common/src/main/antlr4/**</exclude>
<exclude>paimon-core/src/test/resources/compatibility/**</exclude>
<exclude>paimon-python/pypaimon/sample/data/**</exclude>
+ <exclude>**/*.proto</exclude>
</excludes>
</configuration>
</plugin>