apache / apache/iceberg

Fix TIMESTAMP_MILLIS scaling for dictionary-encoded columns in Vectorized Spark Reader

Open
#17,135 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.