[Task]: Use Timestamp logical type in Beam SQL #40454

Activity

1. ahmedabu98

   added

   task

   awaiting triage

   on Oct 7, 2026
2. github-actions

   added

   P2

   on Oct 7, 2026
3. sun-9545sunoj commented on Oct 16, 2026

   

   sun-9545sunoj

   on Oct 16, 2026

   More actions

   Hi @ahmedabu98, I'd like to pick this up. I looked into the current state first and wanted to check the approach with you before starting, because the precondition in the description ("once Beam SQL / Calcite can preserve microsecond precision end-to-end") doesn't seem to hold yet.

   Current state

   • CalciteUtils.TIMESTAMP is still FieldType.DATETIME, which is Joda and millisecond-precision (CalciteUtils.java:119-122).
   • add Timestamp.MICROS for iceberg timestamptz #39592 added partial Timestamp logical type support. toSqlTypeName maps it to Calcite TIMESTAMP, and BeamCalcRel reads and writes it as java.time.Instant.
   • The planner already declares precision 6 for TIMESTAMP (BeamRelDataTypeSystem.getDefaultPrecision / getMaxPrecision). At runtime, though, Calcite's generated code represents TIMESTAMP as a long of epoch millis. Because of that, BeamCalcRel.timestampToCalciteMillis rejects sub-millisecond values with an UnsupportedOperationException.
   • Several components assume TIMESTAMP values are DATETIME: windowing (BeamAggregationRel, TVFSlidingWindowFn, BeamTableFunctionScanRel), MATCH_RECOGNIZE (NFA, CEPLiteral), the Kafka and GenerateSequence table providers, DataCatalog SchemaUtils, and the SQL expansion service. Roughly 27 SQL test files depend on it too.

   Changing the default mapping on its own would make any TIMESTAMP column fail on sub-millisecond input. It would also change the output schema of existing queries (DATETIME → beam:logical_type:timestamp), and that change would be visible to Python/YAML SqlTransform users as well.

   Possible approaches

   1. Microsecond runtime representation first, then flip the default (proposed)

   • PR 1: Carry TIMESTAMP values through the Calcite runtime at microsecond precision. This means updating the Beam↔Calcite conversions in BeamCalcRel and auditing the Calcite builtins that operate on timestamps (comparisons, datetime arithmetic, EXTRACT, FLOOR/CEIL, TIMESTAMPADD/TIMESTAMPDIFF, casts), plus windowing and CEP.
   • PR 2: Switch CalciteUtils.TIMESTAMP to Timestamp.MICROS, migrate the internal DATETIME usages, and update tests and CHANGES.md.
   • This satisfies the issue's precondition. It's also the largest effort and touches generated-code paths.

   2. Opt-in flag

   • Add a planner or pipeline option that maps TIMESTAMP to Timestamp.MICROS and keep DATETIME as the default.
   • Lets users and connectors (e.g. Iceberg timestamptz) adopt the logical type gradually without breaking existing pipelines. Sub-millisecond values would still be rejected until approach 1 lands, so this is mostly an incremental step.

   3. Flip the default now, millisecond semantics

   • Map TIMESTAMP to Timestamp.MICROS immediately and migrate all internal DATETIME usages, while keeping the millisecond runtime and the sub-millisecond rejection.
   • I'd lean against this one. It's a breaking schema change that doesn't yet deliver microsecond precision.

   Questions

   1. Which direction do you prefer? I'd suggest 1, possibly with 2 as a transitional step.
   2. Is the schema change for existing TIMESTAMP columns acceptable with a CHANGES.md note, or should the old behavior stay available behind an option?
   3. Are there connectors (Iceberg, BigQuery, Kafka) or cross-language/YAML consumers you'd like covered in the first PR?

   Happy to adjust based on your feedback. Thanks!