This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 c7b981d01177 [SPARK-55444][SQL] Add Framework support for the
vectorized lazy-decoding gate for TimeType
c7b981d01177 is described below
commit c7b981d0117792bd7a0117f9a22f25fcb293f789
Author: Stevo Mitric <[email protected]>
AuthorDate: Mon Jul 6 22:52:34 2026 +0800
[SPARK-55444][SQL] Add Framework support for the vectorized lazy-decoding
gate for TimeType
## What changes were proposed in this pull request?
Moves the last hardcoded `TimeType` check out of the vectorized reader into
the Types Framework. `VectorizedColumnReader.isLazyDecodingSupported` decided
lazy dictionary decoding for TIME columns via the annotation-keyed
`ParquetVectorUpdaterFactory.isTimeTypeMatched(MICROS|NANOS)`; it now
dispatches framework-first (like `getVectorUpdater`):
- `ParquetTypeOps.supportsLazyDictionaryDecoding(descriptor): Boolean`
(default `false`, opt-in — matching the sibling `isBatchReadSupported`) + Java
entry point `supportsLazyDictionaryDecodingOrNull` (null for non-framework
types).
- `TimeTypeParquetOps` overrides it to `false`: the updater does per-value
micros→nanos conversion + truncation to the requested precision, which lazy
dictionary decoding would bypass.
- `VectorizedColumnReader.isLazyDecodingSupported` dispatches
framework-first; the two `isTimeTypeMatched` arms are removed.
- Removes the now-dead `ParquetVectorUpdaterFactory.isTimeTypeMatched` and
its unused import.
### Why are the changes needed?
`isTimeTypeMatched` was the last hardcoded `TimeType` reference in the
vectorized path after the updater moved into the framework. Routing the
lazy-decoding gate through `ParquetTypeOps` finishes that migration: a
framework type now fully owns its vectorized-read behavior (both the updater
and the lazy-decoding capability) with no `TimeType`-specific code left in
`VectorizedColumnReader` / `ParquetVectorUpdaterFactory`.
The `false` default is fail-safe: a framework type that opts into
vectorized reads (`isBatchReadSupported = true`) with a per-value updater but
forgets to override this would, under a `true` default, silently get lazy
dictionary decoding and return wrong results; `false` makes the worst case a
missed optimization instead. A type whose updater is a plain identity copy can
override to `true` to regain the optimization.
### Does this PR introduce _any_ user-facing change?
No. Reading `TimeType` from Parquet behaves exactly as before; only the
internal routing of the lazy-decoding decision changes.
### How was this patch tested?
`TimeTypeParquetOpsSuite` (unit tests for `supportsLazyDictionaryDecoding`
/ `supportsLazyDictionaryDecodingOrNull`), and `ParquetIOSuite` TIME tests
under `withAllParquetReaders` (dict on/off) exercise the dictionary path
end-to-end.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56991 from stevomitric/stevomitric/parquet-tf-lazy-decoding.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../parquet/ParquetVectorUpdaterFactory.java | 6 -----
.../parquet/VectorizedColumnReader.java | 16 +++++++-----
.../parquet/types/ops/ParquetTypeOps.scala | 29 ++++++++++++++++++++++
.../parquet/types/ops/TimeTypeParquetOps.scala | 7 ++++++
.../types/ops/TimeTypeParquetOpsSuite.scala | 14 +++++++++++
5 files changed, 60 insertions(+), 12 deletions(-)
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 b15777eacc52..155d70685950 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
@@ -24,7 +24,6 @@ import org.apache.parquet.schema.LogicalTypeAnnotation;
import
org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation;
import
org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation;
import
org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
-import
org.apache.parquet.schema.LogicalTypeAnnotation.TimeLogicalTypeAnnotation;
import
org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
import
org.apache.parquet.schema.LogicalTypeAnnotation.UnknownLogicalTypeAnnotation;
import org.apache.parquet.schema.PrimitiveType;
@@ -250,11 +249,6 @@ public class ParquetVectorUpdaterFactory {
annotation.getUnit() == unit;
}
- boolean isTimeTypeMatched(LogicalTypeAnnotation.TimeUnit unit) {
- return logicalTypeAnnotation instanceof TimeLogicalTypeAnnotation
annotation &&
- annotation.getUnit() == unit;
- }
-
boolean isUnsignedIntTypeMatched(int bitWidth) {
return logicalTypeAnnotation instanceof IntLogicalTypeAnnotation
annotation &&
!annotation.isSigned() && annotation.getBitWidth() == bitWidth;
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
index 01f4573557dc..63aa0951313f 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
@@ -39,6 +39,7 @@ import
org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.spark.SparkUnsupportedOperationException;
+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.DataType;
import org.apache.spark.sql.types.Decimal;
@@ -150,6 +151,14 @@ public class VectorizedColumnReader {
private boolean isLazyDecodingSupported(
PrimitiveType.PrimitiveTypeName typeName,
DataType sparkType) {
+ // Types Framework: a framework-managed type decides whether its
dictionary-encoded column can
+ // be lazily decoded (false when its vectorized updater does per-value
processing that lazy
+ // decoding would bypass). Non-framework types fall through to the
built-in cases below.
+ Boolean frameworkDecision =
+
ParquetTypeOps$.MODULE$.supportsLazyDictionaryDecodingOrNull(sparkType,
descriptor);
+ if (frameworkDecision != null) {
+ return frameworkDecision;
+ }
boolean isSupported = false;
// Don't use lazy dictionary decoding if the column needs extra
processing: upcasting or date
// rebasing.
@@ -166,13 +175,8 @@ public class VectorizedColumnReader {
}
case INT64: {
boolean isDecimal = sparkType instanceof DecimalType;
- // TIME columns (both MICROS and NANOS) need per-value processing in
the updater: a unit
- // conversion for MICROS and/or truncation to the requested precision.
Lazy dictionary
- // decoding would bypass the updater, so it must be disabled for them.
boolean needsUpcast = (isDecimal &&
!DecimalType.is64BitDecimalType(sparkType)) ||
- updaterFactory.isTimestampTypeMatched(TimeUnit.MILLIS) ||
- updaterFactory.isTimeTypeMatched(TimeUnit.MICROS) ||
- updaterFactory.isTimeTypeMatched(TimeUnit.NANOS);
+ updaterFactory.isTimestampTypeMatched(TimeUnit.MILLIS);
boolean needsRebase =
updaterFactory.isTimestampTypeMatched(TimeUnit.MICROS) &&
!"CORRECTED".equals(datetimeRebaseMode);
isSupported = !needsUpcast && !needsRebase &&
!needsDecimalScaleRebase(sparkType);
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 4184efa82f84..63209572c94d 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
@@ -214,6 +214,25 @@ private[parquet] trait ParquetTypeOps extends Serializable
{
* @param descriptor the Parquet column descriptor being read
*/
def getVectorUpdater(descriptor: ColumnDescriptor):
Option[ParquetVectorUpdater] = None
+
+ /**
+ * Whether a dictionary-encoded column of this type can be lazily
dictionary-decoded on the
+ * vectorized path (the dictionary is attached to the column vector and
decoded on read) rather
+ * than eagerly decoding every value up front. Consulted (Spark DataType ->
ops) by
+ * `VectorizedColumnReader.isLazyDecodingSupported`.
+ *
+ * Default is false - a type must opt in by overriding to true, matching the
opt-in stance of
+ * [[isBatchReadSupported]]. This is deliberately fail-safe: lazy decoding
bypasses the
+ * per-value work a vectorized updater ([[getVectorUpdater]]) may do (unit
conversion,
+ * truncation, rebasing), so a type that opts into vectorized reads but
forgets to opt out of
+ * lazy decoding when it needs per-value processing would silently return
wrong results. With a
+ * false default the worst case is a missed lazy-decode optimization, not
incorrect data. Types
+ * whose updater is a plain identity copy (no per-value processing) should
override to true to
+ * regain the optimization.
+ *
+ * @param descriptor the Parquet column descriptor being read
+ */
+ def supportsLazyDictionaryDecoding(descriptor: ColumnDescriptor): Boolean =
false
}
/**
@@ -249,4 +268,14 @@ private[parquet] object ParquetTypeOps {
private[parquet] def getVectorUpdaterOrNull(
dt: DataType, descriptor: ColumnDescriptor): ParquetVectorUpdater =
apply(dt).flatMap(_.getVectorUpdater(descriptor)).orNull
+
+ /**
+ * Java-friendly entry point for `VectorizedColumnReader`: whether `dt`'s
dictionary-encoded
+ * column can be lazily decoded, or null if `dt` is not framework-managed
(so the caller keeps
+ * its built-in decision).
+ */
+ private[parquet] def supportsLazyDictionaryDecodingOrNull(
+ dt: DataType, descriptor: ColumnDescriptor): java.lang.Boolean =
+ apply(dt).map(o =>
java.lang.Boolean.valueOf(o.supportsLazyDictionaryDecoding(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 c868716e7fc9..8e6fcee925d7 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
@@ -121,6 +121,13 @@ case class TimeTypeParquetOps(t: TimeType) extends
ParquetTypeOps {
None
}
}
+
+ // TIME decoding converts micros->nanos (for TIME(MICROS)) and truncates to
the requested
+ // precision per value in the updater; lazy dictionary decoding would attach
the raw dictionary
+ // and skip that processing, so it must be disabled. This matches the
fail-safe trait default,
+ // but is stated explicitly because TimeType opts into vectorized reads
(isBatchReadSupported =
+ // true) and the per-value work is exactly why lazy decoding is unsafe here.
+ override def supportsLazyDictionaryDecoding(descriptor: ColumnDescriptor):
Boolean = false
}
private[ops] object TimeTypeParquetOps {
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 363a21e7dec9..70091c6379bc 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
@@ -167,6 +167,20 @@ class TimeTypeParquetOpsSuite extends SparkFunSuite {
assert(ParquetTypeOps.getVectorUpdaterOrNull(IntegerType, null) == null)
}
+ test("supportsLazyDictionaryDecoding is false for TimeType (updater does
per-value work)") {
+ // TIME decoding does per-value micros->nanos + truncation in the updater,
so lazy dictionary
+ // decoding (which would bypass the updater) must stay disabled on the
vectorized path.
+ assert(!TimeTypeParquetOps(timeMicros)
+ .supportsLazyDictionaryDecoding(timeColumn(TimeUnit.MICROS)))
+ assert(!TimeTypeParquetOps(timeNanos)
+ .supportsLazyDictionaryDecoding(timeColumn(TimeUnit.NANOS)))
+ // Java-friendly companion entry point used by VectorizedColumnReader:
FALSE for TimeType,
+ // null for a non-framework type (so the reader keeps its built-in
lazy-decoding decision).
+ assert(ParquetTypeOps.supportsLazyDictionaryDecodingOrNull(
+ timeMicros, timeColumn(TimeUnit.MICROS)) === java.lang.Boolean.FALSE)
+ assert(ParquetTypeOps.supportsLazyDictionaryDecodingOrNull(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]
