Repository navigation
[parquet] Write millisecond timestamps as TIMESTAMP_MICROS when Iceberg compatibility is enabled - #10601
[parquet] Write millisecond timestamps as TIMESTAMP_MICROS when Iceberg compatibility is enabled#106010dunay0 wants to merge 3 commits into
Conversation
…ompatibility is enabled Iceberg requires INT64 TIMESTAMP_MICROS for timestamptz columns, but Paimon writes TIMESTAMP_LTZ(<=3) as TIMESTAMP_MILLIS, which Athena/Trino reject. When metadata.iceberg.storage is not 'disabled', write TIMESTAMP_LTZ(<=3) as TIMESTAMP_MICROS. TIMESTAMP (without time zone) and the default behavior are unchanged. Make the remaining read paths follow each file's Parquet timestamp unit, so MILLIS and MICROS files can be mixed in one table: - footer statistics are decoded using the unit in the file - lazy dictionary decoding is skipped when the file unit differs from the unit of Paimon's compact timestamp vector
| Assertions.assertThat(schema.getType("local_timestamp").getLogicalTypeAnnotation()) | ||
| .isEqualTo( | ||
| LogicalTypeAnnotation.timestampType( | ||
| false, LogicalTypeAnnotation.TimeUnit.MILLIS)); |
There was a problem hiding this comment.
[P1] Encode Iceberg timestamp columns as MICROS too
Iceberg requires INT64 TIMESTAMP_MICROS for both timestamp and timestamptz , not only the zoned type. Paimon's Iceberg schema conversion publishes TIMESTAMP(3) as timestamp , so retaining MILLIS here still produces a Parquet file that strict Iceberg readers reject whenever the table has an ordinary TIMESTAMP(3) column. Please apply the Iceberg-gated MICROS encoding to TIMESTAMP_WITHOUT_TIME_ZONE as well and assert that both columns use MICROS; the non-Iceberg default can remain unchanged.
There was a problem hiding this comment.
When metadata.iceberg.storage is not disabled, TIMESTAMP(<=3) is now written as TIMESTAMP_MICROS, the same as TIMESTAMP_LTZ(<=3). The non-Iceberg default is unchanged.
| reader.forEachRemaining(row -> rows.add(serializer.copy(row))); | ||
| } | ||
| // the predicate only prunes row groups, so the row group holding the match is kept | ||
| Assertions.assertThat(rows).hasSize(3); |
There was a problem hiding this comment.
[P2] Preserve predicate pushdown for the new MICROS representation
This does not currently exercise predicate pushdown: ParquetFilters.timestampColumn expects MILLIS for a precision-3 field, rejects this file's MICROS annotation, and ParquetFilters.convert silently drops the predicate. As a result, all row groups in newly written Iceberg files are scanned for timestamp filters. Please convert timestamp literals using the actual file annotation (while retaining the adjustedToUTC compatibility check) and add a multi-row-group test that verifies a nonmatching row group is pruned.
There was a problem hiding this comment.
ParquetFilters now converts literals using the file's annotated unit and keeps the adjustedToUTC check. A precision <= 3 field reads a MICROS value as floor(v / 1000) ms, so comparisons are widened to the micros of the literal millisecond: = becomes [ms*1000, ms*1000+999].
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. The LTZ physical-encoding correction addresses the Iceberg timestamptz mapping. Current-reader verification passed: 212 Parquet tests with the regular Maven lifecycle; eight additional 12,000-row combinations covering Iceberg/dictionary on/off, batch 17/1024, negative/null timestamps, precision 3/6/9, nested fields, and dictionary → plain-page transitions; independent Iceberg 1.8.1 reads of the emitted files also preserved values. A timestamp-pruning regression remains below.
Production rollout condition: the exact base revision c9a36a8 is also incompatible with the new dictionary-encoded LTZ(3) files, despite already having LongTimestampUpdater. A base/head probe read the same value as 1691827200123000 ms on base versus the correct 1691827200123 ms on head, and base footer stats were wrong by the same factor. Existing Iceberg-enabled tables switch encoding automatically when a writer is upgraded, so the caveat should cover those tables as well as newly enabled ones. All Paimon readers and compactors need the read-side fixes before any writer emits MICROS; reverting the metadata option will not make those files readable by old readers. Reader rollback requires rewriting the new files or restoring a compatible snapshot. Please document that sequence explicitly.
CI is still red: four failed jobs stopped on dependency-transfer connection resets; Flink 1 Common also reports LargeDataITCase.testPartitionedSpillableWriteBuffer (integer-only schema). I have not attributed that failure to this change; the failed jobs still need a successful verification run.
| localZonedTimestampType.getPrecision(), | ||
| repetition, | ||
| true, | ||
| localZonedTimestampMillisAsMicros) |
There was a problem hiding this comment.
[P2] Preserve timestamp predicate pruning for the new MICROS encoding
This changes mirrored LTZ(3) files to MICROS, but ParquetFilters.timestampColumn() still requires MILLIS for logical precision <= 3. It rejects these columns and convert() silently drops the predicate. I verified both an equality and a 10-second range on 120,000 monotonically increasing LTZ(3) events with 64 KiB row groups: both files contain 15 groups, but MILLIS reads only group 7 and emits 7,686 rows, while the new MICROS file gets NoOpFilter, reads all 15 groups and emits 120,000 rows (actual matches remain 1/10). This loses within-file time pruning on a common selective-query path. Please make both timestamp column validation and literal conversion follow the physical file unit, and add a multi-row-group test that asserts actual exclusion; the new single-group predicate test only proves rows are retained.
There was a problem hiding this comment.
- both column validation and literal conversion now follow the file's unit.
- The rollout section is now in
docs/docs/iceberg/data-types.mdand the PR description.
…CROS files - Iceberg requires microseconds for timestamp as well as timestamptz, so the Iceberg-gated MICROS encoding now also applies to TIMESTAMP(<=3). - Timestamp predicate pushdown follows the unit of the file. A precision <= 3 field reads a MICROS value as floor(v / 1000) milliseconds, so comparisons are widened to the micros of the literal millisecond; NOT EQUAL and IN are not pushed down for such columns. - Document that readers and compaction jobs have to be upgraded before writers.
… overflow Statistics of a millisecond field over a microsecond file are floored to the milliseconds the field reads, and literals outside the INT64 microsecond range are no longer pushed down. Tests cover sub-millisecond values at a row-group boundary and the metadata.iceberg.storage values that enable the encoding.
Purpose
Linked issue: close #10600
Iceberg requires
INT64 TIMESTAMP_MICROSfortimestampandtimestamptz, but Paimon writesTIMESTAMP(<=3)andTIMESTAMP_LTZ(<=3)asTIMESTAMP_MILLIS, so strict Iceberg readers such as Athena reject the data files.metadata.iceberg.storageis notdisabled, writeTIMESTAMP(<=3)andTIMESTAMP_LTZ(<=3)asTIMESTAMP_MICROS. The default behavior is unchanged. The key is read literally becausepaimon-formatdoes not depend onpaimon-core.floor(v / 1000)milliseconds, so comparisons are widened to the micros of the literal millisecond and stats are floored to milliseconds.!=andINare not pushed down for such columns, nor are literals outside the INT64 micros range.Rollout: existing Iceberg-enabled tables switch to MICROS as soon as their writer is upgraded. Paimon readers without these read fixes read the new files 1000x too large without an error, so every reader and compaction job must be upgraded before any writer. Disabling the Iceberg option afterwards does not help old readers; rolling them back requires rewriting the new files or restoring an earlier snapshot. This is documented in
docs/docs/iceberg/data-types.md.Tests
ParquetFormatReadWriteTest#testIcebergTimestampMicros: footer units, and values and stats read back from a dictionary-encoded file. Fails without either read-side fix.ParquetFormatReadWriteTest#testIcebergTimestampMicrosFilterPushdown: equality on a file with many row groups prunes the row groups without a match, for both types. Fails without theParquetFilterschange.ParquetFormatReadWriteTest#testMillisFieldOnMicrosFileWithSubMillis:TIMESTAMP(3)fields over aTIMESTAMP(6)file with sub-millisecond values keep the first row of a row group under=and<=, and get floored stats.ParquetFormatReadWriteTest#testIcebergStorageOptionEnablesMicros: whichmetadata.iceberg.storagevalues enable MICROS.ParquetFiltersTest#testMillisTimestampOnMicrosFile: the widened bounds for each operator, and no pushdown for!=,IN, or an out-of-range literal. This replaces theTIMESTAMP(3)over MICROS case oftestTimestampFileUnitMismatchCannotPushDown([parquet] Avoid timestamp predicate pushdown for incompatible file schemas #8797), which asserted no pushdown; the nanos-over-MICROS case is kept.All Parquet tests in
paimon-formatpass.