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

[AbdelrahmanElhawary]

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<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));

    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(); }
        };
    }
  }