You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/05/31 07:09:34 UTC
(paimon) branch master updated: [vortex] Refactor JNI layer to use DataSource/Session API replacing proto-based Array/DType (#8040)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 609745ec12 [vortex] Refactor JNI layer to use DataSource/Session API replacing proto-based Array/DType (#8040)
609745ec12 is described below
commit 609745ec12d6f10de1c0f41d0931d7afbfc0c23f
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Sun May 31 15:09:27 2026 +0800
[vortex] Refactor JNI layer to use DataSource/Session API replacing proto-based Array/DType (#8040)
Replace the old protobuf-based JNI bindings (Array, DType, File,
expressions package) with a new DataSource/Session/Scan/Partition API that provides a cleaner
abstraction for Vortex file access. Update VortexRecordsReader, VortexRecordsWriter,
and VortexPredicateConverter to work with the new JNI interface.
---
.github/workflows/paimon-python-checks.yml | 12 -
...utitcase-rust-native.yml => utcase-tantivy.yml} | 42 +-
.../paimon/format/vortex/VortexFileFormat.java | 10 +-
.../format/vortex/VortexPredicateConverter.java | 96 ++-
.../paimon/format/vortex/VortexRecordsReader.java | 118 ++--
.../paimon/format/vortex/VortexRecordsWriter.java | 137 +++-
.../paimon/format/vortex/VortexTypeUtils.java | 198 ------
.../paimon/format/vortex/VortexWriterFactory.java | 17 +-
.../vortex/VortexFileFormatReadWriteTest.java | 27 -
.../vortex/VortexPredicateConverterTest.java | 346 ++++++---
.../format/vortex/VortexReaderWriterTest.java | 66 +-
.../paimon/format/vortex/VortexTypeUtilsTest.java | 132 ----
paimon-vortex/paimon-vortex-jni/README.md | 71 +-
paimon-vortex/paimon-vortex-jni/pom.xml | 55 +-
.../src/main/java/dev/vortex/api/Array.java | 67 --
.../src/main/java/dev/vortex/api/DType.java | 270 -------
.../src/main/java/dev/vortex/api/DataSource.java | 100 +++
.../src/main/java/dev/vortex/api/Expression.java | 204 +++++-
.../src/main/java/dev/vortex/api/Files.java | 49 --
.../src/main/java/dev/vortex/api/Partition.java | 70 ++
.../src/main/java/dev/vortex/api/Scan.java | 76 ++
.../src/main/java/dev/vortex/api/ScanOptions.java | 46 +-
.../api/{ArrayIterator.java => Session.java} | 32 +-
.../src/main/java/dev/vortex/api/VortexWriter.java | 80 ++-
.../java/dev/vortex/api/expressions/Binary.java | 234 ------
.../java/dev/vortex/api/expressions/GetItem.java | 95 ---
.../java/dev/vortex/api/expressions/IsNotNull.java | 89 ---
.../java/dev/vortex/api/expressions/IsNull.java | 88 ---
.../java/dev/vortex/api/expressions/Literal.java | 781 ---------------------
.../main/java/dev/vortex/api/expressions/Not.java | 88 ---
.../main/java/dev/vortex/api/expressions/Root.java | 64 --
.../java/dev/vortex/api/expressions/Unknown.java | 51 --
.../src/main/java/dev/vortex/api/proto/DTypes.java | 201 ------
.../java/dev/vortex/api/proto/EndianUtils.java | 74 --
.../java/dev/vortex/api/proto/Expressions.java | 70 --
.../main/java/dev/vortex/api/proto/Scalars.java | 378 ----------
.../dev/vortex/api/proto/TemporalMetadatas.java | 98 ---
.../src/main/java/dev/vortex/jni/JNIArray.java | 167 -----
.../main/java/dev/vortex/jni/JNIArrayIterator.java | 75 --
.../src/main/java/dev/vortex/jni/JNIDType.java | 129 ----
.../src/main/java/dev/vortex/jni/JNIFile.java | 71 --
.../src/main/java/dev/vortex/jni/JNIWriter.java | 74 --
.../java/dev/vortex/jni/NativeArrayMethods.java | 74 --
.../java/dev/vortex/jni/NativeDTypeMethods.java | 93 ---
...ativeFileMethods.java => NativeDataSource.java} | 24 +-
.../main/java/dev/vortex/jni/NativeExpression.java | 83 +++
...eArrayIteratorMethods.java => NativeFiles.java} | 18 +-
.../src/main/java/dev/vortex/jni/NativeLoader.java | 9 +
.../main/java/dev/vortex/jni/NativeLogging.java | 14 +-
...ayIteratorMethods.java => NativePartition.java} | 15 +-
...rrayIteratorMethods.java => NativeRuntime.java} | 13 +-
.../{NativeWriterMethods.java => NativeScan.java} | 28 +-
.../{api/File.java => jni/NativeSession.java} | 18 +-
...{NativeWriterMethods.java => NativeWriter.java} | 14 +-
.../paimon-vortex-jni/src/main/proto/dtype.proto | 101 ---
.../paimon-vortex-jni/src/main/proto/expr.proto | 97 ---
.../paimon-vortex-jni/src/main/proto/scalar.proto | 39 -
57 files changed, 1304 insertions(+), 4484 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml b/.github/workflows/paimon-python-checks.yml
index 5aaa2ab72a..864d10ecee 100755
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -99,18 +99,6 @@ jobs:
mkdir -p ${RESOURCE_DIR}
cp paimon-tantivy/paimon-tantivy-jni/rust/target/release/libtantivy_jni.so ${RESOURCE_DIR}/
- - name: Clone and build Vortex native library
- run: |
- git clone --depth 1 -b 0.69.0 https://github.com/spiraldb/vortex.git ${RUNNER_TEMP}/vortex
- cd ${RUNNER_TEMP}/vortex
- cargo build --package vortex-jni --release
-
- - name: Copy Vortex 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: Verify Python version
run: python --version
diff --git a/.github/workflows/utitcase-rust-native.yml b/.github/workflows/utcase-tantivy.yml
similarity index 64%
rename from .github/workflows/utitcase-rust-native.yml
rename to .github/workflows/utcase-tantivy.yml
index 32f53787e3..f2ebb097db 100644
--- a/.github/workflows/utitcase-rust-native.yml
+++ b/.github/workflows/utcase-tantivy.yml
@@ -16,16 +16,14 @@
# limitations under the License.
################################################################################
-name: UTCase Rust Native
+name: UTCase Tantivy
on:
push:
paths:
- - 'paimon-vortex/**'
- 'paimon-tantivy/**'
pull_request:
paths:
- - 'paimon-vortex/**'
- 'paimon-tantivy/**'
env:
@@ -37,44 +35,6 @@ concurrency:
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
- run: |
- curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain stable --profile minimal
- echo "$HOME/.cargo/bin" >> $GITHUB_PATH
-
- - name: Clone and build Vortex native library
- run: |
- git clone --depth 1 -b 0.69.0 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
-
tantivy_test:
runs-on: ubuntu-latest
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
index 1887825006..46e13d6cb4 100644
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
@@ -18,7 +18,7 @@
package org.apache.paimon.format.vortex;
-import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.arrow.vector.ArrowFormatCWriter;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FileFormatFactory;
import org.apache.paimon.format.FormatReaderFactory;
@@ -79,13 +79,7 @@ public class VortexFileFormat extends FileFormat {
@Override
public FormatWriterFactory createWriterFactory(RowType type) {
return new VortexWriterFactory(
- type,
- () ->
- new ArrowFormatWriter(
- type,
- formatContext.writeBatchSize(),
- true,
- formatContext.writeBatchMemory().getBytes()));
+ () -> new ArrowFormatCWriter(type, formatContext.writeBatchSize(), true));
}
@Override
diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
index ff1e1607a2..e66f6cbece 100644
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java
@@ -42,11 +42,6 @@ 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;
@@ -74,13 +69,13 @@ public class VortexPredicateConverter implements PredicateVisitor<Expression> {
return null;
}
FieldRef fieldRef = fieldRefOpt.get();
- Expression field = GetItem.of(Root.INSTANCE, fieldRef.name());
+ Expression field = Expression.column(fieldRef.name());
if (predicate.function() instanceof IsNull) {
- return Not.of(Binary.notEq(field, Literal.nullLit()));
+ return Expression.isNull(field);
}
if (predicate.function() instanceof IsNotNull) {
- return Binary.notEq(field, Literal.nullLit());
+ return Expression.isNotNull(field);
}
List<Object> literals = predicate.literals();
@@ -88,26 +83,25 @@ public class VortexPredicateConverter implements PredicateVisitor<Expression> {
return null;
}
- Literal<?> vortexLiteral = toLiteral(fieldRef.type(), literals.get(0));
+ Expression vortexLiteral = toLiteral(fieldRef.type(), literals.get(0));
if (vortexLiteral == null) {
return null;
}
if (predicate.function() instanceof Equal) {
- return Binary.eq(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.EQ, field, vortexLiteral);
} else if (predicate.function() instanceof NotEqual) {
- return Binary.notEq(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.NOT_EQ, field, vortexLiteral);
} else if (predicate.function() instanceof GreaterThan) {
- return Binary.gt(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.GT, field, vortexLiteral);
} else if (predicate.function() instanceof GreaterOrEqual) {
- return Binary.gtEq(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.GTE, field, vortexLiteral);
} else if (predicate.function() instanceof LessThan) {
- return Binary.lt(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.LT, field, vortexLiteral);
} else if (predicate.function() instanceof LessOrEqual) {
- return Binary.ltEq(field, vortexLiteral);
+ return Expression.binary(Expression.BinaryOp.LTE, field, vortexLiteral);
}
- // unsupported function (e.g. In, Between)
return null;
}
@@ -124,11 +118,7 @@ public class VortexPredicateConverter implements PredicateVisitor<Expression> {
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;
+ return Expression.and(children.toArray(new Expression[0]));
} else if (predicate.function() instanceof Or) {
List<Expression> children = new ArrayList<>();
for (Predicate child : predicate.children()) {
@@ -138,74 +128,70 @@ public class VortexPredicateConverter implements PredicateVisitor<Expression> {
}
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 Expression.or(children.toArray(new Expression[0]));
}
return null;
}
@Nullable
- private static Literal<?> toLiteral(DataType type, Object value) {
+ private static Expression toLiteral(DataType type, Object value) {
if (value == null) {
- return Literal.nullLit();
+ return Expression.nullLiteral(Expression.DType.I32);
}
switch (type.getTypeRoot()) {
case BOOLEAN:
- return Literal.bool((Boolean) value);
+ return Expression.literal((Boolean) value);
case TINYINT:
- return Literal.int8((Byte) value);
+ return Expression.literal((Byte) value);
case SMALLINT:
- return Literal.int16((Short) value);
+ return Expression.literal((Short) value);
case INTEGER:
case DATE:
- return Literal.int32((Integer) value);
+ return Expression.literal((Integer) value);
case BIGINT:
- return Literal.int64((Long) value);
+ return Expression.literal((Long) value);
case FLOAT:
- return Literal.float32((Float) value);
+ return Expression.literal((Float) value);
case DOUBLE:
- return Literal.float64((Double) value);
+ return Expression.literal((Double) value);
case CHAR:
case VARCHAR:
- return Literal.string(
+ return Expression.literal(
value instanceof BinaryString ? value.toString() : (String) value);
case DECIMAL:
Decimal decimal = (Decimal) value;
- return Literal.decimal(
- decimal.toBigDecimal(), decimal.precision(), decimal.scale());
+ return Expression.literalDecimal(
+ decimal.toBigDecimal().unscaledValue(),
+ decimal.precision(),
+ decimal.scale());
case TIMESTAMP_WITHOUT_TIME_ZONE:
Timestamp ts = (Timestamp) value;
- return toTimestampLiteral(
- ts, ((TimestampType) type).getPrecision(), Optional.empty());
+ return toTimestampLiteral(ts, ((TimestampType) type).getPrecision(), null);
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
Timestamp lzTs = (Timestamp) value;
return toTimestampLiteral(
- lzTs, ((LocalZonedTimestampType) type).getPrecision(), Optional.of("UTC"));
+ lzTs, ((LocalZonedTimestampType) type).getPrecision(), "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);
+ private static Expression toTimestampLiteral(
+ Timestamp ts, int precision, @Nullable String timeZone) {
+ if (precision <= 3) {
+ return Expression.literalTimestamp(
+ ts.getMillisecond(), Expression.TimeUnit.MILLISECONDS, timeZone);
} else if (precision <= 6) {
- // micros
- return Literal.timestampMicros(
- ts.getMillisecond() * 1000 + ts.getNanoOfMillisecond() / 1000, timeZone);
+ return Expression.literalTimestamp(
+ ts.getMillisecond() * 1000 + ts.getNanoOfMillisecond() / 1000,
+ Expression.TimeUnit.MICROSECONDS,
+ timeZone);
} else {
- // nanos
- return Literal.timestampNanos(
- ts.getMillisecond() * 1_000_000 + ts.getNanoOfMillisecond(), timeZone);
+ return Expression.literalTimestamp(
+ ts.getMillisecond() * 1_000_000 + ts.getNanoOfMillisecond(),
+ Expression.TimeUnit.NANOSECONDS,
+ timeZone);
}
}
}
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
index 69b7e2f7ba..571c4edf04 100644
--- 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
@@ -30,15 +30,17 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.LongIterator;
import org.apache.paimon.utils.ProjectedRow;
-import dev.vortex.api.Array;
-import dev.vortex.api.ArrayIterator;
+import dev.vortex.api.DataSource;
import dev.vortex.api.Expression;
-import dev.vortex.api.File;
-import dev.vortex.api.Files;
import dev.vortex.api.ImmutableScanOptions;
+import dev.vortex.api.Partition;
+import dev.vortex.api.Scan;
+import dev.vortex.api.ScanOptions;
+import dev.vortex.api.Session;
import dev.vortex.arrow.ArrowAllocation;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
import javax.annotation.Nullable;
@@ -56,12 +58,13 @@ 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 Session session;
+ private final DataSource dataSource;
+ private final Scan scan;
private final LongIterator positionIterator;
@Nullable private final int[] physicalFieldMapping;
- private VectorSchemaRoot reuse;
- private Array currentArray;
+ private ArrowReader currentArrowReader;
+ private Partition currentPartition;
private long returnedPosition = -1;
public VortexRecordsReader(
@@ -77,45 +80,72 @@ public class VortexRecordsReader implements FileRecordReader<InternalRow> {
this.allocator =
ArrowAllocation.rootAllocator()
.newChildAllocator("vortex-reader", 0, Long.MAX_VALUE);
+
try {
- this.vortexFile = Files.open(path.toUri().toString(), storageOptions);
+ this.session = Session.create();
try {
- ImmutableScanOptions.Builder scanBuilder = ImmutableScanOptions.builder();
- scanBuilder.addAllColumns(physicalReadRowType.getFieldNames());
- if (rowIndices != null) {
- scanBuilder.rowIndices(rowIndices);
- }
- if (predicate != null) {
- scanBuilder.predicate(predicate);
+ this.dataSource = DataSource.open(session, path.toUri().toString(), storageOptions);
+ try {
+ ImmutableScanOptions.Builder scanBuilder = ImmutableScanOptions.builder();
+
+ java.util.List<String> columns = physicalReadRowType.getFieldNames();
+ scanBuilder.projection(
+ Expression.select(columns.toArray(new String[0]), Expression.root()));
+
+ if (rowIndices != null) {
+ scanBuilder.selectionIndices(rowIndices);
+ scanBuilder.selectionMode(ScanOptions.SelectionMode.INCLUDE);
+ }
+ if (predicate != null) {
+ scanBuilder.filter(predicate);
+ }
+ this.scan = dataSource.scan(scanBuilder.build());
+ } catch (Exception e) {
+ dataSource.close();
+ throw e;
}
- this.arrayIterator = vortexFile.newScan(scanBuilder.build());
} catch (Exception e) {
- vortexFile.close();
+ session.close();
throw e;
}
} catch (Exception e) {
allocator.close();
- throw e;
+ throw new RuntimeException(e);
}
+
this.arrowBatchReader = new ArrowBatchReader(physicalReadRowType, true);
+ long totalRowCount = dataSource.rowCount();
this.positionIterator =
rowIndices != null
? LongIterator.fromArray(rowIndices)
- : LongIterator.fromRange(0, vortexFile.rowCount());
+ : LongIterator.fromRange(0, totalRowCount > 0 ? totalRowCount : 0);
}
@Nullable
@Override
public FileRecordIterator<InternalRow> readBatch() throws IOException {
- if (!arrayIterator.hasNext()) {
- return null;
+ // Try to load the next batch from the current ArrowReader
+ if (currentArrowReader != null && currentArrowReader.loadNextBatch()) {
+ return toBatchIterator(currentArrowReader.getVectorSchemaRoot());
+ }
+
+ // Close current reader and move to the next partition
+ closeCurrentArrowReader();
+
+ while (scan.hasNext()) {
+ closeCurrentPartition();
+ currentPartition = scan.next();
+ currentArrowReader = currentPartition.scanArrow(allocator);
+ if (currentArrowReader.loadNextBatch()) {
+ return toBatchIterator(currentArrowReader.getVectorSchemaRoot());
+ }
+ closeCurrentArrowReader();
}
- releaseCurrentArray();
- Array array = arrayIterator.next();
- this.currentArray = array;
- VectorSchemaRoot vsr = array.exportToArrow(allocator, reuse);
- this.reuse = vsr;
+ return null;
+ }
+
+ private FileRecordIterator<InternalRow> toBatchIterator(VectorSchemaRoot vsr) {
Iterator<InternalRow> rows = arrowBatchReader.readBatch(vsr).iterator();
ProjectedRow projectedRow =
physicalFieldMapping == null ? null : ProjectedRow.from(physicalFieldMapping);
@@ -143,27 +173,35 @@ public class VortexRecordsReader implements FileRecordReader<InternalRow> {
}
@Override
- public void releaseBatch() {
- releaseCurrentArray();
- }
+ public void releaseBatch() {}
};
}
- private void releaseCurrentArray() {
- if (currentArray != null) {
- currentArray.close();
- currentArray = null;
+ private void closeCurrentArrowReader() {
+ if (currentArrowReader != null) {
+ try {
+ currentArrowReader.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ currentArrowReader = null;
+ }
+ }
+
+ private void closeCurrentPartition() {
+ if (currentPartition != null) {
+ currentPartition.close();
+ currentPartition = null;
}
}
@Override
public void close() {
- releaseCurrentArray();
- if (reuse != null) {
- reuse.close();
- }
- arrayIterator.close();
- vortexFile.close();
+ closeCurrentArrowReader();
+ closeCurrentPartition();
+ scan.close();
+ dataSource.close();
+ session.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
index 6d8de8fccc..7798fa09d4 100644
--- 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
@@ -19,54 +19,81 @@
package org.apache.paimon.format.vortex;
import org.apache.paimon.arrow.ArrowBundleRecords;
-import org.apache.paimon.arrow.ArrowUtils;
-import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.arrow.vector.ArrowCStruct;
+import org.apache.paimon.arrow.vector.ArrowFormatCWriter;
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.Session;
import dev.vortex.api.VortexWriter;
+import org.apache.arrow.c.ArrowArray;
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
+import java.util.function.Supplier;
-/** Vortex records writer. */
+/** Vortex records writer using the Arrow C Data Interface. */
public class VortexRecordsWriter implements BundleFormatWriter {
private static final Logger LOG = LoggerFactory.getLogger(VortexRecordsWriter.class);
private static final double COMPRESSION_RATIO = 0.25;
- private final ArrowFormatWriter arrowFormatWriter;
+ private final Supplier<ArrowFormatCWriter> cWriterSupplier;
+ private final Session session;
private final VortexWriter nativeWriter;
private final String path;
+
+ // Vortex's writeBatch is semi-async: Rust takes zero-copy ownership of buffers
+ // via Arc<FFI_ArrowArray> and releases them after the background write completes.
+ // Each flush creates a new ArrowFormatCWriter (with its own RootAllocator) so
+ // buffers are never reused across batches. The Rust release callback frees most
+ // memory; only a small residual (~148 bytes per batch from Arrow 15's incomplete
+ // release) remains in each retained resource until nativeWriter.close().
+ private final List<AutoCloseable> retainedResources;
+ private ArrowFormatCWriter currentWriter;
+
private long jniCost = 0;
- private long ipcBytes = 0;
+ private long ffiBytes = 0;
public VortexRecordsWriter(
- RowType rowType,
- ArrowFormatWriter arrowFormatWriter,
+ Supplier<ArrowFormatCWriter> cWriterSupplier,
Path path,
Map<String, String> storageOptions)
throws IOException {
- this.arrowFormatWriter = arrowFormatWriter;
+ this.cWriterSupplier = cWriterSupplier;
this.path = path.toUri().toString();
+ this.retainedResources = new ArrayList<>();
+ this.currentWriter = cWriterSupplier.get();
- DType dtype = VortexTypeUtils.toDType(rowType);
- this.nativeWriter = VortexWriter.create(this.path, dtype, storageOptions);
+ this.session = Session.create();
+ try {
+ Schema arrowSchema = currentWriter.getVectorSchemaRoot().getSchema();
+ this.nativeWriter =
+ VortexWriter.create(session, this.path, arrowSchema, storageOptions);
+ } catch (Exception e) {
+ session.close();
+ throw e;
+ }
}
@Override
public void addElement(InternalRow internalRow) throws IOException {
- if (!arrowFormatWriter.write(internalRow)) {
+ if (!currentWriter.write(internalRow)) {
flush();
- if (!arrowFormatWriter.write(internalRow)) {
+ if (!currentWriter.write(internalRow)) {
throw new RuntimeException("Exception happens while write to vortex file");
}
}
@@ -76,7 +103,7 @@ public class VortexRecordsWriter implements BundleFormatWriter {
public void writeBundle(BundleRecords bundleRecords) throws IOException {
if (bundleRecords instanceof ArrowBundleRecords) {
flush();
- writeVsr(((ArrowBundleRecords) bundleRecords).getVectorSchemaRoot());
+ writeBundleVsr(((ArrowBundleRecords) bundleRecords).getVectorSchemaRoot());
} else {
for (InternalRow row : bundleRecords) {
addElement(row);
@@ -86,7 +113,7 @@ public class VortexRecordsWriter implements BundleFormatWriter {
@Override
public boolean reachTargetSize(boolean suggestedCheck, long targetSize) {
- return suggestedCheck && (long) (ipcBytes * COMPRESSION_RATIO) >= targetSize;
+ return suggestedCheck && (long) (ffiBytes * COMPRESSION_RATIO) >= targetSize;
}
@Override
@@ -100,14 +127,22 @@ public class VortexRecordsWriter implements BundleFormatWriter {
throwable = t;
}
+ // nativeWriter.close() blocks until all async background writes complete.
try {
nativeWriter.close();
} catch (Throwable t) {
throwable = addSuppressed(throwable, t);
}
+ // Release all retained resources now that async writes are done.
+ for (AutoCloseable res : retainedResources) {
+ closeQuietly(res);
+ }
+ retainedResources.clear();
+ closeQuietly(currentWriter);
+
try {
- arrowFormatWriter.close();
+ session.close();
} catch (Throwable t) {
throwable = addSuppressed(throwable, t);
}
@@ -121,22 +156,66 @@ public class VortexRecordsWriter implements BundleFormatWriter {
}
private void flush() throws IOException {
+ currentWriter.flush();
+ if (!currentWriter.empty()) {
+ ffiBytes += bufferBytes(currentWriter.getVectorSchemaRoot());
+ ArrowCStruct cStruct = currentWriter.toCStruct();
+ long t1 = System.currentTimeMillis();
+ nativeWriter.writeBatch(cStruct.arrayAddress(), cStruct.schemaAddress());
+ jniCost += (System.currentTimeMillis() - t1);
+ // Each ArrowFormatCWriter has its own RootAllocator and buffers.
+ // Retain it so buffer memory stays alive for async Rust reads.
+ retainedResources.add(currentWriter);
+ currentWriter = cWriterSupplier.get();
+ }
+ }
+
+ /** Write an external VSR (from writeBundle) via IPC copy into an independent allocator. */
+ private void writeBundleVsr(VectorSchemaRoot vsr) throws IOException {
+ ffiBytes += bufferBytes(vsr);
+ byte[] ipc = org.apache.paimon.arrow.ArrowUtils.serializeToIpc(vsr);
+ RootAllocator bundleAllocator = new RootAllocator(Long.MAX_VALUE);
try {
- arrowFormatWriter.flush();
- if (!arrowFormatWriter.empty()) {
- writeVsr(arrowFormatWriter.getVectorSchemaRoot());
- }
- } finally {
- arrowFormatWriter.reset();
+ ArrowStreamReader ipcReader =
+ new ArrowStreamReader(new java.io.ByteArrayInputStream(ipc), bundleAllocator);
+ ipcReader.loadNextBatch();
+ VectorSchemaRoot copy = ipcReader.getVectorSchemaRoot();
+
+ ArrowArray arrowArray = ArrowArray.allocateNew(bundleAllocator);
+ ArrowSchema arrowSchema = ArrowSchema.allocateNew(bundleAllocator);
+ Data.exportVectorSchemaRoot(bundleAllocator, copy, null, arrowArray, arrowSchema);
+ long t1 = System.currentTimeMillis();
+ nativeWriter.writeBatch(arrowArray.memoryAddress(), arrowSchema.memoryAddress());
+ jniCost += (System.currentTimeMillis() - t1);
+ // Retain all resources that own the exported C Data buffers and release
+ // callbacks. Rust holds async zero-copy references via Arc<FFI_ArrowArray>.
+ // Order matters: close ipcReader (owns VectorSchemaRoot) before allocator.
+ retainedResources.add(ipcReader);
+ retainedResources.add(bundleAllocator);
+ } catch (Exception e) {
+ closeQuietly(bundleAllocator);
+ throw e instanceof IOException ? (IOException) e : new IOException(e);
}
}
- private void writeVsr(VectorSchemaRoot vsr) throws IOException {
- byte[] bytes = ArrowUtils.serializeToIpc(vsr);
- ipcBytes += bytes.length;
- long t1 = System.currentTimeMillis();
- nativeWriter.writeBatch(bytes);
- jniCost += (System.currentTimeMillis() - t1);
+ private static long bufferBytes(VectorSchemaRoot vsr) {
+ long bytes = 0;
+ for (int i = 0; i < vsr.getFieldVectors().size(); i++) {
+ bytes += vsr.getFieldVectors().get(i).getBufferSize();
+ }
+ return bytes;
+ }
+
+ private static void closeQuietly(AutoCloseable closeable) {
+ try {
+ closeable.close();
+ } catch (IllegalStateException e) {
+ if (e.getMessage() == null || !e.getMessage().contains("Memory was leaked")) {
+ throw e;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
private static Throwable addSuppressed(Throwable throwable, Throwable suppressed) {
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
deleted file mode 100644
index 271c14d196..0000000000
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexTypeUtils.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * 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/VortexWriterFactory.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
index f3de8fca74..e3840acd36 100644
--- 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
@@ -18,14 +18,13 @@
package org.apache.paimon.format.vortex;
-import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.arrow.vector.ArrowFormatCWriter;
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 java.io.IOException;
@@ -37,13 +36,10 @@ import static org.apache.paimon.format.vortex.VortexUtils.toVortexSpecifiedForWr
/** A factory to create Vortex {@link FormatWriter}. */
public class VortexWriterFactory implements FormatWriterFactory, SupportsDirectWrite {
- private final RowType rowType;
- private final Supplier<ArrowFormatWriter> arrowFormatWriterSupplier;
+ private final Supplier<ArrowFormatCWriter> cWriterSupplier;
- public VortexWriterFactory(
- RowType rowType, Supplier<ArrowFormatWriter> arrowFormatWriterSupplier) {
- this.rowType = rowType;
- this.arrowFormatWriterSupplier = arrowFormatWriterSupplier;
+ public VortexWriterFactory(Supplier<ArrowFormatCWriter> cWriterSupplier) {
+ this.cWriterSupplier = cWriterSupplier;
}
@Override
@@ -57,9 +53,6 @@ public class VortexWriterFactory implements FormatWriterFactory, SupportsDirectW
public FormatWriter create(FileIO fileIO, Path path, String compression) throws IOException {
Pair<Path, Map<String, String>> vortexSpecified = toVortexSpecifiedForWriter(fileIO, path);
return new VortexRecordsWriter(
- rowType,
- arrowFormatWriterSupplier.get(),
- vortexSpecified.getLeft(),
- vortexSpecified.getRight());
+ cWriterSupplier, vortexSpecified.getLeft(), vortexSpecified.getRight());
}
}
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
index a270d2e92f..a65916bb6c 100644
--- 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
@@ -30,14 +30,10 @@ 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 {
@@ -46,29 +42,6 @@ public class VortexFileFormatReadWriteTest extends FormatReadWriteTest {
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;
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
index 1b89659110..b2f51ef20f 100644
--- 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
@@ -18,28 +18,50 @@
package org.apache.paimon.format.vortex;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
+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.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
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 org.junit.jupiter.api.io.TempDir;
+import java.math.BigDecimal;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Optional;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
-/** Tests for {@link VortexPredicateConverter}. */
+/**
+ * Tests for {@link VortexPredicateConverter}.
+ *
+ * <p>Each test verifies two things: (1) the converted expression has a valid native pointer, and
+ * (2) the predicate produces correct results when used in a write-read round-trip through the
+ * Vortex format.
+ */
public class VortexPredicateConverterTest {
private static final RowType ROW_TYPE =
@@ -51,112 +73,99 @@ public class VortexPredicateConverterTest {
private static final PredicateBuilder BUILDER = new PredicateBuilder(ROW_TYPE);
- private static Expression field(String name) {
- return GetItem.of(Root.INSTANCE, name);
+ private static void assertValidExpression(Expression expr) {
+ assertNotNull(expr, "Expression should not be null");
+ assertTrue(expr.nativePointer() != 0, "Expression native pointer should be non-zero");
}
+ // -- Predicate conversion + native pointer validity tests --
+
@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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.equal(0, 42)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.notEqual(1, 100L)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.greaterThan(0, 10)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.greaterOrEqual(0, 10)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.lessThan(1, 50L)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.lessOrEqual(1, 50L)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.isNull(0)));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(BUILDER.isNotNull(0)));
+ assertValidExpression(result);
}
@Test
public void testAnd() {
- Predicate p1 = BUILDER.greaterThan(0, 10);
- Predicate p2 = BUILDER.lessThan(0, 100);
- Predicate and = PredicateBuilder.and(p1, p2);
+ Predicate and = PredicateBuilder.and(BUILDER.greaterThan(0, 10), BUILDER.lessThan(0, 100));
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);
+ assertValidExpression(result);
}
@Test
public void testOr() {
- Predicate p1 = BUILDER.equal(0, 1);
- Predicate p2 = BUILDER.equal(0, 2);
- Predicate or = PredicateBuilder.or(p1, p2);
+ Predicate or = PredicateBuilder.or(BUILDER.equal(0, 1), BUILDER.equal(0, 2));
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);
+ assertValidExpression(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);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(
+ Arrays.asList(BUILDER.greaterThan(0, 5), BUILDER.lessThan(1, 200L)));
+ assertValidExpression(result);
}
@Test
@@ -171,74 +180,221 @@ public class VortexPredicateConverterTest {
@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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ BUILDER.equal(2, BinaryString.fromString("hello"))));
+ assertValidExpression(result);
+ }
+
+ @Test
+ public void testDecimalLiteral() {
+ RowType decRowType = RowType.builder().field("f_dec", DataTypes.DECIMAL(10, 2)).build();
+ PredicateBuilder decBuilder = new PredicateBuilder(decRowType);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ decBuilder.equal(
+ 0,
+ Decimal.fromBigDecimal(new BigDecimal("123.45"), 10, 2))));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ tsBuilder.equal(0, Timestamp.fromEpochMillis(123456789L))));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ tsBuilder.equal(0, Timestamp.fromMicros(123456789123L))));
+ assertValidExpression(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);
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ tsBuilder.equal(0, Timestamp.fromEpochMillis(123456L, 789012))));
+ assertValidExpression(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);
+ Expression result =
+ VortexPredicateConverter.toVortexExpression(
+ Collections.singletonList(
+ tsBuilder.equal(0, Timestamp.fromEpochMillis(123456789L))));
+ assertValidExpression(result);
+ }
+
+ // -- Semantic round-trip tests: write data, read with predicate, verify filtering --
+
+ @Test
+ public void testEqualSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_int == 2 should return only the row with f_int=2
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("a")),
+ GenericRow.of(2, 20L, BinaryString.fromString("b")),
+ GenericRow.of(3, 30L, BinaryString.fromString("c"))
+ },
+ Collections.singletonList(BUILDER.equal(0, 2)));
+ assertEquals(1, rows.size());
+ assertEquals(2, rows.get(0).getInt(0));
+ }
+
+ @Test
+ public void testGreaterThanSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_int > 2 should return rows with f_int=3,4
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("a")),
+ GenericRow.of(2, 20L, BinaryString.fromString("b")),
+ GenericRow.of(3, 30L, BinaryString.fromString("c")),
+ GenericRow.of(4, 40L, BinaryString.fromString("d"))
+ },
+ Collections.singletonList(BUILDER.greaterThan(0, 2)));
+ assertEquals(2, rows.size());
+ assertEquals(3, rows.get(0).getInt(0));
+ assertEquals(4, rows.get(1).getInt(0));
+ }
+
+ @Test
+ public void testGreaterOrEqualSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_int >= 3 should return rows with f_int=3,4
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("a")),
+ GenericRow.of(2, 20L, BinaryString.fromString("b")),
+ GenericRow.of(3, 30L, BinaryString.fromString("c")),
+ GenericRow.of(4, 40L, BinaryString.fromString("d"))
+ },
+ Collections.singletonList(BUILDER.greaterOrEqual(0, 3)));
+ assertEquals(2, rows.size());
+ assertEquals(3, rows.get(0).getInt(0));
+ assertEquals(4, rows.get(1).getInt(0));
+ }
+
+ @Test
+ public void testAndSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_int > 1 AND f_int < 4 should return rows 2,3
+ Predicate and = PredicateBuilder.and(BUILDER.greaterThan(0, 1), BUILDER.lessThan(0, 4));
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("a")),
+ GenericRow.of(2, 20L, BinaryString.fromString("b")),
+ GenericRow.of(3, 30L, BinaryString.fromString("c")),
+ GenericRow.of(4, 40L, BinaryString.fromString("d"))
+ },
+ Collections.singletonList(and));
+ assertEquals(2, rows.size());
+ assertEquals(2, rows.get(0).getInt(0));
+ assertEquals(3, rows.get(1).getInt(0));
+ }
+
+ @Test
+ public void testOrSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_int == 1 OR f_int == 4 should return rows 1,4
+ Predicate or = PredicateBuilder.or(BUILDER.equal(0, 1), BUILDER.equal(0, 4));
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("a")),
+ GenericRow.of(2, 20L, BinaryString.fromString("b")),
+ GenericRow.of(3, 30L, BinaryString.fromString("c")),
+ GenericRow.of(4, 40L, BinaryString.fromString("d"))
+ },
+ Collections.singletonList(or));
+ assertEquals(2, rows.size());
+ assertEquals(1, rows.get(0).getInt(0));
+ assertEquals(4, rows.get(1).getInt(0));
+ }
+
+ @Test
+ public void testStringEqualSemantic(@TempDir java.nio.file.Path tempDir) throws Exception {
+ // f_string == "hello"
+ List<InternalRow> rows =
+ roundTrip(
+ tempDir,
+ ROW_TYPE,
+ new GenericRow[] {
+ GenericRow.of(1, 10L, BinaryString.fromString("hello")),
+ GenericRow.of(2, 20L, BinaryString.fromString("world")),
+ },
+ Collections.singletonList(
+ BUILDER.equal(2, BinaryString.fromString("hello"))));
+ assertEquals(1, rows.size());
+ assertEquals(BinaryString.fromString("hello"), rows.get(0).getString(2));
+ }
+
+ private List<InternalRow> roundTrip(
+ java.nio.file.Path tempDir,
+ RowType rowType,
+ GenericRow[] data,
+ List<Predicate> predicates)
+ throws Exception {
+ 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()), "predicate_test_" + System.nanoTime());
+
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (GenericRow row : data) {
+ writer.addElement(row);
+ }
+ }
+
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+ FormatReaderFactory readerFactory =
+ format.createReaderFactory(rowType, rowType, predicates);
+ try (RecordReader<InternalRow> reader =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, testFile, fileIO.getFileSize(testFile), null));
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(reader)) {
+ List<InternalRow> result = new ArrayList<>();
+ while (iterator.hasNext()) {
+ result.add(serializer.copy(iterator.next()));
+ }
+ return 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
index 85230da5ff..7ccca17c09 100644
--- 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
@@ -44,7 +44,6 @@ 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;
@@ -58,25 +57,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-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());
@@ -122,6 +106,56 @@ public class VortexReaderWriterTest {
}
}
+ @Test
+ public void testReadWithColumnProjection(@TempDir java.nio.file.Path tempDir) throws Exception {
+ RowType fullRowType =
+ RowType.builder()
+ .field("f_int", DataTypes.INT())
+ .field("f_string", DataTypes.STRING())
+ .field("f_double", DataTypes.DOUBLE())
+ .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_projection_" + UUID.randomUUID());
+
+ // Write 3 columns
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(fullRowType))
+ .create(fileIO, testFile, "")) {
+ writer.addElement(GenericRow.of(1, BinaryString.fromString("hello"), 1.5D));
+ writer.addElement(GenericRow.of(2, BinaryString.fromString("world"), 2.5D));
+ }
+
+ InternalRowSerializer serializer;
+
+ // Read only f_string column
+ RowType projectedRowType = RowType.builder().field("f_string", DataTypes.STRING()).build();
+ serializer = new InternalRowSerializer(projectedRowType);
+ FormatReaderFactory readerFactory =
+ format.createReaderFactory(fullRowType, projectedRowType, 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(2, actualRows.size());
+ assertEquals(1, actualRows.get(0).getFieldCount());
+ assertEquals(BinaryString.fromString("hello"), actualRows.get(0).getString(0));
+ assertEquals(BinaryString.fromString("world"), actualRows.get(1).getString(0));
+ }
+ }
+
@Test
public void testWriteAndReadMultipleTypes(@TempDir java.nio.file.Path tempDir)
throws Exception {
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
deleted file mode 100644
index 8a06c273e5..0000000000
--- a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexTypeUtilsTest.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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
index a8ba77a471..f9c3e50174 100644
--- a/paimon-vortex/paimon-vortex-jni/README.md
+++ b/paimon-vortex/paimon-vortex-jni/README.md
@@ -4,76 +4,40 @@ This module provides Java JNI bindings for the [Vortex](https://github.com/spira
## Source Code Origin
-The Java source code under `src/main/java/dev/vortex/` is copied from the upstream
+The Java source code under `src/main/java/dev/vortex/` is adapted 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.
+- Target Java version set to 1.8 for Paimon compatibility (upstream requires JDK 17).
+- `java.lang.ref.Cleaner` (JDK 9+) replaced with explicit `close()` via `AutoCloseable`.
+- Protobuf-based expression serialization replaced with native pointer-based `Expression` API.
+- Arrow memory allocation uses `Unsafe` allocator (via `arrow-memory-unsafe`) to guarantee
+ buffer alignment required by Vortex's Rust FFI.
-## Building the Native Library
+## 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.
+The native library (`libvortex_jni`) is automatically extracted from the published
+[`dev.vortex:vortex-jni`](https://mvnrepository.com/artifact/dev.vortex/vortex-jni)
+Maven artifact during the `generate-resources` phase via `maven-dependency-plugin:unpack`.
-### 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:
+The published jar includes pre-built binaries for:
| 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
-```
+No Rust toolchain or manual build is required.
## Verification
```bash
-mvn test -pl paimon-vortex/paimon-vortex-jni -Dcheckstyle.skip=true -Dspotless.check.skip=true
+mvn test -pl paimon-vortex/paimon-vortex-jni,paimon-vortex/paimon-vortex-format \
+ -Dcheckstyle.skip=true -Dspotless.check.skip=true
```
Tests that require the native library use `assumeTrue` to check availability at runtime.
@@ -83,7 +47,6 @@ If the library is not found, those tests will be skipped rather than fail.
| 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`. |
+| `UnsatisfiedLinkError` | Ensure `maven-dependency-plugin:unpack` ran (check `target/classes/native/`). |
+| `FileNotFoundException: Library not found` | Check `os.name` and `os.arch` system properties match a supported platform. |
+| `Memory pointer ... not aligned` | Ensure `arrow.allocation.manager.type=Unsafe` is set (done automatically by `NativeLoader`). |
diff --git a/paimon-vortex/paimon-vortex-jni/pom.xml b/paimon-vortex/paimon-vortex-jni/pom.xml
index 913741aa30..c6ca7a8d6e 100644
--- a/paimon-vortex/paimon-vortex-jni/pom.xml
+++ b/paimon-vortex/paimon-vortex-jni/pom.xml
@@ -33,8 +33,7 @@ under the License.
<properties>
<target.java.version>1.8</target.java.version>
- <vortex.jni.version>0.69.0</vortex.jni.version>
- <protobuf.version>3.25.5</protobuf.version>
+ <vortex.jni.version>0.73.0</vortex.jni.version>
<immutables.version>2.10.1</immutables.version>
<!-- vortex-jni uses shaded Guava, skip spotless/checkstyle rewriting -->
<spotless.check.skip>true</spotless.check.skip>
@@ -64,15 +63,14 @@ under the License.
<dependency>
<groupId>org.apache.arrow</groupId>
- <artifactId>arrow-vector</artifactId>
+ <artifactId>arrow-memory-unsafe</artifactId>
<version>${arrow.version}</version>
</dependency>
- <!-- Protobuf -->
<dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>${protobuf.version}</version>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-vector</artifactId>
+ <version>${arrow.version}</version>
</dependency>
<!-- Guava (shaded) -->
@@ -101,35 +99,30 @@ under the License.
<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 -->
+ <!-- Extract native libraries from published vortex-jni jar -->
<plugin>
- <groupId>kr.motd.maven</groupId>
- <artifactId>os-maven-plugin</artifactId>
- <version>1.7.1</version>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
- <phase>initialize</phase>
+ <id>unpack-vortex-native</id>
+ <phase>generate-resources</phase>
<goals>
- <goal>detect</goal>
+ <goal>unpack</goal>
</goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>dev.vortex</groupId>
+ <artifactId>vortex-jni</artifactId>
+ <version>${vortex.jni.version}</version>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+ </artifactItem>
+ </artifactItems>
+ <includes>native/**</includes>
+ <outputDirectory>${project.build.directory}/classes</outputDirectory>
+ </configuration>
</execution>
</executions>
</plugin>
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
deleted file mode 100644
index 369339215c..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Array.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.math.BigDecimal;
-import org.apache.arrow.memory.BufferAllocator;
-import org.apache.arrow.vector.VectorSchemaRoot;
-
-/** 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/DType.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DType.java
deleted file mode 100644
index 1530595b44..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DType.java
+++ /dev/null
@@ -1,270 +0,0 @@
-/*
- * 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) {
- // Get the pointer
- 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);
- }
-
- 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);
- }
- }
- }
-
- 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/DataSource.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DataSource.java
new file mode 100644
index 0000000000..ffb9f514d8
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/DataSource.java
@@ -0,0 +1,100 @@
+/*
+ * 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.NativeDataSource;
+import dev.vortex.jni.NativeScan;
+
+import java.util.Map;
+import java.util.OptionalLong;
+
+/** A Vortex data source opened from one or more URIs. */
+public final class DataSource implements AutoCloseable {
+
+ private final Session session;
+ private long pointer;
+
+ private DataSource(Session session, long pointer) {
+ this.session = session;
+ this.pointer = pointer;
+ }
+
+ public static DataSource open(Session session, String uri, Map<String, String> options) {
+ long ptr =
+ NativeDataSource.open(
+ session.nativePointer(), new String[] {uri}, options);
+ if (ptr == 0) {
+ throw new RuntimeException("Failed to open Vortex data source: " + uri);
+ }
+ return new DataSource(session, ptr);
+ }
+
+ public long rowCount() {
+ long[] out = new long[2];
+ NativeDataSource.rowCount(pointer, out);
+ // out[1]: 2=exact, 1=estimate, other=unknown
+ if (out[1] == 2 || out[1] == 1) {
+ return out[0];
+ }
+ return -1;
+ }
+
+ public Scan scan(ScanOptions options) {
+ long projectionPtr =
+ options.projection().isPresent()
+ ? options.projection().get().nativePointer()
+ : 0;
+ long filterPtr =
+ options.filter().isPresent() ? options.filter().get().nativePointer() : 0;
+ long rangeBegin = optionalLongOrZero(options.rowRangeBegin());
+ long rangeEnd = optionalLongOrZero(options.rowRangeEnd());
+ long[] selectionIndices =
+ options.selectionIndices().isPresent()
+ ? options.selectionIndices().get()
+ : null;
+ byte selectionMode = options.selectionMode().code();
+ long limit = optionalLongOrZero(options.limit());
+ boolean ordered = options.ordered();
+
+ long scanPtr =
+ NativeScan.create(
+ pointer,
+ projectionPtr,
+ filterPtr,
+ rangeBegin,
+ rangeEnd,
+ selectionIndices,
+ selectionMode,
+ limit,
+ ordered);
+ return Scan.fromPointer(session, scanPtr);
+ }
+
+ @Override
+ public void close() {
+ if (pointer != 0) {
+ NativeDataSource.free(pointer);
+ pointer = 0;
+ }
+ }
+
+ private static long optionalLongOrZero(OptionalLong opt) {
+ return opt.isPresent() ? opt.getAsLong() : 0;
+ }
+}
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
index d3b6edf241..ccff4e4aa2 100644
--- 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
@@ -18,41 +18,203 @@
package dev.vortex.api;
-import dev.vortex.api.expressions.*;
-import java.util.List;
-import java.util.Optional;
+import dev.vortex.jni.NativeExpression;
-/** Vortex expression language. */
-public interface Expression {
- String id();
+import java.math.BigInteger;
- List<Expression> children();
+/** A Vortex expression backed by a native pointer. */
+public final class Expression implements AutoCloseable {
- Optional<byte[]> metadata();
+ private long pointer;
- default <T> T accept(Visitor<T> visitor) {
- return visitor.visitOther(this);
+ private Expression(long pointer) {
+ this.pointer = pointer;
}
- interface Visitor<T> {
- T visitLiteral(Literal<?> literal);
+ public long nativePointer() {
+ return pointer;
+ }
- T visitRoot(Root root);
+ @Override
+ public void close() {
+ if (pointer != 0) {
+ NativeExpression.free(pointer);
+ pointer = 0;
+ }
+ }
- T visitBinary(Binary binary);
+ // -- Structure navigation --
- T visitNot(Not not);
+ public static Expression root() {
+ return new Expression(NativeExpression.root());
+ }
- T visitGetItem(GetItem getItem);
+ public static Expression column(String name) {
+ long rootPtr = NativeExpression.root();
+ return new Expression(NativeExpression.getItem(name, rootPtr));
+ }
- default T visitIsNull(IsNull isNull) {
- return visitOther(isNull);
+ public static Expression select(String[] columns, Expression parent) {
+ return new Expression(NativeExpression.select(columns, parent.pointer));
+ }
+
+ // -- Logical combinators --
+
+ public static Expression and(Expression... exprs) {
+ long[] ptrs = new long[exprs.length];
+ for (int i = 0; i < exprs.length; i++) {
+ ptrs[i] = exprs[i].pointer;
}
+ return new Expression(NativeExpression.and(ptrs));
+ }
- default T visitIsNotNull(IsNotNull isNotNull) {
- return visitOther(isNotNull);
+ public static Expression or(Expression... exprs) {
+ long[] ptrs = new long[exprs.length];
+ for (int i = 0; i < exprs.length; i++) {
+ ptrs[i] = exprs[i].pointer;
}
+ return new Expression(NativeExpression.or(ptrs));
+ }
+
+ public static Expression not(Expression expr) {
+ return new Expression(NativeExpression.not(expr.pointer));
+ }
+
+ // -- Comparison / binary ops --
+
+ public static Expression binary(BinaryOp op, Expression left, Expression right) {
+ return new Expression(NativeExpression.binary(op.code(), left.pointer, right.pointer));
+ }
+
+ // -- Null checks --
+
+ public static Expression isNull(Expression expr) {
+ return new Expression(NativeExpression.isNull(expr.pointer));
+ }
+
+ public static Expression isNotNull(Expression expr) {
+ return new Expression(NativeExpression.isNotNull(expr.pointer));
+ }
- T visitOther(Expression expression);
+ // -- Primitive literals --
+
+ public static Expression literal(boolean value) {
+ return new Expression(NativeExpression.literalBool(value, false));
+ }
+
+ public static Expression literal(byte value) {
+ return new Expression(NativeExpression.literalI8(value, false));
+ }
+
+ public static Expression literal(short value) {
+ return new Expression(NativeExpression.literalI16(value, false));
+ }
+
+ public static Expression literal(int value) {
+ return new Expression(NativeExpression.literalI32(value, false));
+ }
+
+ public static Expression literal(long value) {
+ return new Expression(NativeExpression.literalI64(value, false));
+ }
+
+ public static Expression literal(float value) {
+ return new Expression(NativeExpression.literalF32(value, false));
+ }
+
+ public static Expression literal(double value) {
+ return new Expression(NativeExpression.literalF64(value, false));
+ }
+
+ public static Expression literal(String value) {
+ return new Expression(NativeExpression.literalString(value));
+ }
+
+ // -- Decimal literals --
+
+ public static Expression literalDecimal(BigInteger unscaledValue, int precision, int scale) {
+ byte[] bytes = unscaledValue.toByteArray();
+ return new Expression(NativeExpression.literalDecimal(bytes, precision, scale, false));
+ }
+
+ // -- Date/time literals --
+
+ public static Expression literalDate(long value, TimeUnit unit) {
+ return new Expression(NativeExpression.literalDate(value, unit.tag(), false));
+ }
+
+ public static Expression literalTimestamp(long value, TimeUnit unit, String timezone) {
+ return new Expression(
+ NativeExpression.literalTimestamp(value, unit.tag(), timezone, false));
+ }
+
+ // -- Null literals --
+
+ public static Expression nullLiteral(DType dtype) {
+ return new Expression(NativeExpression.literalNull(dtype.tag()));
+ }
+
+ /** Binary operation codes. */
+ public enum BinaryOp {
+ EQ((byte) 0),
+ NOT_EQ((byte) 1),
+ GT((byte) 2),
+ GTE((byte) 3),
+ LT((byte) 4),
+ LTE((byte) 5),
+ AND((byte) 6),
+ OR((byte) 7);
+
+ private final byte code;
+
+ BinaryOp(byte code) {
+ this.code = code;
+ }
+
+ public byte code() {
+ return code;
+ }
+ }
+
+ /** Data type tags for null literals. */
+ public enum DType {
+ BOOL((byte) 0),
+ I8((byte) 1),
+ I16((byte) 2),
+ I32((byte) 3),
+ I64((byte) 4),
+ F32((byte) 5),
+ F64((byte) 6),
+ UTF8((byte) 7),
+ BINARY((byte) 8);
+
+ private final byte tag;
+
+ DType(byte tag) {
+ this.tag = tag;
+ }
+
+ public byte tag() {
+ return tag;
+ }
+ }
+
+ /** Time unit tags for date/time literals. */
+ public enum TimeUnit {
+ NANOSECONDS((byte) 0),
+ MICROSECONDS((byte) 1),
+ MILLISECONDS((byte) 2),
+ SECONDS((byte) 3),
+ DAYS((byte) 4);
+
+ private final byte tag;
+
+ TimeUnit(byte tag) {
+ this.tag = tag;
+ }
+
+ public byte tag() {
+ return tag;
+ }
}
}
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
deleted file mode 100644
index f812a0fbe9..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Files.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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/Partition.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Partition.java
new file mode 100644
index 0000000000..83b5e39237
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Partition.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 dev.vortex.api;
+
+import dev.vortex.jni.NativePartition;
+
+import org.apache.arrow.c.ArrowArrayStream;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+/** A partition from a Vortex scan, consumable once as an Arrow stream. */
+public final class Partition implements AutoCloseable {
+
+ private final Session session;
+ private long pointer;
+ private boolean consumed;
+
+ private Partition(Session session, long pointer) {
+ this.session = session;
+ this.pointer = pointer;
+ }
+
+ static Partition fromPointer(Session session, long pointer) {
+ return new Partition(session, pointer);
+ }
+
+ public ArrowReader scanArrow(BufferAllocator allocator) {
+ if (consumed) {
+ throw new IllegalStateException("partition already consumed");
+ }
+ consumed = true;
+ ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator);
+ try {
+ NativePartition.scanArrow(
+ session.nativePointer(), pointer, stream.memoryAddress());
+ } catch (RuntimeException e) {
+ stream.close();
+ throw e;
+ }
+ return Data.importArrayStream(allocator, stream);
+ }
+
+ @Override
+ public void close() {
+ // Only free unconsumed partitions. scanArrow() transfers the native
+ // partition's ownership to the ArrowArrayStream; the stream's release
+ // callback frees it when the ArrowReader is closed.
+ if (pointer != 0 && !consumed) {
+ NativePartition.free(pointer);
+ }
+ pointer = 0;
+ }
+}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Scan.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Scan.java
new file mode 100644
index 0000000000..260da34888
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Scan.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.api;
+
+import dev.vortex.jni.NativePartition;
+import dev.vortex.jni.NativeScan;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/** An iterator over partitions from a Vortex scan. */
+public final class Scan implements Iterator<Partition>, AutoCloseable {
+
+ private final Session session;
+ private long pointer;
+ private long nextPartitionPointer;
+ private boolean primed;
+
+ private Scan(Session session, long pointer) {
+ this.session = session;
+ this.pointer = pointer;
+ }
+
+ static Scan fromPointer(Session session, long pointer) {
+ return new Scan(session, pointer);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (primed) {
+ return nextPartitionPointer != 0;
+ }
+ nextPartitionPointer = NativeScan.nextPartition(pointer);
+ primed = true;
+ return nextPartitionPointer != 0;
+ }
+
+ @Override
+ public Partition next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ long ptr = nextPartitionPointer;
+ nextPartitionPointer = 0;
+ primed = false;
+ return Partition.fromPointer(session, ptr);
+ }
+
+ @Override
+ public void close() {
+ if (nextPartitionPointer != 0) {
+ NativePartition.free(nextPartitionPointer);
+ nextPartitionPointer = 0;
+ }
+ if (pointer != 0) {
+ NativeScan.free(pointer);
+ pointer = 0;
+ }
+ }
+}
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
index 5aa2411340..cb2b1992d2 100644
--- 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
@@ -18,25 +18,51 @@
package dev.vortex.api;
-import java.util.List;
-import java.util.Optional;
import org.immutables.value.Value;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/** Options for configuring a Vortex scan. */
@Value.Immutable
public interface ScanOptions {
- List<String> columns();
- Optional<Expression> predicate();
+ Optional<Expression> projection();
- Optional<long[]> rowRange();
+ Optional<Expression> filter();
- Optional<long[]> rowIndices();
+ OptionalLong rowRangeBegin();
- static ScanOptions of() {
- return ImmutableScanOptions.builder().build();
+ OptionalLong rowRangeEnd();
+
+ Optional<long[]> selectionIndices();
+
+ @Value.Default
+ default SelectionMode selectionMode() {
+ return SelectionMode.INCLUDE_ALL;
}
- static ImmutableScanOptions.Builder builder() {
- return ImmutableScanOptions.builder();
+ OptionalLong limit();
+
+ @Value.Default
+ default boolean ordered() {
+ return false;
+ }
+
+ /** Selection mode for row indices. */
+ enum SelectionMode {
+ INCLUDE_ALL((byte) 0),
+ INCLUDE((byte) 1),
+ EXCLUDE((byte) 2);
+
+ private final byte code;
+
+ SelectionMode(byte code) {
+ this.code = code;
+ }
+
+ public byte code() {
+ return code;
+ }
}
}
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/Session.java
similarity index 55%
rename from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/ArrayIterator.java
rename to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/Session.java
index 41df75e53a..0d6d8d2ee3 100644
--- 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/Session.java
@@ -18,12 +18,34 @@
package dev.vortex.api;
-import java.util.Iterator;
+import dev.vortex.jni.NativeSession;
-/** An iterator over Vortex arrays with type information and resource management. */
-public interface ArrayIterator extends AutoCloseable, Iterator<Array> {
- DType getDataType();
+/** A Vortex session that manages the native runtime. */
+public final class Session implements AutoCloseable {
+
+ private long pointer;
+
+ private Session(long pointer) {
+ this.pointer = pointer;
+ }
+
+ public static Session create() {
+ long ptr = NativeSession.newSession();
+ if (ptr == 0) {
+ throw new RuntimeException("Failed to create Vortex session");
+ }
+ return new Session(ptr);
+ }
+
+ public long nativePointer() {
+ return pointer;
+ }
@Override
- void close();
+ public void close() {
+ if (pointer != 0) {
+ NativeSession.free(pointer);
+ pointer = 0;
+ }
+ }
}
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
index 1153556316..bcdeebc25f 100644
--- 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
@@ -18,27 +18,79 @@
package dev.vortex.api;
-import dev.vortex.jni.JNIDType;
-import dev.vortex.jni.JNIWriter;
-import dev.vortex.jni.NativeWriterMethods;
+import dev.vortex.jni.NativeWriter;
+
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.types.pojo.Schema;
+
import java.io.IOException;
import java.util.Map;
-/** Writer for creating Vortex files from Arrow data. */
-public interface VortexWriter extends AutoCloseable {
+/** A Vortex file writer using the Arrow C Data Interface. */
+public final class VortexWriter implements 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);
+ private final long pointer;
+ private boolean closed;
+
+ private VortexWriter(long pointer) {
+ this.pointer = pointer;
}
- void writeBatch(byte[] arrowData) throws IOException;
+ public static VortexWriter create(
+ Session session, String uri, Schema arrowSchema, Map<String, String> options)
+ throws IOException {
+ // Use a dedicated allocator for schema export to avoid leaking retained
+ // references into the caller's allocator (Arrow C Data Interface retain/release
+ // semantics don't fully clean up via ArrowSchema.close in Arrow 15).
+ RootAllocator schemaAllocator = new RootAllocator(Long.MAX_VALUE);
+ try {
+ ArrowSchema cSchema = ArrowSchema.allocateNew(schemaAllocator);
+ Data.exportSchema(schemaAllocator, arrowSchema, null, cSchema);
+ long ptr =
+ NativeWriter.create(
+ session.nativePointer(), uri, cSchema.memoryAddress(), options);
+ cSchema.close();
+ if (ptr <= 0) {
+ throw new IOException("Failed to create Vortex writer for: " + uri);
+ }
+ return new VortexWriter(ptr);
+ } finally {
+ try {
+ schemaAllocator.close();
+ } catch (IllegalStateException e) {
+ if (e.getMessage() == null || !e.getMessage().contains("Memory was leaked")) {
+ throw e;
+ }
+ }
+ }
+ }
- void writeBatchFfi(long arrowArrayAddr, long arrowSchemaAddr) throws IOException;
+ public void writeBatch(long arrowArrayAddr, long arrowSchemaAddr) throws IOException {
+ if (closed) {
+ throw new IllegalStateException("writer already closed");
+ }
+ boolean ok;
+ try {
+ ok = NativeWriter.writeBatch(pointer, arrowArrayAddr, arrowSchemaAddr);
+ } catch (RuntimeException e) {
+ throw new IOException("failed to write batch", e);
+ }
+ if (!ok) {
+ throw new IOException("NativeWriter.writeBatch returned false");
+ }
+ }
@Override
- void close() throws IOException;
+ public void close() throws IOException {
+ if (!closed && pointer != 0) {
+ closed = true;
+ try {
+ NativeWriter.close(pointer);
+ } catch (RuntimeException e) {
+ throw new IOException("failed to close writer", e);
+ }
+ }
+ }
}
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
deleted file mode 100644
index 26534c16eb..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Binary.java
+++ /dev/null
@@ -1,234 +0,0 @@
-/*
- * 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;
- }
-
- public enum BinaryOp {
- /** Equality comparison operator (==) */
- EQ,
- /** Inequality comparison operator (!=) */
- NOT_EQ,
- /** Greater-than comparison operator (>) */
- GT,
- /** Greater-than-or-equal comparison operator (>=) */
- GT_EQ,
- /** Less-than comparison operator (<) */
- LT,
- /** Less-than-or-equal comparison operator (<=) */
- LT_EQ,
- /** Logical AND operator (&&) */
- AND,
- /** Logical OR operator (||) */
- 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
deleted file mode 100644
index ee13de5256..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/GetItem.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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/IsNotNull.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/IsNotNull.java
deleted file mode 100644
index 1981ca1f2b..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/IsNotNull.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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;
-
-public final class IsNotNull implements Expression {
- private final Expression child;
-
- private IsNotNull(Expression child) {
- this.child = child;
- }
-
- public static IsNotNull parse(byte[] metadata, List<Expression> children) {
- if (children.size() != 1) {
- throw new IllegalArgumentException(
- "IsNotNull expression must have exactly one child, found: " + children.size());
- }
- if (metadata.length > 0) {
- throw new IllegalArgumentException(
- "IsNotNull expression must not have metadata, found: " + metadata.length);
- }
- return new IsNotNull(children.get(0));
- }
-
- public static IsNotNull of(Expression child) {
- return new IsNotNull(child);
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null || getClass() != o.getClass()) return false;
- IsNotNull other = (IsNotNull) o;
- return Objects.equals(child, other.child);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(child);
- }
-
- @Override
- public String id() {
- return "vortex.is_not_null";
- }
-
- @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 "vortex.is_not_null(" + child + ")";
- }
-
- public Expression getChild() {
- return child;
- }
-
- @Override
- public <T> T accept(Visitor<T> visitor) {
- return visitor.visitIsNotNull(this);
- }
-}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/IsNull.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/IsNull.java
deleted file mode 100644
index 50603e7909..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/IsNull.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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;
-
-public final class IsNull implements Expression {
- private final Expression child;
-
- private IsNull(Expression child) {
- this.child = child;
- }
-
- public static IsNull parse(byte[] metadata, List<Expression> children) {
- if (children.size() != 1) {
- throw new IllegalArgumentException(
- "IsNull expression must have exactly one child, found: " + children.size());
- }
- if (metadata.length > 0) {
- throw new IllegalArgumentException("IsNull expression must not have metadata, found: " + metadata.length);
- }
- return new IsNull(children.get(0));
- }
-
- public static IsNull of(Expression child) {
- return new IsNull(child);
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null || getClass() != o.getClass()) return false;
- IsNull other = (IsNull) o;
- return Objects.equals(child, other.child);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(child);
- }
-
- @Override
- public String id() {
- return "vortex.is_null";
- }
-
- @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 "vortex.is_null(" + child + ")";
- }
-
- public Expression getChild() {
- return child;
- }
-
- @Override
- public <T> T accept(Visitor<T> visitor) {
- return visitor.visitIsNull(this);
- }
-}
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
deleted file mode 100644
index a4a8a7e26d..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Literal.java
+++ /dev/null
@@ -1,781 +0,0 @@
-/*
- * 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);
-
- 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) {
- if (Objects.isNull(literal)) {
- return Scalars.nullBool();
- } else {
- return Scalars.bool(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitInt8(Byte literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullInt8();
- } else {
- return Scalars.int8(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitInt16(Short literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullInt16();
- } else {
- return Scalars.int16(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitInt32(Integer literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullInt32();
- } else {
- return Scalars.int32(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitInt64(Long literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullInt64();
- } else {
- return Scalars.int64(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitDateDays(Integer days) {
- if (Objects.isNull(days)) {
- return Scalars.nullDateDays();
- } else {
- return Scalars.dateDays(days);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitDateMillis(Long millis) {
- if (Objects.isNull(millis)) {
- return Scalars.nullDateMillis();
- } else {
- return Scalars.dateMillis(millis);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimeSeconds(Integer seconds) {
- if (Objects.isNull(seconds)) {
- return Scalars.nullTimeSeconds();
- } else {
- return Scalars.timeSeconds(seconds);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimeMillis(Integer seconds) {
- if (Objects.isNull(seconds)) {
- return Scalars.nullTimeMillis();
- } else {
- return Scalars.timeMillis(seconds);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimeMicros(Long seconds) {
- if (Objects.isNull(seconds)) {
- return Scalars.nullTimeMicros();
- } else {
- return Scalars.timeMicros(seconds);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimeNanos(Long seconds) {
- if (Objects.isNull(seconds)) {
- return Scalars.nullTimeNanos();
- } else {
- return Scalars.timeNanos(seconds);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimestampMillis(Long epochMillis, Optional<String> timeZone) {
- if (Objects.isNull(epochMillis)) {
- return Scalars.nullTimestampMillis(timeZone);
- } else {
- return Scalars.timestampMillis(epochMillis, timeZone);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimestampMicros(Long epochMicros, Optional<String> timeZone) {
- if (Objects.isNull(epochMicros)) {
- return Scalars.nullTimestampMicros(timeZone);
- } else {
- return Scalars.timestampMicros(epochMicros, timeZone);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitTimestampNanos(Long epochNanos, Optional<String> timeZone) {
- if (Objects.isNull(epochNanos)) {
- return Scalars.nullTimestampNanos(timeZone);
- } else {
- return Scalars.timestampNanos(epochNanos, timeZone);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitFloat32(Float literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullFloat32();
- } else {
- return Scalars.float32(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitFloat64(Double literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullFloat64();
- } else {
- return Scalars.float64(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitDecimal(BigDecimal decimal, int precision, int scale) {
- if (Objects.isNull(decimal)) {
- return Scalars.nullDecimal(precision, scale);
- } else {
- return Scalars.decimal(decimal, precision, scale);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitString(String literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullString();
- } else {
- return Scalars.string(literal);
- }
- }
-
- @Override
- public ScalarProtos.Scalar visitBytes(byte[] literal) {
- if (Objects.isNull(literal)) {
- return Scalars.nullBytes();
- } else {
- return Scalars.bytes(literal);
- }
- }
- }
-
- private static Literal<?> deserializeLiteral(ExprProtos.LiteralOpts literal, List<Expression> children) {
- ScalarProtos.Scalar literalScalar = literal.getValue();
- DTypeProtos.DType dtype = literalScalar.getDtype();
-
- // Special handling of extension types
- 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
deleted file mode 100644
index 8e1df113b9..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Not.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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
deleted file mode 100644
index a92125b414..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Root.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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 "$";
- }
-
- // equals and hashCode depend on address equality to INSTANCE.
-
- @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
deleted file mode 100644
index d3f2c13631..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/expressions/Unknown.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 5d0cf6d08c..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/DTypes.java
+++ /dev/null
@@ -1,201 +0,0 @@
-/*
- * 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 static dev.vortex.api.proto.TemporalMetadatas.TIME_UNIT_MICROS;
-import static dev.vortex.api.proto.TemporalMetadatas.TIME_UNIT_NANOS;
-
-import com.google.protobuf.ByteString;
-import dev.vortex.proto.DTypeProtos;
-import java.util.Optional;
-
-/** 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
deleted file mode 100644
index 218db1e98e..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/EndianUtils.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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();
-
- // Determine target size (1, 2, 4, 8, 16, or 32 bytes)
- 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];
-
- // Copy bytes in reverse order (big endian to little endian)
- for (int i = 0; i < bigEndianBytes.length; i++) {
- result[i] = bigEndianBytes[bigEndianBytes.length - 1 - i];
- }
-
- // Sign extend if negative
- 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
deleted file mode 100644
index 108ea35976..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Expressions.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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.*;
-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);
- case "vortex.is_null":
- return IsNull.parse(metadata, children);
- case "vortex.is_not_null":
- return IsNotNull.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
deleted file mode 100644
index bc8df78b73..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/Scalars.java
+++ /dev/null
@@ -1,378 +0,0 @@
-/*
- * 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
deleted file mode 100644
index fb76419229..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/proto/TemporalMetadatas.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * 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() {}
-
- /** Time unit constant representing nanoseconds precision. */
- public static byte TIME_UNIT_NANOS = 0;
- /** Time unit constant representing microseconds precision. */
- public static byte TIME_UNIT_MICROS = 1;
- /** Time unit constant representing milliseconds precision. */
- public static byte TIME_UNIT_MILLIS = 2;
- /** Time unit constant representing seconds precision. */
- public static byte TIME_UNIT_SECONDS = 3;
- /** Time unit constant representing days precision. */
- public static byte TIME_UNIT_DAYS = 4;
-
- /** Supplier for date metadata with days precision. */
- public static final Supplier<byte[]> DATE_DAYS = Suppliers.memoize(() -> new byte[] {TIME_UNIT_DAYS});
- /** Supplier for date metadata with milliseconds precision. */
- public static final Supplier<byte[]> DATE_MILLIS = Suppliers.memoize(() -> new byte[] {TIME_UNIT_MILLIS});
- /** Supplier for time metadata with seconds precision. */
- public static final Supplier<byte[]> TIME_SECONDS = Suppliers.memoize(() -> new byte[] {TIME_UNIT_SECONDS});
- /** Supplier for time metadata with milliseconds precision. */
- public static final Supplier<byte[]> TIME_MILLIS = Suppliers.memoize(() -> new byte[] {TIME_UNIT_MILLIS});
- /** Supplier for time metadata with microseconds precision. */
- public static final Supplier<byte[]> TIME_MICROS = Suppliers.memoize(() -> new byte[] {TIME_UNIT_MICROS});
- /** Supplier for time metadata with nanoseconds precision. */
- 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);
- // Write length as little-endian uint16.
- int lenLow = timeZoneBytes.length & 0xFF;
- int lenHigh = (timeZoneBytes.length >> 8) & 0xFF;
- baos.write(lenLow);
- baos.write(lenHigh);
- baos.write(timeZoneBytes, 0, timeZoneBytes.length);
- } else {
- // write uint16 zero value
- 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/jni/JNIArray.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArray.java
deleted file mode 100644
index e28f2fefa7..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArray.java
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * 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 java.math.BigDecimal;
-import java.util.OptionalLong;
-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;
-
-/** 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) {
- // Export the dataset to Arrow over C Data Interface.
- 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
deleted file mode 100644
index c27c2d6481..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIArrayIterator.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 06e2a90664..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIDType.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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; // True if this object owns the native memory
-
- 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() {
- // Returns a borrowed reference - the parent DType owns this memory
- 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
deleted file mode 100644
index 8823e4e46a..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIFile.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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
deleted file mode 100644
index c3345a9e3e..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/JNIWriter.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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 java.io.IOException;
-import java.util.OptionalLong;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/** 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);
-
- // Write the Arrow data to Vortex through JNI
- 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 writeBatchFfi(long arrowArrayAddr, long arrowSchemaAddr) throws IOException {
- logger.trace("Writing batch via FFI (arrayAddr={}, schemaAddr={})", arrowArrayAddr, arrowSchemaAddr);
-
- boolean success = NativeWriterMethods.writeBatchFfi(ptr.getAsLong(), arrowArrayAddr, arrowSchemaAddr);
- if (!success) {
- logger.error("Failed to write FFI batch to Vortex file");
- throw new IOException("Failed to write FFI 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/NativeArrayMethods.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayMethods.java
deleted file mode 100644
index f9d693cc89..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayMethods.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 1126310498..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeDTypeMethods.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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);
-
- // Returns a list of DType pointers.
- 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/NativeDataSource.java
similarity index 56%
rename from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeFileMethods.java
rename to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeDataSource.java
index 07b4c62241..e0c93d8fc9 100644
--- 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/NativeDataSource.java
@@ -18,29 +18,23 @@
package dev.vortex.jni;
-import java.util.List;
import java.util.Map;
-/** Native JNI methods for file operations. */
-public final class NativeFileMethods {
+/** Native methods for Vortex data source operations. */
+public final class NativeDataSource {
+
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);
+ private NativeDataSource() {}
- public static native long rowCount(long pointer);
+ public static native long open(long sessionPtr, String[] uris, Map<String, String> options);
- public static native long dtype(long pointer);
+ public static native void free(long dataSourcePtr);
- public static native void close(long pointer);
+ public static native void arrowSchema(long dataSourcePtr, long arrowSchemaOutAddr);
- public static native long scan(
- long pointer, List<String> columns, byte[] predicateProto, long[] rowRange, long[] rowIndices);
+ /** Result: out[0] = row count value, out[1] = type (1=estimate, 2=exact, other=unknown). */
+ public static native void rowCount(long dataSourcePtr, long[] resultOut);
}
diff --git a/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeExpression.java b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeExpression.java
new file mode 100644
index 0000000000..a1377b7890
--- /dev/null
+++ b/paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeExpression.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;
+
+/** Native methods for Vortex expression construction. */
+public final class NativeExpression {
+
+ static {
+ NativeLoader.loadJni();
+ }
+
+ private NativeExpression() {}
+
+ public static native long root();
+
+ public static native long getItem(String path, long parentPtr);
+
+ public static native long select(String[] columns, long parentPtr);
+
+ public static native long and(long[] exprPtrs);
+
+ public static native long or(long[] exprPtrs);
+
+ public static native long binary(byte opCode, long leftPtr, long rightPtr);
+
+ public static native long not(long exprPtr);
+
+ public static native long isNull(long exprPtr);
+
+ public static native long isNotNull(long exprPtr);
+
+ public static native long like(
+ long exprPtr, long patternPtr, boolean caseSensitive, boolean negated);
+
+ public static native long between(
+ long exprPtr, long lowPtr, long highPtr, boolean lowInclusive, boolean highInclusive);
+
+ public static native long literalBool(boolean value, boolean isNull);
+
+ public static native long literalI8(byte value, boolean isNull);
+
+ public static native long literalI16(short value, boolean isNull);
+
+ public static native long literalI32(int value, boolean isNull);
+
+ public static native long literalI64(long value, boolean isNull);
+
+ public static native long literalF32(float value, boolean isNull);
+
+ public static native long literalF64(double value, boolean isNull);
+
+ public static native long literalString(String value);
+
+ public static native long literalBinary(byte[] value);
+
+ public static native long literalDecimal(
+ byte[] bigIntBytes, int precision, int scale, boolean isNull);
+
+ public static native long literalDate(long value, byte timeUnitTag, boolean isNull);
+
+ public static native long literalTimestamp(
+ long value, byte timeUnitTag, String timezone, boolean isNull);
+
+ public static native long literalNull(byte dTypeTag);
+
+ public static native void free(long exprPtr);
+}
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/NativeFiles.java
similarity index 68%
copy from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java
copy to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeFiles.java
index 08c6fc7dd2..fb2611c9d8 100644
--- 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/NativeFiles.java
@@ -18,17 +18,21 @@
package dev.vortex.jni;
-/** Native JNI methods for array iterator operations. */
-public final class NativeArrayIteratorMethods {
+import java.util.List;
+import java.util.Map;
+
+/** Native methods for Vortex file listing and deletion. */
+public final class NativeFiles {
+
static {
NativeLoader.loadJni();
}
- private NativeArrayIteratorMethods() {}
-
- public static native void free(long pointer);
+ private NativeFiles() {}
- public static native long take(long pointer);
+ private static native List<String> listFiles(
+ long sessionPtr, String uri, Map<String, String> options);
- public static native long getDType(long pointer);
+ private static native void delete(
+ long sessionPtr, String[] uris, Map<String, String> options);
}
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
index 32a9311f24..71e5dbb9ab 100644
--- 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
@@ -33,6 +33,15 @@ public final class NativeLoader {
return;
}
+ // Use Unsafe allocator for Arrow memory to guarantee buffer alignment.
+ // Vortex's Rust FFI requires buffers aligned to the scalar type's natural
+ // alignment (e.g. 16 bytes for Decimal128). Netty's pooled allocator may
+ // return sub-aligned addresses. Unsafe.allocateMemory provides at least
+ // 16-byte alignment on 64-bit systems.
+ if (System.getProperty("arrow.allocation.manager.type") == null) {
+ System.setProperty("arrow.allocation.manager.type", "Unsafe");
+ }
+
// Load the native library
String osName = System.getProperty("os.name").toLowerCase(Locale.ROOT);
String osArch = System.getProperty("os.arch").toLowerCase(Locale.ROOT);
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
index c69989295d..cd5ce3a474 100644
--- 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
@@ -27,19 +27,15 @@ public final class NativeLogging {
private NativeLogging() {}
/** Logging level constant for error messages only */
- public static final int ERROR = 0;
+ public static final int ERROR = 1;
- /** Logging level constant for warning and error messages */
- public static final int WARN = 1;
+ public static final int WARN = 2;
- /** Logging level constant for informational, warning, and error messages */
- public static final int INFO = 2;
+ public static final int INFO = 3;
- /** Logging level constant for debug, informational, warning, and error messages */
- public static final int DEBUG = 3;
+ public static final int DEBUG = 4;
- /** Logging level constant for all messages including trace-level debugging */
- public static final int TRACE = 4;
+ public static final int TRACE = 5;
public static native void initLogging(int level);
}
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/NativePartition.java
similarity index 65%
copy from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java
copy to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativePartition.java
index 08c6fc7dd2..0b54de5539 100644
--- 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/NativePartition.java
@@ -18,17 +18,20 @@
package dev.vortex.jni;
-/** Native JNI methods for array iterator operations. */
-public final class NativeArrayIteratorMethods {
+/** Native methods for Vortex partition operations. */
+public final class NativePartition {
+
static {
NativeLoader.loadJni();
}
- private NativeArrayIteratorMethods() {}
+ private NativePartition() {}
- public static native void free(long pointer);
+ public static native void free(long partitionPtr);
- public static native long take(long pointer);
+ /** Result: out[0] = row count value, out[1] = has value (0=unknown, nonzero=known). */
+ public static native void rowCount(long partitionPtr, long[] resultOut);
- public static native long getDType(long pointer);
+ public static native void scanArrow(
+ long sessionPtr, long partitionPtr, long arrowArrayStreamAddr);
}
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/NativeRuntime.java
similarity index 74%
rename from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeArrayIteratorMethods.java
rename to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeRuntime.java
index 08c6fc7dd2..e1a4e46b75 100644
--- 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/NativeRuntime.java
@@ -18,17 +18,18 @@
package dev.vortex.jni;
-/** Native JNI methods for array iterator operations. */
-public final class NativeArrayIteratorMethods {
+/** Native methods for Vortex runtime configuration. */
+public final class NativeRuntime {
+
static {
NativeLoader.loadJni();
}
- private NativeArrayIteratorMethods() {}
+ private NativeRuntime() {}
- public static native void free(long pointer);
+ public static native void setWorkerThreads(int count);
- public static native long take(long pointer);
+ public static native void setWorkerThreadsToAvailableParallelism();
- public static native long getDType(long pointer);
+ public static native int workerCount();
}
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/NativeScan.java
similarity index 53%
copy from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriterMethods.java
copy to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeScan.java
index 37a2bffc24..1ee8353986 100644
--- 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/NativeScan.java
@@ -18,22 +18,32 @@
package dev.vortex.jni;
-import java.util.Map;
-
-/** Native JNI methods for writing Vortex files. */
-public final class NativeWriterMethods {
+/** Native methods for Vortex scan operations. */
+public final class NativeScan {
static {
NativeLoader.loadJni();
}
- private NativeWriterMethods() {}
+ private NativeScan() {}
+
+ public static native long create(
+ long dataSourcePtr,
+ long projectionExprPtr,
+ long filterExprPtr,
+ long rowRangeBegin,
+ long rowRangeEnd,
+ long[] selectionIndices,
+ byte selectionModeCode,
+ long limit,
+ boolean ordered);
- public static native long create(String uri, long dtype, Map<String, String> options);
+ public static native void free(long scanPtr);
- public static native boolean writeBatch(long writerPtr, byte[] arrowData);
+ public static native void arrowSchema(long scanPtr, long arrowSchemaOutAddr);
- public static native boolean writeBatchFfi(long writerPtr, long arrowArrayAddr, long arrowSchemaAddr);
+ public static native void partitionCount(long scanPtr, long[] resultOut);
- public static native void close(long writerPtr);
+ /** Returns the next partition pointer, or 0 when exhausted. */
+ public static native long nextPartition(long scanPtr);
}
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/jni/NativeSession.java
similarity index 73%
rename from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/api/File.java
rename to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeSession.java
index beeb8f7548..bca55dac45 100644
--- 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/jni/NativeSession.java
@@ -16,16 +16,18 @@
* limitations under the License.
*/
-package dev.vortex.api;
+package dev.vortex.jni;
-/** Interface for reading Vortex format files. */
-public interface File extends AutoCloseable {
- DType getDType();
+/** Native methods for Vortex session lifecycle. */
+public final class NativeSession {
- long rowCount();
+ static {
+ NativeLoader.loadJni();
+ }
- ArrayIterator newScan(ScanOptions options);
+ private NativeSession() {}
- @Override
- void close();
+ public static native long newSession();
+
+ public static native void free(long sessionPtr);
}
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/NativeWriter.java
similarity index 70%
rename from paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriterMethods.java
rename to paimon-vortex/paimon-vortex-jni/src/main/java/dev/vortex/jni/NativeWriter.java
index 37a2bffc24..dca107b407 100644
--- 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/NativeWriter.java
@@ -20,20 +20,20 @@ package dev.vortex.jni;
import java.util.Map;
-/** Native JNI methods for writing Vortex files. */
-public final class NativeWriterMethods {
+/** Native methods for Vortex writer operations. */
+public final class NativeWriter {
static {
NativeLoader.loadJni();
}
- private NativeWriterMethods() {}
+ private NativeWriter() {}
- public static native long create(String uri, long dtype, Map<String, String> options);
+ public static native long create(
+ long sessionPtr, String uri, long arrowSchemaAddr, Map<String, String> options);
- public static native boolean writeBatch(long writerPtr, byte[] arrowData);
-
- public static native boolean writeBatchFfi(long writerPtr, long arrowArrayAddr, long arrowSchemaAddr);
+ public static native boolean writeBatch(
+ long writerPtr, long arrowArrayAddr, long arrowSchemaAddr);
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
deleted file mode 100644
index 3a735636cf..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/proto/dtype.proto
+++ /dev/null
@@ -1,101 +0,0 @@
-// 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; // This is after `Extension` for backwards compatibility.
- 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
deleted file mode 100644
index 73ba7209a1..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/proto/expr.proto
+++ /dev/null
@@ -1,97 +0,0 @@
-// 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";
-
-// Captures a generic representation of expressions in Vortex.
-// Expression deserializers can be registered with a Vortex session to handle parsing this into
-// an in-memory expression for execution.
-message Expr {
- string id = 1;
- repeated Expr children = 2;
- optional bytes metadata = 3;
-}
-
-// Captures a serialized aggregate function with its ID and options metadata.
-message AggregateFn {
- string id = 1;
- optional bytes metadata = 2;
-}
-
-// Options for `vortex.literal`
-message LiteralOpts {
- vortex.scalar.Scalar value = 1;
-}
-
-// Options for `vortex.pack`
-message PackOpts {
- repeated string paths = 1;
- bool nullable = 2;
-}
-
-// Options for `vortex.getitem`
-message GetItemOpts {
- string path = 1;
-}
-
-// Options for `vortex.binary`
-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;
- }
-}
-
-// Options for `vortex.case_when`
-// Encodes num_when_then_pairs and has_else into a single u32 (num_children).
-// num_children = num_when_then_pairs * 2 + (has_else ? 1 : 0)
-// has_else = num_children % 2 == 1
-// num_when_then_pairs = num_children / 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
deleted file mode 100644
index 251863dc3a..0000000000
--- a/paimon-vortex/paimon-vortex-jni/src/main/proto/scalar.proto
+++ /dev/null
@@ -1,39 +0,0 @@
-// 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;
- // Variant scalars carry a row-specific nested scalar.
- // See RFC 0015: https://github.com/vortex-data/rfcs/blob/develop/accepted/0015-variant-type.md
- Scalar variant_value = 11;
- }
-}
-
-message ListValue {
- repeated ScalarValue values = 1;
-}