You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2025/08/21 12:40:17 UTC
(iceberg) branch main updated: Arrow: Add nanosec precision timestamp (#13562)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new efc27f9dee Arrow: Add nanosec precision timestamp (#13562)
efc27f9dee is described below
commit efc27f9deed0511653ffb94612a8bfcd1489df42
Author: Nándor Kollár <na...@users.noreply.github.com>
AuthorDate: Thu Aug 21 14:40:11 2025 +0200
Arrow: Add nanosec precision timestamp (#13562)
---
.../org/apache/iceberg/arrow/ArrowSchemaUtil.java | 6 +
.../iceberg/arrow/DictEncodedArrowConverter.java | 25 ++
.../iceberg/arrow/vectorized/ArrowReader.java | 12 +-
.../iceberg/arrow/vectorized/ColumnVector.java | 1 +
.../GenericArrowVectorAccessorFactory.java | 42 ++--
.../arrow/vectorized/VectorizedArrowReader.java | 263 ++++++++++++++-------
.../apache/iceberg/arrow/TestArrowSchemaUtil.java | 12 +-
.../iceberg/arrow/vectorized/TestArrowReader.java | 151 +++++++++++-
8 files changed, 396 insertions(+), 116 deletions(-)
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java b/arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
index c70e4ff3ff..75a9786457 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
@@ -171,6 +171,12 @@ public class ArrowSchemaUtil {
TimeUnit.MICROSECOND,
((Types.TimestampType) primitive).shouldAdjustToUTC() ? "UTC" : null);
break;
+ case TIMESTAMP_NANO:
+ arrowType =
+ new ArrowType.Timestamp(
+ TimeUnit.NANOSECOND,
+ ((Types.TimestampNanoType) primitive).shouldAdjustToUTC() ? "UTC" : null);
+ break;
case DATE:
arrowType = new ArrowType.Date(DateUnit.DAY);
break;
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/DictEncodedArrowConverter.java b/arrow/src/main/java/org/apache/iceberg/arrow/DictEncodedArrowConverter.java
index 7169bda28f..f1e781c347 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/DictEncodedArrowConverter.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/DictEncodedArrowConverter.java
@@ -31,6 +31,8 @@ import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.TimeMicroVector;
import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampNanoVector;
import org.apache.arrow.vector.TimeStampVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
@@ -56,6 +58,8 @@ public class DictEncodedArrowConverter {
return toDecimalVector(vectorHolder, accessor);
} else if (Type.TypeID.TIMESTAMP.equals(vectorHolder.icebergType().typeId())) {
return toTimestampVector(vectorHolder, accessor);
+ } else if (Type.TypeID.TIMESTAMP_NANO.equals(vectorHolder.icebergType().typeId())) {
+ return toTimestampNanoVector(vectorHolder, accessor);
} else if (Type.TypeID.LONG.equals(vectorHolder.icebergType().typeId())) {
return toBigIntVector(vectorHolder, accessor);
} else if (Type.TypeID.FLOAT.equals(vectorHolder.icebergType().typeId())) {
@@ -119,6 +123,27 @@ public class DictEncodedArrowConverter {
return vector;
}
+ private static TimeStampVector toTimestampNanoVector(
+ VectorHolder vectorHolder, ArrowVectorAccessor<?, String, ?, ?> accessor) {
+ TimeStampVector vector;
+ if (((Types.TimestampNanoType) vectorHolder.icebergType()).shouldAdjustToUTC()) {
+ vector =
+ new TimeStampNanoTZVector(
+ vectorHolder.vector().getName(),
+ ArrowSchemaUtil.convert(vectorHolder.icebergField()).getFieldType(),
+ vectorHolder.vector().getAllocator());
+ } else {
+ vector =
+ new TimeStampNanoVector(
+ vectorHolder.vector().getName(),
+ ArrowSchemaUtil.convert(vectorHolder.icebergField()).getFieldType(),
+ vectorHolder.vector().getAllocator());
+ }
+
+ initVector(vector, vectorHolder, idx -> vector.set(idx, accessor.getLong(idx)));
+ return vector;
+ }
+
private static BigIntVector toBigIntVector(
VectorHolder vectorHolder, ArrowVectorAccessor<?, String, ?, ?> accessor) {
BigIntVector vector =
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ArrowReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ArrowReader.java
index a4196cf62c..cd186e0afb 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ArrowReader.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ArrowReader.java
@@ -73,9 +73,12 @@ import org.slf4j.LoggerFactory;
* <li>Iceberg: {@link Types.LongType}, Arrow: {@link MinorType#BIGINT}
* <li>Iceberg: {@link Types.FloatType}, Arrow: {@link MinorType#FLOAT4}
* <li>Iceberg: {@link Types.DoubleType}, Arrow: {@link MinorType#FLOAT8}
+ * <li>Iceberg: {@link Types.DecimalType}, Arrow: {@link MinorType#DECIMAL}
* <li>Iceberg: {@link Types.StringType}, Arrow: {@link MinorType#VARCHAR}
* <li>Iceberg: {@link Types.TimestampType} (both with and without timezone), Arrow: {@link
- * MinorType#TIMEMICRO}
+ * MinorType#TIMESTAMPMICRO} and {@link MinorType#TIMESTAMPMICROTZ}
+ * <li>Iceberg: {@link Types.TimestampNanoType} (both with and without timezone), Arrow: {@link
+ * MinorType#TIMESTAMPNANO} and {@link MinorType#TIMESTAMPNANOTZ}
* <li>Iceberg: {@link Types.BinaryType}, Arrow: {@link MinorType#VARBINARY}
* <li>Iceberg: {@link Types.FixedType}, Arrow: {@link MinorType#FIXEDSIZEBINARY}
* <li>Iceberg: {@link Types.DateType}, Arrow: {@link MinorType#DATEDAY}
@@ -92,8 +95,8 @@ import org.slf4j.LoggerFactory;
* <li>Columns with constant values are physically encoded as a dictionary. The Arrow vector type
* is int32 instead of the type as per the schema. See
* https://github.com/apache/iceberg/issues/2484.
- * <li>Data types: {@link Types.ListType}, {@link Types.MapType}, {@link Types.StructType}, and
- * {@link Types.DecimalType} See https://github.com/apache/iceberg/issues/2485 and
+ * <li>Data types: {@link Types.ListType}, {@link Types.MapType} and {@link Types.StructType} See
+ * https://github.com/apache/iceberg/issues/2485 and
* https://github.com/apache/iceberg/issues/2486.
* <li>Delete files are not supported. See https://github.com/apache/iceberg/issues/2487.
* </ul>
@@ -115,7 +118,8 @@ public class ArrowReader extends CloseableGroup {
TypeID.UUID,
TypeID.TIME,
TypeID.DECIMAL,
- TypeID.FIXED);
+ TypeID.FIXED,
+ TypeID.TIMESTAMP_NANO);
private final Schema schema;
private final FileIO io;
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ColumnVector.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ColumnVector.java
index d2e5c18215..649ec2c9e2 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ColumnVector.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ColumnVector.java
@@ -40,6 +40,7 @@ import org.apache.iceberg.types.Types;
* <li>{@link Types.BinaryType}
* <li>{@link Types.FixedType}
* <li>{@link Types.TimestampType} (with and without timezone)
+ * <li>{@link Types.TimestampNanoType} (with and without timezone)
* <li>{@link Types.DateType}
* <li>{@link Types.TimeType}
* <li>{@link Types.UUIDType}
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactory.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactory.java
index 2e24ce2e87..beadefd16f 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactory.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactory.java
@@ -38,6 +38,9 @@ import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.TimeMicroVector;
import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampNanoVector;
+import org.apache.arrow.vector.TimeStampVector;
import org.apache.arrow.vector.ValueVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
@@ -119,6 +122,8 @@ public class GenericArrowVectorAccessorFactory<
PrimitiveType primitive) {
Preconditions.checkState(
vector instanceof IntVector, "Dictionary ids should be stored in IntVectors only");
+ // TODO: consider moving this to logical type annotations,
+ // as new Parquet types are added only there.
if (primitive.getOriginalType() != null) {
switch (desc.getPrimitiveType().getOriginalType()) {
case ENUM:
@@ -203,9 +208,13 @@ public class GenericArrowVectorAccessorFactory<
} else if (vector instanceof DateDayVector) {
return new DateAccessor<>((DateDayVector) vector);
} else if (vector instanceof TimeStampMicroTZVector) {
- return new TimestampMicroTzAccessor<>((TimeStampMicroTZVector) vector);
+ return new TimestampAccessor<>((TimeStampMicroTZVector) vector);
} else if (vector instanceof TimeStampMicroVector) {
- return new TimestampMicroAccessor<>((TimeStampMicroVector) vector);
+ return new TimestampAccessor<>((TimeStampMicroVector) vector);
+ } else if (vector instanceof TimeStampNanoVector) {
+ return new TimestampAccessor<>((TimeStampNanoVector) vector);
+ } else if (vector instanceof TimeStampNanoTZVector) {
+ return new TimestampAccessor<>((TimeStampNanoTZVector) vector);
} else if (vector instanceof ListVector) {
ListVector listVector = (ListVector) vector;
return new ArrayAccessor<>(listVector, arrayFactorySupplier.get());
@@ -516,30 +525,17 @@ public class GenericArrowVectorAccessorFactory<
}
}
- private static class TimestampMicroTzAccessor<
- DecimalT, Utf8StringT, ArrayT, ChildVectorT extends AutoCloseable>
- extends ArrowVectorAccessor<DecimalT, Utf8StringT, ArrayT, ChildVectorT> {
-
- private final TimeStampMicroTZVector vector;
-
- TimestampMicroTzAccessor(TimeStampMicroTZVector vector) {
- super(vector);
- this.vector = vector;
- }
-
- @Override
- public final long getLong(int rowId) {
- return vector.get(rowId);
- }
- }
-
- private static class TimestampMicroAccessor<
- DecimalT, Utf8StringT, ArrayT, ChildVectorT extends AutoCloseable>
+ private static class TimestampAccessor<
+ DecimalT,
+ Utf8StringT,
+ ArrayT,
+ TimestampVectorT extends TimeStampVector,
+ ChildVectorT extends AutoCloseable>
extends ArrowVectorAccessor<DecimalT, Utf8StringT, ArrayT, ChildVectorT> {
- private final TimeStampMicroVector vector;
+ private final TimestampVectorT vector;
- TimestampMicroAccessor(TimeStampMicroVector vector) {
+ TimestampAccessor(TimestampVectorT vector) {
super(vector);
this.vector = vector;
}
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 b9509f1999..015ff602bd 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
@@ -19,6 +19,7 @@
package org.apache.iceberg.arrow.vectorized;
import java.util.Map;
+import java.util.Optional;
import org.apache.arrow.memory.ArrowBuf;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.BigIntVector;
@@ -33,6 +34,8 @@ import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.TimeMicroVector;
import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampNanoVector;
import org.apache.arrow.vector.types.FloatingPointPrecision;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
@@ -213,8 +216,8 @@ public class VectorizedArrowReader implements VectorizedReader<VectorHolder> {
allocateDictEncodedVector();
} else {
Field arrowField = ArrowSchemaUtil.convert(getPhysicalType(columnDescriptor, icebergField));
- if (columnDescriptor.getPrimitiveType().getOriginalType() != null) {
- allocateVectorBasedOnOriginalType(columnDescriptor.getPrimitiveType(), arrowField);
+ if (columnDescriptor.getPrimitiveType().getLogicalTypeAnnotation() != null) {
+ allocateVectorBasedOnLogicalType(columnDescriptor.getPrimitiveType(), arrowField);
} else {
allocateVectorBasedOnTypeName(columnDescriptor.getPrimitiveType(), arrowField);
}
@@ -258,89 +261,18 @@ public class VectorizedArrowReader implements VectorizedReader<VectorHolder> {
this.readType = ReadType.DICTIONARY;
}
- private void allocateVectorBasedOnOriginalType(PrimitiveType primitive, Field arrowField) {
- switch (primitive.getOriginalType()) {
- case ENUM:
- case JSON:
- case UTF8:
- case BSON:
- this.vec = arrowField.createVector(rootAlloc);
- // TODO: Possibly use the uncompressed page size info to set the initial capacity
- vec.setInitialCapacity(batchSize * AVERAGE_VARIABLE_WIDTH_RECORD_SIZE);
- vec.allocateNewSafe();
- this.readType = ReadType.VARCHAR;
- this.typeWidth = UNKNOWN_WIDTH;
- break;
- case INT_8:
- case INT_16:
- case INT_32:
- this.vec = arrowField.createVector(rootAlloc);
- ((IntVector) vec).allocateNew(batchSize);
- this.readType = ReadType.INT;
- this.typeWidth = (int) IntVector.TYPE_WIDTH;
- break;
- case DATE:
- this.vec = arrowField.createVector(rootAlloc);
- ((DateDayVector) vec).allocateNew(batchSize);
- this.readType = ReadType.INT;
- this.typeWidth = (int) IntVector.TYPE_WIDTH;
- break;
- case INT_64:
- this.vec = arrowField.createVector(rootAlloc);
- ((BigIntVector) vec).allocateNew(batchSize);
- this.readType = ReadType.LONG;
- this.typeWidth = (int) BigIntVector.TYPE_WIDTH;
- break;
- case TIMESTAMP_MILLIS:
- this.vec = arrowField.createVector(rootAlloc);
- ((BigIntVector) vec).allocateNew(batchSize);
- this.readType = ReadType.TIMESTAMP_MILLIS;
- this.typeWidth = (int) BigIntVector.TYPE_WIDTH;
- break;
- case TIMESTAMP_MICROS:
- this.vec = arrowField.createVector(rootAlloc);
- if (((Types.TimestampType) icebergField.type()).shouldAdjustToUTC()) {
- ((TimeStampMicroTZVector) vec).allocateNew(batchSize);
- } else {
- ((TimeStampMicroVector) vec).allocateNew(batchSize);
- }
- this.readType = ReadType.LONG;
- this.typeWidth = (int) BigIntVector.TYPE_WIDTH;
- break;
- case TIME_MICROS:
- this.vec = arrowField.createVector(rootAlloc);
- ((TimeMicroVector) vec).allocateNew(batchSize);
- this.readType = ReadType.LONG;
- this.typeWidth = (int) TimeMicroVector.TYPE_WIDTH;
- break;
- case DECIMAL:
- this.vec = arrowField.createVector(rootAlloc);
- switch (primitive.getPrimitiveTypeName()) {
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- ((FixedSizeBinaryVector) vec).allocateNew(batchSize);
- this.readType = ReadType.FIXED_LENGTH_DECIMAL;
- this.typeWidth = primitive.getTypeLength();
- break;
- case INT64:
- ((BigIntVector) vec).allocateNew(batchSize);
- this.readType = ReadType.LONG_BACKED_DECIMAL;
- this.typeWidth = (int) BigIntVector.TYPE_WIDTH;
- break;
- case INT32:
- ((IntVector) vec).allocateNew(batchSize);
- this.readType = ReadType.INT_BACKED_DECIMAL;
- this.typeWidth = (int) IntVector.TYPE_WIDTH;
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported logical type: " + primitive.getOriginalType());
- }
+ private void allocateVectorBasedOnLogicalType(PrimitiveType primitive, Field arrowField) {
+ LogicalTypeVisitorResult logicalTypeVisitorResult =
+ primitive
+ .getLogicalTypeAnnotation()
+ .accept(new LogicalTypeVisitor(arrowField, primitive))
+ .orElseThrow(
+ () ->
+ new UnsupportedOperationException(
+ "Unsupported logical type: " + primitive.getOriginalType()));
+ this.readType = logicalTypeVisitorResult.readType;
+ this.typeWidth = logicalTypeVisitorResult.typeWidth;
+ this.vec = logicalTypeVisitorResult.vec;
}
private void allocateVectorBasedOnTypeName(PrimitiveType primitive, Field arrowField) {
@@ -508,6 +440,167 @@ public class VectorizedArrowReader implements VectorizedReader<VectorHolder> {
return reader;
}
+ private static final class LogicalTypeVisitorResult {
+ private final FieldVector vec;
+ private final Integer typeWidth;
+ private final ReadType readType;
+
+ LogicalTypeVisitorResult(FieldVector vec, ReadType readType, Integer typeWidth) {
+ this.vec = vec;
+ this.typeWidth = typeWidth;
+ this.readType = readType;
+ }
+ }
+
+ private final class LogicalTypeVisitor
+ implements LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<LogicalTypeVisitorResult> {
+ private final Field arrowField;
+ private final PrimitiveType primitive;
+
+ LogicalTypeVisitor(Field arrowField, PrimitiveType primitive) {
+ this.arrowField = arrowField;
+ this.primitive = primitive;
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.StringLogicalTypeAnnotation stringLogicalType) {
+ return allocateVectorForEnumJsonBsonString();
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.EnumLogicalTypeAnnotation enumLogicalType) {
+ return allocateVectorForEnumJsonBsonString();
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.UUIDLogicalTypeAnnotation uuidLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ ((FixedSizeBinaryVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.UUID, primitive.getTypeLength()));
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ switch (primitive.getPrimitiveTypeName()) {
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ ((FixedSizeBinaryVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(
+ vector, ReadType.FIXED_LENGTH_DECIMAL, primitive.getTypeLength()));
+ case INT64:
+ ((BigIntVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(
+ vector, ReadType.LONG_BACKED_DECIMAL, (int) BigIntVector.TYPE_WIDTH));
+ case INT32:
+ ((IntVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(
+ vector, ReadType.INT_BACKED_DECIMAL, (int) IntVector.TYPE_WIDTH));
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
+ }
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.DateLogicalTypeAnnotation dateLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ ((DateDayVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.INT, (int) IntVector.TYPE_WIDTH));
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.TimeLogicalTypeAnnotation timeLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ ((TimeMicroVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.LONG, (int) TimeMicroVector.TYPE_WIDTH));
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ switch (timestampLogicalType.getUnit()) {
+ case MILLIS:
+ ((BigIntVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(
+ vector, ReadType.TIMESTAMP_MILLIS, (int) BigIntVector.TYPE_WIDTH));
+ case MICROS:
+ if (((Types.TimestampType) icebergField.type()).shouldAdjustToUTC()) {
+ ((TimeStampMicroTZVector) vector).allocateNew(batchSize);
+ } else {
+ ((TimeStampMicroVector) vector).allocateNew(batchSize);
+ }
+
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.LONG, (int) BigIntVector.TYPE_WIDTH));
+ case NANOS:
+ if (((Types.TimestampNanoType) icebergField.type()).shouldAdjustToUTC()) {
+ ((TimeStampNanoTZVector) vector).allocateNew(batchSize);
+ } else {
+ ((TimeStampNanoVector) vector).allocateNew(batchSize);
+ }
+
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.LONG, (int) BigIntVector.TYPE_WIDTH));
+ }
+
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.IntLogicalTypeAnnotation intLogicalType) {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ int bitWidth = intLogicalType.getBitWidth();
+
+ if (bitWidth == 8 || bitWidth == 16 || bitWidth == 32) {
+ ((IntVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.INT, (int) IntVector.TYPE_WIDTH));
+ } else if (bitWidth == 64) {
+ ((BigIntVector) vector).allocateNew(batchSize);
+ return Optional.of(
+ new LogicalTypeVisitorResult(vector, ReadType.LONG, (int) BigIntVector.TYPE_WIDTH));
+ }
+
+ return Optional.empty();
+ }
+
+ private Optional<LogicalTypeVisitorResult> allocateVectorForEnumJsonBsonString() {
+ FieldVector vector = arrowField.createVector(rootAlloc);
+ // TODO: Possibly use the uncompressed page size info to set the initial capacity
+ vector.setInitialCapacity(batchSize * AVERAGE_VARIABLE_WIDTH_RECORD_SIZE);
+ vector.allocateNewSafe();
+ return Optional.of(new LogicalTypeVisitorResult(vector, ReadType.VARCHAR, UNKNOWN_WIDTH));
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.JsonLogicalTypeAnnotation jsonLogicalType) {
+ return allocateVectorForEnumJsonBsonString();
+ }
+
+ @Override
+ public Optional<LogicalTypeVisitorResult> visit(
+ LogicalTypeAnnotation.BsonLogicalTypeAnnotation bsonLogicalType) {
+ return allocateVectorForEnumJsonBsonString();
+ }
+ }
+
private static final class NullVectorReader extends VectorizedArrowReader {
private static final NullVectorReader INSTANCE = new NullVectorReader();
diff --git a/arrow/src/test/java/org/apache/iceberg/arrow/TestArrowSchemaUtil.java b/arrow/src/test/java/org/apache/iceberg/arrow/TestArrowSchemaUtil.java
index ad132c7bc8..fef0ef4efd 100644
--- a/arrow/src/test/java/org/apache/iceberg/arrow/TestArrowSchemaUtil.java
+++ b/arrow/src/test/java/org/apache/iceberg/arrow/TestArrowSchemaUtil.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.arrow;
import static org.assertj.core.api.Assertions.assertThat;
+import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.iceberg.Schema;
@@ -47,6 +48,7 @@ public class TestArrowSchemaUtil {
private static final String STRING_FIELD = "s";
private static final String DATE_FIELD = "d2";
private static final String TIMESTAMP_FIELD = "ts";
+ private static final String TIMESTAMP_NANO_FIELD = "tsn";
private static final String LONG_FIELD = "l";
private static final String FLOAT_FIELD = "f";
private static final String TIME_FIELD = "tt";
@@ -80,7 +82,9 @@ public class TestArrowSchemaUtil {
MAP_FIELD,
Types.MapType.ofOptional(15, 16, StringType.get(), IntegerType.get())),
Types.NestedField.optional(17, FIXED_WIDTH_BINARY_FIELD, Types.FixedType.ofLength(10)),
- Types.NestedField.optional(18, UUID_FIELD, Types.UUIDType.get()));
+ Types.NestedField.optional(18, UUID_FIELD, Types.UUIDType.get()),
+ Types.NestedField.optional(
+ 19, TIMESTAMP_NANO_FIELD, Types.TimestampNanoType.withZone()));
org.apache.arrow.vector.types.pojo.Schema arrow = ArrowSchemaUtil.convert(iceberg);
@@ -224,6 +228,12 @@ public class TestArrowSchemaUtil {
case TIMESTAMP:
assertThat(field.getName()).isEqualTo(TIMESTAMP_FIELD);
assertThat(arrowType.getTypeID()).isEqualTo(ArrowType.ArrowTypeID.Timestamp);
+ assertThat(((ArrowType.Timestamp) arrowType).getUnit()).isEqualTo(TimeUnit.MICROSECOND);
+ break;
+ case TIMESTAMP_NANO:
+ assertThat(field.getName()).isEqualTo(TIMESTAMP_NANO_FIELD);
+ assertThat(arrowType.getTypeID()).isEqualTo(ArrowType.ArrowTypeID.Timestamp);
+ assertThat(((ArrowType.Timestamp) arrowType).getUnit()).isEqualTo(TimeUnit.NANOSECOND);
break;
case STRING:
assertThat(field.getName()).isEqualTo(STRING_FIELD);
diff --git a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestArrowReader.java b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestArrowReader.java
index 6d6e6e6a8d..8ef149fc86 100644
--- a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestArrowReader.java
+++ b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestArrowReader.java
@@ -52,6 +52,8 @@ import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.TimeMicroVector;
import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampNanoVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
@@ -69,6 +71,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
@@ -79,6 +82,7 @@ import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -126,7 +130,11 @@ public class TestArrowReader {
"decimal",
"decimal_nullable",
"fixed",
- "fixed_nullable");
+ "fixed_nullable",
+ "timestamp_nano",
+ "timestamp_nano_nullable",
+ "timestamp_nano_tz",
+ "timestamp_nano_tz_nullable");
@TempDir private File tempDir;
private HadoopTables tables;
@@ -655,6 +663,46 @@ public class TestArrowReader {
"fixed_nullable",
(records, i) -> records.get(i).getField("fixed_nullable"),
(array, i) -> array.getBinary(i));
+
+ checkColumnarArrayValues(
+ expectedNumRows,
+ expectedRows,
+ batch,
+ columnNameToIndex.get("timestamp_nano"),
+ columnSet,
+ "timestamp_nano",
+ (records, i) -> records.get(i).getField("timestamp_nano"),
+ (array, i) -> timestampFromNanos(array.getLong(i)));
+ checkColumnarArrayValues(
+ expectedNumRows,
+ expectedRows,
+ batch,
+ columnNameToIndex.get("timestamp_nano_nullable"),
+ columnSet,
+ "timestamp_nano_nullable",
+ (records, i) -> records.get(i).getField("timestamp_nano_nullable"),
+ (array, i) -> timestampFromNanos(array.getLong(i)));
+ checkColumnarArrayValues(
+ expectedNumRows,
+ expectedRows,
+ batch,
+ columnNameToIndex.get("timestamp_nano_tz"),
+ columnSet,
+ "timestamp_nano_tz",
+ (records, i) ->
+ timestampToNanos((OffsetDateTime) records.get(i).getField("timestamp_nano_tz")),
+ ColumnVector::getLong);
+ checkColumnarArrayValues(
+ expectedNumRows,
+ expectedRows,
+ batch,
+ columnNameToIndex.get("timestamp_nano_tz_nullable"),
+ columnSet,
+ "timestamp_nano_tz_nullable",
+ (records, i) ->
+ timestampToNanos(
+ (OffsetDateTime) records.get(i).getField("timestamp_nano_tz_nullable")),
+ ColumnVector::getLong);
}
private static void checkColumnarArrayValues(
@@ -722,11 +770,19 @@ public class TestArrowReader {
Types.NestedField.required(26, "decimal", Types.DecimalType.of(9, 2)),
Types.NestedField.optional(27, "decimal_nullable", Types.DecimalType.of(9, 2)),
Types.NestedField.required(28, "fixed", Types.FixedType.ofLength(7)),
- Types.NestedField.optional(29, "fixed_nullable", Types.FixedType.ofLength(7)));
+ Types.NestedField.optional(29, "fixed_nullable", Types.FixedType.ofLength(7)),
+ Types.NestedField.required(30, "timestamp_nano", Types.TimestampNanoType.withoutZone()),
+ Types.NestedField.optional(
+ 31, "timestamp_nano_nullable", Types.TimestampNanoType.withoutZone()),
+ Types.NestedField.required(32, "timestamp_nano_tz", Types.TimestampNanoType.withZone()),
+ Types.NestedField.optional(
+ 33, "timestamp_nano_tz_nullable", Types.TimestampNanoType.withZone()));
PartitionSpec spec = PartitionSpec.builderFor(schema).month("timestamp").build();
- Table table = tables.create(schema, spec, tableLocation);
+ Table table =
+ tables.create(
+ schema, spec, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"), tableLocation);
OverwriteFiles overwrite = table.newOverwrite();
for (int i = 1; i <= 12; i++) {
@@ -814,6 +870,30 @@ public class TestArrowReader {
new Field(
"fixed_nullable",
new FieldType(true, new ArrowType.FixedSizeBinary(7), null),
+ null),
+ new Field(
+ "timestamp_nano",
+ new FieldType(false, MinorType.TIMESTAMPNANO.getType(), null),
+ null),
+ new Field(
+ "timestamp_nano_nullable",
+ new FieldType(true, MinorType.TIMESTAMPNANO.getType(), null),
+ null),
+ new Field(
+ "timestamp_nano_tz",
+ new FieldType(
+ false,
+ new ArrowType.Timestamp(
+ org.apache.arrow.vector.types.TimeUnit.NANOSECOND, "UTC"),
+ null),
+ null),
+ new Field(
+ "timestamp_nano_tz_nullable",
+ new FieldType(
+ true,
+ new ArrowType.Timestamp(
+ org.apache.arrow.vector.types.TimeUnit.NANOSECOND, "UTC"),
+ null),
null));
List<Field> filteredFields =
allFields.stream()
@@ -859,6 +939,14 @@ public class TestArrowReader {
rec.setField("decimal_nullable", new BigDecimal("14.0" + i % 10));
rec.setField("fixed", ("abcdef" + i % 7).getBytes(StandardCharsets.UTF_8));
rec.setField("fixed_nullable", ("abcdef" + i % 7).getBytes(StandardCharsets.UTF_8));
+ rec.setField("timestamp_nano", datetime.plus(i, ChronoUnit.DAYS).plusNanos(i));
+ rec.setField("timestamp_nano_nullable", datetime.plus(i, ChronoUnit.DAYS).plusNanos(i));
+ rec.setField(
+ "timestamp_nano_tz",
+ datetime.plus(i, ChronoUnit.MINUTES).plusNanos(i).atOffset(ZoneOffset.UTC));
+ rec.setField(
+ "timestamp_nano_tz_nullable",
+ datetime.plus(i, ChronoUnit.MINUTES).plusNanos(i).atOffset(ZoneOffset.UTC));
records.add(rec);
}
return records;
@@ -898,6 +986,10 @@ public class TestArrowReader {
rec.setField("decimal_nullable", new BigDecimal("14.20"));
rec.setField("fixed", "abcdefg".getBytes(StandardCharsets.UTF_8));
rec.setField("fixed_nullable", "abcdefg".getBytes(StandardCharsets.UTF_8));
+ rec.setField("timestamp_nano", datetime);
+ rec.setField("timestamp_nano_nullable", datetime);
+ rec.setField("timestamp_nano_tz", datetime.atOffset(ZoneOffset.UTC));
+ rec.setField("timestamp_nano_tz_nullable", datetime.atOffset(ZoneOffset.UTC));
records.add(rec);
}
return records;
@@ -939,6 +1031,11 @@ public class TestArrowReader {
return ChronoUnit.MICROS.between(Instant.EPOCH, instant);
}
+ private static long timestampToNanos(OffsetDateTime value) {
+ Instant instant = value.toInstant();
+ return ChronoUnit.NANOS.between(Instant.EPOCH, instant);
+ }
+
private static LocalDateTime timestampFromMicros(long micros) {
return LocalDateTime.ofEpochSecond(
TimeUnit.MICROSECONDS.toSeconds(micros),
@@ -946,6 +1043,13 @@ public class TestArrowReader {
ZoneOffset.UTC);
}
+ private static LocalDateTime timestampFromNanos(long nanos) {
+ return LocalDateTime.ofEpochSecond(
+ TimeUnit.NANOSECONDS.toSeconds(nanos),
+ (int) TimeUnit.NANOSECONDS.toNanos(nanos % 1000),
+ ZoneOffset.UTC);
+ }
+
private static LocalDate dateFromDay(int day) {
return LocalDate.ofEpochDay(day);
}
@@ -980,6 +1084,11 @@ public class TestArrowReader {
assertEqualsForField(root, columnSet, "decimal_nullable", DecimalVector.class);
assertEqualsForField(root, columnSet, "fixed", FixedSizeBinaryVector.class);
assertEqualsForField(root, columnSet, "fixed_nullable", FixedSizeBinaryVector.class);
+ assertEqualsForField(root, columnSet, "timestamp_nano", TimeStampNanoVector.class);
+ assertEqualsForField(root, columnSet, "timestamp_nano_nullable", TimeStampNanoVector.class);
+ assertEqualsForField(root, columnSet, "timestamp_nano_tz", TimeStampNanoTZVector.class);
+ assertEqualsForField(
+ root, columnSet, "timestamp_nano_tz_nullable", TimeStampNanoTZVector.class);
}
private void assertEqualsForField(
@@ -1238,6 +1347,42 @@ public class TestArrowReader {
"fixed_nullable",
(records, i) -> records.get(i).getField("fixed_nullable"),
(vector, i) -> ((FixedSizeBinaryVector) vector).getObject(i));
+
+ checkVectorValues(
+ expectedNumRows,
+ expectedRows,
+ root,
+ columnSet,
+ "timestamp_nano",
+ (records, i) -> records.get(i).getField("timestamp_nano"),
+ (vector, i) -> ((TimeStampNanoVector) vector).getObject(i));
+ checkVectorValues(
+ expectedNumRows,
+ expectedRows,
+ root,
+ columnSet,
+ "timestamp_nano_nullable",
+ (records, i) -> records.get(i).getField("timestamp_nano_nullable"),
+ (vector, i) -> ((TimeStampNanoVector) vector).getObject(i));
+ checkVectorValues(
+ expectedNumRows,
+ expectedRows,
+ root,
+ columnSet,
+ "timestamp_nano_tz",
+ (records, i) ->
+ timestampToNanos((OffsetDateTime) records.get(i).getField("timestamp_nano_tz")),
+ (vector, i) -> ((TimeStampNanoTZVector) vector).get(i));
+ checkVectorValues(
+ expectedNumRows,
+ expectedRows,
+ root,
+ columnSet,
+ "timestamp_nano_tz_nullable",
+ (records, i) ->
+ timestampToNanos(
+ (OffsetDateTime) records.get(i).getField("timestamp_nano_tz_nullable")),
+ (vector, i) -> ((TimeStampNanoTZVector) vector).get(i));
}
private static void checkVectorValues(