(magnolify-beam) Map Instant to Beam's portable Timestamp logical type

[shnapz]

Beam 2.76.0 changed how IcebergIO represents timestamptz, which broke magnolify.beam.logical.millis. Investigating that turned up two further problems that predate the Beam change. This PR adds Timestamp-based mappings under logical.timestamp, groups the existing encodings under logical.compat, deprecates the bare precision objects in favour of an explicit choice between the two, and deprecates logical.sql.

No schema changes on upgrade. The bare logical.{millis,micros,nanos} objects keep their 0.9.7 behavior exactly; they are deprecated aliases of compat.*. Adopting the new encoding is opt-in, one import at a time. See Migration.

What broke

Beam #39344 (in 2.76.0) changed Iceberg's timestamptz → Beam schema mapping from FieldType.DATETIME to FieldType.logicalType(Timestamp.MICROS). From Beam's CHANGES.md:

[IcebergIO] Reading a timestamptz column will now return a Timestamp.MICROS Beam logical type to preserve microseconds (the old Beam Schema.FieldType#DATETIME primitive type truncates past milliseconds). This may break existing streaming read pipelines. It also breaks Python reads when a timestamptz column is present. Use pipeline option --updateCompatibilityVersion=2.75.0 (or any older version) to keep the old behavior.

FieldType.DATETIME stores org.joda.time.Instant; Timestamp.MICROS stores java.time.Instant. logical.millis's RowField[java.time.Instant] was built as a conversion layer over RowField[joda.Instant], so its FromT was joda.Instant. Reading an IcebergIO Row on 2.76.0 therefore hit the unchecked cast in RowField#fromAny and failed:

java.lang.ClassCastException: class java.time.Instant cannot be cast to class org.joda.time.Instant
  at magnolify.beam.RowField$FromWord$$anon$3.from(RowType.scala:96)
  at magnolify.beam.RowField.fromAny(RowType.scala:69)

Note the scope: the commit touches only sdks/java/io/iceberg/**, and only the read direction. Iceberg writes still accept DATETIME (IcebergUtils.java:77), and no other IO was changed. That bounds what actually had to move.

Two pre-existing bugs found along the way

Both are independent of the Beam 2.76.0 change and are fixed/flagged here.

1. logical.sql.rfSqlInstant throws at write time. SqlTypes.TIMESTAMP is not a distinct type — it is new MicrosInstant() (SqlTypes.java:43), whose own Javadoc says it "should never be used in a native Java context," and whose toBaseType throws AssertionError when getNano() % 1000 != 0. logical.micros carries a comment explaining it avoids MicrosInstant for precisely this reason; logical.sql used it anyway. Now covered by a test that asserts the throw.

2. The test suite could not observe precision handling. shared's arbInstant is Gen.posNum[Long].map(Instant.ofEpochMilli), and Gen.posNum is size-bounded rather than range-bounded (sized(n => c.choose(one, max(fromInt(n), one))), ScalaCheck Gen.scala:1387-1391). With the default size of 100, that generates 100 values, all in the first 100 ms of 1970. Every RowField.id-based instant mapping round-tripped vacuously, so neither bug above was detectable, and a naive RowField.id[Instant] mapping onto Timestamp.MILLIS/MICROS would also have passed while throwing in production.

Changes

logical.timestamp.{millis,micros,nanos} — new

Instant maps to Beam's portable Timestamp logical type at the named precision. Timestamp#toBaseType rejects instants carrying finer precision than the type declares (checkState, Timestamp.java:123) rather than truncating, so these mappings truncate on write via a shared helper:

private def tsInstant(ts: Timestamp, unit: ChronoUnit): RowField[jt.Instant] = {
  implicit val base: RowField[jt.Instant] =
    RowField.id[jt.Instant](_ => FieldType.logicalType(ts))
  RowField.from[jt.Instant](identity)(_.truncatedTo(unit))
}

Discarding excess precision matches what the non-instant mappings at each precision already do. Here joda.Instant and joda.DateTime derive from the java.time base, the reverse of the compat direction.

Timestamp is Beam's go-forward instant type: portable (beam:logical_type:timestamp:v1, which Python's Timestamp now maps to), precision-parameterized, and lossless where the other options are not — FieldType.DATETIME truncates past millis and is joda-bound, and MicrosInstant throws.

logical.compat.{millis,micros,nanos} — new name for the existing encodings

	compat (= 0.9.7)	timestamp
millis	FieldType.DATETIME (joda.Instant)	logicalType(Timestamp.MILLIS)
micros	INT64 (micros since epoch)	logicalType(Timestamp.MICROS)
nanos	logicalType(NanosInstant)	logicalType(Timestamp.NANOS)

compat is not a deprecated holding pen — it is the magnolify-side counterpart to Beam's --updateCompatibilityVersion flag, and for several IOs the only thing that works at all. The flag gates only the read direction (icebergTypeToBeamFieldType and friends take an updateCompatibilityVersion; beamFieldTypeToIcebergFieldType does not), so a pipeline pinned below 2.76.0 gets FieldType.DATETIME back from IcebergIO — which only compat.millis can read. Pair the flag with compat, or omit both; setting the flag while using timestamp.* is the one broken combination.

It is also required for connectors that hardcode FieldType.DATETIME regardless of the flag. As of Beam 2.76.0, within sdks/java/io that is jdbc, google-cloud-platform, clickhouse, delta, hcatalog, iceberg, singlestore and amazon-web-services2; DATETIME is additionally hardcoded by core and by the arrow, avro, sql and sql-datacatalog extensions.

That list was built at the v2.76.0 tag restricted to /src/main/, counting only code that produces a DATETIME field. Connectors that merely accept one are excluded: CsvIO's hits are membership in VALID_FIELD_TYPE_SET plus a parse consumer, and ProtoSchemaTranslator.java:92 states that protobuf Timestamp cannot be translated to DATETIME at all.

One caveat on how to read it: it enumerates connectors that name FieldType.DATETIME. Beam's schema inference also maps any joda Instant field to DATETIME (FieldTypeDescriptors.java:53), so a connector whose element type has joda fields produces it without naming it — KafkaIO's KafkaSourceDescriptor (:35,59,69) is one.

On the name. compat rather than joda because only one of the three is joda-backed: compat.micros is a raw INT64 and compat.nanos is NanosInstant, which is java.time.Instant-backed. The three share no representation, only the fact that this is what 0.9.7 emitted, so the grouping is named for the compatibility it provides. timestamp is named after the Beam type it produces.

Bare logical.{millis,micros,nanos} — deprecated, behavior unchanged

Each extends the same private[logical] trait as its compat counterpart, so they cannot drift:

@deprecated("Renamed to `compat.millis` … This object is unchanged …", "0.9.8")
object millis extends MillisCompat

This is the part that matters most for safety. Repurposing these names would have been a silent change: millis.rfInstantMillis keeps its exact signature (RowField[java.time.Instant]), so downstream code would compile unchanged and write different bytes, with no tooling signal. Given that the original bug reached a worker as a ClassCastException rather than failing at compile or submit time, shipping a second silent behavior change seemed like the wrong trade. Now the only signal is a deprecation warning that names both alternatives.

Three tests pin each bare object to its 0.9.7 FieldType, and a fourth asserts schema equality against the matching compat object so a future edit to one grouping and not the other fails.

The encoding you pick determines which Beam IOs you can write to

This is the real constraint in the PR, and the reason the new encodings are opt-in rather than default. All verified at the v2.76.0 tag. These are Beam's IOs consuming a PCollection<Row> — magnolify's own avro, bigquery and parquet modules target GenericRecord/TableRow/Parquet directly and never see a Beam Schema, so they are unaffected:

	IcebergIO	BigQueryIO	Avro ext	managed JDBC	Kafka JSON	Kafka AVRO	Beam SQL
timestamp.millis (p3)	✗	✗	✗	✗	✗	✗	✓ †
timestamp.micros (p6)	✓	✗	✗	✗	✗	✗	✓ †
timestamp.nanos (p9)	✗	✓	✓	✗	✗	✓	✓ †
compat.millis (DATETIME)	✓	✓	✓	✓	✓	✓	✓

† accepted at any precision, but Beam SQL round-trips timestamps through millis (BeamCalcRel.java:463), silently dropping anything finer.

if (precision == Timestamp.MICROS.getArgument()) { type = Types.TimestampType.withZone(); }
else { throw new UnsupportedOperationException("Unsupported Timestamp precision: " + precision); }

IcebergUtils.java:227-234 — not gated by any compatibility flag.

• BigQueryIO and the Avro extension require 9 and throw otherwise: BigQueryUtils.java:591-596, BeamRowToStorageApiProto.java:252-260, AvroUtils.java:1227-1232. Kafka AVRO goes through the same Avro check (KafkaWriteSchemaTransformProvider.java:201).
• The managed JDBC sinks have no Timestamp branch. Unknown logical types fall back to the base type (JdbcUtil.java:336-338), and Timestamp's base is a ROW, which is not a writable JDBC type — RuntimeException("ROW in schema is not supported while writing").
• Kafka JSON fails late: RowJson recurses into the ROW base type, finds INT64/INT16, passes schema validation, then throws ClassCastException per element (RowJson.java:176-180, :593).

A note on weight: Iceberg is the only case anyone hits today. The JDBC and Kafka rows are reachable rather than exercised — a schema transform needs a schema-bearing PCollection<Row>, and since RowCoder is a SchemaCoder (PCollection.java:318-319), the usual setCoder(RowCoder.of(rowType.schema)) idiom produces one, which saveAsManaged(sink, …) will hand to any managed sink. But nothing routes to those sinks that way in practice; Scio's own JDBC support never touches Beam Row. They are listed because they establish that no Timestamp precision is universal, which is the argument for keeping the new encodings opt-in.

Two consequences: no timestamp.* precision reaches every sink, and compat.millis is the only Instant encoding every schema-aware Beam IO accepts. Records needing multiple destinations should use compat.millis or per-destination RowTypes. timestamp.nanos being Iceberg-rejected is not a regression — NanosInstant was never writable there either (it is not a PassThroughLogicalType and is absent from BEAM_LOGICAL_TYPES_TO_ICEBERG_TYPES, so it hits throw new RuntimeException("Unsupported Beam logical type …") at IcebergUtils.java:236).

timestamp.millis has no write destination at all, and is kept for symmetry and for reading data that is already precision 3 — a BigQuery TIMESTAMP(12) column under --picosecondTimestampMapping=MILLIS (BigQueryUtils.java:474-493).

Documented on the objects themselves and in docs/beam.md.

Reads tolerate a precision mismatch

Verified by running it, not by reading source — two RowTypes over one case class, each reading the other's output:

timestamp.millis.from(microsRow) = OK -> 1970-01-01T00:16:40.123456Z   (full micros, no error)
timestamp.nanos.from(microsRow)  = OK -> 1970-01-01T00:16:40.123456Z
compat.millis.from(microsRow)    = THREW ClassCastException: java.time.Instant
                                   cannot be cast to org.joda.time.Instant

A Timestamp field surfaces as a java.time.Instant at any precision, and truncation lives only in to, so a reader whose declared precision differs from the data silently returns whatever was stored. Only a compat.* reader against Timestamp data fails loudly. Worth knowing because a green read proves less than it appears to.

Reads are also where the groupings are least interchangeable, since most IOs still produce DATETIME. Beam's BigQueryIO is the clearest case: precision 9 on write, but DATETIME on read for an ordinary TIMESTAMP column. No single import round-trips it.

NonInstantTypes.scala and CompatTypes.scala — new

LocalTime, LocalDateTime, and Duration are represented identically regardless of instant encoding, because Beam offers no joda schema type for them at all — SqlTypes.TIME/DATETIME and NanosDuration are the only mappings and all return java.time types. Those mappings live in three private[logical] traits in NonInstantTypes.scala, shared by all three groupings. CompatTypes.scala holds the three 0.9.7 instant encodings, shared by compat.* and the deprecated bare objects.

Separate files are required: tlFatalWarnings rejects traits declared inside package objects. This mirrors parquet/logical/TimeTypes.scala.

logical.sql — deprecated

Three of its four members (DATE, TIME, DATETIME) are the same LogicalType instances logical.date and the precision objects already use; the fourth throws. Deprecated rather than removed, per review preference — removal in a later release.

Docs

docs/beam.md's "Time and dates" section described the pre-change model and recommended logical.sql without qualification. Rewritten around the two groupings, with the per-IO matrix, a compat section and an Iceberg section. docs/mapping.md is unchanged: java.time.Instant and org.joda.time.DateTime still reach DATETIME, INT64 and ROW, since all three encodings remain available.

Migration

Before	After
any logical.{millis,micros,nanos}._	no change required — same schema, same bytes; deprecation warning asks you to pick a grouping
logical.millis._ and want the warning gone, no behavior change	logical.compat.millis._
logical.millis._ reading IcebergIO on Beam ≥ 2.76	logical.timestamp.micros._ — this is the ClassCastException above
logical.millis._ writing to Iceberg, want microseconds	logical.timestamp.micros._
logical.millis._ with --updateCompatibilityVersion below 2.76.0	logical.compat.millis._ — the flag and compat go together
logical.micros._ reading IcebergIO on Beam ≥ 2.76	logical.timestamp.micros._
logical.sql._	logical.date._ plus a timestamp.* or compat.* object

Switching from compat.X to timestamp.X changes the wire format, so existing data needs rewriting or a compat reader.

Binary compatibility

This targets 0.9.8, so tlBaseVersion stays at 0.9 and MiMa runs for real in CI against every 0.9.x artifact — no manual baseline pinning, and the guarantee stays enforced for every future commit rather than resting on a one-off command:

sbt 'show beam/mimaPreviousArtifacts' beam/mimaReportBinaryIssues
# => Set(magnolify-beam:0.9.0 … 0.9.7)   (8 artifacts)
# => no binary issues

Keeping the bare objects as objects that inherit from private[logical] traits is what makes this work: members moved into an inherited trait keep their signatures and their owning class, so the 0.9 call sites still link.

An earlier revision of this PR bumped tlBaseVersion to 0.10. That is no longer needed — the change is source- and binary-compatible, so early-semver puts it in the 0.9 line, and staying there is what keeps MiMa non-vacuous. Note build.sbt:185 still carries tlVersionIntroduced := Map("3" -> "0.10.0") from #1258; that predates this PR and is a separate question for whoever cuts the release with Scala 3 support.

Testing

beam 55/55 on 2.13 and 2.12; avro 40/40 and parquet 216/216 unaffected; scalafmt clean; site/mdoc compiles all 17 docs files.

26 new tests:

• 14 targeted assertions — boundary-value truncation and the exact logical type and precision for each timestamp.* object; the FieldType for each compat.* object (previously untested, so representation changes could regress silently); the 0.9.7 FieldType for each deprecated bare object plus schema equality against compat; and one asserting logical.sql throws AssertionError on a sub-microsecond instant, documenting the deprecation rationale.
• 12 round-trip properties across timestamp.* and compat.* for both java.time and joda records, over a generator described below.

Closing the precision blind spot

Because Gen.posNum is size-bounded (see above), the shared arbInstant yields 100 values inside the first 100 ms of 1970 — and everything derived from it collapses too: arbLocalTime always has hour, minute and second 0; arbLocalDateTime is always 1970-01-01. Every pre-existing instant round-trip property therefore held vacuously: nothing was ever truncated and the epoch boundary was never crossed. That is why neither the Timestamp-throws behavior nor the logical.sql bug was detectable.

Widening arbInstant itself is out of scope here and cannot be done alone — it lives in shared, and TimeSpec.scala:37-39 asserts java → unit → joda → unit → java == identity, which is false for micros/nanos (the joda leg is millis-only) and passes only because of the weak generator. Fixing the generator requires fixing that expectation in the same commit, plus avro and parquet's round-trips. Scoped out in BEAM_TIMESTAMP_REMAINING_SCOPE.md. Instead this adds a beam-local generator spanning 1900–2100 at full nanosecond precision, and asserts the contract that actually holds:

private def truncatesTo(rt: RowType[JavaInstant], unit: ChronoUnit): Prop =
  Prop.forAll(preciseInstants) { i =>
    rt.from(rt.to(JavaInstant(i))).i == i.truncatedTo(unit)
  }

Note this is deliberately not roundtrip(i) == i. Each precision discards excess, so the real invariant is truncation to the declared unit. Instant.truncatedTo floors, which is what makes pre-epoch instants correct against Timestamp's non-negative-subseconds representation.

All mappings satisfy it, including pre-epoch — so the negative-timestamp hazard called out in Timestamp's Javadoc does not bite here, and there is now a regression test saying so.

Both new assertion styles verified by mutation

• Precision assertions check getLogicalType.getArgument, not getIdentifier. Timestamp.IDENTIFIER is one shared constant across MILLIS/MICROS/NANOS (Timestamp.java:61), so an identifier-only assertion passes for any precision. Pointing timestamp.millis at Timestamp.NANOS while keeping ChronoUnit.MILLIS passes an identifier check but fails the precision check.
• Changing timestamp.millis's truncation unit to ChronoUnit.SECONDS is falsified in one generated case (2065-06-17T23:01:44.265996624Z).

Deliberate non-changes

logical.date left alone. Folding LocalDate into the precision objects would make the common import date._ + import timestamp.millis._ pairing an ambiguous-implicit error. Note that neither avro nor parquet has a logical.date object — both put LocalDate in base implicits (AvroType.scala:351, ParquetField.scala:658), so aligning with house style means moving it there, which is a separate change.

Global arbInstant not widened to nanosecond precision. See "Closing the precision blind spot" above — beam's gap is closed with a module-local generator. Doing it globally means fixing shared/TimeSpec's java→joda→java identity expectation and auditing avro and parquet's precision expectations, all in one commit. The same vacuous-property problem exists in those modules today. Scoped out in BEAM_TIMESTAMP_REMAINING_SCOPE.md, along with the Long-nanos range limits in the nanos joda paths, the missing non-instant precision coverage, and the open question of whether micros should be built on MicrosInstant + truncation for broader IO support.

nanos.rfJodaLocalDateTimeMicros keeps its misleading name. Misnamed since 0.9.7; renaming breaks source compatibility. Flagged with a comment.

Joda support retained. It was added deliberately across six modules (#1234), and 10 Beam IO connectors still emit FieldType.DATETIME, so a joda-backed reader remains necessary. Dropping it from beam alone would make it the only module without joda support.

Not validated against a running pipeline. Every read/write claim above comes from reading Beam source at v2.76.0 or from local RowType experiments. No IcebergIO pipeline was run. BEAM_TIMESTAMP_REMAINING_SCOPE.md §6 lists what that leaves open, including a Beam-side gap in BeamRowWrapper that would only show up on partitioned timestamptz writes.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 95.82%. Comparing base (5fb4845) to head (26881aa).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1392      +/-   ##
==========================================
+ Coverage   95.78%   95.82%   +0.04%     
==========================================
  Files          58       60       +2     
  Lines        2183     2205      +22     
  Branches      173      181       +8     
==========================================
+ Hits         2091     2113      +22     
  Misses         92       92              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[shnapz]

propose to remove it because it mixes 3 different precisions, and we already implement these types

[clairemcginty]

Makes sense for IcebergIO; my concern is that Row is the output type for many Beam transformations, for example beam sql and all the new managed IOs (which includes IcebergIO). Are Joda timestamp types used in any of those other IOs/will we break those use cases?

I slightly prefer the package name magnolify.beam.logical.joda over magnolify.beam.logical.legacy but it's not a blocking concern

[shnapz]

@clairemcginty Beam's change is narrow - only changed what IcebergIO reads so timestampz comes back as Timestamp.MICROS instead of DATETIME. Iceberg write is still accepting DATETIME, no other IOs were touched. Beam SQL is fine.

But you are spot on about this PR's blast radius - if we introduce Beam's Timestamp these managed IOs are affected (on write):

• Iceberg needs precision 6 (micros)
• BQ, Kafka Avro need precision 9 (nanos)
• Kafka JSON, postgres, mysq, sqlserver don't handle Timetamp precision at all.

But Iceberg is the only "live" case, because all other sinks are reachable through classic Scio IOs (that don't touchBeam Row, e.g. scio-jdbc). So I changed this, it's now opt-in:

• logical.timestamp.{millis,micros,nanos} — portable Timestamp encodings (new)
• logical.compat.{millis,micros,nanos} — the 0.9 encodings (your legacy, renamed)
• bare logical.{millis,micros,nanos} — unchanged 0.9 behavior, deprecated aliases of compat.*

Upgrading to 0.10 changes no schema; you get a deprecation warning asking you to pick a grouping

[shnapz]

we don't break it anymore