This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 88fa2af227f2 [SPARK-55444][SQL] Introduce and Route TimeType to
Parquet vectorized read through the Types Framework
88fa2af227f2 is described below
commit 88fa2af227f2f85f2cd7bfd8d61d76c1198aca78
Author: Stevo Mitric <[email protected]>
AuthorDate: Wed Jul 1 21:53:34 2026 +0200
[SPARK-55444][SQL] Introduce and Route TimeType to Parquet vectorized read
through the Types Framework
### What changes were proposed in this pull request?
Follow-up to the Types Framework Phase 3a Parquet work
([SPARK-55444](https://issues.apache.org/jira/browse/SPARK-55444)). It moves
`TimeType`'s **vectorized** Parquet decoder into the framework, so the
framework now owns all of `TimeType`'s Parquet read/write paths (schema, write,
row-based read, and now vectorized read).
- Adds a framework dispatch hook at the top of
`ParquetVectorUpdaterFactory.getUpdater` (before the `switch (typeName)`):
`ParquetTypeOps.getVectorUpdaterOrNull(sparkType, descriptor)`. A
framework-managed type returning a `Some` updater short-circuits the factory's
built-in cases; everything else returns `null` and falls through unchanged.
- Adds the trait extension point
`ParquetTypeOps.getVectorUpdater(descriptor): Option[ParquetVectorUpdater]`
(default `None`), symmetric with the existing `newConverter` row hook and
dispatched through the same `apply(dt)` registration.
- `TimeTypeParquetOps` owns the updater: a Scala `TimeVectorUpdater`
replacing the removed factory `TimeUpdater` (INT64 micros/nanos-of-day →
nanos-of-day, truncated to the requested precision).
- Unifies the read guard: `getVectorUpdater` validates the descriptor via a
newly-extracted `isCompatibleParquetType` — the same predicate the row path's
`requireCompatibleParquetType` now delegates to. Incompatible encodings (INT32
TIME(MILLIS), raw INT64, INT64 TIMESTAMP) return `None` and fall through to the
factory's clean `SchemaColumnConvertNotSupportedException` instead of silently
mis-reading. Both readers now accept/reject the identical encoding set.
- Truncation reuses the shared, table-backed
`DateTimeUtils.truncateTimeToPrecision` (the same call the row path uses) — no
duplicate factor table.
- Renames the stale `ParquetVectorUpdaterBenchmark` case label to
`TimeVectorUpdater (TimeType)`.
### Why are the changes needed?
The vectorized Parquet **decoder** was the last `TimeType` updater living
outside the framework: `ParquetVectorUpdaterFactory.getUpdater` had `instanceof
TimeType` arms in its INT64 case, so `ParquetTypeOps.isBatchReadSupported =
true` for `TimeType` was only safe because of out-of-framework factory code.
Moving it in makes that flag honest and lets future framework types supply
their own batch updater with no factory changes.
(`VectorizedColumnReader.isLazyDecodingSupported`'s TIME c [...]
### Does this PR introduce _any_ user-facing change?
No. `TimeVectorUpdater` performs the same conversion + precision truncation
as the removed factory `TimeUpdater`, and `isCompatibleParquetType` preserves
the existing accept/reject behavior on both readers.
### How was this patch tested?
- `TimeTypeParquetOpsSuite`: dispatch tests plus a vectorized reject test
(INT32 TIME(MILLIS) / raw INT64 / INT64 TIMESTAMP read as `TimeType` →
`getVectorUpdater` returns `None`).
- `ParquetIOSuite` TIME tests under `withAllParquetReaders` (vectorized +
row, dict on/off, MICROS/NANOS, `isAdjustedToUTC`, read-side precision
truncation) exercise the relocated decode end-to-end.
- `ParquetVectorizedSuite` passes unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56661 from stevomitric/stevomitric/parquet-tf-vectorized-read.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
...ParquetVectorUpdaterBenchmark-jdk21-results.txt | 1 +
...ParquetVectorUpdaterBenchmark-jdk25-results.txt | 1 +
.../ParquetVectorUpdaterBenchmark-results.txt | 1 +
.../parquet/ParquetVectorUpdaterFactory.java | 89 ++----------------
.../parquet/types/ops/ParquetTypeOps.scala | 69 +++++++++-----
.../parquet/types/ops/TimeTypeParquetOps.scala | 104 +++++++++++++++++----
.../datasources/parquet/ParquetIOSuite.scala | 73 +++++++++++++++
.../parquet/ParquetVectorUpdaterBenchmark.scala | 4 +-
.../parquet/ParquetVectorUpdaterSuite.scala | 28 ++++++
.../types/ops/TimeTypeParquetOpsSuite.scala | 42 ++++++++-
10 files changed, 285 insertions(+), 127 deletions(-)
diff --git
a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
index 79f08a7c511c..a8967151ec31 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
@@ -28,6 +28,7 @@ IntegerToLongUpdater 0
0
IntegerToDoubleUpdater 0 0
0 6120.9 0.2 1.0X
FloatToDoubleUpdater 0 0
0 2527.0 0.4 0.4X
DateToTimestampNTZUpdater 1 1
0 935.1 1.1 0.2X
+TimeVectorUpdater (TimeType) 1 1
0 1228.5 0.8 0.2X
DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 5823.3 0.2 0.9X
diff --git
a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
index 985b53c7d9fd..4303c196bfda 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
@@ -28,6 +28,7 @@ IntegerToLongUpdater 0
0
IntegerToDoubleUpdater 0 0
0 6568.8 0.2 1.3X
FloatToDoubleUpdater 0 0
0 3189.9 0.3 0.6X
DateToTimestampNTZUpdater 1 1
0 884.2 1.1 0.2X
+TimeVectorUpdater (TimeType) 1 1
0 1115.4 0.9 0.2X
DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 5089.5 0.2 1.0X
diff --git a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
index 95c02e4dd9e9..7930d1b63bcc 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
@@ -28,6 +28,7 @@ IntegerToLongUpdater 1
1
IntegerToDoubleUpdater 1 1
0 1556.5 0.6 1.2X
FloatToDoubleUpdater 1 1
0 1418.2 0.7 1.1X
DateToTimestampNTZUpdater 2 2
0 605.1 1.7 0.5X
+TimeVectorUpdater (TimeType) 1 1
0 942.2 1.1 0.7X
DowncastLongUpdater (INT64 -> Decimal(9,2)) 1 1
0 1287.2 0.8 1.0X
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
index 544a60a21222..b15777eacc52 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
@@ -34,6 +34,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils;
import org.apache.spark.sql.catalyst.util.RebaseDateTime;
import org.apache.spark.sql.execution.datasources.DataSourceUtils;
import
org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException;
+import
org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeOps$;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
import org.apache.spark.sql.types.*;
@@ -78,6 +79,13 @@ public class ParquetVectorUpdaterFactory {
return new NullTypeUpdater();
}
+ // Types Framework: a framework-managed type provides its own vectorized
updater.
+ ParquetVectorUpdater frameworkUpdater =
+ ParquetTypeOps$.MODULE$.getVectorUpdaterOrNull(sparkType, descriptor);
+ if (frameworkUpdater != null) {
+ return frameworkUpdater;
+ }
+
switch (typeName) {
case BOOLEAN -> {
if (sparkType == DataTypes.BooleanType) {
@@ -165,17 +173,6 @@ public class ParquetVectorUpdaterFactory {
return new LongUpdater();
} else if (canReadAsDecimal(descriptor, sparkType)) {
return new LongToDecimalUpdater(descriptor, (DecimalType) sparkType);
- } else if (sparkType instanceof TimeType &&
- isTimeTypeMatched(LogicalTypeAnnotation.TimeUnit.NANOS)) {
- // TIME(NANOS) is stored as nanoseconds since midnight, matching the
internal
- // representation, so no unit conversion is needed; the decoded
value is truncated to
- // the requested precision (consistent with the row-based
ParquetRowConverter path).
- return new TimeUpdater(((TimeType) sparkType).precision(), /*
fileStoresNanos = */ true);
- } else if (sparkType instanceof TimeType &&
- isTimeTypeMatched(LogicalTypeAnnotation.TimeUnit.MICROS)) {
- // TIME(MICROS) is converted to nanoseconds, then truncated to the
requested precision
- // (consistent with the row-based ParquetRowConverter path).
- return new TimeUpdater(((TimeType) sparkType).precision(), /*
fileStoresNanos = */ false);
}
}
case FLOAT -> {
@@ -922,76 +919,6 @@ public class ParquetVectorUpdaterFactory {
}
}
- // Reads an INT64 TIME column into the internal nanoseconds-since-midnight
representation and
- // truncates it to the requested TimeType precision. `fileStoresNanos`
selects the on-disk unit:
- // TIME(NANOS) stores nanos directly (identity), TIME(MICROS) stores micros
(converted to nanos).
- // Mirrors the row-based ParquetRowConverter path so the vectorized and
non-vectorized readers
- // agree on the decoded value, including when the requested precision is
lower than the on-disk
- // value's precision.
- // 10^k for k in [0, 9] (TimeType.NANOS_PRECISION), indexed by the
truncation scale
- // (NANOS_PRECISION - p), used to truncate a nanosecond TIME value to the
requested
- // fractional-second precision. Length - 1 equals TimeType.NANOS_PRECISION.
- private static final long[] TIME_TRUNCATION_FACTORS = {
- 1L, 10L, 100L, 1_000L, 10_000L, 100_000L,
- 1_000_000L, 10_000_000L, 100_000_000L, 1_000_000_000L
- };
-
- private static class TimeUpdater implements ParquetVectorUpdater {
- // The truncation step for the requested precision. The precision is
constant per column, so the
- // factor is looked up once here rather than recomputed per value via the
math.pow in
- // DateTimeUtils.truncateTimeToPrecision (this is the vectorized hot loop).
- private final long truncationFactor;
- private final boolean fileStoresNanos;
-
- TimeUpdater(int precision, boolean fileStoresNanos) {
- this.fileStoresNanos = fileStoresNanos;
- // scale = NANOS_PRECISION - precision; NANOS_PRECISION ==
factors.length - 1.
- int scale = TIME_TRUNCATION_FACTORS.length - 1 - precision;
- this.truncationFactor = TIME_TRUNCATION_FACTORS[scale];
- }
-
- private long toTruncatedNanos(long value) {
- long nanos = fileStoresNanos ? value :
DateTimeUtils.microsToNanos(value);
- // Equivalent to DateTimeUtils.truncateTimeToPrecision with the factor
hoisted.
- return (nanos / truncationFactor) * truncationFactor;
- }
-
- @Override
- public void readValues(
- int total,
- int offset,
- WritableColumnVector values,
- VectorizedValuesReader valuesReader) {
- valuesReader.readLongs(total, values, offset);
- for (int i = 0; i < total; i++) {
- values.putLong(offset + i, toTruncatedNanos(values.getLong(offset +
i)));
- }
- }
-
- @Override
- public void skipValues(int total, VectorizedValuesReader valuesReader) {
- valuesReader.skipLongs(total);
- }
-
- @Override
- public void readValue(
- int offset,
- WritableColumnVector values,
- VectorizedValuesReader valuesReader) {
- values.putLong(offset, toTruncatedNanos(valuesReader.readLong()));
- }
-
- @Override
- public void decodeSingleDictionaryId(
- int offset,
- WritableColumnVector values,
- WritableColumnVector dictionaryIds,
- Dictionary dictionary) {
- long value = dictionary.decodeToLong(dictionaryIds.getDictId(offset));
- values.putLong(offset, toTruncatedNanos(value));
- }
- }
-
private static class FloatUpdater implements ParquetVectorUpdater {
@Override
public void readValues(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
index 598ec75fb45f..4184efa82f84 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
@@ -19,13 +19,14 @@ package
org.apache.spark.sql.execution.datasources.parquet.types.ops
import java.time.ZoneId
+import org.apache.parquet.column.ColumnDescriptor
import org.apache.parquet.io.api.{Converter, RecordConsumer}
import org.apache.parquet.schema.Type
import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
-import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetToSparkSchemaConverter}
+import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetToSparkSchemaConverter, ParquetVectorUpdater}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, StructType,
TimestampLTZNanosType, TimestampNTZNanosType, TimeType}
@@ -40,13 +41,12 @@ import org.apache.spark.sql.types.{DataType, StructType,
TimestampLTZNanosType,
* - Schema conversion: Spark DataType -> Parquet schema type (write path)
* - Value write: writing values to Parquet RecordConsumer
* - Row-based read: creating Parquet converters for reading into InternalRow
- * - Type gates: declaring Parquet support (supportDataType) and the
vectorized-read
- * capability flag (isBatchReadSupported)
+ * - Vectorized read: the batch capability flag (isBatchReadSupported) and
the batch
+ * updater (getVectorUpdater)
+ * - Type gates: declaring Parquet support (supportDataType)
* - Schema clipping: declaring internal struct schema for column pruning
*
- * NOT yet on the trait (deferred to follow-ups): vectorized-read batch
updaters and
- * filter-pushdown predicates. Only the isBatchReadSupported capability gate
exists today;
- * the actual vectorized updater and filter predicate hooks are not
implemented here.
+ * NOT yet on the trait (deferred to follow-ups): filter-pushdown predicates.
*
* DISPATCH PATTERN: Framework FIRST at all integration sites. Each Parquet
infrastructure
* method wraps itself with:
@@ -167,23 +167,6 @@ private[parquet] trait ParquetTypeOps extends Serializable
{
*/
def supportDataType: Boolean = true
- /**
- * Whether vectorized (batch) reading is supported for this type.
- * Used by ParquetUtils.isBatchReadSupported. Default is false - types must
opt in
- * by overriding to true. When false, Spark uses the row-based read path
(newConverter)
- * which is always available.
- *
- * PRECONDITION: there is no framework vectorized-read hook yet, so
returning true is only
- * safe for a type the legacy Java vectorized path
(ParquetVectorUpdaterFactory /
- * VectorizedColumnReader) already handles. A new type that returns true
without that
- * legacy support would route into a factory that does not recognize it.
TimeType is safe
- * here precisely because the legacy path handles it; until the vectorized
integration
- * (follow-up) lands, other types should leave this false.
- *
- * @param sqlConf the active SQL configuration
- */
- def isBatchReadSupported(sqlConf: SQLConf): Boolean = false
-
// ==================== Schema Clipping (Struct-Backed Types)
====================
/**
@@ -200,6 +183,37 @@ private[parquet] trait ParquetTypeOps extends Serializable
{
* Primitive types return None (no sub-fields to clip).
*/
def parquetStructSchema: Option[StructType] = None
+
+ // ==================== Vectorized Read ====================
+
+ /**
+ * Whether vectorized (batch) reading is supported for this type.
+ * Used by ParquetUtils.isBatchReadSupported. Default is false - types must
opt in
+ * by overriding to true. When false, Spark uses the row-based read path
(newConverter)
+ * which is always available.
+ *
+ * A type that returns true must also supply a batch decoder via
[[getVectorUpdater]]
+ * (dispatched from ParquetVectorUpdaterFactory.getUpdater); otherwise the
vectorized factory
+ * would not recognize it. TimeType returns true and overrides
getVectorUpdater accordingly.
+ *
+ * @param sqlConf the active SQL configuration
+ */
+ def isBatchReadSupported(sqlConf: SQLConf): Boolean = false
+
+ /**
+ * The vectorized (batch) [[ParquetVectorUpdater]] for this type, or None to
fall back to the
+ * built-in `ParquetVectorUpdaterFactory`. Dispatched (Spark DataType ->
ops) at the top of
+ * `ParquetVectorUpdaterFactory.getUpdater`, before its built-in cases.
+ *
+ * This and [[isBatchReadSupported]] form a two-way contract: returning Some
here without also
+ * returning true from isBatchReadSupported leaves the vectorized path
unreachable (the row path
+ * is used), while returning true from isBatchReadSupported without
supplying an updater here
+ * (and without legacy factory support for the type) routes into a factory
that does not
+ * recognize the type. A framework type that opts into vectorized reads must
do both.
+ *
+ * @param descriptor the Parquet column descriptor being read
+ */
+ def getVectorUpdater(descriptor: ColumnDescriptor):
Option[ParquetVectorUpdater] = None
}
/**
@@ -226,4 +240,13 @@ private[parquet] object ParquetTypeOps {
case _ => None
}
}
+
+ /**
+ * Java-friendly entry point for `ParquetVectorUpdaterFactory`: the
framework vectorized
+ * updater for `dt`, or null if `dt` is not framework-managed (so the
factory falls through
+ * to its built-in updaters).
+ */
+ private[parquet] def getVectorUpdaterOrNull(
+ dt: DataType, descriptor: ColumnDescriptor): ParquetVectorUpdater =
+ apply(dt).flatMap(_.getVectorUpdater(descriptor)).orNull
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
index f155d35378a8..c868716e7fc9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
import org.apache.parquet.io.api.{Converter, RecordConsumer}
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
@@ -26,7 +27,8 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.errors.QueryExecutionErrors
-import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetPrimitiveConverter}
+import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetPrimitiveConverter, ParquetVectorUpdater,
VectorizedValuesReader}
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, TimeType}
@@ -104,6 +106,21 @@ case class TimeTypeParquetOps(t: TimeType) extends
ParquetTypeOps {
// ==================== Vectorized Read ====================
override def isBatchReadSupported(sqlConf: SQLConf): Boolean = true
+
+ // Only a canonical INT64 TIME(MICROS)/TIME(NANOS) column can be
vectorized-decoded as TimeType.
+ // Return None for anything else (INT32 TIME(MILLIS), raw INT64, INT64
TIMESTAMP, ...) so the
+ // factory falls through to its clean
SchemaColumnConvertNotSupportedException instead of silently
+ // mis-reading. This is the same compatibility check the row path uses
+ // (requireCompatibleParquetType), unifying the read guard across both
readers. The on-disk unit
+ // (MICROS vs NANOS) and the requested precision drive the conversion +
truncation.
+ override def getVectorUpdater(descriptor: ColumnDescriptor):
Option[ParquetVectorUpdater] = {
+ val parquetType = descriptor.getPrimitiveType
+ if (TimeTypeParquetOps.isCompatibleParquetType(parquetType)) {
+ Some(new TimeVectorUpdater(t.precision,
TimeTypeParquetOps.isNanosTime(parquetType)))
+ } else {
+ None
+ }
+ }
}
private[ops] object TimeTypeParquetOps {
@@ -122,33 +139,82 @@ private[ops] object TimeTypeParquetOps {
}
/**
- * Validates that a Parquet field can be decoded as TimeType. TimeType is
written as INT64 with
- * TIME(MICROS, isAdjustedToUTC=false) for precision 0..6 and TIME(NANOS,
isAdjustedToUTC=false)
- * for precision 7..9. On read, any INT64 TIME(MICROS) or TIME(NANOS) column
is accepted
- * regardless of the isAdjustedToUTC flag: Spark's zone-less TimeType
decodes the raw
- * time-of-day identically either way, matching the legacy
ParquetRowConverter guard (see
- * SPARK-57416). Any other encoding (raw INT64, INT32 TIME(MILLIS), INT64
TIMESTAMP(_),
- * decimal-annotated, etc.) cannot be decoded as TimeType - throw the same
error as the legacy
- * ParquetRowConverter path so reads fail loudly instead of silently
misinterpreting bytes.
+ * Whether a Parquet field is a canonical INT64 TIME(MICROS) (precision
0..6) or TIME(NANOS)
+ * (precision 7..9) column - the only encodings Spark can decode as
TimeType. The isAdjustedToUTC
+ * flag is intentionally ignored: Spark's TimeType is zone-less local time,
so the raw
+ * time-of-day decodes identically either way, matching the legacy
ParquetRowConverter guard
+ * (see SPARK-57416). Any other encoding (raw INT64, INT32 TIME(MILLIS),
INT64 TIMESTAMP(_),
+ * decimal-annotated, etc.) is incompatible. Shared by both the row-based
and vectorized read
+ * paths so they accept/reject the same set.
*/
- private[ops] def requireCompatibleParquetType(
- sparkType: TimeType, parquetType: Type): Unit = {
- val ok = parquetType.isPrimitive &&
+ private[ops] def isCompatibleParquetType(parquetType: Type): Boolean =
+ parquetType.isPrimitive &&
parquetType.asPrimitiveType.getPrimitiveTypeName == INT64 &&
(parquetType.getLogicalTypeAnnotation match {
case t: LogicalTypeAnnotation.TimeLogicalTypeAnnotation =>
- // Accept both MICROS (precision 0..6) and NANOS (precision 7..9),
and both
- // isAdjustedToUTC=false and =true. Spark's TimeType is zone-less
local time, so the
- // UTC-adjustment flag carries no extra information on read: the raw
time-of-day value
- // decodes identically either way. Mirroring the legacy
ParquetRowConverter guard keeps
- // the framework row-based read path consistent with the legacy and
vectorized readers.
- // SPARK-57416.
t.getUnit == TimeUnit.MICROS || t.getUnit == TimeUnit.NANOS
case _ => false
})
- if (!ok) {
+
+ /**
+ * Validates that a Parquet field can be decoded as TimeType on the
row-based path, throwing the
+ * same error as the legacy ParquetRowConverter so incompatible reads fail
loudly instead of
+ * silently misinterpreting bytes. See [[isCompatibleParquetType]] for the
accepted encodings.
+ */
+ private[ops] def requireCompatibleParquetType(
+ sparkType: TimeType, parquetType: Type): Unit = {
+ if (!isCompatibleParquetType(parquetType)) {
throw QueryExecutionErrors.cannotCreateParquetConverterForDataTypeError(
sparkType, parquetType.toString)
}
}
}
+
+/**
+ * Vectorized (batch) updater for TimeType: reads an INT64 TIME column into
the internal
+ * nanos-of-day representation and truncates it to the requested precision.
`fileStoresNanos`
+ * selects the on-disk unit - TIME(NANOS) stores nanos (identity),
TIME(MICROS) stores micros
+ * (converted to nanos). Mirrors the row-based newConverter path so the
vectorized and row-based
+ * readers agree; replaces the former
`ParquetVectorUpdaterFactory.TimeUpdater`, now owned by the
+ * type's ops.
+ */
+private[ops] class TimeVectorUpdater(precision: Int, fileStoresNanos: Boolean)
+ extends ParquetVectorUpdater {
+ private def toTruncatedNanos(value: Long): Long = {
+ val nanos = if (fileStoresNanos) value else
DateTimeUtils.microsToNanos(value)
+ // Same conversion + truncation as the row-based newConverter path, via
the shared
+ // (table-backed) DateTimeUtils.truncateTimeToPrecision, so both readers
stay in lock-step.
+ DateTimeUtils.truncateTimeToPrecision(nanos, precision)
+ }
+
+ override def readValues(
+ total: Int,
+ offset: Int,
+ values: WritableColumnVector,
+ valuesReader: VectorizedValuesReader): Unit = {
+ valuesReader.readLongs(total, values, offset)
+ var i = 0
+ while (i < total) {
+ values.putLong(offset + i, toTruncatedNanos(values.getLong(offset + i)))
+ i += 1
+ }
+ }
+
+ override def skipValues(total: Int, valuesReader: VectorizedValuesReader):
Unit =
+ valuesReader.skipLongs(total)
+
+ override def readValue(
+ offset: Int,
+ values: WritableColumnVector,
+ valuesReader: VectorizedValuesReader): Unit =
+ values.putLong(offset, toTruncatedNanos(valuesReader.readLong()))
+
+ override def decodeSingleDictionaryId(
+ offset: Int,
+ values: WritableColumnVector,
+ dictionaryIds: WritableColumnVector,
+ dictionary: Dictionary): Unit = {
+ val value = dictionary.decodeToLong(dictionaryIds.getDictId(offset))
+ values.putLong(offset, toTruncatedNanos(value))
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
index 6b41da841e56..4439062ccfdc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
@@ -2009,6 +2009,79 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
+ test("SPARK-55444: vectorized read rejects an incompatible encoding
requested as TimeType") {
+ // TimeTypeParquetOps.getVectorUpdater returns None for any encoding other
than INT64
+ // TIME(MICROS/NANOS), so the vectorized factory falls through to a clean
+ // SchemaColumnConvertNotSupportedException (surfaced as FAILED_READ_FILE)
instead of
+ // silently mis-decoding - e.g. running readLongs over an INT32 column.
This is the
+ // end-to-end counterpart of the unit-level reject assertions in
TimeTypeParquetOpsSuite,
+ // pinned on the actual vectorized reader (the path the descriptor guard
protects).
+ val readSchema = new StructType().add("c", TimeType())
+ // (Parquet column definition, writer of one matching-primitive value for
that column)
+ val cases: Seq[(String, SimpleGroup => Unit)] = Seq(
+ ("required int32 c(TIME(MILLIS,false));", _.add(0, 0)), // wrong
primitive
+ ("required int64 c;", _.add(0, 0L)), // no TIME
annotation
+ ("required int64 c(TIMESTAMP(MICROS,false));", _.add(0, 0L)) // wrong
annotation
+ )
+ for ((column, addValue) <- cases) {
+ val schema = MessageTypeParser.parseMessageType(s"message root {\n
$column\n}")
+ withTempDir { dir =>
+ val tablePath = new
Path(s"${dir.getCanonicalPath}/incompatible.parquet")
+ val writer = createParquetWriter(schema, tablePath, dictionaryEnabled
= false)
+ (0 until 10).foreach { _ =>
+ val record = new SimpleGroup(schema)
+ addValue(record)
+ writer.write(record)
+ }
+ writer.close
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ val e = intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(tablePath.toString).collect()
+ }
+ assert(e.getCondition ===
"FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH",
+ s"[$column] expected a clean column-type-mismatch, got
${e.getCondition}")
+ }
+ }
+ }
+ }
+
+ test("SPARK-55444: vectorized read of a nullable TIME(NANOS) column
(single-value path)") {
+ // A nullable (OPTIONAL) column interleaves def-levels, splitting value
runs into sub-batch
+ // lengths that drive VectorizedRleValuesReader down the runLen == 1 path
- i.e.
+ // TimeVectorUpdater.readValue / decodeSingleDictionaryId. The existing
REQUIRED TIME tests
+ // only exercise the bulk readValues path, so this closes the single-value
decode gap (and,
+ // via withAllParquetReaders, cross-checks the row-based reader on nulls).
+ val schema = MessageTypeParser.parseMessageType(
+ """message root {
+ | optional int64 time_nanos(TIME(NANOS,false));
+ |}""".stripMargin)
+ val readSchema = new StructType().add("time_nanos",
TimeType(TimeType.NANOS_PRECISION))
+ val lt = LocalTime.of(23, 59, 59, 123456789)
+
+ for (dictEnabled <- Seq(true, false)) {
+ withTempDir { dir =>
+ val tablePath = new
Path(s"${dir.getCanonicalPath}/times_nullable.parquet")
+ val numRecords = 100
+ val writer = createParquetWriter(schema, tablePath, dictionaryEnabled
= dictEnabled)
+ (0 until numRecords).foreach { i =>
+ val record = new SimpleGroup(schema)
+ // Every 7th row is null (the field is simply omitted): interleaves
null/value runs.
+ if (i % 7 != 0) record.add(0, localTime(23, 59, 59, 123456, 789))
+ writer.write(record)
+ }
+ writer.close
+
+ withAllParquetReaders {
+ val df = spark.read.schema(readSchema).parquet(tablePath.toString)
+ assertResult(df.schema)(readSchema)
+ val expected = (0 until numRecords).map { i => if (i % 7 == 0)
Row(null) else Row(lt) }
+ checkAnswer(df, expected)
+ }
+ }
+ }
+ }
+
// Deterministic INT32 sample shared by the INT32 widening tests below.
Mixes sign,
// zero, and MIN/MAX boundaries to catch sign-extension and precision
regressions.
private def widenSampleAt(i: Int): Int = i % 5 match {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterBenchmark.scala
index a0664fa5780c..f620d5ee4239 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterBenchmark.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterBenchmark.scala
@@ -45,7 +45,7 @@ import org.apache.spark.sql.types._
* (Boolean, Byte, Short, Integer, Long, Float, Double, Binary).
* B. Type-converting Updaters -- per-row read+convert+write loops.
* `IntegerToLong`, `IntegerToDouble`, `FloatToDouble`,
`DateToTimestampNTZ`,
- * `DowncastLong`, `LongAsNanos`.
+ * `DowncastLong`, `TimeVectorUpdater`.
* C. Rebase Updaters -- date/timestamp legacy-calendar rebase variants.
* `IntegerWithRebase` (DATE), `LongWithRebase` (TIMESTAMP_MICROS),
* `LongAsMicros`, `DateToTimestampNTZWithRebase`, `LongAsMicrosRebase`.
@@ -264,7 +264,7 @@ object ParquetVectorUpdaterBenchmark extends BenchmarkBase {
TimestampNTZType,
descriptor(PrimitiveTypeName.INT32, LogicalTypeAnnotation.dateType()),
longVec, intBytes)
- addReadValuesCase(benchmark, "LongAsNanosUpdater (TimeType)",
+ addReadValuesCase(benchmark, "TimeVectorUpdater (TimeType)",
TimeType(),
descriptor(PrimitiveTypeName.INT64,
LogicalTypeAnnotation.timeType(false,
LogicalTypeAnnotation.TimeUnit.MICROS)),
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterSuite.scala
index 13e9376fb6fa..85a41d5ed296 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterSuite.scala
@@ -28,6 +28,7 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import
org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
import org.apache.spark.sql.types._
@@ -493,4 +494,31 @@ class ParquetVectorUpdaterSuite extends SparkFunSuite {
-86_400_000_000L) // 1969-12-31 00:00:00 UTC
assert(readViaDateToTimestampNTZUpdater(input) === expected)
}
+
+ test("SPARK-55444: factory rejects an incompatible TIME encoding (no silent
mis-read)") {
+ // getVectorUpdaterOrNull returns null for any encoding other than INT64
TIME(MICROS/NANOS),
+ // so ParquetVectorUpdaterFactory.getUpdater - the entry point the
vectorized reader calls -
+ // falls through to a clean SchemaColumnConvertNotSupportedException
rather than silently
+ // decoding (e.g. readLongs over an INT32 column). Guarding the factory
entry point directly
+ // means a future broad TimeType arm / catch-all that reintroduced the
silent mis-read would
+ // fail here, not just at the framework-hook unit level.
+ val incompatible = Seq(
+ // INT32 TIME(MILLIS): wrong primitive width.
+ Types.primitive(PrimitiveTypeName.INT32, Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.timeType(false,
LogicalTypeAnnotation.TimeUnit.MILLIS))
+ .named("col"),
+ // raw INT64 with no TIME annotation.
+ Types.primitive(PrimitiveTypeName.INT64,
Repetition.OPTIONAL).named("col"),
+ // INT64 carrying a TIMESTAMP (not TIME) annotation.
+ Types.primitive(PrimitiveTypeName.INT64, Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.timestampType(false,
LogicalTypeAnnotation.TimeUnit.MICROS))
+ .named("col")
+ )
+ incompatible.foreach { pt =>
+ val desc = new ColumnDescriptor(Array("col"), pt, 0, 1)
+ intercept[SchemaColumnConvertNotSupportedException] {
+ newFactory(desc).getUpdater(desc, TimeType())
+ }
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
index fa44297103ea..363a21e7dec9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
@@ -17,16 +17,20 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import org.apache.parquet.column.ColumnDescriptor
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.{INT32, INT64}
import org.apache.parquet.schema.Type.Repetition.REQUIRED
import org.apache.spark.{SparkFunSuite, SparkRuntimeException}
-import org.apache.spark.sql.types.TimeType
+import org.apache.spark.sql.types.{IntegerType, TimeType}
/**
- * Unit tests for [[TimeTypeParquetOps.requireCompatibleParquetType]].
+ * Unit tests for [[TimeTypeParquetOps]]'s Parquet read guards - both the
row-based
+ * [[TimeTypeParquetOps.requireCompatibleParquetType]] and the vectorized-read
+ * getVectorUpdater / getVectorUpdaterOrNull dispatch, which share the same
+ * compatible-encoding check so the two readers accept and reject the same set.
*
* TimeType is stored in Parquet as INT64 TIME(MICROS, isAdjustedToUTC=false)
for precision
* 0..6 and INT64 TIME(NANOS, isAdjustedToUTC=false) for precision 7..9. The
read-path guard
@@ -129,6 +133,40 @@ class TimeTypeParquetOpsSuite extends SparkFunSuite {
// the TIME annotation; the raw-INT64 / TIMESTAMP / DECIMAL / group tests
// above already exercise the !isPrimitive and "non-TIME annotation"
branches.
+ // ---------- vectorized read updater ----------
+
+ private def timeColumn(unit: TimeUnit): ColumnDescriptor =
+ new ColumnDescriptor(
+ Array("c"),
+ Types.primitive(INT64,
REQUIRED).as(LogicalTypeAnnotation.timeType(false, unit)).named("c"),
+ 0, 0)
+
+ test("getVectorUpdater returns a framework updater for TimeType (micros and
nanos)") {
+
assert(TimeTypeParquetOps(timeMicros).getVectorUpdater(timeColumn(TimeUnit.MICROS)).isDefined)
+
assert(TimeTypeParquetOps(timeNanos).getVectorUpdater(timeColumn(TimeUnit.NANOS)).isDefined)
+ // Java-friendly companion entry point used by ParquetVectorUpdaterFactory.
+ assert(ParquetTypeOps.getVectorUpdaterOrNull(timeMicros,
timeColumn(TimeUnit.MICROS)) != null)
+ }
+
+ test("getVectorUpdater returns None for incompatible encodings (clean
reject, vectorized path)") {
+ val int32Millis = Types.primitive(INT32, REQUIRED)
+ .as(LogicalTypeAnnotation.timeType(false, TimeUnit.MILLIS)).named("c")
+ val rawInt64 = Types.primitive(INT64, REQUIRED).named("c")
+ val int64Timestamp = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(false,
TimeUnit.MICROS)).named("c")
+ // None -> the factory falls through to a clean
SchemaColumnConvertNotSupportedException,
+ // matching the row-path reject set (requireCompatibleParquetType),
instead of silently
+ // mis-decoding (e.g. readLongs over an INT32 column).
+ Seq(int32Millis, rawInt64, int64Timestamp).foreach { field =>
+ val descriptor = new ColumnDescriptor(Array("c"), field, 0, 0)
+
assert(TimeTypeParquetOps(timeMicros).getVectorUpdater(descriptor).isEmpty)
+ }
+ }
+
+ test("getVectorUpdaterOrNull returns null for non-framework types") {
+ assert(ParquetTypeOps.getVectorUpdaterOrNull(IntegerType, null) == null)
+ }
+
// ---------- helper ----------
private def assertRejects(sparkType: TimeType, field: Type): Unit = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
