From a08e811fde5831f37b006c012fd094986680bcbe Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Wed, 16 Sep 2026 15:59:38 -0400 Subject: [PATCH 1/7] (magnolify-beam) Map Instant to Beam's portable Timestamp logical type --- .../beam/logical/NonInstantTypes.scala | 94 +++++++++ .../magnolify/beam/logical/package.scala | 183 ++++++++++-------- .../scala/magnolify/beam/RowTypeSuite.scala | 94 ++++++++- build.sbt | 2 +- 4 files changed, 294 insertions(+), 79 deletions(-) create mode 100644 beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala diff --git a/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala b/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala new file mode 100644 index 00000000..4ff6dea6 --- /dev/null +++ b/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala @@ -0,0 +1,94 @@ +/* + * Copyright 2024 Spotify AB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package magnolify.beam.logical + +import magnolify.beam.RowField +import magnolify.shared.Time._ +import org.apache.beam.sdk.schemas.Schema.FieldType +import org.apache.beam.sdk.schemas.logicaltypes +import org.joda.time as joda + +import java.time as jt + +// Mappings that Beam represents identically regardless of the instant encoding, shared +// between the default precision objects and their `legacy` counterparts. + +private[logical] trait MillisNonInstant { + implicit val rfLocalTimeMillis: RowField[jt.LocalTime] = + RowField.from[Int](millisToLocalTime)(millisFromLocalTime) + implicit val rfJodaLocalTimeMillis: RowField[joda.LocalTime] = + RowField.from[Int](millisToJodaLocalTime)(millisFromJodaLocalTime) + + implicit val rfLocalDateTimeMillis: RowField[jt.LocalDateTime] = + RowField.id[jt.LocalDateTime](_ => FieldType.logicalType(new logicaltypes.DateTime())) + implicit val rfJodaLocalDateTimeMillis: RowField[joda.LocalDateTime] = + RowField.from[jt.LocalDateTime](ldt => millisToJodaLocalDateTime(millisFromLocalDateTime(ldt)))( + ldt => millisToLocalDateTime(millisFromJodaLocalDateTime(ldt)) + ) + + implicit val rfDurationMillis: RowField[jt.Duration] = + RowField.from[Long](millisToDuration)(millisFromDuration) + implicit val rfJodaDurationMillis: RowField[joda.Duration] = + RowField.from[Long](millisToJodaDuration)(millisFromJodaDuration) +} + +private[logical] trait MicrosNonInstant { + implicit val rfLocalTimeMicros: RowField[jt.LocalTime] = + RowField.from[Long](microsToLocalTime)(microsFromLocalTime) + // joda.LocalTime only has millisecond resolution, so excess precision is discarded + implicit val rfJodaLocalTimeMicros: RowField[joda.LocalTime] = + RowField.from[Long](microsToJodaLocalTime)(microsFromJodaLocalTime) + + implicit val rfLocalDateTimeMicros: RowField[jt.LocalDateTime] = + RowField.from[Long](microsToLocalDateTime)(microsFromLocalDateTime) + // joda.LocalDateTime has millisecond precision, excess precision discarded + implicit val rfJodaLocalDateTimeMicros: RowField[joda.LocalDateTime] = + RowField.from[Long](microsToJodaLocalDateTime)(microsFromJodaLocalDateTime) + + implicit val rfDurationMicros: RowField[jt.Duration] = + RowField.from[Long](microsToDuration)(microsFromDuration) + // joda.Duration has millisecond precision, excess precision discarded + implicit val rfJodaDurationMicros: RowField[joda.Duration] = + RowField.from[Long](microsToJodaDuration)(microsFromJodaDuration) +} + +private[logical] trait NanosNonInstant { + implicit val rfLocalTimeNanos: RowField[jt.LocalTime] = + RowField.id[jt.LocalTime](_ => FieldType.logicalType(new logicaltypes.Time())) + // joda.LocalTime only has millisecond resolution, so excess precision is discarded + implicit val rfJodaLocalTimeNanos: RowField[joda.LocalTime] = + RowField.from[jt.LocalTime](lt => nanosToJodaLocalTime(nanosFromLocalTime(lt)))(lt => + nanosToLocalTime(nanosFromJodaLocalTime(lt)) + ) + + implicit val rfLocalDateTimeNanos: RowField[jt.LocalDateTime] = + RowField.from[Long](nanosToLocalDateTime)(nanosFromLocalDateTime) + // joda.LocalDateTime has millisecond precision, excess precision discarded + // NOTE: misnamed `Micros` since 0.9; kept for source/binary compatibility + implicit val rfJodaLocalDateTimeMicros: RowField[joda.LocalDateTime] = + RowField.from[jt.LocalDateTime](ldt => nanosToJodaLocalDateTime(nanosFromLocalDateTime(ldt)))( + ldt => nanosToLocalDateTime(nanosFromJodaLocalDateTime(ldt)) + ) + + implicit val rfDurationNanos: RowField[jt.Duration] = + RowField.id[jt.Duration](_ => FieldType.logicalType(new logicaltypes.NanosDuration())) + // joda.Duration has millisecond precision, excess precision discarded + implicit val rfJodaDurationNanos: RowField[joda.Duration] = + RowField.from[jt.Duration](d => nanosToJodaDuration(nanosFromDuration(d)))(d => + nanosToDuration(nanosFromJodaDuration(d)) + ) +} diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index 4ca2728b..c12da988 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -18,11 +18,12 @@ package magnolify.beam import org.apache.beam.sdk.schemas.logicaltypes import org.apache.beam.sdk.schemas.Schema.FieldType -import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes +import org.apache.beam.sdk.schemas.logicaltypes.{SqlTypes, Timestamp} import org.joda.time as joda import org.joda.time.chrono.ISOChronology import java.time as jt +import java.time.temporal.ChronoUnit package object logical { import magnolify.shared.Time._ @@ -34,104 +35,132 @@ package object logical { RowField.from[jt.LocalDate](localDateToJodaLocalDate)(jodaLocalDateToLocalDate) } - object millis { + // Timestamp#toBaseType throws when an instant carries finer precision than the type + // declares, so writes truncate instead. Excess precision is discarded, matching the + // behavior of the non-instant mappings at each precision. + 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)) + } + + /** + * Millisecond-precision temporal mappings. + * + * `Instant` maps to Beam's portable `Timestamp.MILLIS` logical type. Prior to 0.10 it mapped to + * `FieldType.DATETIME`, which is backed by `org.joda.time.Instant`; see [[legacy.millis]]. + */ + object millis extends MillisNonInstant { implicit lazy val rfInstantMillis: RowField[jt.Instant] = - RowField.from[joda.Instant](i => millisToInstant(millisFromJodaInstant(i)))(i => - millisToJodaInstant(millisFromInstant(i)) - ) + tsInstant(Timestamp.MILLIS, ChronoUnit.MILLIS) implicit val rfJodaInstantMillis: RowField[joda.Instant] = - RowField.id[joda.Instant](_ => FieldType.DATETIME) + RowField.from[jt.Instant](i => millisToJodaInstant(millisFromInstant(i)))(i => + millisToInstant(millisFromJodaInstant(i)) + )(rfInstantMillis) implicit val rfJodaDateTimeMillis: RowField[joda.DateTime] = - RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant) - - implicit val rfLocalTimeMillis: RowField[jt.LocalTime] = - RowField.from[Int](millisToLocalTime)(millisFromLocalTime) - implicit val rfJodaLocalTimeMillis: RowField[joda.LocalTime] = - RowField.from[Int](millisToJodaLocalTime)(millisFromJodaLocalTime) - - implicit val rfLocalDateTimeMillis: RowField[jt.LocalDateTime] = - RowField.id[jt.LocalDateTime](_ => FieldType.logicalType(new logicaltypes.DateTime())) - implicit val rfJodaLocalDateTimeMillis: RowField[joda.LocalDateTime] = - RowField.from[jt.LocalDateTime](ldt => - millisToJodaLocalDateTime(millisFromLocalDateTime(ldt)) - )(ldt => millisToLocalDateTime(millisFromJodaLocalDateTime(ldt))) - - implicit val rfDurationMillis: RowField[jt.Duration] = - RowField.from[Long](millisToDuration)(millisFromDuration) - implicit val rfJodaDurationMillis: RowField[joda.Duration] = - RowField.from[Long](millisToJodaDuration)(millisFromJodaDuration) + RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant)( + rfJodaInstantMillis + ) } - object micros { - // NOTE: logicaltypes.MicrosInstant() cannot be used as it throws assertion - // errors when greater-than-microsecond precision data is used - implicit val rfInstantMicros: RowField[jt.Instant] = - RowField.from[Long](microsToInstant)(microsFromInstant) + /** + * Microsecond-precision temporal mappings. + * + * `Instant` maps to Beam's portable `Timestamp.MICROS` logical type. This is the encoding + * IcebergIO produces for `timestamptz` as of Beam 2.76.0. Prior to 0.10 it mapped to a raw + * `INT64` of microseconds since epoch; see [[legacy.micros]]. + */ + object micros extends MicrosNonInstant { + implicit lazy val rfInstantMicros: RowField[jt.Instant] = + tsInstant(Timestamp.MICROS, ChronoUnit.MICROS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantMicros: RowField[joda.Instant] = - RowField.from[Long](microsToJodaInstant)(microsFromJodaInstant) + RowField.from[jt.Instant](i => microsToJodaInstant(microsFromInstant(i)))(i => + microsToInstant(microsFromJodaInstant(i)) + )(rfInstantMicros) // joda.DateTime only has millisecond resolution, so excess precision is discarded implicit val rfJodaDateTimeMicros: RowField[joda.DateTime] = - RowField.from[Long](microsToJodaDateTime)(microsFromJodaDateTime) - - implicit val rfLocalTimeMicros: RowField[jt.LocalTime] = - RowField.from[Long](microsToLocalTime)(microsFromLocalTime) - // joda.LocalTime only has millisecond resolution, so excess precision is discarded - implicit val rfJodaLocalTimeMicros: RowField[joda.LocalTime] = - RowField.from[Long](microsToJodaLocalTime)(microsFromJodaLocalTime) - - implicit val rfLocalDateTimeMicros: RowField[jt.LocalDateTime] = - RowField.from[Long](microsToLocalDateTime)(microsFromLocalDateTime) - // joda.LocalDateTime has millisecond precision, excess precision discarded - implicit val rfJodaLocalDateTimeMicros: RowField[joda.LocalDateTime] = - RowField.from[Long](microsToJodaLocalDateTime)(microsFromJodaLocalDateTime) - - implicit val rfDurationMicros: RowField[jt.Duration] = - RowField.from[Long](microsToDuration)(microsFromDuration) - // joda.Duration has millisecond precision, excess precision discarded - implicit val rfJodaDurationMicros: RowField[joda.Duration] = - RowField.from[Long](microsToJodaDuration)(microsFromJodaDuration) + RowField.from[jt.Instant](i => microsToJodaDateTime(microsFromInstant(i)))(dt => + microsToInstant(microsFromJodaDateTime(dt)) + )(rfInstantMicros) } - object nanos { - implicit val rfInstantNanos: RowField[jt.Instant] = - RowField.id[jt.Instant](_ => FieldType.logicalType(new logicaltypes.NanosInstant())) + /** + * Nanosecond-precision temporal mappings. + * + * `Instant` maps to Beam's portable `Timestamp.NANOS` logical type. Prior to 0.10 it mapped to + * the SDK-local `NanosInstant` logical type; see [[legacy.nanos]]. + */ + object nanos extends NanosNonInstant { + implicit lazy val rfInstantNanos: RowField[jt.Instant] = + tsInstant(Timestamp.NANOS, ChronoUnit.NANOS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantNanos: RowField[joda.Instant] = RowField.from[jt.Instant](i => nanosToJodaInstant(nanosFromInstant(i)))(i => nanosToInstant(nanosFromJodaInstant(i)) - ) + )(rfInstantNanos) // joda.DateTime only has millisecond resolution implicit val rfJodaDateTimeNanos: RowField[joda.DateTime] = RowField.from[jt.Instant](i => nanosToJodaDateTime(nanosFromInstant(i)))(i => nanosToInstant(nanosFromJodaDateTime(i)) - ) - - implicit val rfLocalTimeNanos: RowField[jt.LocalTime] = - RowField.id[jt.LocalTime](_ => FieldType.logicalType(new logicaltypes.Time())) - // joda.LocalTime only has millisecond resolution, so excess precision is discarded - implicit val rfJodaLocalTimeNanos: RowField[joda.LocalTime] = - RowField.from[jt.LocalTime](lt => nanosToJodaLocalTime(nanosFromLocalTime(lt)))(lt => - nanosToLocalTime(nanosFromJodaLocalTime(lt)) - ) - - implicit val rfLocalDateTimeNanos: RowField[jt.LocalDateTime] = - RowField.from[Long](nanosToLocalDateTime)(nanosFromLocalDateTime) - // joda.LocalDateTime has millisecond precision, excess precision discarded - implicit val rfJodaLocalDateTimeMicros: RowField[joda.LocalDateTime] = - RowField.from[jt.LocalDateTime](ldt => nanosToJodaLocalDateTime(nanosFromLocalDateTime(ldt)))( - ldt => nanosToLocalDateTime(nanosFromJodaLocalDateTime(ldt)) - ) + )(rfInstantNanos) + } - implicit val rfDurationNanos: RowField[jt.Duration] = - RowField.id[jt.Duration](_ => FieldType.logicalType(new logicaltypes.NanosDuration())) - // joda.Duration has millisecond precision, excess precision discarded - implicit val rfJodaDurationNanos: RowField[joda.Duration] = - RowField.from[jt.Duration](d => nanosToJodaDuration(nanosFromDuration(d)))(d => - nanosToDuration(nanosFromJodaDuration(d)) - ) + /** + * Instant encodings used before 0.10. + * + * Use these to read Rows produced by Beam IO connectors that still emit `FieldType.DATETIME` (as + * of 2.76.0: amazon-web-services2, clickhouse, csv, delta, google-cloud-platform, hcatalog, + * iceberg, jdbc, kafka, singlestore), or by a pipeline pinned via `--updateCompatibilityVersion`. + * Non-instant mappings are identical to the defaults. + */ + object legacy { + object millis extends MillisNonInstant { + implicit lazy val rfInstantMillis: RowField[jt.Instant] = + RowField.from[joda.Instant](i => millisToInstant(millisFromJodaInstant(i)))(i => + millisToJodaInstant(millisFromInstant(i)) + ) + implicit val rfJodaInstantMillis: RowField[joda.Instant] = + RowField.id[joda.Instant](_ => FieldType.DATETIME) + implicit val rfJodaDateTimeMillis: RowField[joda.DateTime] = + RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant) + } + + object micros extends MicrosNonInstant { + // NOTE: logicaltypes.MicrosInstant() cannot be used as it throws assertion + // errors when greater-than-microsecond precision data is used + implicit val rfInstantMicros: RowField[jt.Instant] = + RowField.from[Long](microsToInstant)(microsFromInstant) + // joda.Instant has millisecond precision, excess precision discarded + implicit val rfJodaInstantMicros: RowField[joda.Instant] = + RowField.from[Long](microsToJodaInstant)(microsFromJodaInstant) + // joda.DateTime only has millisecond resolution, so excess precision is discarded + implicit val rfJodaDateTimeMicros: RowField[joda.DateTime] = + RowField.from[Long](microsToJodaDateTime)(microsFromJodaDateTime) + } + + object nanos extends NanosNonInstant { + implicit val rfInstantNanos: RowField[jt.Instant] = + RowField.id[jt.Instant](_ => FieldType.logicalType(new logicaltypes.NanosInstant())) + // joda.Instant has millisecond precision, excess precision discarded + implicit val rfJodaInstantNanos: RowField[joda.Instant] = + RowField.from[jt.Instant](i => nanosToJodaInstant(nanosFromInstant(i)))(i => + nanosToInstant(nanosFromJodaInstant(i)) + ) + // joda.DateTime only has millisecond resolution + implicit val rfJodaDateTimeNanos: RowField[joda.DateTime] = + RowField.from[jt.Instant](i => nanosToJodaDateTime(nanosFromInstant(i)))(i => + nanosToInstant(nanosFromJodaDateTime(i)) + ) + } } + @deprecated( + "SqlTypes.DATE/TIME/DATETIME duplicate `date` and the precision objects, and " + + "SqlTypes.TIMESTAMP is MicrosInstant, which throws on sub-microsecond instants. " + + "Use `date` plus one of millis/micros/nanos instead.", + "0.10.0" + ) object sql { implicit val rfSqlLocalTime: RowField[jt.LocalTime] = RowField.id(_ => FieldType.logicalType(SqlTypes.TIME)) diff --git a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala index f7d6049d..a4ca3e10 100644 --- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala +++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala @@ -27,6 +27,7 @@ import magnolify.test.ADT import magnolify.test.MagnolifySuite import magnolify.test.Simple.* import org.apache.beam.sdk.schemas.Schema +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp import org.apache.beam.sdk.values.Row import org.joda.time as joda import org.scalacheck.{Arbitrary, Gen, Prop} @@ -34,6 +35,7 @@ import org.scalacheck.{Arbitrary, Gen, Prop} import java.nio.ByteBuffer import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime} import java.util.UUID +import scala.annotation.nowarn import scala.reflect.ClassTag import scala.jdk.CollectionConverters.* @@ -117,6 +119,91 @@ class RowTypeSuite extends MagnolifySuite { testNamed[JodaTime]("JodaNanos") } + { + import magnolify.beam.logical.legacy.millis.* + testNamed[JavaTime]("JavaLegacyMillis") + testNamed[JodaTime]("JodaLegacyMillis") + } + + { + import magnolify.beam.logical.legacy.micros.* + testNamed[JavaTime]("JavaLegacyMicros") + testNamed[JodaTime]("JodaLegacyMicros") + } + + { + import magnolify.beam.logical.legacy.nanos.* + testNamed[JavaTime]("JavaLegacyNanos") + testNamed[JodaTime]("JodaLegacyNanos") + } + + // The shared arbInstant only generates millisecond precision, so the roundtrip properties + // above cannot observe how each precision handles a finer-grained Instant. Pin that here: + // Timestamp#toBaseType throws rather than silently truncating, so these mappings must + // truncate on write themselves. + private val subMicro = Instant.ofEpochSecond(1000L, 123456789L) + private def instantField(rt: RowType[JavaInstant]): Schema.FieldType = + rt.schema.getField("i").getType + private def roundtrip(rt: RowType[JavaInstant]): Instant = + rt.from(rt.to(JavaInstant(subMicro))).i + + { + import magnolify.beam.logical.millis.* + val rt = RowType[JavaInstant] + test("millis truncates sub-millisecond instants rather than throwing") { + assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123000000L)) + } + test("millis maps Instant to the portable Timestamp logical type") { + assertEquals(instantField(rt).getLogicalType.getIdentifier, Timestamp.IDENTIFIER) + } + } + + { + import magnolify.beam.logical.micros.* + val rt = RowType[JavaInstant] + test("micros truncates sub-microsecond instants rather than throwing") { + assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123456000L)) + } + test("micros maps Instant to the portable Timestamp logical type") { + assertEquals(instantField(rt).getLogicalType.getIdentifier, Timestamp.IDENTIFIER) + } + } + + { + import magnolify.beam.logical.nanos.* + val rt = RowType[JavaInstant] + test("nanos preserves full instant precision") { + assertEquals(roundtrip(rt), subMicro) + } + } + + { + import magnolify.beam.logical.legacy.millis.* + test("legacy millis keeps the joda-backed DATETIME primitive") { + assertEquals(instantField(RowType[JavaInstant]), Schema.FieldType.DATETIME) + } + } + + { + import magnolify.beam.logical.legacy.micros.* + test("legacy micros keeps the raw INT64 encoding") { + assertEquals(instantField(RowType[JavaInstant]), Schema.FieldType.INT64) + } + } + + // Documents why `sql` is deprecated: SqlTypes.TIMESTAMP is MicrosInstant, whose + // toBaseType throws on sub-microsecond precision. + { + @nowarn("cat=deprecation") + val rt = { + import magnolify.beam.logical.sql.* + RowType[JavaInstant] + } + test("deprecated sql mapping throws on sub-microsecond instants") { + intercept[AssertionError](rt.to(JavaInstant(subMicro))) + } + } + { implicit val bst: RowType[LowerCamel] = RowType[LowerCamel](CaseMapper(_.toUpperCase)) @@ -148,7 +235,11 @@ class RowTypeSuite extends MagnolifySuite { } { - import magnolify.beam.logical.sql.* + @nowarn("cat=deprecation") + implicit val bst: RowType[Sql] = { + import magnolify.beam.logical.sql.* + RowType[Sql] + } test[Sql] } @@ -236,6 +327,7 @@ case class Sql( ) case class JavaDate(d: LocalDate) case class JodaDate(jd: joda.LocalDate) +case class JavaInstant(i: Instant) case class JavaTime( i: Instant, dt: LocalDateTime, diff --git a/build.sbt b/build.sbt index d32e6cc4..8c25122b 100644 --- a/build.sbt +++ b/build.sbt @@ -49,7 +49,7 @@ val tensorflowVersion = "1.1.0" val tensorflowMetadataVersion = "1.16.1" // project -ThisBuild / tlBaseVersion := "0.9" +ThisBuild / tlBaseVersion := "0.10" ThisBuild / organization := "com.spotify" ThisBuild / organizationName := "Spotify AB" ThisBuild / startYear := Some(2016) From 83b181fd9fb9ae2b23fd25e47c8a1725132d8079 Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Wed, 16 Sep 2026 16:43:30 -0400 Subject: [PATCH 2/7] fixes --- .../magnolify/beam/logical/package.scala | 54 ++++++++++++++----- .../scala/magnolify/beam/RowTypeSuite.scala | 29 ++++++++-- docs/beam.md | 21 ++++++-- docs/mapping.md | 4 +- 4 files changed, 83 insertions(+), 25 deletions(-) diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index c12da988..fa8a65f8 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -47,11 +47,18 @@ package object logical { /** * Millisecond-precision temporal mappings. * - * `Instant` maps to Beam's portable `Timestamp.MILLIS` logical type. Prior to 0.10 it mapped to - * `FieldType.DATETIME`, which is backed by `org.joda.time.Instant`; see [[legacy.millis]]. + * `Instant` maps to Beam's portable `Timestamp.MILLIS` logical type. Instants carrying finer + * precision are truncated on write, because `Timestamp` rejects them rather than rounding. + * + * Not writable to Iceberg: its schema conversion accepts only `Timestamp.MICROS` (precision 6) + * and throws `UnsupportedOperationException` otherwise. For Iceberg use [[micros]], or + * [[legacy.millis]] when the pipeline pins `--updateCompatibilityVersion` below 2.76.0. + * + * Prior to 0.10 this mapped to `FieldType.DATETIME`, backed by `org.joda.time.Instant`; see + * [[legacy.millis]]. */ object millis extends MillisNonInstant { - implicit lazy val rfInstantMillis: RowField[jt.Instant] = + implicit val rfInstantMillis: RowField[jt.Instant] = tsInstant(Timestamp.MILLIS, ChronoUnit.MILLIS) implicit val rfJodaInstantMillis: RowField[joda.Instant] = RowField.from[jt.Instant](i => millisToJodaInstant(millisFromInstant(i)))(i => @@ -66,12 +73,17 @@ package object logical { /** * Microsecond-precision temporal mappings. * - * `Instant` maps to Beam's portable `Timestamp.MICROS` logical type. This is the encoding - * IcebergIO produces for `timestamptz` as of Beam 2.76.0. Prior to 0.10 it mapped to a raw - * `INT64` of microseconds since epoch; see [[legacy.micros]]. + * `Instant` maps to Beam's portable `Timestamp.MICROS` logical type. Instants carrying finer + * precision are truncated on write, because `Timestamp` rejects them rather than rounding. + * + * This is the encoding IcebergIO both produces and accepts for `timestamptz` as of Beam 2.76.0, + * making it the right choice for Iceberg — unless the pipeline pins + * `--updateCompatibilityVersion` below 2.76.0, in which case see [[legacy.millis]]. + * + * Prior to 0.10 this mapped to a raw `INT64` of microseconds since epoch; see [[legacy.micros]]. */ object micros extends MicrosNonInstant { - implicit lazy val rfInstantMicros: RowField[jt.Instant] = + implicit val rfInstantMicros: RowField[jt.Instant] = tsInstant(Timestamp.MICROS, ChronoUnit.MICROS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantMicros: RowField[joda.Instant] = @@ -88,11 +100,16 @@ package object logical { /** * Nanosecond-precision temporal mappings. * - * `Instant` maps to Beam's portable `Timestamp.NANOS` logical type. Prior to 0.10 it mapped to - * the SDK-local `NanosInstant` logical type; see [[legacy.nanos]]. + * `Instant` maps to Beam's portable `Timestamp.NANOS` logical type, which holds the full + * precision of `java.time.Instant`, so nothing is truncated. + * + * Not writable to Iceberg, which accepts only `Timestamp.MICROS`; use [[micros]] instead. The + * pre-0.10 `NanosInstant` encoding was not Iceberg-writable either, so this is not a regression. + * + * Prior to 0.10 this mapped to the SDK-local `NanosInstant` logical type; see [[legacy.nanos]]. */ object nanos extends NanosNonInstant { - implicit lazy val rfInstantNanos: RowField[jt.Instant] = + implicit val rfInstantNanos: RowField[jt.Instant] = tsInstant(Timestamp.NANOS, ChronoUnit.NANOS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantNanos: RowField[joda.Instant] = @@ -107,11 +124,20 @@ package object logical { } /** - * Instant encodings used before 0.10. + * Instant encodings used before 0.10: joda-backed `FieldType.DATETIME` at millis, a raw `INT64` + * of microseconds at micros, and the SDK-local `NanosInstant` at nanos. + * + * These are the magnolify-side counterpart to Beam's `--updateCompatibilityVersion` flag. A + * pipeline pinned below 2.76.0 gets `FieldType.DATETIME` back from IcebergIO, which only + * [[legacy.millis]] can read; the default objects expect the portable `Timestamp` type and will + * not match. Pair the flag with `legacy`, or omit both — mixing them is the one broken + * combination. + * + * Also needed for connectors that emit `FieldType.DATETIME` irrespective of the flag. As of Beam + * 2.76.0, within `sdks/java/io` that is amazon-web-services2, clickhouse, csv, delta, + * google-cloud-platform, hcatalog, iceberg, jdbc and singlestore; `DATETIME` is additionally + * produced by core and by the arrow, avro, protobuf and sql-datacatalog extensions. * - * Use these to read Rows produced by Beam IO connectors that still emit `FieldType.DATETIME` (as - * of 2.76.0: amazon-web-services2, clickhouse, csv, delta, google-cloud-platform, hcatalog, - * iceberg, jdbc, kafka, singlestore), or by a pipeline pinned via `--updateCompatibilityVersion`. * Non-instant mappings are identical to the defaults. */ object legacy { diff --git a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala index a4ca3e10..b82bc757 100644 --- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala +++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala @@ -27,6 +27,7 @@ import magnolify.test.ADT import magnolify.test.MagnolifySuite import magnolify.test.Simple.* import org.apache.beam.sdk.schemas.Schema +import org.apache.beam.sdk.schemas.logicaltypes import org.apache.beam.sdk.schemas.logicaltypes.Timestamp import org.apache.beam.sdk.values.Row import org.joda.time as joda @@ -146,6 +147,13 @@ class RowTypeSuite extends MagnolifySuite { rt.schema.getField("i").getType private def roundtrip(rt: RowType[JavaInstant]): Instant = rt.from(rt.to(JavaInstant(subMicro))).i + // Timestamp.IDENTIFIER is one shared constant across MILLIS/MICROS/NANOS, so asserting it + // alone cannot distinguish precisions. getArgument carries the precision. + private def timestampPrecision(rt: RowType[JavaInstant]): Int = { + val lt = instantField(rt).getLogicalType + assertEquals(lt.getIdentifier, Timestamp.IDENTIFIER) + lt.getArgument[Integer].intValue + } { import magnolify.beam.logical.millis.* @@ -153,8 +161,8 @@ class RowTypeSuite extends MagnolifySuite { test("millis truncates sub-millisecond instants rather than throwing") { assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123000000L)) } - test("millis maps Instant to the portable Timestamp logical type") { - assertEquals(instantField(rt).getLogicalType.getIdentifier, Timestamp.IDENTIFIER) + test("millis maps Instant to Timestamp at precision 3") { + assertEquals(timestampPrecision(rt), 3) } } @@ -164,8 +172,8 @@ class RowTypeSuite extends MagnolifySuite { test("micros truncates sub-microsecond instants rather than throwing") { assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123456000L)) } - test("micros maps Instant to the portable Timestamp logical type") { - assertEquals(instantField(rt).getLogicalType.getIdentifier, Timestamp.IDENTIFIER) + test("micros maps Instant to Timestamp at precision 6") { + assertEquals(timestampPrecision(rt), 6) } } @@ -175,6 +183,9 @@ class RowTypeSuite extends MagnolifySuite { test("nanos preserves full instant precision") { assertEquals(roundtrip(rt), subMicro) } + test("nanos maps Instant to Timestamp at precision 9") { + assertEquals(timestampPrecision(rt), 9) + } } { @@ -191,6 +202,16 @@ class RowTypeSuite extends MagnolifySuite { } } + { + import magnolify.beam.logical.legacy.nanos.* + test("legacy nanos keeps the SDK-local NanosInstant logical type") { + assertEquals( + instantField(RowType[JavaInstant]).getLogicalType.getIdentifier, + new logicaltypes.NanosInstant().getIdentifier + ) + } + } + // Documents why `sql` is deprecated: SqlTypes.TIMESTAMP is MicrosInstant, whose // toBaseType throws on sub-microsecond precision. { diff --git a/docs/beam.md b/docs/beam.md index 5c47a227..1911db40 100644 --- a/docs/beam.md +++ b/docs/beam.md @@ -31,19 +31,30 @@ Java and joda `LocalDate` types are available via `import magnolify.beam.logical For date-time, instants, and durations, use `import magnolify.beam.logical.millis.*`, `import magnolify.beam.logical.micros.*` or `import magnolify.beam.logical.nanos.*` as appropriate for your use-case. Note that joda types have only millisecond resolution, so excess precision will be discarded when used with `micros` or `nanos`. -Where possible, Beam logical types are used and joda types defer to these implementations: +Where possible, Beam logical types are used and joda types defer to the java.time implementations: -* Beam's `DATETIME` primitive type maps to the millisecond-precision java and joda `Instant`s and the joda `DateTime`. +* Beam's portable `Timestamp` logical type is used for java and joda `Instant` and the joda `DateTime`, at the precision of the object you import: `Timestamp.MILLIS`, `Timestamp.MICROS` or `Timestamp.NANOS`. * The `DateTime` logical type is used for millisecond-precision java and joda `LocalDateTime` -* The `NanosInstant` logical type is used for nanosecond-precision java and joda `Instant` * The `Time` logical type is used for nanosecond-precision java and joda `LocalTime` * The `NanosDuration` logical type is used for java and joda `Duration` -Beam's `MicrosInstant` should not be used as it throws exceptions when presented with greater-than-microsecond precision data. +`Timestamp` rejects instants carrying finer precision than it declares rather than rounding them, so `millis` and `micros` truncate on write. An `Instant` with nanosecond precision written via `micros` reads back truncated to microseconds. Use `nanos` to preserve it. + +Beam's `MicrosInstant` should not be used as it throws exceptions when presented with greater-than-microsecond precision data. `Timestamp.MICROS` is the safe equivalent. + +### Pre-0.10 encodings + +Before 0.10, `Instant` mapped to Beam's joda-backed `DATETIME` primitive under `millis`, a raw `INT64` of microseconds under `micros`, and the `NanosInstant` logical type under `nanos`. Those encodings are still available via `import magnolify.beam.logical.legacy.millis.*` (or `legacy.micros`, `legacy.nanos`). Non-instant mappings are identical to the defaults. + +Use `legacy` when reading Rows that are still `DATETIME`-encoded — either because the connector emits them (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, csv, delta, hcatalog, iceberg, singlestore and amazon-web-services2), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `legacy` is the counterpart to that flag: pair them, or omit both. Setting the flag while using the default objects is the one combination that will not work. + +### Iceberg + +Beam 2.76.0 changed IcebergIO's `timestamptz` mapping from `DATETIME` to `Timestamp.MICROS` in order to stop truncating microseconds. Use `micros` — it is the only precision IcebergIO accepts on write, and what it produces on read. `millis` and `nanos` are not Iceberg-writable and will fail schema conversion with `UnsupportedOperationException`. If the pipeline pins `--updateCompatibilityVersion` below 2.76.0, use `legacy.millis` instead. ## SQL types -SQL-compatible logical types are supported via `import magnolify.beam.logical.sql.*` +**Deprecated since 0.10.** `magnolify.beam.logical.sql`'s `DATE`, `TIME` and `DATETIME` members duplicate those in `logical.date` and the precision objects, and its `TIMESTAMP` member is Beam's `MicrosInstant`, which throws on sub-microsecond instants. Use `logical.date` plus one of `millis`/`micros`/`nanos` instead. ## Case mapping diff --git a/docs/mapping.md b/docs/mapping.md index 48c2fa0e..0dd595fd 100644 --- a/docs/mapping.md +++ b/docs/mapping.md @@ -23,13 +23,13 @@ | `Iterable[T]`2 | `array[T]` | `ITERABLE` | `REPEATED` | x | `Array` | `REPEATED`13 | `repeated` | Size >= 0 | | Nested | `record` | `ROW` | `STRUCT` | Flat8 | `Entity` | Group | `Message` | Flat8 | | `Map[K, V]` | `map[V]`15 | `MAP` | x | x | x | x | `map` | x | -| `java.time.Instant` | `long`11 | `DATETIME`, `INT64`, `ROW`17 | `TIMESTAMP` | x | `Timestamp` | `LOGICAL[TIMESTAMP]`9 | x | x | +| `java.time.Instant` | `long`11 | `ROW`17 | `TIMESTAMP` | x | `Timestamp` | `LOGICAL[TIMESTAMP]`9 | x | x | | `java.time.LocalDateTime` | `long`11 | `ROW`, `INT64`17 | `DATETIME` | x | x | `LOGICAL[TIMESTAMP]`9 | x | x | | `java.time.OffsetTime` | x | x | x | x | x | `LOGICAL[TIME]`9 | x | x | | `java.time.LocalTime` | `long`11 | `INT32`, `INT64`17 | `TIME` | x | x | `LOGICAL[TIME]`9 | x | x | | `java.time.LocalDate` | `int`11 | `INT64`17 | `DATE` | x | x | `LOGICAL[DATE]`9 | x | x | | `org.joda.time.LocalDate` | `int`11 | `INT64`17 | x | x | x | x | x | x | -| `org.joda.time.DateTime` | `int`11 | `DATETIME`, `INT64`, `ROW`17 | x | x | x | x | x | x | +| `org.joda.time.DateTime` | `int`11 | `ROW`17 | x | x | x | x | x | x | | `org.joda.time.LocalTime` | `int`11 | `INT32`, `INT64`17 | x | x | x | x | x | x | | `java.util.UUID` | `string`4 | `ROW`18 | x | ByteString (16 bytes) | x | `FIXED[16]` | x | x | | `(Long, Long, Long)`12 | `fixed[12]` | x | x | x | x | x | x | x | From fb66b984d68b2876070e99ff218573f28a5227a8 Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Wed, 16 Sep 2026 17:24:06 -0400 Subject: [PATCH 3/7] add tests --- .../scala/magnolify/beam/RowTypeSuite.scala | 41 ++++++++++++++++--- 1 file changed, 35 insertions(+), 6 deletions(-) diff --git a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala index b82bc757..fb4b13aa 100644 --- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala +++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala @@ -34,6 +34,7 @@ import org.joda.time as joda import org.scalacheck.{Arbitrary, Gen, Prop} import java.nio.ByteBuffer +import java.time.temporal.ChronoUnit import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime} import java.util.UUID import scala.annotation.nowarn @@ -138,10 +139,9 @@ class RowTypeSuite extends MagnolifySuite { testNamed[JodaTime]("JodaLegacyNanos") } - // The shared arbInstant only generates millisecond precision, so the roundtrip properties - // above cannot observe how each precision handles a finer-grained Instant. Pin that here: // Timestamp#toBaseType throws rather than silently truncating, so these mappings must - // truncate on write themselves. + // truncate on write themselves. `subMicro` pins the exact boundary values; `preciseInstants` + // below generalizes it. private val subMicro = Instant.ofEpochSecond(1000L, 123456789L) private def instantField(rt: RowType[JavaInstant]): Schema.FieldType = rt.schema.getField("i").getType @@ -155,6 +155,22 @@ class RowTypeSuite extends MagnolifySuite { lt.getArgument[Integer].intValue } + // The shared `arbInstant` generates only millisecond precision and only positive epochs, so the + // roundtrip properties above hold vacuously: nothing is ever truncated and the epoch boundary is + // never crossed. These cover both. Note the contract asserted is *not* `roundtrip(i) == i` -- + // each precision discards excess, so the real invariant is truncation to the declared unit. + // `Instant.truncatedTo` floors, which matters pre-epoch and matches `Timestamp`'s + // non-negative-subseconds representation. + private val preciseInstants: Gen[Instant] = for { + seconds <- Gen.chooseNum(-2208988800L, 4102444800L) // 1900-01-01 .. 2100-01-01 + nanos <- Gen.chooseNum(0, 999999999) + } yield Instant.ofEpochSecond(seconds, nanos.toLong) + + private def truncatesTo(rt: RowType[JavaInstant], unit: ChronoUnit): Prop = + Prop.forAll(preciseInstants) { i => + rt.from(rt.to(JavaInstant(i))).i == i.truncatedTo(unit) + } + { import magnolify.beam.logical.millis.* val rt = RowType[JavaInstant] @@ -164,6 +180,7 @@ class RowTypeSuite extends MagnolifySuite { test("millis maps Instant to Timestamp at precision 3") { assertEquals(timestampPrecision(rt), 3) } + property("millis truncates to millis across the epoch")(truncatesTo(rt, ChronoUnit.MILLIS)) } { @@ -175,6 +192,7 @@ class RowTypeSuite extends MagnolifySuite { test("micros maps Instant to Timestamp at precision 6") { assertEquals(timestampPrecision(rt), 6) } + property("micros truncates to micros across the epoch")(truncatesTo(rt, ChronoUnit.MICROS)) } { @@ -186,30 +204,41 @@ class RowTypeSuite extends MagnolifySuite { test("nanos maps Instant to Timestamp at precision 9") { assertEquals(timestampPrecision(rt), 9) } + property("nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) } { import magnolify.beam.logical.legacy.millis.* + val rt = RowType[JavaInstant] test("legacy millis keeps the joda-backed DATETIME primitive") { - assertEquals(instantField(RowType[JavaInstant]), Schema.FieldType.DATETIME) + assertEquals(instantField(rt), Schema.FieldType.DATETIME) } + property("legacy millis truncates to millis across the epoch")( + truncatesTo(rt, ChronoUnit.MILLIS) + ) } { import magnolify.beam.logical.legacy.micros.* + val rt = RowType[JavaInstant] test("legacy micros keeps the raw INT64 encoding") { - assertEquals(instantField(RowType[JavaInstant]), Schema.FieldType.INT64) + assertEquals(instantField(rt), Schema.FieldType.INT64) } + property("legacy micros truncates to micros across the epoch")( + truncatesTo(rt, ChronoUnit.MICROS) + ) } { import magnolify.beam.logical.legacy.nanos.* + val rt = RowType[JavaInstant] test("legacy nanos keeps the SDK-local NanosInstant logical type") { assertEquals( - instantField(RowType[JavaInstant]).getLogicalType.getIdentifier, + instantField(rt).getLogicalType.getIdentifier, new logicaltypes.NanosInstant().getIdentifier ) } + property("legacy nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) } // Documents why `sql` is deprecated: SqlTypes.TIMESTAMP is MicrosInstant, whose From 23e004a37016c11cddef126d571185d90ef5c09d Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Wed, 16 Sep 2026 18:17:41 -0400 Subject: [PATCH 4/7] fix docs --- .../magnolify/beam/logical/package.scala | 27 ++++++++++++++----- docs/beam.md | 25 ++++++++++++++++- 2 files changed, 44 insertions(+), 8 deletions(-) diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index fa8a65f8..c010bfc9 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -50,9 +50,10 @@ package object logical { * `Instant` maps to Beam's portable `Timestamp.MILLIS` logical type. Instants carrying finer * precision are truncated on write, because `Timestamp` rejects them rather than rounding. * - * Not writable to Iceberg: its schema conversion accepts only `Timestamp.MICROS` (precision 6) - * and throws `UnsupportedOperationException` otherwise. For Iceberg use [[micros]], or - * [[legacy.millis]] when the pipeline pins `--updateCompatibilityVersion` below 2.76.0. + * Not writable by any Beam IO that validates `Timestamp` precision: IcebergIO requires precision + * 6 and BigQueryIO and the Avro extension require 9, so precision 3 is rejected by all three. Use + * [[micros]] for Iceberg, [[nanos]] for BigQuery or Avro, or [[legacy.millis]] when the pipeline + * pins `--updateCompatibilityVersion` below 2.76.0. See [[legacy]] for the portability trade-off. * * Prior to 0.10 this mapped to `FieldType.DATETIME`, backed by `org.joda.time.Instant`; see * [[legacy.millis]]. @@ -80,6 +81,9 @@ package object logical { * making it the right choice for Iceberg — unless the pipeline pins * `--updateCompatibilityVersion` below 2.76.0, in which case see [[legacy.millis]]. * + * Not writable to BigQuery or via the Avro extension, which require precision 9; use [[nanos]] + * there. No single precision object satisfies both Iceberg and BigQuery — see [[legacy]]. + * * Prior to 0.10 this mapped to a raw `INT64` of microseconds since epoch; see [[legacy.micros]]. */ object micros extends MicrosNonInstant { @@ -101,7 +105,8 @@ package object logical { * Nanosecond-precision temporal mappings. * * `Instant` maps to Beam's portable `Timestamp.NANOS` logical type, which holds the full - * precision of `java.time.Instant`, so nothing is truncated. + * precision of `java.time.Instant`, so nothing is truncated. This is the precision BigQueryIO and + * the Avro extension require. * * Not writable to Iceberg, which accepts only `Timestamp.MICROS`; use [[micros]] instead. The * pre-0.10 `NanosInstant` encoding was not Iceberg-writable either, so this is not a regression. @@ -133,10 +138,18 @@ package object logical { * not match. Pair the flag with `legacy`, or omit both — mixing them is the one broken * combination. * - * Also needed for connectors that emit `FieldType.DATETIME` irrespective of the flag. As of Beam - * 2.76.0, within `sdks/java/io` that is amazon-web-services2, clickhouse, csv, delta, + * Also needed for connectors that hardcode `FieldType.DATETIME` irrespective of the flag. As of + * Beam 2.76.0, within `sdks/java/io` that is amazon-web-services2, clickhouse, delta, * google-cloud-platform, hcatalog, iceberg, jdbc and singlestore; `DATETIME` is additionally - * produced by core and by the arrow, avro, protobuf and sql-datacatalog extensions. + * hardcoded by core and by the arrow, avro, sql and sql-datacatalog extensions. Beyond those, + * schema inference maps any joda `Instant` field to `DATETIME` (`FieldTypeDescriptors`), so a + * connector whose element type has joda fields produces it without naming the type — KafkaIO's + * `KafkaSourceDescriptor` is one. + * + * `legacy` is also the only option that keeps a single `Instant` mapping writable across IOs. The + * default objects each satisfy exactly one validating write path — Iceberg requires `Timestamp` + * precision 6, BigQueryIO and the Avro extension require 9 — whereas `DATETIME` is accepted by + * all of them. Choosing a precision now means choosing a destination. * * Non-instant mappings are identical to the defaults. */ diff --git a/docs/beam.md b/docs/beam.md index 1911db40..8231a344 100644 --- a/docs/beam.md +++ b/docs/beam.md @@ -42,11 +42,34 @@ Where possible, Beam logical types are used and joda types defer to the java.tim Beam's `MicrosInstant` should not be used as it throws exceptions when presented with greater-than-microsecond precision data. `Timestamp.MICROS` is the safe equivalent. +### Choosing a precision: which IOs accept which + +Beam IOs that validate `Timestamp` precision do not agree on one, so the precision you import determines which IOs you can write to. As of Beam 2.76.0: + +| | IcebergIO | BigQueryIO | Avro extension | +|---|---|---|---| +| `millis` (`Timestamp.MILLIS`, precision 3) | ✗ | ✗ | ✗ | +| `micros` (`Timestamp.MICROS`, precision 6) | **✓** | ✗ | ✗ | +| `nanos` (`Timestamp.NANOS`, precision 9) | ✗ | **✓** | **✓** | +| `legacy.*` (`DATETIME`) | ✓ | ✓ | ✓ | + +IcebergIO requires precision 6 and throws `UnsupportedOperationException` otherwise; BigQueryIO and Beam's Avro extension require precision 9 and throw `IllegalArgumentException`/`RuntimeException` otherwise. + +Note the consequence: **no single precision object is writable to both Iceberg and BigQuery.** Before 0.10, `millis` mapped `Instant` to `DATETIME`, which all of them accept, so one import served every destination. If you need one record type to reach both, either keep `legacy.*` or define per-destination `RowType`s. + +### Reading + +Reads are more forgiving than writes, in a way worth knowing about. A `Timestamp` field surfaces as a `java.time.Instant` at any precision, so a reader whose declared precision is *finer or coarser* than the data still succeeds — it simply returns whatever precision the writer stored. Reading an Iceberg `timestamptz` (micros) with `millis` yields a full microsecond `Instant`, with no truncation and no error; truncation applies on write only. + +The one combination that fails loudly is a `legacy.*` reader against `Timestamp`-encoded data, which throws `ClassCastException: java.time.Instant cannot be cast to org.joda.time.Instant`. That is the pairing contract below. + ### Pre-0.10 encodings Before 0.10, `Instant` mapped to Beam's joda-backed `DATETIME` primitive under `millis`, a raw `INT64` of microseconds under `micros`, and the `NanosInstant` logical type under `nanos`. Those encodings are still available via `import magnolify.beam.logical.legacy.millis.*` (or `legacy.micros`, `legacy.nanos`). Non-instant mappings are identical to the defaults. -Use `legacy` when reading Rows that are still `DATETIME`-encoded — either because the connector emits them (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, csv, delta, hcatalog, iceberg, singlestore and amazon-web-services2), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `legacy` is the counterpart to that flag: pair them, or omit both. Setting the flag while using the default objects is the one combination that will not work. +Use `legacy` when reading Rows that are still `DATETIME`-encoded — either because the connector hardcodes it (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, delta, hcatalog, iceberg, singlestore and amazon-web-services2, plus core and the arrow, avro, sql and sql-datacatalog extensions), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `legacy` is the counterpart to that flag: pair them, or omit both. Setting the flag while using the default objects is the one combination that will not work. + +Beyond the connectors that name `DATETIME` explicitly, Beam's schema inference maps any joda `Instant` field to `DATETIME`, so a connector whose element type has joda fields produces it too — KafkaIO's `KafkaSourceDescriptor` is one. ### Iceberg From 1cdced519f5d34b8c7a132b01355a784948b1563 Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Wed, 16 Sep 2026 18:26:42 -0400 Subject: [PATCH 5/7] fix docs --- .../scala/magnolify/beam/logical/package.scala | 5 +++-- docs/beam.md | 14 +++++++++----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index c010bfc9..14e928e5 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -81,8 +81,9 @@ package object logical { * making it the right choice for Iceberg — unless the pipeline pins * `--updateCompatibilityVersion` below 2.76.0, in which case see [[legacy.millis]]. * - * Not writable to BigQuery or via the Avro extension, which require precision 9; use [[nanos]] - * there. No single precision object satisfies both Iceberg and BigQuery — see [[legacy]]. + * Not writable via Beam's BigQueryIO or Avro extension, which require precision 9; use [[nanos]] + * there. No single precision object satisfies both — see [[legacy]]. (This concerns Beam's own + * BigQueryIO on `Row`; magnolify's `bigquery` module converts to `TableRow` and is unaffected.) * * Prior to 0.10 this mapped to a raw `INT64` of microseconds since epoch; see [[legacy.micros]]. */ diff --git a/docs/beam.md b/docs/beam.md index 8231a344..21decfa5 100644 --- a/docs/beam.md +++ b/docs/beam.md @@ -42,20 +42,24 @@ Where possible, Beam logical types are used and joda types defer to the java.tim Beam's `MicrosInstant` should not be used as it throws exceptions when presented with greater-than-microsecond precision data. `Timestamp.MICROS` is the safe equivalent. -### Choosing a precision: which IOs accept which +### Choosing a precision: which Beam IOs accept which -Beam IOs that validate `Timestamp` precision do not agree on one, so the precision you import determines which IOs you can write to. As of Beam 2.76.0: +**Scope:** this section is about **Beam's own IOs consuming a `PCollection`** — that is, where the output of `RowType[T]` ends up. It says nothing about the IOs of frameworks built on Beam, which may share a name but not a code path. In Scio, for instance, only `IcebergIO`/`ManagedIO` take Beam `Row`s; its `BigQueryIO` and `AvroIO` route through magnolify's `bigquery` and `avro` modules, which target `TableRow` and `GenericRecord` directly, never a Beam `Schema`, and are unaffected by anything below. -| | IcebergIO | BigQueryIO | Avro extension | +Beam IOs that validate `Timestamp` precision do not agree on one, so the precision you import determines which of them you can write to. As of Beam 2.76.0: + +| | Beam `IcebergIO` | Beam `BigQueryIO` | Beam Avro extension | |---|---|---|---| | `millis` (`Timestamp.MILLIS`, precision 3) | ✗ | ✗ | ✗ | | `micros` (`Timestamp.MICROS`, precision 6) | **✓** | ✗ | ✗ | | `nanos` (`Timestamp.NANOS`, precision 9) | ✗ | **✓** | **✓** | | `legacy.*` (`DATETIME`) | ✓ | ✓ | ✓ | -IcebergIO requires precision 6 and throws `UnsupportedOperationException` otherwise; BigQueryIO and Beam's Avro extension require precision 9 and throw `IllegalArgumentException`/`RuntimeException` otherwise. +`IcebergIO` requires precision 6 and throws `UnsupportedOperationException` otherwise (`IcebergUtils.java:227-234`); `BigQueryIO` and the Avro extension require precision 9 and throw `IllegalArgumentException`/`RuntimeException` otherwise (`BigQueryUtils.java:593-596`, `BeamRowToStorageApiProto.java:257-260`, `AvroUtils.java:1229-1232`). + +Note the consequence: **no single precision object is writable to both Beam's `IcebergIO` and its `BigQueryIO`.** Before 0.10, `millis` mapped `Instant` to `DATETIME`, which all of them accept, so one import served every destination. If you need one record type to reach both, either keep `legacy.*` or define per-destination `RowType`s. -Note the consequence: **no single precision object is writable to both Iceberg and BigQuery.** Before 0.10, `millis` mapped `Instant` to `DATETIME`, which all of them accept, so one import served every destination. If you need one record type to reach both, either keep `legacy.*` or define per-destination `RowType`s. +The Avro row is about `beam-sdks-java-extensions-avro` converting a Beam `Schema` to an Avro `Schema`. If you want Avro output from a case class, use magnolify's `avro` module instead — it has no Beam dependency and none of these constraints. ### Reading From b55f3ea57134eda3f9d749060244730464e71f77 Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Thu, 17 Sep 2026 12:22:06 -0400 Subject: [PATCH 6/7] fix after review --- .../magnolify/beam/logical/CompatTypes.scala | 75 ++++++ .../beam/logical/NonInstantTypes.scala | 6 +- .../magnolify/beam/logical/package.scala | 240 +++++++++--------- .../scala/magnolify/beam/RowTypeSuite.scala | 137 +++++++--- docs/beam.md | 69 +++-- docs/mapping.md | 4 +- 6 files changed, 349 insertions(+), 182 deletions(-) create mode 100644 beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala diff --git a/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala b/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala new file mode 100644 index 00000000..cf25752f --- /dev/null +++ b/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala @@ -0,0 +1,75 @@ +/* + * Copyright 2026 Spotify AB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package magnolify.beam.logical + +import magnolify.beam.RowField +import magnolify.shared.Time._ +import org.apache.beam.sdk.schemas.Schema.FieldType +import org.apache.beam.sdk.schemas.logicaltypes +import org.joda.time as joda +import org.joda.time.chrono.ISOChronology + +import java.time as jt + +// The pre-0.10 `Instant` encodings, shared by `compat.*` and by the deprecated bare +// `millis`/`micros`/`nanos` objects so that the two cannot drift apart. +// +// Note these three do not share a representation -- millis is the joda-backed `DATETIME` +// primitive, micros is a raw `INT64` and nanos is the SDK-local `NanosInstant` logical type. +// What they have in common is only that this is what 0.9 produced, which is why the grouping +// is named for its history rather than for an encoding. + +private[logical] trait MillisCompat extends MillisNonInstant { + // Layered on the joda mapping below, so the typeclass's internal representation is joda. + // `lazy` because it resolves `rfJodaInstantMillis`, which is declared after it. + implicit lazy val rfInstantMillis: RowField[jt.Instant] = + RowField.from[joda.Instant](i => millisToInstant(millisFromJodaInstant(i)))(i => + millisToJodaInstant(millisFromInstant(i)) + ) + implicit val rfJodaInstantMillis: RowField[joda.Instant] = + RowField.id[joda.Instant](_ => FieldType.DATETIME) + implicit val rfJodaDateTimeMillis: RowField[joda.DateTime] = + RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant) +} + +private[logical] trait MicrosCompat extends MicrosNonInstant { + // NOTE: logicaltypes.MicrosInstant() cannot be used as it throws assertion + // errors when greater-than-microsecond precision data is used + implicit val rfInstantMicros: RowField[jt.Instant] = + RowField.from[Long](microsToInstant)(microsFromInstant) + // joda.Instant has millisecond precision, excess precision discarded + implicit val rfJodaInstantMicros: RowField[joda.Instant] = + RowField.from[Long](microsToJodaInstant)(microsFromJodaInstant) + // joda.DateTime only has millisecond resolution, so excess precision is discarded + implicit val rfJodaDateTimeMicros: RowField[joda.DateTime] = + RowField.from[Long](microsToJodaDateTime)(microsFromJodaDateTime) +} + +private[logical] trait NanosCompat extends NanosNonInstant { + implicit val rfInstantNanos: RowField[jt.Instant] = + RowField.id[jt.Instant](_ => FieldType.logicalType(new logicaltypes.NanosInstant())) + // joda.Instant has millisecond precision, excess precision discarded + implicit val rfJodaInstantNanos: RowField[joda.Instant] = + RowField.from[jt.Instant](i => nanosToJodaInstant(nanosFromInstant(i)))(i => + nanosToInstant(nanosFromJodaInstant(i)) + ) + // joda.DateTime only has millisecond resolution + implicit val rfJodaDateTimeNanos: RowField[joda.DateTime] = + RowField.from[jt.Instant](i => nanosToJodaDateTime(nanosFromInstant(i)))(i => + nanosToInstant(nanosFromJodaDateTime(i)) + ) +} diff --git a/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala b/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala index 4ff6dea6..ea5812aa 100644 --- a/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala +++ b/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala @@ -24,8 +24,10 @@ import org.joda.time as joda import java.time as jt -// Mappings that Beam represents identically regardless of the instant encoding, shared -// between the default precision objects and their `legacy` counterparts. +// Mappings that Beam represents identically regardless of the instant encoding, shared by +// `timestamp.*`, `compat.*` and the deprecated bare `millis`/`micros`/`nanos` objects. +// "NonInstant" means invariant across those groupings at a given precision -- all six of +// these mappings do still vary by precision. private[logical] trait MillisNonInstant { implicit val rfLocalTimeMillis: RowField[jt.LocalTime] = diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index 14e928e5..30f9bea3 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -45,160 +45,164 @@ package object logical { } /** - * Millisecond-precision temporal mappings. - * - * `Instant` maps to Beam's portable `Timestamp.MILLIS` logical type. Instants carrying finer - * precision are truncated on write, because `Timestamp` rejects them rather than rounding. - * - * Not writable by any Beam IO that validates `Timestamp` precision: IcebergIO requires precision - * 6 and BigQueryIO and the Avro extension require 9, so precision 3 is rejected by all three. Use - * [[micros]] for Iceberg, [[nanos]] for BigQuery or Avro, or [[legacy.millis]] when the pipeline - * pins `--updateCompatibilityVersion` below 2.76.0. See [[legacy]] for the portability trade-off. - * - * Prior to 0.10 this mapped to `FieldType.DATETIME`, backed by `org.joda.time.Instant`; see - * [[legacy.millis]]. - */ - object millis extends MillisNonInstant { - implicit val rfInstantMillis: RowField[jt.Instant] = - tsInstant(Timestamp.MILLIS, ChronoUnit.MILLIS) - implicit val rfJodaInstantMillis: RowField[joda.Instant] = - RowField.from[jt.Instant](i => millisToJodaInstant(millisFromInstant(i)))(i => - millisToInstant(millisFromJodaInstant(i)) - )(rfInstantMillis) - implicit val rfJodaDateTimeMillis: RowField[joda.DateTime] = - RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant)( - rfJodaInstantMillis - ) - } - - /** - * Microsecond-precision temporal mappings. - * - * `Instant` maps to Beam's portable `Timestamp.MICROS` logical type. Instants carrying finer - * precision are truncated on write, because `Timestamp` rejects them rather than rounding. + * Temporal mappings that encode `Instant` with Beam's portable `Timestamp` logical type + * (`beam:logical_type:timestamp:v1`) at the precision of the object you import. * * This is the encoding IcebergIO both produces and accepts for `timestamptz` as of Beam 2.76.0, - * making it the right choice for Iceberg — unless the pipeline pins - * `--updateCompatibilityVersion` below 2.76.0, in which case see [[legacy.millis]]. - * - * Not writable via Beam's BigQueryIO or Avro extension, which require precision 9; use [[nanos]] - * there. No single precision object satisfies both — see [[legacy]]. (This concerns Beam's own - * BigQueryIO on `Row`; magnolify's `bigquery` module converts to `TableRow` and is unaffected.) - * - * Prior to 0.10 this mapped to a raw `INT64` of microseconds since epoch; see [[legacy.micros]]. - */ - object micros extends MicrosNonInstant { - implicit val rfInstantMicros: RowField[jt.Instant] = - tsInstant(Timestamp.MICROS, ChronoUnit.MICROS) - // joda.Instant has millisecond precision, excess precision discarded - implicit val rfJodaInstantMicros: RowField[joda.Instant] = - RowField.from[jt.Instant](i => microsToJodaInstant(microsFromInstant(i)))(i => - microsToInstant(microsFromJodaInstant(i)) - )(rfInstantMicros) - // joda.DateTime only has millisecond resolution, so excess precision is discarded - implicit val rfJodaDateTimeMicros: RowField[joda.DateTime] = - RowField.from[jt.Instant](i => microsToJodaDateTime(microsFromInstant(i)))(dt => - microsToInstant(microsFromJodaDateTime(dt)) - )(rfInstantMicros) - } - - /** - * Nanosecond-precision temporal mappings. + * which is the reason these mappings exist. * - * `Instant` maps to Beam's portable `Timestamp.NANOS` logical type, which holds the full - * precision of `java.time.Instant`, so nothing is truncated. This is the precision BigQueryIO and - * the Avro extension require. + * Beam's IOs do not agree on a precision, so the object you import determines which of them you + * can write to. As of Beam 2.76.0: IcebergIO requires [[timestamp.micros]]; BigQueryIO and the + * Avro extension require [[timestamp.nanos]]; managed JDBC (postgres/mysql/sqlserver) and Kafka + * with `JSON` format accept no `Timestamp` precision at all and need [[compat]]. Beam SQL accepts + * any precision but round-trips values through milliseconds, discarding anything finer. * - * Not writable to Iceberg, which accepts only `Timestamp.MICROS`; use [[micros]] instead. The - * pre-0.10 `NanosInstant` encoding was not Iceberg-writable either, so this is not a regression. + * `Timestamp` rejects instants carrying finer precision than it declares rather than rounding + * them, so [[timestamp.millis]] and [[timestamp.micros]] truncate on write. * - * Prior to 0.10 this mapped to the SDK-local `NanosInstant` logical type; see [[legacy.nanos]]. + * Non-instant mappings are identical to [[compat]]. */ - object nanos extends NanosNonInstant { - implicit val rfInstantNanos: RowField[jt.Instant] = - tsInstant(Timestamp.NANOS, ChronoUnit.NANOS) - // joda.Instant has millisecond precision, excess precision discarded - implicit val rfJodaInstantNanos: RowField[joda.Instant] = - RowField.from[jt.Instant](i => nanosToJodaInstant(nanosFromInstant(i)))(i => - nanosToInstant(nanosFromJodaInstant(i)) - )(rfInstantNanos) - // joda.DateTime only has millisecond resolution - implicit val rfJodaDateTimeNanos: RowField[joda.DateTime] = - RowField.from[jt.Instant](i => nanosToJodaDateTime(nanosFromInstant(i)))(i => - nanosToInstant(nanosFromJodaDateTime(i)) - )(rfInstantNanos) - } + object timestamp { - /** - * Instant encodings used before 0.10: joda-backed `FieldType.DATETIME` at millis, a raw `INT64` - * of microseconds at micros, and the SDK-local `NanosInstant` at nanos. - * - * These are the magnolify-side counterpart to Beam's `--updateCompatibilityVersion` flag. A - * pipeline pinned below 2.76.0 gets `FieldType.DATETIME` back from IcebergIO, which only - * [[legacy.millis]] can read; the default objects expect the portable `Timestamp` type and will - * not match. Pair the flag with `legacy`, or omit both — mixing them is the one broken - * combination. - * - * Also needed for connectors that hardcode `FieldType.DATETIME` irrespective of the flag. As of - * Beam 2.76.0, within `sdks/java/io` that is amazon-web-services2, clickhouse, delta, - * google-cloud-platform, hcatalog, iceberg, jdbc and singlestore; `DATETIME` is additionally - * hardcoded by core and by the arrow, avro, sql and sql-datacatalog extensions. Beyond those, - * schema inference maps any joda `Instant` field to `DATETIME` (`FieldTypeDescriptors`), so a - * connector whose element type has joda fields produces it without naming the type — KafkaIO's - * `KafkaSourceDescriptor` is one. - * - * `legacy` is also the only option that keeps a single `Instant` mapping writable across IOs. The - * default objects each satisfy exactly one validating write path — Iceberg requires `Timestamp` - * precision 6, BigQueryIO and the Avro extension require 9 — whereas `DATETIME` is accepted by - * all of them. Choosing a precision now means choosing a destination. - * - * Non-instant mappings are identical to the defaults. - */ - object legacy { + /** + * `Instant` maps to `Timestamp.MILLIS`. Instants carrying finer precision are truncated on + * write. + * + * No Beam IO accepts precision 3 on write: IcebergIO requires 6, BigQueryIO and the Avro + * extension require 9. Its use is reading data that is already precision 3 — notably a BigQuery + * `TIMESTAMP(12)` column under `--picosecondTimestampMapping=MILLIS`. To *write* + * millisecond-precision instants, use [[compat.millis]], whose `DATETIME` encoding every + * schema-aware Beam IO accepts. + */ object millis extends MillisNonInstant { - implicit lazy val rfInstantMillis: RowField[jt.Instant] = - RowField.from[joda.Instant](i => millisToInstant(millisFromJodaInstant(i)))(i => - millisToJodaInstant(millisFromInstant(i)) - ) + implicit val rfInstantMillis: RowField[jt.Instant] = + tsInstant(Timestamp.MILLIS, ChronoUnit.MILLIS) implicit val rfJodaInstantMillis: RowField[joda.Instant] = - RowField.id[joda.Instant](_ => FieldType.DATETIME) + RowField.from[jt.Instant](i => millisToJodaInstant(millisFromInstant(i)))(i => + millisToInstant(millisFromJodaInstant(i)) + )(rfInstantMillis) implicit val rfJodaDateTimeMillis: RowField[joda.DateTime] = - RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant) + RowField.from[joda.Instant](_.toDateTime(ISOChronology.getInstanceUTC))(_.toInstant)( + rfJodaInstantMillis + ) } + /** + * `Instant` maps to `Timestamp.MICROS`. Instants carrying finer precision are truncated on + * write. + * + * This is what IcebergIO produces and accepts for `timestamptz` as of Beam 2.76.0, making it + * the right choice for Iceberg — unless the pipeline pins `--updateCompatibilityVersion` below + * 2.76.0, in which case use [[compat.millis]]. + * + * Not writable via Beam's BigQueryIO or Avro extension, which require precision 9; use + * [[timestamp.nanos]] there. No single precision satisfies both. (This concerns Beam's own + * BigQueryIO on `Row`; magnolify's `bigquery` module converts to `TableRow` and is unaffected.) + */ object micros extends MicrosNonInstant { - // NOTE: logicaltypes.MicrosInstant() cannot be used as it throws assertion - // errors when greater-than-microsecond precision data is used implicit val rfInstantMicros: RowField[jt.Instant] = - RowField.from[Long](microsToInstant)(microsFromInstant) + tsInstant(Timestamp.MICROS, ChronoUnit.MICROS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantMicros: RowField[joda.Instant] = - RowField.from[Long](microsToJodaInstant)(microsFromJodaInstant) + RowField.from[jt.Instant](i => microsToJodaInstant(microsFromInstant(i)))(i => + microsToInstant(microsFromJodaInstant(i)) + )(rfInstantMicros) // joda.DateTime only has millisecond resolution, so excess precision is discarded implicit val rfJodaDateTimeMicros: RowField[joda.DateTime] = - RowField.from[Long](microsToJodaDateTime)(microsFromJodaDateTime) + RowField.from[jt.Instant](i => microsToJodaDateTime(microsFromInstant(i)))(dt => + microsToInstant(microsFromJodaDateTime(dt)) + )(rfInstantMicros) } + /** + * `Instant` maps to `Timestamp.NANOS`, which holds the full precision of `java.time.Instant`, + * so nothing is truncated. This is the precision Beam's BigQueryIO and Avro extension require. + * + * Not writable to Iceberg, which accepts only `Timestamp.MICROS`; use [[timestamp.micros]] + * instead. The pre-0.10 `NanosInstant` encoding was not Iceberg-writable either, so this is no + * regression. + */ object nanos extends NanosNonInstant { implicit val rfInstantNanos: RowField[jt.Instant] = - RowField.id[jt.Instant](_ => FieldType.logicalType(new logicaltypes.NanosInstant())) + tsInstant(Timestamp.NANOS, ChronoUnit.NANOS) // joda.Instant has millisecond precision, excess precision discarded implicit val rfJodaInstantNanos: RowField[joda.Instant] = RowField.from[jt.Instant](i => nanosToJodaInstant(nanosFromInstant(i)))(i => nanosToInstant(nanosFromJodaInstant(i)) - ) + )(rfInstantNanos) // joda.DateTime only has millisecond resolution implicit val rfJodaDateTimeNanos: RowField[joda.DateTime] = RowField.from[jt.Instant](i => nanosToJodaDateTime(nanosFromInstant(i)))(i => nanosToInstant(nanosFromJodaDateTime(i)) - ) + )(rfInstantNanos) } } + /** + * The `Instant` encodings magnolify produced before 0.10: the joda-backed `FieldType.DATETIME` + * primitive at [[compat.millis]], a raw `INT64` of microseconds at [[compat.micros]], and the + * SDK-local `NanosInstant` logical type at [[compat.nanos]]. + * + * These three share no representation — only their history. The grouping is named for the + * compatibility it provides rather than for an encoding, because there is no encoding common to + * all three. + * + * [[compat.millis]] is the magnolify-side counterpart to Beam's `--updateCompatibilityVersion` + * flag. A pipeline pinned below 2.76.0 gets `FieldType.DATETIME` back from IcebergIO, which only + * [[compat.millis]] can read; [[timestamp]] expects the portable `Timestamp` type and will not + * match. Pair the flag with `compat.millis`, or omit both — mixing them is the one broken + * combination. + * + * `DATETIME` is also the only `Instant` encoding that every schema-aware Beam IO accepts, so + * [[compat.millis]] is not merely a migration aid. It is required for connectors that hardcode + * `DATETIME`: as of Beam 2.76.0, within `sdks/java/io` that is amazon-web-services2, clickhouse, + * delta, google-cloud-platform, hcatalog, iceberg, jdbc and singlestore, plus core and the arrow, + * avro, sql and sql-datacatalog extensions. Beyond those, schema inference maps any joda + * `Instant` field to `DATETIME` (`FieldTypeDescriptors`), so a connector whose element type has + * joda fields produces it without naming the type — KafkaIO's `KafkaSourceDescriptor` is one. And + * it is the only option for the managed JDBC sinks and for Kafka with `JSON` format, neither of + * which handles any `Timestamp` precision. + * + * Non-instant mappings are identical to [[timestamp]]. + */ + object compat { + object millis extends MillisCompat + object micros extends MicrosCompat + object nanos extends NanosCompat + } + + @deprecated( + "Renamed to `compat.millis` so the encoding it produces is explicit. This object is " + + "unchanged: `Instant` still maps to the joda-backed `FieldType.DATETIME` primitive. " + + "Use `timestamp.micros` for IcebergIO on Beam 2.76.0+, or `compat.millis` to keep this " + + "encoding.", + "0.10.0" + ) + object millis extends MillisCompat + + @deprecated( + "Renamed to `compat.micros` so the encoding it produces is explicit. This object is " + + "unchanged: `Instant` still maps to a raw `INT64` of microseconds since epoch. " + + "Use `timestamp.micros` for Beam's portable `Timestamp` logical type, or `compat.micros` " + + "to keep this encoding.", + "0.10.0" + ) + object micros extends MicrosCompat + + @deprecated( + "Renamed to `compat.nanos` so the encoding it produces is explicit. This object is " + + "unchanged: `Instant` still maps to the SDK-local `NanosInstant` logical type. " + + "Use `timestamp.nanos` for Beam's portable `Timestamp` logical type, or `compat.nanos` " + + "to keep this encoding.", + "0.10.0" + ) + object nanos extends NanosCompat + @deprecated( "SqlTypes.DATE/TIME/DATETIME duplicate `date` and the precision objects, and " + "SqlTypes.TIMESTAMP is MicrosInstant, which throws on sub-microsecond instants. " + - "Use `date` plus one of millis/micros/nanos instead.", + "Use `date` plus one of timestamp.{millis,micros,nanos} or compat.{millis,micros,nanos} " + + "instead.", "0.10.0" ) object sql { diff --git a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala index fb4b13aa..1adb3b97 100644 --- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala +++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala @@ -104,39 +104,39 @@ class RowTypeSuite extends MagnolifySuite { } { - import magnolify.beam.logical.millis.* - testNamed[JavaTime]("JavaMillis") - testNamed[JodaTime]("JodaMillis") + import magnolify.beam.logical.timestamp.millis.* + testNamed[JavaTime]("JavaTimestampMillis") + testNamed[JodaTime]("JodaTimestampMillis") } { - import magnolify.beam.logical.micros.* - testNamed[JavaTime]("JavaMicros") - testNamed[JodaTime]("JodaMicros") + import magnolify.beam.logical.timestamp.micros.* + testNamed[JavaTime]("JavaTimestampMicros") + testNamed[JodaTime]("JodaTimestampMicros") } { - import magnolify.beam.logical.nanos.* - testNamed[JavaTime]("JavaNanos") - testNamed[JodaTime]("JodaNanos") + import magnolify.beam.logical.timestamp.nanos.* + testNamed[JavaTime]("JavaTimestampNanos") + testNamed[JodaTime]("JodaTimestampNanos") } { - import magnolify.beam.logical.legacy.millis.* - testNamed[JavaTime]("JavaLegacyMillis") - testNamed[JodaTime]("JodaLegacyMillis") + import magnolify.beam.logical.compat.millis.* + testNamed[JavaTime]("JavaCompatMillis") + testNamed[JodaTime]("JodaCompatMillis") } { - import magnolify.beam.logical.legacy.micros.* - testNamed[JavaTime]("JavaLegacyMicros") - testNamed[JodaTime]("JodaLegacyMicros") + import magnolify.beam.logical.compat.micros.* + testNamed[JavaTime]("JavaCompatMicros") + testNamed[JodaTime]("JodaCompatMicros") } { - import magnolify.beam.logical.legacy.nanos.* - testNamed[JavaTime]("JavaLegacyNanos") - testNamed[JodaTime]("JodaLegacyNanos") + import magnolify.beam.logical.compat.nanos.* + testNamed[JavaTime]("JavaCompatNanos") + testNamed[JodaTime]("JodaCompatNanos") } // Timestamp#toBaseType throws rather than silently truncating, so these mappings must @@ -172,73 +172,132 @@ class RowTypeSuite extends MagnolifySuite { } { - import magnolify.beam.logical.millis.* + import magnolify.beam.logical.timestamp.millis.* val rt = RowType[JavaInstant] - test("millis truncates sub-millisecond instants rather than throwing") { + test("timestamp.millis truncates sub-millisecond instants rather than throwing") { assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123000000L)) } - test("millis maps Instant to Timestamp at precision 3") { + test("timestamp.millis maps Instant to Timestamp at precision 3") { assertEquals(timestampPrecision(rt), 3) } - property("millis truncates to millis across the epoch")(truncatesTo(rt, ChronoUnit.MILLIS)) + property("timestamp.millis truncates to millis across the epoch")( + truncatesTo(rt, ChronoUnit.MILLIS) + ) } { - import magnolify.beam.logical.micros.* + import magnolify.beam.logical.timestamp.micros.* val rt = RowType[JavaInstant] - test("micros truncates sub-microsecond instants rather than throwing") { + test("timestamp.micros truncates sub-microsecond instants rather than throwing") { assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123456000L)) } - test("micros maps Instant to Timestamp at precision 6") { + test("timestamp.micros maps Instant to Timestamp at precision 6") { assertEquals(timestampPrecision(rt), 6) } - property("micros truncates to micros across the epoch")(truncatesTo(rt, ChronoUnit.MICROS)) + property("timestamp.micros truncates to micros across the epoch")( + truncatesTo(rt, ChronoUnit.MICROS) + ) } { - import magnolify.beam.logical.nanos.* + import magnolify.beam.logical.timestamp.nanos.* val rt = RowType[JavaInstant] - test("nanos preserves full instant precision") { + test("timestamp.nanos preserves full instant precision") { assertEquals(roundtrip(rt), subMicro) } - test("nanos maps Instant to Timestamp at precision 9") { + test("timestamp.nanos maps Instant to Timestamp at precision 9") { assertEquals(timestampPrecision(rt), 9) } - property("nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) + property("timestamp.nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) } { - import magnolify.beam.logical.legacy.millis.* + import magnolify.beam.logical.compat.millis.* val rt = RowType[JavaInstant] - test("legacy millis keeps the joda-backed DATETIME primitive") { + test("compat.millis keeps the joda-backed DATETIME primitive") { assertEquals(instantField(rt), Schema.FieldType.DATETIME) } - property("legacy millis truncates to millis across the epoch")( + property("compat.millis truncates to millis across the epoch")( truncatesTo(rt, ChronoUnit.MILLIS) ) } { - import magnolify.beam.logical.legacy.micros.* + import magnolify.beam.logical.compat.micros.* val rt = RowType[JavaInstant] - test("legacy micros keeps the raw INT64 encoding") { + test("compat.micros keeps the raw INT64 encoding") { assertEquals(instantField(rt), Schema.FieldType.INT64) } - property("legacy micros truncates to micros across the epoch")( + property("compat.micros truncates to micros across the epoch")( truncatesTo(rt, ChronoUnit.MICROS) ) } { - import magnolify.beam.logical.legacy.nanos.* + import magnolify.beam.logical.compat.nanos.* val rt = RowType[JavaInstant] - test("legacy nanos keeps the SDK-local NanosInstant logical type") { + test("compat.nanos keeps the SDK-local NanosInstant logical type") { assertEquals( instantField(rt).getLogicalType.getIdentifier, new logicaltypes.NanosInstant().getIdentifier ) } - property("legacy nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) + property("compat.nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) + } + + // The whole point of deprecating rather than repurposing the bare objects: upgrading to 0.10 + // must not silently change the schema of code that still compiles. These pin the bare objects + // to the 0.9 encodings, field-for-field identical to their `compat` counterparts above. + { + @nowarn("cat=deprecation") + val bareMillis = { + import magnolify.beam.logical.millis.* + RowType[JavaInstant] + } + @nowarn("cat=deprecation") + val bareMicros = { + import magnolify.beam.logical.micros.* + RowType[JavaInstant] + } + @nowarn("cat=deprecation") + val bareNanos = { + import magnolify.beam.logical.nanos.* + RowType[JavaInstant] + } + + test("deprecated millis still produces the 0.9 DATETIME encoding") { + assertEquals(instantField(bareMillis), Schema.FieldType.DATETIME) + } + test("deprecated micros still produces the 0.9 raw INT64 encoding") { + assertEquals(instantField(bareMicros), Schema.FieldType.INT64) + } + test("deprecated nanos still produces the 0.9 NanosInstant encoding") { + assertEquals( + instantField(bareNanos).getLogicalType.getIdentifier, + new logicaltypes.NanosInstant().getIdentifier + ) + } + + // Not merely "not a Timestamp" -- assert the schema matches `compat`, so a future edit that + // touches one grouping and not the other fails here. + val compatMillis = { + import magnolify.beam.logical.compat.millis.* + RowType[JavaInstant] + } + val compatMicros = { + import magnolify.beam.logical.compat.micros.* + RowType[JavaInstant] + } + val compatNanos = { + import magnolify.beam.logical.compat.nanos.* + RowType[JavaInstant] + } + + test("deprecated objects are schema-identical to their compat counterparts") { + assertEquals(bareMillis.schema, compatMillis.schema) + assertEquals(bareMicros.schema, compatMicros.schema) + assertEquals(bareNanos.schema, compatNanos.schema) + } } // Documents why `sql` is deprecated: SqlTypes.TIMESTAMP is MicrosInstant, whose diff --git a/docs/beam.md b/docs/beam.md index 21decfa5..0a2de50b 100644 --- a/docs/beam.md +++ b/docs/beam.md @@ -28,60 +28,87 @@ Enum-like types map to the Beam logical [Enum type]((https://beam.apache.org/doc Java and joda `LocalDate` types are available via `import magnolify.beam.logical.date.*` -For date-time, instants, and durations, use `import magnolify.beam.logical.millis.*`, `import magnolify.beam.logical.micros.*` or `import magnolify.beam.logical.nanos.*` as appropriate for your use-case. +For date-time, instants, and durations, choose a **grouping** — which decides how `Instant` is encoded — and a **precision**: + +* `import magnolify.beam.logical.timestamp.millis.*` (or `.micros`, `.nanos`) maps `Instant` to Beam's portable `Timestamp` logical type. This is what IcebergIO requires as of Beam 2.76.0. +* `import magnolify.beam.logical.compat.millis.*` (or `.micros`, `.nanos`) keeps the encodings magnolify produced before 0.10: the joda-backed `DATETIME` primitive at `millis`, a raw `INT64` of microseconds at `micros`, and the SDK-local `NanosInstant` logical type at `nanos`. + +The bare `magnolify.beam.logical.millis.*`/`.micros`/`.nanos` objects still exist and still produce exactly what they produced in 0.9. They are deprecated aliases for the matching `compat` object, so **upgrading to 0.10 changes no schema until you change an import** — you get a deprecation warning telling you to pick a grouping explicitly. + Note that joda types have only millisecond resolution, so excess precision will be discarded when used with `micros` or `nanos`. Where possible, Beam logical types are used and joda types defer to the java.time implementations: -* Beam's portable `Timestamp` logical type is used for java and joda `Instant` and the joda `DateTime`, at the precision of the object you import: `Timestamp.MILLIS`, `Timestamp.MICROS` or `Timestamp.NANOS`. +* Beam's portable `Timestamp` logical type is used for java and joda `Instant` and the joda `DateTime` under `timestamp.*`, at the precision of the object you import: `Timestamp.MILLIS`, `Timestamp.MICROS` or `Timestamp.NANOS`. * The `DateTime` logical type is used for millisecond-precision java and joda `LocalDateTime` * The `Time` logical type is used for nanosecond-precision java and joda `LocalTime` * The `NanosDuration` logical type is used for java and joda `Duration` -`Timestamp` rejects instants carrying finer precision than it declares rather than rounding them, so `millis` and `micros` truncate on write. An `Instant` with nanosecond precision written via `micros` reads back truncated to microseconds. Use `nanos` to preserve it. +`LocalTime`, `LocalDateTime` and `Duration` are encoded identically under `timestamp` and `compat` at a given precision — Beam offers no joda schema type for them, so there is nothing to differ about. Only `Instant`, joda `Instant` and joda `DateTime` differ between the two groupings. + +`Timestamp` rejects instants carrying finer precision than it declares rather than rounding them, so `timestamp.millis` and `timestamp.micros` truncate on write. An `Instant` with nanosecond precision written via `timestamp.micros` reads back truncated to microseconds. Use `timestamp.nanos` to preserve it. Beam's `MicrosInstant` should not be used as it throws exceptions when presented with greater-than-microsecond precision data. `Timestamp.MICROS` is the safe equivalent. -### Choosing a precision: which Beam IOs accept which +### Choosing an encoding: which Beam IOs accept which **Scope:** this section is about **Beam's own IOs consuming a `PCollection`** — that is, where the output of `RowType[T]` ends up. It says nothing about the IOs of frameworks built on Beam, which may share a name but not a code path. In Scio, for instance, only `IcebergIO`/`ManagedIO` take Beam `Row`s; its `BigQueryIO` and `AvroIO` route through magnolify's `bigquery` and `avro` modules, which target `TableRow` and `GenericRecord` directly, never a Beam `Schema`, and are unaffected by anything below. -Beam IOs that validate `Timestamp` precision do not agree on one, so the precision you import determines which of them you can write to. As of Beam 2.76.0: +Beam's IOs do not agree on a `Timestamp` precision, and several do not handle the type at all, so the grouping and precision you import determine where you can write. Verified against Beam 2.76.0: + +| | `IcebergIO` | `BigQueryIO` | Avro extension | managed JDBC | Kafka `JSON` | Kafka `AVRO` | Beam SQL | +|---|---|---|---|---|---|---|---| +| `timestamp.millis` (precision 3) | ✗ | ✗ | ✗ | ✗ | ✗ | ✗ | ✓ † | +| `timestamp.micros` (precision 6) | **✓** | ✗ | ✗ | ✗ | ✗ | ✗ | ✓ † | +| `timestamp.nanos` (precision 9) | ✗ | **✓** | **✓** | ✗ | ✗ | **✓** | ✓ † | +| `compat.millis` (`DATETIME`) | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | + +† accepted at any precision, but Beam SQL round-trips timestamps through milliseconds (`BeamCalcRel.java:463`), so anything finer is silently dropped. -| | Beam `IcebergIO` | Beam `BigQueryIO` | Beam Avro extension | -|---|---|---|---| -| `millis` (`Timestamp.MILLIS`, precision 3) | ✗ | ✗ | ✗ | -| `micros` (`Timestamp.MICROS`, precision 6) | **✓** | ✗ | ✗ | -| `nanos` (`Timestamp.NANOS`, precision 9) | ✗ | **✓** | **✓** | -| `legacy.*` (`DATETIME`) | ✓ | ✓ | ✓ | +Where the rejections come from: -`IcebergIO` requires precision 6 and throws `UnsupportedOperationException` otherwise (`IcebergUtils.java:227-234`); `BigQueryIO` and the Avro extension require precision 9 and throw `IllegalArgumentException`/`RuntimeException` otherwise (`BigQueryUtils.java:593-596`, `BeamRowToStorageApiProto.java:257-260`, `AvroUtils.java:1229-1232`). +* `IcebergIO` requires precision 6 and throws `UnsupportedOperationException` otherwise (`IcebergUtils.java:227-234`). +* `BigQueryIO` and the Avro extension require precision 9 (`BigQueryUtils.java:591-596`, `BeamRowToStorageApiProto.java:252-260`, `AvroUtils.java:1227-1232`). Kafka with `AVRO` format goes through the same Avro check (`KafkaWriteSchemaTransformProvider.java:201`). +* The managed JDBC sinks (`postgres`, `mysql`, `sqlserver`) have no `Timestamp` branch at all. Unknown logical types fall back to their base type (`JdbcUtil.java:336-338`), and `Timestamp`'s base type is a `ROW`, which is not a writable JDBC type — so it throws `RuntimeException("ROW in schema is not supported while writing")`. +* Kafka with `JSON` format fails *late*: `RowJson` recurses into the `ROW` base type, finds only `INT64`/`INT16` inside, and so passes schema validation — then throws `ClassCastException` per element when it tries to cast the `java.time.Instant` to a `Row` (`RowJson.java:176-180`, `:593`). -Note the consequence: **no single precision object is writable to both Beam's `IcebergIO` and its `BigQueryIO`.** Before 0.10, `millis` mapped `Instant` to `DATETIME`, which all of them accept, so one import served every destination. If you need one record type to reach both, either keep `legacy.*` or define per-destination `RowType`s. +How you would hit the JDBC and Kafka rows, since they are less obvious than Iceberg: Beam's schema transforms require a schema-bearing `PCollection`, and `RowCoder` is a `SchemaCoder`, so `pcoll.setCoder(RowCoder.of(rowType.schema))` is enough to make one. Any managed sink then accepts it. -The Avro row is about `beam-sdks-java-extensions-avro` converting a Beam `Schema` to an Avro `Schema`. If you want Avro output from a case class, use magnolify's `avro` module instead — it has no Beam dependency and none of these constraints. +Two consequences worth stating plainly: + +**No `timestamp.*` precision reaches every sink, and some sinks accept none of them.** `compat.millis` is the only `Instant` encoding every schema-aware Beam IO accepts, because `DATETIME` is a primitive they all understand. If one record type needs multiple destinations, use `compat.millis` or define per-destination `RowType`s. + +**`compat.micros` and `compat.nanos` are not general escape hatches.** `compat.micros` is a plain `INT64`, so it is structurally accepted nearly everywhere but lands as an integer column rather than a timestamp. `compat.nanos` is `NanosInstant`, whose base type is also a `ROW`, so it hits the same walls as `timestamp.*` — it is rejected by Iceberg (`IcebergUtils.java:236`) and was never writable there. + +The Avro column is about `beam-sdks-java-extensions-avro` converting a Beam `Schema` to an Avro `Schema`. If you want Avro output from a case class, use magnolify's `avro` module instead — it has no Beam dependency and none of these constraints. ### Reading -Reads are more forgiving than writes, in a way worth knowing about. A `Timestamp` field surfaces as a `java.time.Instant` at any precision, so a reader whose declared precision is *finer or coarser* than the data still succeeds — it simply returns whatever precision the writer stored. Reading an Iceberg `timestamptz` (micros) with `millis` yields a full microsecond `Instant`, with no truncation and no error; truncation applies on write only. +Reads are more forgiving than writes, in a way worth knowing about. A `Timestamp` field surfaces as a `java.time.Instant` at any precision, so a reader whose declared precision is *finer or coarser* than the data still succeeds — it simply returns whatever precision the writer stored. Reading an Iceberg `timestamptz` (micros) with `timestamp.millis` yields a full microsecond `Instant`, with no truncation and no error; truncation applies on write only. -The one combination that fails loudly is a `legacy.*` reader against `Timestamp`-encoded data, which throws `ClassCastException: java.time.Instant cannot be cast to org.joda.time.Instant`. That is the pairing contract below. +The one combination that fails loudly is a `compat.*` reader against `Timestamp`-encoded data, which throws `ClassCastException: java.time.Instant cannot be cast to org.joda.time.Instant`. That is the pairing contract below. -### Pre-0.10 encodings +Reading is also where the groupings are least interchangeable, because most IOs still *produce* `DATETIME`. Beam's `BigQueryIO` is the clearest case: it demands precision 9 on write, but on read it returns `DATETIME` for an ordinary `TIMESTAMP` column and only yields a `Timestamp` logical type for a `TIMESTAMP(12)` column, under `--picosecondTimestampMapping` (`BigQueryUtils.java:474-493`). So no single import round-trips it — write with `timestamp.nanos`, read with `compat.millis`. -Before 0.10, `Instant` mapped to Beam's joda-backed `DATETIME` primitive under `millis`, a raw `INT64` of microseconds under `micros`, and the `NanosInstant` logical type under `nanos`. Those encodings are still available via `import magnolify.beam.logical.legacy.millis.*` (or `legacy.micros`, `legacy.nanos`). Non-instant mappings are identical to the defaults. +### The `compat` encodings -Use `legacy` when reading Rows that are still `DATETIME`-encoded — either because the connector hardcodes it (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, delta, hcatalog, iceberg, singlestore and amazon-web-services2, plus core and the arrow, avro, sql and sql-datacatalog extensions), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `legacy` is the counterpart to that flag: pair them, or omit both. Setting the flag while using the default objects is the one combination that will not work. +`compat.millis`, `compat.micros` and `compat.nanos` hold the `Instant` encodings magnolify produced before 0.10 — the joda-backed `DATETIME` primitive, a raw `INT64` of microseconds, and the `NanosInstant` logical type respectively. Non-instant mappings are identical to `timestamp.*`. + +The name is deliberately about compatibility rather than representation: those three share no encoding, only the fact that this is what 0.9 emitted. `compat` is also not a deprecated holding pen — `compat.millis` is the correct and often the *only* choice for the destinations listed above, and stays correct as long as those IOs emit and accept `DATETIME`. + +Use `compat.millis` when reading Rows that are still `DATETIME`-encoded — either because the connector hardcodes it (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, delta, hcatalog, iceberg, singlestore and amazon-web-services2, plus core and the arrow, avro, sql and sql-datacatalog extensions), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `compat.millis` is the counterpart to that flag: pair them, or omit both. Setting the flag while using `timestamp.*` is the one combination that will not work. Beyond the connectors that name `DATETIME` explicitly, Beam's schema inference maps any joda `Instant` field to `DATETIME`, so a connector whose element type has joda fields produces it too — KafkaIO's `KafkaSourceDescriptor` is one. ### Iceberg -Beam 2.76.0 changed IcebergIO's `timestamptz` mapping from `DATETIME` to `Timestamp.MICROS` in order to stop truncating microseconds. Use `micros` — it is the only precision IcebergIO accepts on write, and what it produces on read. `millis` and `nanos` are not Iceberg-writable and will fail schema conversion with `UnsupportedOperationException`. If the pipeline pins `--updateCompatibilityVersion` below 2.76.0, use `legacy.millis` instead. +Beam 2.76.0 changed IcebergIO's `timestamptz` mapping from `DATETIME` to `Timestamp.MICROS` in order to stop truncating microseconds. Use `timestamp.micros` — it is the only `Timestamp` precision IcebergIO accepts on write, and what it produces on read. `timestamp.millis` and `timestamp.nanos` will fail schema conversion with `UnsupportedOperationException`. If the pipeline pins `--updateCompatibilityVersion` below 2.76.0, use `compat.millis` instead. + +Note the change was to the **read** path only; IcebergIO still accepts `DATETIME` on write (`IcebergUtils.java:77`), so `compat.millis` remains a valid way to write Iceberg — it just truncates to milliseconds, which is what prompted the Beam change in the first place. ## SQL types -**Deprecated since 0.10.** `magnolify.beam.logical.sql`'s `DATE`, `TIME` and `DATETIME` members duplicate those in `logical.date` and the precision objects, and its `TIMESTAMP` member is Beam's `MicrosInstant`, which throws on sub-microsecond instants. Use `logical.date` plus one of `millis`/`micros`/`nanos` instead. +**Deprecated since 0.10.** `magnolify.beam.logical.sql`'s `DATE`, `TIME` and `DATETIME` members duplicate those in `logical.date` and the precision objects, and its `TIMESTAMP` member is Beam's `MicrosInstant`, which throws on sub-microsecond instants. Use `logical.date` plus one of `timestamp.{millis,micros,nanos}` or `compat.{millis,micros,nanos}` instead. ## Case mapping diff --git a/docs/mapping.md b/docs/mapping.md index 0dd595fd..48c2fa0e 100644 --- a/docs/mapping.md +++ b/docs/mapping.md @@ -23,13 +23,13 @@ | `Iterable[T]`2 | `array[T]` | `ITERABLE` | `REPEATED` | x | `Array` | `REPEATED`13 | `repeated` | Size >= 0 | | Nested | `record` | `ROW` | `STRUCT` | Flat8 | `Entity` | Group | `Message` | Flat8 | | `Map[K, V]` | `map[V]`15 | `MAP` | x | x | x | x | `map` | x | -| `java.time.Instant` | `long`11 | `ROW`17 | `TIMESTAMP` | x | `Timestamp` | `LOGICAL[TIMESTAMP]`9 | x | x | +| `java.time.Instant` | `long`11 | `DATETIME`, `INT64`, `ROW`17 | `TIMESTAMP` | x | `Timestamp` | `LOGICAL[TIMESTAMP]`9 | x | x | | `java.time.LocalDateTime` | `long`11 | `ROW`, `INT64`17 | `DATETIME` | x | x | `LOGICAL[TIMESTAMP]`9 | x | x | | `java.time.OffsetTime` | x | x | x | x | x | `LOGICAL[TIME]`9 | x | x | | `java.time.LocalTime` | `long`11 | `INT32`, `INT64`17 | `TIME` | x | x | `LOGICAL[TIME]`9 | x | x | | `java.time.LocalDate` | `int`11 | `INT64`17 | `DATE` | x | x | `LOGICAL[DATE]`9 | x | x | | `org.joda.time.LocalDate` | `int`11 | `INT64`17 | x | x | x | x | x | x | -| `org.joda.time.DateTime` | `int`11 | `ROW`17 | x | x | x | x | x | x | +| `org.joda.time.DateTime` | `int`11 | `DATETIME`, `INT64`, `ROW`17 | x | x | x | x | x | x | | `org.joda.time.LocalTime` | `int`11 | `INT32`, `INT64`17 | x | x | x | x | x | x | | `java.util.UUID` | `string`4 | `ROW`18 | x | ByteString (16 bytes) | x | `FIXED[16]` | x | x | | `(Long, Long, Long)`12 | `fixed[12]` | x | x | x | x | x | x | x | From 26881aad82c48df7f387cf5bb8df760387b79332 Mon Sep 17 00:00:00 2001 From: Andrew Kabas Date: Thu, 17 Sep 2026 12:27:53 -0400 Subject: [PATCH 7/7] retarget to 0.9.x --- .../scala/magnolify/beam/logical/CompatTypes.scala | 4 ++-- .../main/scala/magnolify/beam/logical/package.scala | 12 ++++++------ .../src/test/scala/magnolify/beam/RowTypeSuite.scala | 10 +++++----- build.sbt | 2 +- docs/beam.md | 10 +++++----- 5 files changed, 19 insertions(+), 19 deletions(-) diff --git a/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala b/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala index cf25752f..f62ab059 100644 --- a/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala +++ b/beam/src/main/scala/magnolify/beam/logical/CompatTypes.scala @@ -25,12 +25,12 @@ import org.joda.time.chrono.ISOChronology import java.time as jt -// The pre-0.10 `Instant` encodings, shared by `compat.*` and by the deprecated bare +// The pre-0.9.8 `Instant` encodings, shared by `compat.*` and by the deprecated bare // `millis`/`micros`/`nanos` objects so that the two cannot drift apart. // // Note these three do not share a representation -- millis is the joda-backed `DATETIME` // primitive, micros is a raw `INT64` and nanos is the SDK-local `NanosInstant` logical type. -// What they have in common is only that this is what 0.9 produced, which is why the grouping +// What they have in common is only that this is what 0.9.7 produced, which is why the grouping // is named for its history rather than for an encoding. private[logical] trait MillisCompat extends MillisNonInstant { diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala index 30f9bea3..dc362066 100644 --- a/beam/src/main/scala/magnolify/beam/logical/package.scala +++ b/beam/src/main/scala/magnolify/beam/logical/package.scala @@ -119,7 +119,7 @@ package object logical { * so nothing is truncated. This is the precision Beam's BigQueryIO and Avro extension require. * * Not writable to Iceberg, which accepts only `Timestamp.MICROS`; use [[timestamp.micros]] - * instead. The pre-0.10 `NanosInstant` encoding was not Iceberg-writable either, so this is no + * instead. The pre-0.9.8 `NanosInstant` encoding was not Iceberg-writable either, so this is no * regression. */ object nanos extends NanosNonInstant { @@ -139,7 +139,7 @@ package object logical { } /** - * The `Instant` encodings magnolify produced before 0.10: the joda-backed `FieldType.DATETIME` + * The `Instant` encodings magnolify produced through 0.9.7: the joda-backed `FieldType.DATETIME` * primitive at [[compat.millis]], a raw `INT64` of microseconds at [[compat.micros]], and the * SDK-local `NanosInstant` logical type at [[compat.nanos]]. * @@ -176,7 +176,7 @@ package object logical { "unchanged: `Instant` still maps to the joda-backed `FieldType.DATETIME` primitive. " + "Use `timestamp.micros` for IcebergIO on Beam 2.76.0+, or `compat.millis` to keep this " + "encoding.", - "0.10.0" + "0.9.8" ) object millis extends MillisCompat @@ -185,7 +185,7 @@ package object logical { "unchanged: `Instant` still maps to a raw `INT64` of microseconds since epoch. " + "Use `timestamp.micros` for Beam's portable `Timestamp` logical type, or `compat.micros` " + "to keep this encoding.", - "0.10.0" + "0.9.8" ) object micros extends MicrosCompat @@ -194,7 +194,7 @@ package object logical { "unchanged: `Instant` still maps to the SDK-local `NanosInstant` logical type. " + "Use `timestamp.nanos` for Beam's portable `Timestamp` logical type, or `compat.nanos` " + "to keep this encoding.", - "0.10.0" + "0.9.8" ) object nanos extends NanosCompat @@ -203,7 +203,7 @@ package object logical { "SqlTypes.TIMESTAMP is MicrosInstant, which throws on sub-microsecond instants. " + "Use `date` plus one of timestamp.{millis,micros,nanos} or compat.{millis,micros,nanos} " + "instead.", - "0.10.0" + "0.9.8" ) object sql { implicit val rfSqlLocalTime: RowField[jt.LocalTime] = diff --git a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala index 1adb3b97..09273595 100644 --- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala +++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala @@ -245,9 +245,9 @@ class RowTypeSuite extends MagnolifySuite { property("compat.nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS)) } - // The whole point of deprecating rather than repurposing the bare objects: upgrading to 0.10 + // The whole point of deprecating rather than repurposing the bare objects: upgrading to 0.9.8 // must not silently change the schema of code that still compiles. These pin the bare objects - // to the 0.9 encodings, field-for-field identical to their `compat` counterparts above. + // to the 0.9.7 encodings, field-for-field identical to their `compat` counterparts above. { @nowarn("cat=deprecation") val bareMillis = { @@ -265,13 +265,13 @@ class RowTypeSuite extends MagnolifySuite { RowType[JavaInstant] } - test("deprecated millis still produces the 0.9 DATETIME encoding") { + test("deprecated millis still produces the 0.9.7 DATETIME encoding") { assertEquals(instantField(bareMillis), Schema.FieldType.DATETIME) } - test("deprecated micros still produces the 0.9 raw INT64 encoding") { + test("deprecated micros still produces the 0.9.7 raw INT64 encoding") { assertEquals(instantField(bareMicros), Schema.FieldType.INT64) } - test("deprecated nanos still produces the 0.9 NanosInstant encoding") { + test("deprecated nanos still produces the 0.9.7 NanosInstant encoding") { assertEquals( instantField(bareNanos).getLogicalType.getIdentifier, new logicaltypes.NanosInstant().getIdentifier diff --git a/build.sbt b/build.sbt index 8c25122b..d32e6cc4 100644 --- a/build.sbt +++ b/build.sbt @@ -49,7 +49,7 @@ val tensorflowVersion = "1.1.0" val tensorflowMetadataVersion = "1.16.1" // project -ThisBuild / tlBaseVersion := "0.10" +ThisBuild / tlBaseVersion := "0.9" ThisBuild / organization := "com.spotify" ThisBuild / organizationName := "Spotify AB" ThisBuild / startYear := Some(2016) diff --git a/docs/beam.md b/docs/beam.md index 0a2de50b..fef6f4fc 100644 --- a/docs/beam.md +++ b/docs/beam.md @@ -31,9 +31,9 @@ Java and joda `LocalDate` types are available via `import magnolify.beam.logical For date-time, instants, and durations, choose a **grouping** — which decides how `Instant` is encoded — and a **precision**: * `import magnolify.beam.logical.timestamp.millis.*` (or `.micros`, `.nanos`) maps `Instant` to Beam's portable `Timestamp` logical type. This is what IcebergIO requires as of Beam 2.76.0. -* `import magnolify.beam.logical.compat.millis.*` (or `.micros`, `.nanos`) keeps the encodings magnolify produced before 0.10: the joda-backed `DATETIME` primitive at `millis`, a raw `INT64` of microseconds at `micros`, and the SDK-local `NanosInstant` logical type at `nanos`. +* `import magnolify.beam.logical.compat.millis.*` (or `.micros`, `.nanos`) keeps the encodings magnolify produced through 0.9.7: the joda-backed `DATETIME` primitive at `millis`, a raw `INT64` of microseconds at `micros`, and the SDK-local `NanosInstant` logical type at `nanos`. -The bare `magnolify.beam.logical.millis.*`/`.micros`/`.nanos` objects still exist and still produce exactly what they produced in 0.9. They are deprecated aliases for the matching `compat` object, so **upgrading to 0.10 changes no schema until you change an import** — you get a deprecation warning telling you to pick a grouping explicitly. +The bare `magnolify.beam.logical.millis.*`/`.micros`/`.nanos` objects still exist and still produce exactly what they produced in 0.9.7. They are deprecated aliases for the matching `compat` object, so **upgrading to 0.9.8 changes no schema until you change an import** — you get a deprecation warning telling you to pick a grouping explicitly. Note that joda types have only millisecond resolution, so excess precision will be discarded when used with `micros` or `nanos`. @@ -92,9 +92,9 @@ Reading is also where the groupings are least interchangeable, because most IOs ### The `compat` encodings -`compat.millis`, `compat.micros` and `compat.nanos` hold the `Instant` encodings magnolify produced before 0.10 — the joda-backed `DATETIME` primitive, a raw `INT64` of microseconds, and the `NanosInstant` logical type respectively. Non-instant mappings are identical to `timestamp.*`. +`compat.millis`, `compat.micros` and `compat.nanos` hold the `Instant` encodings magnolify produced through 0.9.7 — the joda-backed `DATETIME` primitive, a raw `INT64` of microseconds, and the `NanosInstant` logical type respectively. Non-instant mappings are identical to `timestamp.*`. -The name is deliberately about compatibility rather than representation: those three share no encoding, only the fact that this is what 0.9 emitted. `compat` is also not a deprecated holding pen — `compat.millis` is the correct and often the *only* choice for the destinations listed above, and stays correct as long as those IOs emit and accept `DATETIME`. +The name is deliberately about compatibility rather than representation: those three share no encoding, only the fact that this is what 0.9.7 emitted. `compat` is also not a deprecated holding pen — `compat.millis` is the correct and often the *only* choice for the destinations listed above, and stays correct as long as those IOs emit and accept `DATETIME`. Use `compat.millis` when reading Rows that are still `DATETIME`-encoded — either because the connector hardcodes it (as of Beam 2.76.0 that includes jdbc, google-cloud-platform, clickhouse, delta, hcatalog, iceberg, singlestore and amazon-web-services2, plus core and the arrow, avro, sql and sql-datacatalog extensions), or because the pipeline pins Beam's `--updateCompatibilityVersion` below 2.76.0. `compat.millis` is the counterpart to that flag: pair them, or omit both. Setting the flag while using `timestamp.*` is the one combination that will not work. @@ -108,7 +108,7 @@ Note the change was to the **read** path only; IcebergIO still accepts `DATETIME ## SQL types -**Deprecated since 0.10.** `magnolify.beam.logical.sql`'s `DATE`, `TIME` and `DATETIME` members duplicate those in `logical.date` and the precision objects, and its `TIMESTAMP` member is Beam's `MicrosInstant`, which throws on sub-microsecond instants. Use `logical.date` plus one of `timestamp.{millis,micros,nanos}` or `compat.{millis,micros,nanos}` instead. +**Deprecated since 0.9.8.** `magnolify.beam.logical.sql`'s `DATE`, `TIME` and `DATETIME` members duplicate those in `logical.date` and the precision objects, and its `TIMESTAMP` member is Beam's `MicrosInstant`, which throws on sub-microsecond instants. Use `logical.date` plus one of `timestamp.{millis,micros,nanos}` or `compat.{millis,micros,nanos}` instead. ## Case mapping