Skip to content

VortexArrowColumnVector supports conversion from all arrow types that Spark can support #8771

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
robert3005 merged 1 commit into from
Jul 15, 2026
Merged

VortexArrowColumnVector supports conversion from all arrow types that Spark can support #8771

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions java/vortex-jni/src/test/java/dev/vortex/api/TestMinimal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.DecimalVector;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ViewVarCharVector;
import org.apache.arrow.vector.ipc.ArrowReader;
Expand Down Expand Up @@ -155,8 +154,8 @@ public void testProjectedScan() throws Exception {

List<Person> people = readAll(ds, options, allocator, batch -> {
List<Person> results = new ArrayList<>();
VarCharVector names = (VarCharVector) batch.getVector("Name");
VarCharVector states = (VarCharVector) batch.getVector("State");
ViewVarCharVector names = (ViewVarCharVector) batch.getVector("Name");
ViewVarCharVector states = (ViewVarCharVector) batch.getVector("State");
for (int i = 0; i < batch.getRowCount(); i++) {
String name = names.isNull(i) ? null : new String(names.get(i), UTF_8);
String state = states.isNull(i) ? null : new String(states.get(i), UTF_8);
Expand Down Expand Up @@ -272,9 +271,9 @@ private static List<Person> readAll(

private static List<Person> readFullBatch(VectorSchemaRoot root) {
List<Person> result = new ArrayList<>();
VarCharVector names = (VarCharVector) root.getVector("Name");
ViewVarCharVector names = (ViewVarCharVector) root.getVector("Name");
FieldVector salaries = root.getVector("Salary");
VarCharVector states = (VarCharVector) root.getVector("State");
ViewVarCharVector states = (ViewVarCharVector) root.getVector("State");

for (int i = 0; i < root.getRowCount(); i++) {
String name = names.isNull(i) ? null : new String(names.get(i), UTF_8);
Expand Down
3 changes: 2 additions & 1 deletion java/vortex-jni/src/test/java/dev/vortex/jni/JNIWriterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ViewVarCharVector;
import org.apache.arrow.vector.complex.StructVector;
import org.apache.arrow.vector.ipc.ArrowReader;
import org.apache.arrow.vector.types.pojo.ArrowType;
Expand Down Expand Up @@ -192,7 +193,7 @@ public void testWriteBatch() throws IOException {
try (ArrowReader reader = p.scanArrow(allocator)) {
reader.loadNextBatch();
VectorSchemaRoot resultRoot = reader.getVectorSchemaRoot();
VarCharVector nameOut = (VarCharVector) resultRoot.getVector("name");
ViewVarCharVector nameOut = (ViewVarCharVector) resultRoot.getVector("name");
IntVector ageOut = (IntVector) resultRoot.getVector("age");
assertEquals("Alice", nameOut.getObject(0).toString());
assertEquals("Bob", nameOut.getObject(1).toString());
Expand Down
111 changes: 77 additions & 34 deletions java/vortex-spark/src/main/java/dev/vortex/spark/ArrowUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
package dev.vortex.spark;

import dev.vortex.relocated.org.apache.arrow.vector.types.DateUnit;
import dev.vortex.relocated.org.apache.arrow.vector.types.FloatingPointPrecision;
import dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit;
import dev.vortex.relocated.org.apache.arrow.vector.types.pojo.ArrowType;
import dev.vortex.relocated.org.apache.arrow.vector.types.pojo.Field;
Expand All @@ -18,17 +17,27 @@
* Utility class for converting Arrow types to Spark SQL data types.
*
* <p>This class provides static methods to convert Arrow field definitions and type definitions into their
* corresponding Spark SQL DataType representations. It handles the mapping between Arrow's type system and Spark's type
* system, including complex types like structs and arrays.
* corresponding Spark SQL DataType representations. The mapping matches Spark 4.1's own {@code ArrowUtils}, extended
* where Vortex produces types Spark's mapping does not cover:
*
* <ul>
* <li>the Arrow view types map to their logical Spark types: {@code Utf8View} to {@code StringType},
* {@code BinaryView} to {@code BinaryType}, and {@code ListView} to {@code ArrayType}
* <li>timestamps of every unit map to {@code TimestampType}/{@code TimestampNTZType};
* {@link dev.vortex.spark.read.VortexArrowColumnVector} normalizes non-microsecond values on read
* </ul>
*
* <p>The one intentional divergence from Spark 4.1 is Arrow's Time type: Spark's {@code TimeType} only exists in Spark
* 4.1+, and this module compiles a single source set against both Spark 3.5 and 4.1, so Time is rejected.
*/
public final class ArrowUtils {
private ArrowUtils() {}

/**
* Converts an Arrow Field to a Spark SQL DataType.
*
* <p>This method handles complex types like structs and arrays by recursively converting their child fields. For
* primitive types, it delegates to {@link #fromArrowType(ArrowType)}.
* <p>This method handles nested types like structs, lists and maps by recursively converting their child fields.
* For non-nested types, it delegates to {@link #fromArrowType(ArrowType)}.
*
* @param field the Arrow field to convert
* @return the corresponding Spark SQL DataType
Expand All @@ -43,11 +52,19 @@ public static DataType fromArrowField(Field field) {
return new StructField(child.getName(), dt, child.isNullable(), Metadata.empty());
})
.collect(Collectors.toList()));
case List: {
case List:
case ListView: {
Field elementField = field.getChildren().get(0);
DataType elementType = fromArrowField(elementField);
return DataTypes.createArrayType(elementType, elementField.isNullable());
}
case Map: {
Field entries = field.getChildren().get(0);
Field keyField = entries.getChildren().get(0);
Field valueField = entries.getChildren().get(1);
return DataTypes.createMapType(
fromArrowField(keyField), fromArrowField(valueField), valueField.isNullable());
}
Comment on lines +61 to +67

@a10y a10y Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we never return this right? guess it's just there for later when vortex adds map type

@robert3005 robert3005 Jul 15, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, we never end up here

default:
return fromArrowType(field.getType());
}
Expand All @@ -56,40 +73,44 @@ public static DataType fromArrowField(Field field) {
/**
* Converts an Arrow type to a Spark SQL DataType.
*
* <p>This method maps primitive Arrow types to their corresponding Spark SQL types. It supports most common Arrow
* types including integers, floating point numbers, strings, binary data, dates, timestamps, decimals, and nulls.
* <p>This method maps non-nested Arrow types to their corresponding Spark SQL types, following Spark 4.1's own
* Arrow type mapping plus the view types (see the class documentation).
*
* @param dt the Arrow type to convert
* @return the corresponding Spark SQL DataType
* @throws UnsupportedOperationException if the Arrow type configuration is not supported
* @throws RuntimeException if the Arrow type is not recognized
* @throws UnsupportedOperationException if the Arrow type has no Spark representation
*/
public static DataType fromArrowType(ArrowType dt) {
switch (dt.getTypeID()) {
case Bool:
return DataTypes.BooleanType;
case Int: {
ArrowType.Int intType = (ArrowType.Int) dt;
if (intType.getIsSigned() && intType.getBitWidth() == 8) {
return DataTypes.ByteType;
} else if (intType.getIsSigned() && intType.getBitWidth() == 16) {
return DataTypes.ShortType;
} else if (intType.getIsSigned() && intType.getBitWidth() == 32) {
return DataTypes.IntegerType;
} else if (intType.getIsSigned() && intType.getBitWidth() == 64) {
return DataTypes.LongType;
} else {
throw new UnsupportedOperationException("Unsupported Arrow type: " + dt);
if (!intType.getIsSigned()) {
throw new UnsupportedOperationException("Unsupported Arrow unsigned integer type: " + dt);
}
switch (intType.getBitWidth()) {
case 8:
return DataTypes.ByteType;
case 16:
return DataTypes.ShortType;
case 32:
return DataTypes.IntegerType;
case 64:
return DataTypes.LongType;
default:
throw new UnsupportedOperationException("Unsupported Arrow integer bit width: " + dt);
}
}
case FloatingPoint: {
ArrowType.FloatingPoint floatType = (ArrowType.FloatingPoint) dt;
if (floatType.getPrecision() == FloatingPointPrecision.SINGLE) {
return DataTypes.FloatType;
} else if (floatType.getPrecision() == FloatingPointPrecision.DOUBLE) {
return DataTypes.DoubleType;
} else {
throw new UnsupportedOperationException("Unsupported Arrow type: " + dt);
switch (floatType.getPrecision()) {
case SINGLE:
return DataTypes.FloatType;
case DOUBLE:
return DataTypes.DoubleType;
default:
throw new UnsupportedOperationException("Unsupported Arrow float precision: " + dt);
}
}
case Decimal: {
Expand All @@ -98,30 +119,52 @@ public static DataType fromArrowType(ArrowType dt) {
}
case Utf8:
case LargeUtf8:
case Utf8View:
return DataTypes.StringType;
case Binary:
case LargeBinary:
case BinaryView:
return DataTypes.BinaryType;
case Date: {
ArrowType.Date dateType = (ArrowType.Date) dt;
if (dateType.getUnit() == DateUnit.DAY) {
return DataTypes.DateType;
} else {
throw new UnsupportedOperationException("Unsupported Arrow type: " + dt);
}
throw new UnsupportedOperationException("Unsupported Arrow date unit: " + dt);
}
case Timestamp: {
// Spark timestamps are physically microseconds; VortexArrowColumnVector's accessor
// normalizes second/millisecond/nanosecond values on read.
ArrowType.Timestamp ts = (ArrowType.Timestamp) dt;
if (ts.getUnit() == TimeUnit.MICROSECOND) {
return ts.getTimezone() != null ? DataTypes.TimestampType : DataTypes.TimestampNTZType;
} else {
throw new UnsupportedOperationException("Unsupported Arrow type: " + dt);
}
return ts.getTimezone() != null ? DataTypes.TimestampType : DataTypes.TimestampNTZType;
}
case Null:
return DataTypes.NullType;
case Interval: {
ArrowType.Interval interval = (ArrowType.Interval) dt;
switch (interval.getUnit()) {
case YEAR_MONTH:
return DataTypes.createYearMonthIntervalType();
case MONTH_DAY_NANO:
return DataTypes.CalendarIntervalType;
default:
throw new UnsupportedOperationException("Unsupported Arrow interval unit: " + dt);
}
}
case Duration: {
ArrowType.Duration duration = (ArrowType.Duration) dt;
if (duration.getUnit() != TimeUnit.MICROSECOND) {
throw new UnsupportedOperationException("Unsupported Arrow duration unit: " + dt);
}
return DataTypes.createDayTimeIntervalType();
}
case Time:
// Spark 3.5 has no TIME type (TimeType only exists in Spark 4.1+), and this
// module compiles a single source set against both Spark versions.
throw new UnsupportedOperationException("Arrow Time type has no Spark 3.5 representation: " + dt);
default:
throw new IllegalArgumentException("Unsupported Arrow type: " + dt);
throw new UnsupportedOperationException(
"Unsupported Arrow type: " + dt + " (type id: " + dt.getTypeID() + ")");
}
}
}
Loading
Loading