AbdelrahmanElhawary opened a new issue, #17135:
URL: https://github.com/apache/iceberg/issues/17135
### 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<ColumnPath,
ColumnChunkMetaData> 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<ColumnPath,
ColumnChunkMetaData> metadata) {
ColumnChunkMetaData chunkMetaData =
metadata.get(ColumnPath.get(columnDescriptor.getPath()));
this.dictionary =
vectorizedColumnIterator.setRowGroupInfo(
source.getPageReader(columnDescriptor),
!ParquetUtil.hasNonDictionaryPages(chunkMetaData));
// Modern, non-deprecated check using LogicalTypeAnnotation
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(); }
};
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
