diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java index fdaf6d261d7b..2a5fccfb5632 100644 --- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java +++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java @@ -891,6 +891,11 @@ private static boolean isNull(VectorHolder holder, int index) { private static BigIntVector allocateBigIntVector(Field field, int valueCount) { BigIntVector vector = (BigIntVector) field.createVector(ArrowAllocation.rootAllocator()); vector.allocateNew(valueCount); + ArrowBuf validityBuffer = vector.getValidityBuffer(); + for (int i = 0; i < valueCount; i += 1) { + BitVectorHelper.setBit(validityBuffer, i); + } + return vector; } diff --git a/build.gradle b/build.gradle index 182ce3f70c19..c613d793f91a 100644 --- a/build.gradle +++ b/build.gradle @@ -436,17 +436,12 @@ project(':iceberg-data') { implementation project(':iceberg-core') compileOnly project(':iceberg-parquet') compileOnly project(':iceberg-orc') - compileOnly project(':iceberg-vortex') compileOnly(libs.hadoop3.common) { exclude group: 'commons-beanutils' exclude group: 'org.apache.avro', module: 'avro' exclude group: 'org.slf4j', module: 'slf4j-log4j12' } - implementation(libs.vortex.jni) { - exclude group: 'com.google.protobuf', module: 'protobuf-java' - } - implementation("${libs.orc.core.get().module}:${libs.versions.orc.get()}:nohive") { exclude group: 'org.apache.hadoop' exclude group: 'commons-lang' @@ -473,6 +468,7 @@ project(':iceberg-data') { testImplementation project(path: ':iceberg-core', configuration: 'testArtifacts') testImplementation project(path: ':iceberg-orc', configuration: 'testArtifacts') testImplementation(testFixtures(project(':iceberg-parquet'))) + testRuntimeOnly project(':iceberg-vortex') } test { @@ -977,9 +973,11 @@ project(':iceberg-vortex') { implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow') implementation project(':iceberg-core') implementation project(':iceberg-common') - implementation(libs.vortex.jni) { - exclude group: 'com.google.protobuf', module: 'protobuf-java' + // Compile against the shaded artifact so Vortex APIs use relocated Arrow signatures. + compileOnly("${libs.vortex.jni.get().module}:${libs.vortex.jni.get().version}:all") { + transitive = false } + runtimeOnly("${libs.vortex.jni.get().module}:${libs.vortex.jni.get().version}:all") // VortexFormatModel exposes Arrow types in its public API, so Arrow must be // visible to consumers of iceberg-vortex. api(libs.arrow.vector) { @@ -992,6 +990,12 @@ project(':iceberg-vortex') { exclude group: 'io.netty', module: 'netty-common' exclude group: 'com.google.code.findbugs', module: 'jsr305' } + implementation(libs.arrow.memory.netty) { + exclude group: 'io.netty', module: 'netty-buffer' + exclude group: 'io.netty', module: 'netty-common' + exclude group: 'com.google.code.findbugs', module: 'jsr305' + } + runtimeOnly libs.netty.buffer annotationProcessor libs.immutables.value compileOnly libs.immutables.value diff --git a/core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java b/core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java index e1e93aa1fd07..b0a2ab8b8b63 100644 --- a/core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java +++ b/core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java @@ -58,6 +58,7 @@ private FormatModelRegistry() {} ImmutableList.of( "org.apache.iceberg.data.GenericFormatModels", "org.apache.iceberg.arrow.vectorized.ArrowFormatModels", + "org.apache.iceberg.vortex.VortexFormatModels", "org.apache.iceberg.flink.data.FlinkFormatModels", "org.apache.iceberg.spark.source.SparkFormatModels"); diff --git a/data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java b/data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java index efbdcb70aa4a..af7176f7a1f6 100644 --- a/data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java +++ b/data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java @@ -25,12 +25,9 @@ import org.apache.iceberg.data.orc.GenericOrcWriter; import org.apache.iceberg.data.parquet.GenericParquetReaders; import org.apache.iceberg.data.parquet.GenericParquetWriter; -import org.apache.iceberg.data.vortex.GenericVortexReader; -import org.apache.iceberg.data.vortex.GenericVortexWriter; import org.apache.iceberg.formats.FormatModelRegistry; import org.apache.iceberg.orc.ORCFormatModel; import org.apache.iceberg.parquet.ParquetFormatModel; -import org.apache.iceberg.vortex.VortexFormatModel; public class GenericFormatModels { public static void register() { @@ -65,14 +62,6 @@ public static void register() { GenericOrcReader.buildReader(icebergSchema, fileSchema, idToConstant))); FormatModelRegistry.register(ORCFormatModel.forPositionDeletes()); - - FormatModelRegistry.register( - VortexFormatModel.create( - Record.class, - Void.class, - (icebergSchema, fileSchema, engineSchema) -> - GenericVortexWriter.buildWriter(icebergSchema), - (VortexFormatModel.ReaderFunction) GenericVortexReader::buildReader)); } private GenericFormatModels() {} diff --git a/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java b/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java index 01d6bf92c210..73b21dcd2b9c 100644 --- a/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java +++ b/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java @@ -287,6 +287,9 @@ public void testPositionDeleteWriter() throws IOException { @TestTemplate public void testPositionDeleteWriterWithRow() throws IOException { + assumeThat(fileFormat) + .as("Vortex does not support position deletes") + .isNotEqualTo(FileFormat.VORTEX); FileWriterFactory writerFactory = newWriterFactory(table.schema(), table.schema()); // write a data file diff --git a/data/src/test/java/org/apache/iceberg/io/TestPartitioningWriters.java b/data/src/test/java/org/apache/iceberg/io/TestPartitioningWriters.java index bc6e2d027b4a..597d32684b50 100644 --- a/data/src/test/java/org/apache/iceberg/io/TestPartitioningWriters.java +++ b/data/src/test/java/org/apache/iceberg/io/TestPartitioningWriters.java @@ -100,6 +100,9 @@ public void testClusteredDataWriterNoRecords() throws IOException { @TestTemplate public void testClusteredDataWriterMultiplePartitions() throws IOException { + assumeThat(fileFormat) + .as("Vortex does not yet inject partition values supplied via idToConstant") + .isNotEqualTo(FileFormat.VORTEX); table.updateSpec().addField(Expressions.ref("data")).commit(); FileWriterFactory writerFactory = newWriterFactory(table.schema()); @@ -549,6 +552,9 @@ public void testFanoutDataWriterNoRecords() throws IOException { @TestTemplate public void testFanoutDataWriterMultiplePartitions() throws IOException { + assumeThat(fileFormat) + .as("Vortex does not yet inject partition values supplied via idToConstant") + .isNotEqualTo(FileFormat.VORTEX); table.updateSpec().addField(Expressions.ref("data")).commit(); FileWriterFactory writerFactory = newWriterFactory(table.schema()); diff --git a/data/src/test/java/org/apache/iceberg/io/TestRollingFileWriters.java b/data/src/test/java/org/apache/iceberg/io/TestRollingFileWriters.java index 706635d3597b..7af32ded9b64 100644 --- a/data/src/test/java/org/apache/iceberg/io/TestRollingFileWriters.java +++ b/data/src/test/java/org/apache/iceberg/io/TestRollingFileWriters.java @@ -105,6 +105,9 @@ public void testRollingDataWriterNoRecords() throws IOException { @TestTemplate public void testRollingDataWriterSplitData() throws IOException { + assumeThat(fileFormat) + .as("VortexFileAppender.length() returns 0 until the appender is closed") + .isNotEqualTo(FileFormat.VORTEX); FileWriterFactory writerFactory = newWriterFactory(table.schema()); RollingDataWriter writer = new RollingDataWriter<>( @@ -148,6 +151,9 @@ public void testRollingEqualityDeleteWriterNoRecords() throws IOException { @TestTemplate public void testRollingEqualityDeleteWriterSplitDeletes() throws IOException { + assumeThat(fileFormat) + .as("VortexFileAppender.length() returns 0 until the appender is closed") + .isNotEqualTo(FileFormat.VORTEX); List equalityFieldIds = ImmutableList.of(table.schema().findField("id").fieldId()); Schema equalityDeleteRowSchema = table.schema().select("id"); FileWriterFactory writerFactory = diff --git a/flink/v1.20/flink-runtime/runtime-deps.txt b/flink/v1.20/flink-runtime/runtime-deps.txt index 75fcd8ec2fa0..ebf8a55b48ff 100644 --- a/flink/v1.20/flink-runtime/runtime-deps.txt +++ b/flink/v1.20/flink-runtime/runtime-deps.txt @@ -1,26 +1,11 @@ com.fasterxml.jackson.core:jackson-annotations:2.21 com.fasterxml.jackson.core:jackson-core:2.21 com.fasterxml.jackson.core:jackson-databind:2.21 -com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.21 com.github.ben-manes.caffeine:caffeine:2.9 com.github.luben:zstd-jni:1.5 -com.google.errorprone:error_prone_annotations:2.47 -com.google.flatbuffers:flatbuffers-java:25.2 -com.google.guava:failureaccess:1.0 -com.google.guava:guava:33.6 -com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava -com.google.j2objc:j2objc-annotations:3.1 +com.google.errorprone:error_prone_annotations:2.10 dev.failsafe:failsafe:3.3 -dev.vortex:vortex-jni:0.72 io.airlift:aircompressor:2.0 -io.netty:netty-buffer:4.2 -io.netty:netty-common:4.2 -org.apache.arrow:arrow-c-data:19.0 -org.apache.arrow:arrow-format:19.0 -org.apache.arrow:arrow-memory-core:19.0 -org.apache.arrow:arrow-memory-netty-buffer-patch:19.0 -org.apache.arrow:arrow-memory-netty:19.0 -org.apache.arrow:arrow-vector:19.0 org.apache.avro:avro:1.12 org.apache.datasketches:datasketches-java:6.2 org.apache.datasketches:datasketches-memory:3.0 @@ -39,7 +24,6 @@ org.apache.parquet:parquet-jackson:1.17 org.apache.parquet:parquet-variant:1.17 org.checkerframework:checker-qual:3.19 org.eclipse.microprofile.openapi:microprofile-openapi-api:4.1 -org.jspecify:jspecify:1.0 org.locationtech.jts:jts-core:1.20 org.projectnessie.nessie:nessie-client:0.107 org.projectnessie.nessie:nessie-model:0.107 diff --git a/flink/v2.0/flink-runtime/runtime-deps.txt b/flink/v2.0/flink-runtime/runtime-deps.txt index 75fcd8ec2fa0..ebf8a55b48ff 100644 --- a/flink/v2.0/flink-runtime/runtime-deps.txt +++ b/flink/v2.0/flink-runtime/runtime-deps.txt @@ -1,26 +1,11 @@ com.fasterxml.jackson.core:jackson-annotations:2.21 com.fasterxml.jackson.core:jackson-core:2.21 com.fasterxml.jackson.core:jackson-databind:2.21 -com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.21 com.github.ben-manes.caffeine:caffeine:2.9 com.github.luben:zstd-jni:1.5 -com.google.errorprone:error_prone_annotations:2.47 -com.google.flatbuffers:flatbuffers-java:25.2 -com.google.guava:failureaccess:1.0 -com.google.guava:guava:33.6 -com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava -com.google.j2objc:j2objc-annotations:3.1 +com.google.errorprone:error_prone_annotations:2.10 dev.failsafe:failsafe:3.3 -dev.vortex:vortex-jni:0.72 io.airlift:aircompressor:2.0 -io.netty:netty-buffer:4.2 -io.netty:netty-common:4.2 -org.apache.arrow:arrow-c-data:19.0 -org.apache.arrow:arrow-format:19.0 -org.apache.arrow:arrow-memory-core:19.0 -org.apache.arrow:arrow-memory-netty-buffer-patch:19.0 -org.apache.arrow:arrow-memory-netty:19.0 -org.apache.arrow:arrow-vector:19.0 org.apache.avro:avro:1.12 org.apache.datasketches:datasketches-java:6.2 org.apache.datasketches:datasketches-memory:3.0 @@ -39,7 +24,6 @@ org.apache.parquet:parquet-jackson:1.17 org.apache.parquet:parquet-variant:1.17 org.checkerframework:checker-qual:3.19 org.eclipse.microprofile.openapi:microprofile-openapi-api:4.1 -org.jspecify:jspecify:1.0 org.locationtech.jts:jts-core:1.20 org.projectnessie.nessie:nessie-client:0.107 org.projectnessie.nessie:nessie-model:0.107 diff --git a/flink/v2.1/flink-runtime/runtime-deps.txt b/flink/v2.1/flink-runtime/runtime-deps.txt index 75fcd8ec2fa0..ebf8a55b48ff 100644 --- a/flink/v2.1/flink-runtime/runtime-deps.txt +++ b/flink/v2.1/flink-runtime/runtime-deps.txt @@ -1,26 +1,11 @@ com.fasterxml.jackson.core:jackson-annotations:2.21 com.fasterxml.jackson.core:jackson-core:2.21 com.fasterxml.jackson.core:jackson-databind:2.21 -com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.21 com.github.ben-manes.caffeine:caffeine:2.9 com.github.luben:zstd-jni:1.5 -com.google.errorprone:error_prone_annotations:2.47 -com.google.flatbuffers:flatbuffers-java:25.2 -com.google.guava:failureaccess:1.0 -com.google.guava:guava:33.6 -com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava -com.google.j2objc:j2objc-annotations:3.1 +com.google.errorprone:error_prone_annotations:2.10 dev.failsafe:failsafe:3.3 -dev.vortex:vortex-jni:0.72 io.airlift:aircompressor:2.0 -io.netty:netty-buffer:4.2 -io.netty:netty-common:4.2 -org.apache.arrow:arrow-c-data:19.0 -org.apache.arrow:arrow-format:19.0 -org.apache.arrow:arrow-memory-core:19.0 -org.apache.arrow:arrow-memory-netty-buffer-patch:19.0 -org.apache.arrow:arrow-memory-netty:19.0 -org.apache.arrow:arrow-vector:19.0 org.apache.avro:avro:1.12 org.apache.datasketches:datasketches-java:6.2 org.apache.datasketches:datasketches-memory:3.0 @@ -39,7 +24,6 @@ org.apache.parquet:parquet-jackson:1.17 org.apache.parquet:parquet-variant:1.17 org.checkerframework:checker-qual:3.19 org.eclipse.microprofile.openapi:microprofile-openapi-api:4.1 -org.jspecify:jspecify:1.0 org.locationtech.jts:jts-core:1.20 org.projectnessie.nessie:nessie-client:0.107 org.projectnessie.nessie:nessie-model:0.107 diff --git a/kafka-connect/kafka-connect-runtime/runtime-deps.txt b/kafka-connect/kafka-connect-runtime/runtime-deps.txt index b200a94028d4..95ef88a87969 100644 --- a/kafka-connect/kafka-connect-runtime/runtime-deps.txt +++ b/kafka-connect/kafka-connect-runtime/runtime-deps.txt @@ -54,9 +54,9 @@ com.google.cloud:google-cloud-storage:2.68 com.google.code.findbugs:jsr305:3.0 com.google.code.gson:gson:2.13 com.google.errorprone:error_prone_annotations:2.48 -com.google.flatbuffers:flatbuffers-java:25.2 +com.google.flatbuffers:flatbuffers-java:24.3 com.google.guava:failureaccess:1.0 -com.google.guava:guava:33.6 +com.google.guava:guava:33.5 com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava com.google.http-client:google-http-client-apache-v2:2.1 com.google.http-client:google-http-client-appengine:2.1 @@ -73,13 +73,12 @@ com.microsoft.azure:msal4j-persistence-extension:1.3 com.microsoft.azure:msal4j:1.23 com.sun.xml.bind:jaxb-impl:2.2 commons-cli:commons-cli:1.9 -commons-codec:commons-codec:1.21 +commons-codec:commons-codec:1.19 commons-io:commons-io:2.20 commons-logging:commons-logging:1.2 commons-net:commons-net:3.9 commons-pool:commons-pool:1.6 dev.failsafe:failsafe:3.3 -dev.vortex:vortex-jni:0.72 dnsjava:dnsjava:3.6 io.airlift:aircompressor:2.0 io.dropwizard.metrics:metrics-core:3.2 @@ -150,12 +149,11 @@ javax.xml.bind:jaxb-api:2.2 javax.xml.stream:stax-api:1.0-2 net.java.dev.jna:jna-platform:5.17 net.java.dev.jna:jna:5.17 -org.apache.arrow:arrow-c-data:19.0 -org.apache.arrow:arrow-format:19.0 -org.apache.arrow:arrow-memory-core:19.0 -org.apache.arrow:arrow-memory-netty-buffer-patch:19.0 -org.apache.arrow:arrow-memory-netty:19.0 -org.apache.arrow:arrow-vector:19.0 +org.apache.arrow:arrow-format:17.0 +org.apache.arrow:arrow-memory-core:17.0 +org.apache.arrow:arrow-memory-netty-buffer-patch:17.0 +org.apache.arrow:arrow-memory-netty:17.0 +org.apache.arrow:arrow-vector:17.0 org.apache.avro:avro:1.12 org.apache.commons:commons-collections4:4.4 org.apache.commons:commons-compress:1.28 diff --git a/spark/v3.5/spark-runtime/baseline-class-uniqueness.lock b/spark/v3.5/spark-runtime/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v3.5/spark-runtime/baseline-class-uniqueness.lock +++ b/spark/v3.5/spark-runtime/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/spark/v3.5/spark/baseline-class-uniqueness.lock b/spark/v3.5/spark/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v3.5/spark/baseline-class-uniqueness.lock +++ b/spark/v3.5/spark/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/spark/v4.0/spark-runtime/baseline-class-uniqueness.lock b/spark/v4.0/spark-runtime/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v4.0/spark-runtime/baseline-class-uniqueness.lock +++ b/spark/v4.0/spark-runtime/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/spark/v4.0/spark/baseline-class-uniqueness.lock b/spark/v4.0/spark/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v4.0/spark/baseline-class-uniqueness.lock +++ b/spark/v4.0/spark/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/spark/v4.1/spark-runtime/baseline-class-uniqueness.lock b/spark/v4.1/spark-runtime/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v4.1/spark-runtime/baseline-class-uniqueness.lock +++ b/spark/v4.1/spark-runtime/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/spark/v4.1/spark/baseline-class-uniqueness.lock b/spark/v4.1/spark/baseline-class-uniqueness.lock index 53a154c69d97..e07065be3f77 100644 --- a/spark/v4.1/spark/baseline-class-uniqueness.lock +++ b/spark/v4.1/spark/baseline-class-uniqueness.lock @@ -15,13 +15,3 @@ - io.netty.buffer.NettyArrowBuf - io.netty.buffer.PooledByteBufAllocatorL - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator -[dev.vortex:vortex-jni, dev.vortex:vortex-jni (classifier=all)] - - dev.vortex.api.DataSource - - dev.vortex.api.Expression - - dev.vortex.api.ImmutableScanOptions - - dev.vortex.api.Partition - - dev.vortex.api.Scan - - dev.vortex.api.Session - - dev.vortex.api.VortexWriter - - dev.vortex.arrow.ArrowAllocation - - dev.vortex.jni.NativeLoader diff --git a/vortex/baseline-class-uniqueness.lock b/vortex/baseline-class-uniqueness.lock new file mode 100644 index 000000000000..e07065be3f77 --- /dev/null +++ b/vortex/baseline-class-uniqueness.lock @@ -0,0 +1,17 @@ +# Danger! Multiple jars contain identically named classes. This may cause different behaviour depending on classpath ordering. +# Run ./gradlew checkClassUniqueness --fix to update this file + +## runtimeClasspath +[com.google.guava:guava, dev.vortex:vortex-jni (classifier=all)] + - com.google.thirdparty.publicsuffix.PublicSuffixPatterns + - com.google.thirdparty.publicsuffix.PublicSuffixTrie + - com.google.thirdparty.publicsuffix.PublicSuffixTrie$ChunksCharSequence + - com.google.thirdparty.publicsuffix.PublicSuffixType +[dev.vortex:vortex-jni (classifier=all), org.apache.arrow:arrow-c-data] + - org.apache.arrow.c.ArrayStreamExporter$ExportedArrayStreamPrivateData + - org.apache.arrow.c.jni.JniWrapper +[dev.vortex:vortex-jni (classifier=all), org.apache.arrow:arrow-memory-netty-buffer-patch] + - io.netty.buffer.ExpandableByteBuf + - io.netty.buffer.NettyArrowBuf + - io.netty.buffer.PooledByteBufAllocatorL + - io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexArrowBridge.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexArrowBridge.java new file mode 100644 index 000000000000..53bc9cb592b0 --- /dev/null +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexArrowBridge.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.vortex; + +import dev.vortex.arrow.ArrowAllocation; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.VectorSchemaRoot; + +final class VortexArrowBridge { + private static final RootAllocator ARROW_ALLOCATOR = new RootAllocator(Long.MAX_VALUE); + + private VortexArrowBridge() {} + + static BufferAllocator arrowAllocator() { + return ARROW_ALLOCATOR; + } + + static dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator() { + return ArrowAllocation.rootAllocator(); + } + + static VectorSchemaRoot importVortexRoot( + dev.vortex.relocated.org.apache.arrow.vector.VectorSchemaRoot vortexRoot, + dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator, + BufferAllocator arrowAllocator) { + try (dev.vortex.relocated.org.apache.arrow.c.ArrowArray vortexArray = + dev.vortex.relocated.org.apache.arrow.c.ArrowArray.allocateNew(vortexAllocator); + dev.vortex.relocated.org.apache.arrow.c.ArrowSchema vortexSchema = + dev.vortex.relocated.org.apache.arrow.c.ArrowSchema.allocateNew(vortexAllocator)) { + dev.vortex.relocated.org.apache.arrow.c.Data.exportVectorSchemaRoot( + vortexAllocator, vortexRoot, null, vortexArray, vortexSchema); + + try (ArrowArray arrowArray = ArrowArray.wrap(vortexArray.memoryAddress())) { + ArrowSchema arrowSchema = ArrowSchema.wrap(vortexSchema.memoryAddress()); + return Data.importVectorSchemaRoot(arrowAllocator, arrowArray, arrowSchema, null); + } + } + } +} diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexFileAppender.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFileAppender.java index 1b92026375f1..8cdc4a6f07c2 100644 --- a/vortex/src/main/java/org/apache/iceberg/vortex/VortexFileAppender.java +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFileAppender.java @@ -19,7 +19,6 @@ package org.apache.iceberg.vortex; import dev.vortex.api.VortexWriter; -import dev.vortex.arrow.ArrowAllocation; import java.io.IOException; import org.apache.arrow.c.ArrowArray; import org.apache.arrow.c.ArrowSchema; @@ -66,11 +65,7 @@ class VortexFileAppender implements FileAppender { MetricsConfig metricsConfig) { this.writer = writer; this.valueWriter = valueWriter; - // Use Vortex's shared root allocator: VortexWriter performs allocations against this - // and the native side manages lifetime via Cleaner references. Owning our own - // RootAllocator and closing it eagerly trips strict leak checks when Vortex retains - // small per-writer state. - this.allocator = allocator != null ? allocator : ArrowAllocation.rootAllocator(); + this.allocator = allocator != null ? allocator : VortexArrowBridge.arrowAllocator(); this.root = VectorSchemaRoot.create(arrowSchema, this.allocator); this.batchSize = batchSize; this.outputFile = outputFile; @@ -135,8 +130,7 @@ public void close() throws IOException { writer.close(); } finally { root.close(); - // Don't close `allocator`: it is Vortex's shared root allocator and is managed for - // the lifetime of the process. + // Don't close `allocator`: it is shared for the lifetime of the process. closed = true; } } diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModel.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModel.java index 3638367f0456..58c3f5b94fb5 100644 --- a/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModel.java +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModel.java @@ -20,7 +20,6 @@ import dev.vortex.api.Session; import dev.vortex.api.VortexWriter; -import dev.vortex.arrow.ArrowAllocation; import dev.vortex.jni.NativeRuntime; import java.io.IOException; import java.nio.ByteBuffer; @@ -221,6 +220,8 @@ public FileAppender build() throws IOException { private FileAppender buildAppender(org.apache.iceberg.Schema writeSchema) throws IOException { Schema arrowSchema = VortexSchemas.toArrowSchema(writeSchema); + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema vortexSchema = + VortexSchemas.toVortexArrowSchema(writeSchema); VortexValueWriter valueWriter = (VortexValueWriter) writerFunction.write(writeSchema, arrowSchema, engineSchema); @@ -234,10 +235,12 @@ private FileAppender buildAppender(org.apache.iceberg.Schema writeSchema) // Apply worker-thread setting on this executor JVM before any Vortex native work begins. NativeRuntime.setWorkerThreads(workerThreads); - BufferAllocator allocator = ArrowAllocation.rootAllocator(); + BufferAllocator allocator = VortexArrowBridge.arrowAllocator(); + dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator = + VortexArrowBridge.vortexAllocator(); Session session = Session.create(); VortexWriter vortexWriter = - VortexWriter.create(session, uri, arrowSchema, properties, allocator); + VortexWriter.create(session, uri, vortexSchema, properties, vortexAllocator); return new VortexFileAppender<>( vortexWriter, diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModels.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModels.java new file mode 100644 index 000000000000..8d7f21309ba2 --- /dev/null +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexFormatModels.java @@ -0,0 +1,38 @@ +/* + * 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.iceberg.vortex; + +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.vortex.GenericVortexReader; +import org.apache.iceberg.data.vortex.GenericVortexWriter; +import org.apache.iceberg.formats.FormatModelRegistry; + +public class VortexFormatModels { + public static void register() { + FormatModelRegistry.register( + VortexFormatModel.create( + Record.class, + Void.class, + (icebergSchema, fileSchema, engineSchema) -> + GenericVortexWriter.buildWriter(icebergSchema), + (VortexFormatModel.ReaderFunction) GenericVortexReader::buildReader)); + } + + private VortexFormatModels() {} +} diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexIterable.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexIterable.java index 16c73ededd41..d0d680b983b0 100644 --- a/vortex/src/main/java/org/apache/iceberg/vortex/VortexIterable.java +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexIterable.java @@ -24,7 +24,6 @@ import dev.vortex.api.Scan; import dev.vortex.api.ScanOptions; import dev.vortex.api.Session; -import dev.vortex.arrow.ArrowAllocation; import dev.vortex.jni.NativeRuntime; import java.io.IOException; import java.util.List; @@ -34,7 +33,6 @@ import java.util.function.Function; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.vector.VectorSchemaRoot; -import org.apache.arrow.vector.ipc.ArrowReader; import org.apache.iceberg.Schema; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.io.CloseableGroup; @@ -90,13 +88,18 @@ public CloseableIterator iterator() { Map properties = VortexFileUtil.resolveInputProperties(inputFile); DataSource dataSource = DataSource.open(session, uri, properties); - BufferAllocator allocator = ArrowAllocation.rootAllocator(); - org.apache.arrow.vector.types.pojo.Schema fileArrowSchema = dataSource.arrowSchema(allocator); + dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator = + VortexArrowBridge.vortexAllocator(); + BufferAllocator allocator = VortexArrowBridge.arrowAllocator(); + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema vortexArrowSchema = + dataSource.arrowSchema(vortexAllocator); + org.apache.arrow.vector.types.pojo.Schema fileArrowSchema = + VortexSchemas.toArrowSchema(vortexArrowSchema); Optional scanFilter = filterPredicate.map( icebergExpression -> { - Schema icebergFileSchema = VortexSchemas.convert(fileArrowSchema); + Schema icebergFileSchema = VortexSchemas.convert(vortexArrowSchema); return ConvertFilterToVortex.convert(icebergFileSchema, icebergExpression); }); @@ -113,7 +116,8 @@ public CloseableIterator iterator() { Scan scan = dataSource.scan(optionsBuilder.build()); Preconditions.checkNotNull(scan, "scan"); - PartitionBatchIterator batchIterator = new PartitionBatchIterator(scan, allocator); + PartitionBatchIterator batchIterator = + new PartitionBatchIterator(scan, vortexAllocator, allocator); if (rowReaderFunc != null) { VortexRowReader rowFunction = rowReaderFunc.apply(fileArrowSchema); @@ -127,14 +131,19 @@ public CloseableIterator iterator() { /** Iterator that pulls Arrow {@link VectorSchemaRoot} batches across Vortex partitions. */ static class PartitionBatchIterator implements CloseableIterator { private final Scan scan; + private final dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator; private final BufferAllocator allocator; - private ArrowReader currentReader; + private dev.vortex.relocated.org.apache.arrow.vector.ipc.ArrowReader currentReader; private VectorSchemaRoot currentRoot; private boolean hasPending = false; private boolean exhausted = false; - PartitionBatchIterator(Scan scan, BufferAllocator allocator) { + PartitionBatchIterator( + Scan scan, + dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator vortexAllocator, + BufferAllocator allocator) { this.scan = scan; + this.vortexAllocator = vortexAllocator; this.allocator = allocator; } @@ -147,6 +156,7 @@ public boolean hasNext() { return false; } try { + closeCurrentRoot(); while (true) { if (currentReader == null) { if (!scan.hasNext()) { @@ -154,10 +164,12 @@ public boolean hasNext() { return false; } Partition partition = scan.next(); - currentReader = partition.scanArrow(allocator); - currentRoot = currentReader.getVectorSchemaRoot(); + currentReader = partition.scanArrow(vortexAllocator); } if (currentReader.loadNextBatch()) { + currentRoot = + VortexArrowBridge.importVortexRoot( + currentReader.getVectorSchemaRoot(), vortexAllocator, allocator); hasPending = true; return true; } else { @@ -183,11 +195,19 @@ public VectorSchemaRoot next() { @Override public void close() throws IOException { + closeCurrentRoot(); if (currentReader != null) { currentReader.close(); currentReader = null; } - // Don't close `allocator`: it is Vortex's shared root allocator. + // Don't close shared allocators; they are managed for the lifetime of the process. + } + + private void closeCurrentRoot() { + if (currentRoot != null) { + currentRoot.close(); + currentRoot = null; + } } } diff --git a/vortex/src/main/java/org/apache/iceberg/vortex/VortexSchemas.java b/vortex/src/main/java/org/apache/iceberg/vortex/VortexSchemas.java index 386230479903..ba0f5d43c4ee 100644 --- a/vortex/src/main/java/org/apache/iceberg/vortex/VortexSchemas.java +++ b/vortex/src/main/java/org/apache/iceberg/vortex/VortexSchemas.java @@ -59,7 +59,26 @@ public static Schema convert(org.apache.arrow.vector.types.pojo.Schema arrowSche return new Schema(columns); } - /** Convert an Iceberg Schema to an Arrow Schema suitable for {@code VortexWriter.create}. */ + /** Convert a Vortex file's relocated Arrow schema to Iceberg. */ + public static Schema convert( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema arrowSchema) { + List fields = + arrowSchema.getFields(); + List columns = Lists.newArrayListWithExpectedSize(fields.size()); + for (int fieldId = 0; fieldId < fields.size(); fieldId++) { + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field = fields.get(fieldId); + Type icebergType = toIcebergType(field); + if (field.isNullable()) { + columns.add(optional(fieldId, field.getName(), icebergType)); + } else { + columns.add(required(fieldId, field.getName(), icebergType)); + } + } + + return new Schema(columns); + } + + /** Convert an Iceberg Schema to an Arrow Schema suitable for local Arrow vectors. */ public static org.apache.arrow.vector.types.pojo.Schema toArrowSchema(Schema icebergSchema) { ImmutableList.Builder fields = ImmutableList.builder(); for (Types.NestedField column : icebergSchema.columns()) { @@ -69,6 +88,33 @@ public static org.apache.arrow.vector.types.pojo.Schema toArrowSchema(Schema ice return new org.apache.arrow.vector.types.pojo.Schema(fields.build()); } + /** + * Convert a relocated Vortex Arrow schema to an Arrow Schema suitable for local Arrow vectors. + */ + public static org.apache.arrow.vector.types.pojo.Schema toArrowSchema( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema vortexSchema) { + ImmutableList.Builder fields = ImmutableList.builder(); + for (dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field : + vortexSchema.getFields()) { + fields.add(toArrowField(field)); + } + + return new org.apache.arrow.vector.types.pojo.Schema( + fields.build(), vortexSchema.getCustomMetadata()); + } + + /** Convert an Iceberg Schema to a relocated Arrow Schema suitable for {@code VortexWriter}. */ + public static dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema toVortexArrowSchema( + Schema icebergSchema) { + ImmutableList.Builder fields = + ImmutableList.builder(); + for (Types.NestedField column : icebergSchema.columns()) { + fields.add(toVortexArrowField(column.name(), column.type(), column.isOptional())); + } + + return new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema(fields.build()); + } + private static Field toArrowField(String name, Type type, boolean nullable) { return switch (type.typeId()) { case BOOLEAN -> new Field(name, new FieldType(nullable, ArrowType.Bool.INSTANCE, null), null); @@ -175,14 +221,260 @@ yield new Field( }; } + private static Field toArrowField( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field) { + List children = + field.getChildren(); + List arrowChildren = null; + if (!children.isEmpty()) { + ImmutableList.Builder arrowChildBuilder = ImmutableList.builder(); + for (dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field child : children) { + arrowChildBuilder.add(toArrowField(child)); + } + + arrowChildren = arrowChildBuilder.build(); + } + + return new Field( + field.getName(), + new FieldType(field.isNullable(), toArrowType(field.getType()), null, field.getMetadata()), + arrowChildren); + } + + @SuppressWarnings("checkstyle:CyclomaticComplexity") + private static ArrowType toArrowType( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType arrowType) { + if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Null) { + return ArrowType.Null.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Struct) { + return ArrowType.Struct.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.List) { + return ArrowType.List.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.LargeList) { + return ArrowType.LargeList.INSTANCE; + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeList list) { + return new ArrowType.FixedSizeList(list.getListSize()); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Bool) { + return ArrowType.Bool.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Int intType) { + return new ArrowType.Int(intType.getBitWidth(), intType.getIsSigned()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint fpType) { + return new ArrowType.FloatingPoint( + FloatingPointPrecision.valueOf(fpType.getPrecision().name())); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Utf8) { + return ArrowType.Utf8.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.LargeUtf8) { + return ArrowType.LargeUtf8.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Binary) { + return ArrowType.Binary.INSTANCE; + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.LargeBinary) { + return ArrowType.LargeBinary.INSTANCE; + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary fixed) { + return new ArrowType.FixedSizeBinary(fixed.getByteWidth()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Decimal decimal) { + return new ArrowType.Decimal( + decimal.getPrecision(), decimal.getScale(), decimal.getBitWidth()); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Date date) { + return new ArrowType.Date(DateUnit.valueOf(date.getUnit().name())); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Time time) { + return new ArrowType.Time(TimeUnit.valueOf(time.getUnit().name()), time.getBitWidth()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Timestamp timestamp) { + return new ArrowType.Timestamp( + TimeUnit.valueOf(timestamp.getUnit().name()), timestamp.getTimezone()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.ExtensionType ext) { + return toArrowType(ext.storageType()); + } + + throw new UnsupportedOperationException( + "Unsupported Vortex Arrow type for Arrow conversion: " + arrowType); + } + + private static dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field toVortexArrowField( + String name, Type type, boolean nullable) { + return switch (type.typeId()) { + case BOOLEAN -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Bool(), + nullable); + case INTEGER -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Int( + Integer.SIZE, true), + nullable); + case LONG -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Int( + Long.SIZE, true), + nullable); + case FLOAT -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint( + dev.vortex.relocated.org.apache.arrow.vector.types.FloatingPointPrecision.SINGLE), + nullable); + case DOUBLE -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint( + dev.vortex.relocated.org.apache.arrow.vector.types.FloatingPointPrecision.DOUBLE), + nullable); + case STRING -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Utf8(), + nullable); + case BINARY -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Binary(), + nullable); + case FIXED -> { + Types.FixedType fixedType = (Types.FixedType) type; + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary( + fixedType.length()), + nullable); + } + case DECIMAL -> { + Types.DecimalType decimalType = (Types.DecimalType) type; + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Decimal( + decimalType.precision(), decimalType.scale(), 128), + nullable); + } + case UUID -> { + Map extMetadata = + ImmutableMap.of( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.ExtensionType + .EXTENSION_METADATA_KEY_NAME, + UUID_EXTENSION_NAME, + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.ExtensionType + .EXTENSION_METADATA_KEY_METADATA, + ""); + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary( + 16), + nullable, + extMetadata, + null); + } + case DATE -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Date( + dev.vortex.relocated.org.apache.arrow.vector.types.DateUnit.DAY), + nullable); + case TIME -> + toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Time( + dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit.MICROSECOND, + Long.SIZE), + nullable); + case TIMESTAMP -> { + Types.TimestampType tsType = (Types.TimestampType) type; + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Timestamp( + dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit.MICROSECOND, + tsType.shouldAdjustToUTC() ? "UTC" : null), + nullable); + } + case TIMESTAMP_NANO -> { + Types.TimestampNanoType tsNanoType = (Types.TimestampNanoType) type; + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Timestamp( + dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit.NANOSECOND, + tsNanoType.shouldAdjustToUTC() ? "UTC" : null), + nullable); + } + case LIST -> { + Types.ListType listType = (Types.ListType) type; + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field elementField = + toVortexArrowField("element", listType.elementType(), listType.isElementOptional()); + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.List(), + nullable, + null, + ImmutableList.of(elementField)); + } + case STRUCT -> { + Types.StructType structType = (Types.StructType) type; + ImmutableList.Builder + children = ImmutableList.builder(); + for (Types.NestedField field : structType.fields()) { + children.add(toVortexArrowField(field.name(), field.type(), field.isOptional())); + } + + yield toVortexArrowField( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Struct(), + nullable, + null, + children.build()); + } + default -> + throw new UnsupportedOperationException( + "Unsupported Iceberg type for Arrow conversion: " + type); + }; + } + + private static dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field toVortexArrowField( + String name, + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType arrowType, + boolean nullable) { + return toVortexArrowField(name, arrowType, nullable, null, null); + } + + private static dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field toVortexArrowField( + String name, + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType arrowType, + boolean nullable, + Map metadata, + List children) { + return new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field( + name, + new dev.vortex.relocated.org.apache.arrow.vector.types.pojo.FieldType( + nullable, arrowType, null, metadata), + children); + } + private static Type toIcebergType(Field field) { // UUID is conveyed as the {@code arrow.uuid} extension over - // FixedSizeBinary(16). Check the - // metadata directly so this works whether or not the extension is registered - // with - // ExtensionTypeRegistry (i.e. whether arrowType deserialized to ExtensionType - // or stayed as - // FixedSizeBinary). + // FixedSizeBinary(16). Check metadata directly so this works whether or not + // the extension is registered with ExtensionTypeRegistry. if (isUuidField(field)) { return Types.UUIDType.get(); } @@ -203,6 +495,38 @@ private static Type toIcebergType(Field field) { return toIcebergSimpleType(arrowType); } + private static Type toIcebergType( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field) { + if (isUuidField(field)) { + return Types.UUIDType.get(); + } + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType arrowType = field.getType(); + if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Int intType) { + return intType.getBitWidth() <= Integer.SIZE ? Types.IntegerType.get() : Types.LongType.get(); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint fpType) { + return toIcebergFloatingPoint(fpType); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Decimal decType) { + return Types.DecimalType.of(decType.getPrecision(), decType.getScale()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary fixed) { + return Types.FixedType.ofLength(fixed.getByteWidth()); + } else if (arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Timestamp tsType) { + return toIcebergTimestamp(tsType); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.List) { + return toIcebergList(field); + } + return toIcebergSimpleType(arrowType); + } + private static Type toIcebergSimpleType(ArrowType arrowType) { if (arrowType instanceof ArrowType.Null) { return Types.UnknownType.get(); @@ -221,6 +545,36 @@ private static Type toIcebergSimpleType(ArrowType arrowType) { throw new UnsupportedOperationException("Unsupported Arrow type: " + arrowType); } + private static Type toIcebergSimpleType( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType arrowType) { + if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Null) { + return Types.UnknownType.get(); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Bool) { + return Types.BooleanType.get(); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Utf8 + || arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.LargeUtf8) { + return Types.StringType.get(); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Binary + || arrowType + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.LargeBinary) { + return Types.BinaryType.get(); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Date) { + return Types.DateType.get(); + } else if (arrowType + instanceof dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Time) { + return Types.TimeType.get(); + } + throw new UnsupportedOperationException("Unsupported Arrow type: " + arrowType); + } + private static Type toIcebergFloatingPoint(ArrowType.FloatingPoint fpType) { return switch (fpType.getPrecision()) { case SINGLE -> Types.FloatType.get(); @@ -230,6 +584,16 @@ private static Type toIcebergFloatingPoint(ArrowType.FloatingPoint fpType) { }; } + private static Type toIcebergFloatingPoint( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint fpType) { + return switch (fpType.getPrecision()) { + case SINGLE -> Types.FloatType.get(); + case DOUBLE -> Types.DoubleType.get(); + case HALF -> + throw new UnsupportedOperationException("Half-precision floats are not supported"); + }; + } + private static Type toIcebergTimestamp(ArrowType.Timestamp tsType) { boolean isNano = tsType.getUnit() == TimeUnit.NANOSECOND; if (tsType.getTimezone() == null) { @@ -238,6 +602,16 @@ private static Type toIcebergTimestamp(ArrowType.Timestamp tsType) { return isNano ? Types.TimestampNanoType.withZone() : Types.TimestampType.withZone(); } + private static Type toIcebergTimestamp( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.Timestamp tsType) { + boolean isNano = + tsType.getUnit() == dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit.NANOSECOND; + if (tsType.getTimezone() == null) { + return isNano ? Types.TimestampNanoType.withoutZone() : Types.TimestampType.withoutZone(); + } + return isNano ? Types.TimestampNanoType.withZone() : Types.TimestampType.withZone(); + } + private static Type toIcebergList(Field field) { Field elementField = field.getChildren().get(0); Type innerType = toIcebergType(elementField); @@ -246,6 +620,16 @@ private static Type toIcebergList(Field field) { : Types.ListType.ofRequired(0, innerType); } + private static Type toIcebergList( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field) { + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field elementField = + field.getChildren().get(0); + Type innerType = toIcebergType(elementField); + return elementField.isNullable() + ? Types.ListType.ofOptional(0, innerType) + : Types.ListType.ofRequired(0, innerType); + } + /** * True when {@code field} carries the {@code arrow.uuid} extension marker. Checking the field * metadata works whether or not {@link ArrowType.ExtensionType} was deserialized by the registry. @@ -257,4 +641,19 @@ public static boolean isUuidField(Field field) { return UUID_EXTENSION_NAME.equals( field.getMetadata().get(ArrowType.ExtensionType.EXTENSION_METADATA_KEY_NAME)); } + + public static boolean isUuidField( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field field) { + if (field.getType() + instanceof + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.ExtensionType ext) { + return UUID_EXTENSION_NAME.equals(ext.extensionName()); + } + return UUID_EXTENSION_NAME.equals( + field + .getMetadata() + .get( + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType.ExtensionType + .EXTENSION_METADATA_KEY_NAME)); + } } diff --git a/vortex/src/test/java/org/apache/iceberg/vortex/TestVortexUuid.java b/vortex/src/test/java/org/apache/iceberg/vortex/TestVortexUuid.java index 51da5b2b0fb3..283905f74ece 100644 --- a/vortex/src/test/java/org/apache/iceberg/vortex/TestVortexUuid.java +++ b/vortex/src/test/java/org/apache/iceberg/vortex/TestVortexUuid.java @@ -52,6 +52,20 @@ public void testArrowUuidExtensionMapsBackToIcebergUuid() { assertThat(roundTrip.findType("uid")).isEqualTo(Types.UUIDType.get()); } + @Test + public void testRelocatedArrowUuidExtensionMapsBackToIcebergUuid() { + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Schema vortexSchema = + VortexSchemas.toVortexArrowSchema(SCHEMA); + dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field uuidField = + vortexSchema.findField("uid"); + + assertThat(VortexSchemas.isUuidField(uuidField)).isTrue(); + assertThat(VortexSchemas.convert(vortexSchema).findType("uid")).isEqualTo(Types.UUIDType.get()); + assertThat( + VortexSchemas.isUuidField(VortexSchemas.toArrowSchema(vortexSchema).findField("uid"))) + .isTrue(); + } + @Test public void testWriteAndReadUuidRoundTrip() { UUID uuid1 = UUID.fromString("00000000-0000-0000-0000-000000000001"); @@ -68,7 +82,8 @@ public void testWriteAndReadUuidRoundTrip() { root.setRowCount(2); // Read back using the GenericVortexReader's primitive dispatch through GenericVortexReaders. - VortexRowReader reader = GenericVortexReader.buildReader(SCHEMA, arrowSchema); + VortexRowReader reader = + GenericVortexReader.buildReader(SCHEMA, VortexSchemas.toArrowSchema(SCHEMA)); Record row0 = reader.read(root, 0); Record row1 = reader.read(root, 1);