Fix TIMESTAMP_MILLIS scaling for dictionary-encoded columns in Vectorized Spark Reader
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 16h
- Merged PRs (30d)
- 129
Description
### Apache Iceberg version
1.11.0
### Query engine
Spark
### Problem Context
Apache Iceberg supports reading Parquet files with `TIMESTAMP_MILLIS` annotations by converting the millisecond values to Iceberg's internal microsecond representation ($\times 1000$). While this scaling logic works correctly for `PLAIN` encoded pages (via `TimestampMillisReader`), it is completely bypassed when a column is entirely dictionary-encoded (e.g., columns with duplicate or low-cardinality values, like a batch extraction timestamp).
When this optimization occurs, the values are displayed as incorrect dates in the year 1970 because raw millisecond values are treated as microseconds.
### Root Cause Analysis
In `VectorizedArrowReader#read`, the engine checks whether a column segment produces a dictionary-encoded vector:
```
boolean dictEncoded = vectorizedColumnIterator.producesDictionaryEncodedVector();
if (vectorizedColumnIterator.hasNext()) {
if (dictEncoded) {
vectorizedColumnIterator.dictionaryBatchReader().nextBatch(vec, -1, nullabilityHolder);
} else {
switch (readType) { ... }
}
}
```
If `dictEncoded` is true, the reader completely bypasses the type-specific switch statement—which normally maps to `ReadType.TIMESTAMP_MILLIS` and uses the correct `TimestampMillisReader`. Instead, it shortcuts by populating a generic `IntVector` with raw dictionary IDs and attaches the raw Parquet `Dictionary` object straight to the `VectorHolder` returned to Spark.
When Spark eventually decodes these IDs via` Dictionary#decodeToLong(id)`, it receives unscaled milliseconds from the raw Parquet metadata, resulting in corrupted timestamps.
### Solution
The fix intercepts the raw Parquet `Dictionary` inside `VectorizedArrowReader#setRowGroupInfo` right after initialization.
If the column's modern `LogicalTypeAnnotation` indicates it is a `TIMESTAMP` with `TimeUnit.MILLIS` precision, the dictionary is wrapped in a proxy wrapper. This proxy intercepts calls to` decodeToLong(int id)` and scales the returned values to microseconds.
This approach resolves the bug gracefully:
It fixes the issue on the optimized dictionary-passthrough path.
### Changes
`VectorizedArrowReader.java`: Added a check using ### LogicalTypeAnnotation to detect `TimeUnit.MILLIS` timestamps inside `setRowGroupInfo`.
Wrapped this.dictionary in an anonymous proxy class that applies the `* 1000L bit-shift` multiplier inside `decodeToLong`.
### How to Test
Write an Iceberg table where a timestamp column contains identical values (forcing Parquet's writer optimization to select `PLAIN_DICTIONARY` encoding instead of `PLAIN`).
Read the table using Spark with vectorization enabled (`spark.sql.iceberg.vectorized_read.enabled=true`).
Before Fix: Values display as 1970-01-21...
After Fix: Values accurately decode to their current, modern calendar dates.
### Exact code change
From :
```
@Override
public void setRowGroupInfo(PageReadStore source, Map metadata) {
ColumnChunkMetaData chunkMetaData = metadata.get(ColumnPath.get(columnDescriptor.getPath()));
this.dictionary =
vectorizedColumnIterator.setRowGroupInfo(
source.getPageReader(columnDescriptor),
!ParquetUtil.hasNonDictionaryPages(chunkMetaData));
}
```
To :
```
@Override
public void setRowGroupInfo(PageReadStore source, Map metadata) {
ColumnChunkMetaData chunkMetaData = metadata.get(ColumnPath.get(columnDescriptor.getPath()));
this.dictionary =
vectorizedColumnIterator.setRowGroupInfo(
source.getPageReader(columnDescriptor),
!ParquetUtil.hasNonDictionaryPages(chunkMetaData));
boolean isTimestampMillis = false;
if (columnDescriptor != null && columnDescriptor.getPrimitiveType() != null) {
org.apache.parquet.schema.LogicalTypeAnnotation annotation =
columnDescriptor.getPrimitiveType().getLogicalTypeAnnotation();
if (annotation instanceof org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampAnnotation =
(org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) annotation;
isTimestampMillis = timestampAnnotation.getUnit() == org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.MILLIS;
}
}
if (this.dictionary != null && isTimestampMillis) {
final Dictionary backingDictionary = this.dictionary;
this.dictionary = new Dictionary(backingDictionary.getEncoding()) {
@Override
public long decodeToLong(int id) {
return backingDictionary.decodeToLong(id) * 1000L;
}
@Override
public int decodeToInt(int id) { return backingDictionary.decodeToInt(id); }
@Override
public float decodeToFloat(int id) { return backingDictionary.decodeToFloat(id); }
@Override
public double decodeToDouble(int id) { return backingDictionary.decodeToDouble(id); }
@Override
public org.apache.parquet.io.api.Binary decodeToBinary(int id) { return backingDictionary.decodeToBinary(id); }
@Override
public int getMaxId() { return backingDictionary.getMaxId(); }
};
}
}
```
Contributor guide
Research direction
Start in VectorizedArrowReader.java at setRowGroupInfo, then trace how dictionaryBatchReader and the VectorHolder are used by the vectorized Spark reader. Create an Iceberg table with identical timestamp values so Parquet uses dictionary encoding, read it with spark.sql.iceberg.vectorized_read.enabled=true, and verify TIMESTAMP_MILLIS values remain modern dates rather than appearing in 1970.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 74/100