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..f62ab059
--- /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.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.7 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
new file mode 100644
index 00000000..ea5812aa
--- /dev/null
+++ b/beam/src/main/scala/magnolify/beam/logical/NonInstantTypes.scala
@@ -0,0 +1,96 @@
+/*
+ * 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 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] =
+    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..dc362066 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,176 @@ package object logical {
       RowField.from[jt.LocalDate](localDateToJodaLocalDate)(jodaLocalDateToLocalDate)
   }
 
-  object millis {
-    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)
-
-    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)
+  // 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))
   }
 
-  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)
-    // 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)
-
-    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)
+  /**
+   * 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,
+   * which is the reason these mappings exist.
+   *
+   * 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.
+   *
+   * `Timestamp` rejects instants carrying finer precision than it declares rather than rounding
+   * them, so [[timestamp.millis]] and [[timestamp.micros]] truncate on write.
+   *
+   * Non-instant mappings are identical to [[compat]].
+   */
+  object timestamp {
+
+    /**
+     * `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 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
+        )
+    }
+
+    /**
+     * `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 {
+      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)
+    }
+
+    /**
+     * `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.9.8 `NanosInstant` encoding was not Iceberg-writable either, so this is no
+     * regression.
+     */
+    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 nanos {
-    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))
-      )
-
-    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))
-      )
-
-    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))
-      )
+  /**
+   * 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]].
+   *
+   * 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.9.8"
+  )
+  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.9.8"
+  )
+  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.9.8"
+  )
+  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 timestamp.{millis,micros,nanos} or compat.{millis,micros,nanos} " +
+      "instead.",
+    "0.9.8"
+  )
   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..09273595 100644
--- a/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala
+++ b/beam/src/test/scala/magnolify/beam/RowTypeSuite.scala
@@ -27,13 +27,17 @@ 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
 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
 import scala.reflect.ClassTag
 import scala.jdk.CollectionConverters.*
 
@@ -100,21 +104,213 @@ 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.compat.millis.*
+    testNamed[JavaTime]("JavaCompatMillis")
+    testNamed[JodaTime]("JodaCompatMillis")
+  }
+
+  {
+    import magnolify.beam.logical.compat.micros.*
+    testNamed[JavaTime]("JavaCompatMicros")
+    testNamed[JodaTime]("JodaCompatMicros")
+  }
+
+  {
+    import magnolify.beam.logical.compat.nanos.*
+    testNamed[JavaTime]("JavaCompatNanos")
+    testNamed[JodaTime]("JodaCompatNanos")
+  }
+
+  // Timestamp#toBaseType throws rather than silently truncating, so these mappings must
+  // 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
+  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
+  }
+
+  // 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.timestamp.millis.*
+    val rt = RowType[JavaInstant]
+    test("timestamp.millis truncates sub-millisecond instants rather than throwing") {
+      assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123000000L))
+    }
+    test("timestamp.millis maps Instant to Timestamp at precision 3") {
+      assertEquals(timestampPrecision(rt), 3)
+    }
+    property("timestamp.millis truncates to millis across the epoch")(
+      truncatesTo(rt, ChronoUnit.MILLIS)
+    )
+  }
+
+  {
+    import magnolify.beam.logical.timestamp.micros.*
+    val rt = RowType[JavaInstant]
+    test("timestamp.micros truncates sub-microsecond instants rather than throwing") {
+      assertEquals(roundtrip(rt), Instant.ofEpochSecond(1000L, 123456000L))
+    }
+    test("timestamp.micros maps Instant to Timestamp at precision 6") {
+      assertEquals(timestampPrecision(rt), 6)
+    }
+    property("timestamp.micros truncates to micros across the epoch")(
+      truncatesTo(rt, ChronoUnit.MICROS)
+    )
+  }
+
+  {
+    import magnolify.beam.logical.timestamp.nanos.*
+    val rt = RowType[JavaInstant]
+    test("timestamp.nanos preserves full instant precision") {
+      assertEquals(roundtrip(rt), subMicro)
+    }
+    test("timestamp.nanos maps Instant to Timestamp at precision 9") {
+      assertEquals(timestampPrecision(rt), 9)
+    }
+    property("timestamp.nanos preserves nanos across the epoch")(truncatesTo(rt, ChronoUnit.NANOS))
+  }
+
+  {
+    import magnolify.beam.logical.compat.millis.*
+    val rt = RowType[JavaInstant]
+    test("compat.millis keeps the joda-backed DATETIME primitive") {
+      assertEquals(instantField(rt), Schema.FieldType.DATETIME)
+    }
+    property("compat.millis truncates to millis across the epoch")(
+      truncatesTo(rt, ChronoUnit.MILLIS)
+    )
+  }
+
+  {
+    import magnolify.beam.logical.compat.micros.*
+    val rt = RowType[JavaInstant]
+    test("compat.micros keeps the raw INT64 encoding") {
+      assertEquals(instantField(rt), Schema.FieldType.INT64)
+    }
+    property("compat.micros truncates to micros across the epoch")(
+      truncatesTo(rt, ChronoUnit.MICROS)
+    )
+  }
+
+  {
+    import magnolify.beam.logical.compat.nanos.*
+    val rt = RowType[JavaInstant]
+    test("compat.nanos keeps the SDK-local NanosInstant logical type") {
+      assertEquals(
+        instantField(rt).getLogicalType.getIdentifier,
+        new logicaltypes.NanosInstant().getIdentifier
+      )
+    }
+    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.9.8
+  // must not silently change the schema of code that still compiles. These pin the bare objects
+  // to the 0.9.7 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.7 DATETIME encoding") {
+      assertEquals(instantField(bareMillis), Schema.FieldType.DATETIME)
+    }
+    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.7 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
+  // 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)))
+    }
   }
 
   {
@@ -148,7 +344,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 +436,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/docs/beam.md b/docs/beam.md
index 5c47a227..fef6f4fc 100644
--- a/docs/beam.md
+++ b/docs/beam.md
@@ -28,22 +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 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.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`.
 
-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` 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 `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.
+`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 an encoding: which Beam IOs accept which
+
+**Scope:** this section is about **Beam's own IOs consuming a `PCollection<Row>`** — 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'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.
+
+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 (`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`).
+
+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<Row>`, and `RowCoder` is a `SchemaCoder`, so `pcoll.setCoder(RowCoder.of(rowType.schema))` is enough to make one. Any managed sink then accepts it.
+
+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 `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 `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.
+
+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`.
+
+### The `compat` encodings
+
+`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.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.
+
+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 `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
 
-SQL-compatible logical types are supported via `import magnolify.beam.logical.sql.*`
+**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