diff --git a/avro/src/main/scala/magnolify/avro/AvroType.scala b/avro/src/main/scala/magnolify/avro/AvroType.scala
index 7ede81ffe..d4f853cf7 100644
--- a/avro/src/main/scala/magnolify/avro/AvroType.scala
+++ b/avro/src/main/scala/magnolify/avro/AvroType.scala
@@ -221,7 +221,7 @@ object AvroField {
     override def from(v: String)(cm: CaseMapper): String = v
     override def to(v: String)(cm: CaseMapper): String = v
   }
-
+f
   implicit def afEnum[T](implicit et: EnumType[T], lp: shapeless.LowPriority): AvroField[T] =
     // Avro 1.9+ added a type parameter for `GenericEnumSymbol`, breaking 1.8 compatibility
     // Some reader, i.e. `AvroParquetReader` reads enums as `Utf8`
diff --git a/avro/src/main/scala/magnolify/avro/logical/package.scala b/avro/src/main/scala/magnolify/avro/logical/package.scala
index a6d0d7473..35620cc56 100644
--- a/avro/src/main/scala/magnolify/avro/logical/package.scala
+++ b/avro/src/main/scala/magnolify/avro/logical/package.scala
@@ -25,106 +25,61 @@ import java.time.format.{DateTimeFormatter, DateTimeFormatterBuilder}
 import java.util.concurrent.TimeUnit
 
 package object logical {
+  import magnolify.shared.Time._
   // Duplicate implementation from org.apache.avro.data.TimeConversions
   // to support both 1.8 (joda-time based) and 1.9+ (java-time based)
   object micros {
-    private def toTimestampMicros(microsFromEpoch: Long): Instant = {
-      val epochSeconds = microsFromEpoch / 1000000L
-      val nanoAdjustment = (microsFromEpoch % 1000000L) * 1000L;
-      Instant.ofEpochSecond(epochSeconds, nanoAdjustment)
-    }
-
-    private def fromTimestampMicros(instant: Instant): Long = {
-      val seconds = instant.getEpochSecond
-      val nanos = instant.getNano
-      if (seconds < 0 && nanos > 0) {
-        val micros = Math.multiplyExact(seconds + 1, 1000000L)
-        val adjustment = (nanos / 1000L) - 1000000
-        Math.addExact(micros, adjustment)
-      } else {
-        val micros = Math.multiplyExact(seconds, 1000000L)
-        Math.addExact(micros, nanos / 1000L)
-      }
-    }
-
     implicit val afTimestampMicros: AvroField[Instant] =
-      AvroField.logicalType[Long](LogicalTypes.timestampMicros())(toTimestampMicros)(
-        fromTimestampMicros
+      AvroField.logicalType[Long](LogicalTypes.timestampMicros())(microsToInstant)(
+        microsFromInstant
       )
 
     implicit val afTimeMicros: AvroField[LocalTime] =
-      AvroField.logicalType[Long](LogicalTypes.timeMicros()) { us =>
-        LocalTime.ofNanoOfDay(TimeUnit.MICROSECONDS.toNanos(us))
-      } { time =>
-        TimeUnit.NANOSECONDS.toMicros(time.toNanoOfDay)
-      }
+      AvroField.logicalType[Long](LogicalTypes.timeMicros())(microsToLocalTime)(microsFromLocalTime)
 
     // `LogicalTypes.localTimestampMicros()` is Avro 1.10
     implicit val afLocalTimestampMicros: AvroField[LocalDateTime] =
-      AvroField.logicalType[Long](new LogicalType("local-timestamp-micros")) { microsFromEpoch =>
-        val instant = toTimestampMicros(microsFromEpoch)
-        LocalDateTime.ofInstant(instant, ZoneOffset.UTC)
-      } { timestamp =>
-        val instant = timestamp.toInstant(ZoneOffset.UTC)
-        fromTimestampMicros(instant)
-      }
+      AvroField.logicalType[Long](new LogicalType("local-timestamp-micros"))(microsToLocalDateTime)(
+        microsFromLocalDateTime
+      )
 
     // avro 1.8 uses joda-time
     implicit val afJodaTimestampMicros: AvroField[joda.DateTime] =
-      AvroField.logicalType[Long](LogicalTypes.timestampMicros()) { microsFromEpoch =>
-        new joda.DateTime(microsFromEpoch / 1000, joda.DateTimeZone.UTC)
-      } { timestamp =>
-        1000 * timestamp.getMillis
-      }
+      AvroField.logicalType[Long](LogicalTypes.timestampMicros())(microsToJodaDateTime)(
+        microsFromJodaDateTime
+      )
 
     implicit val afJodaTimeMicros: AvroField[joda.LocalTime] =
-      AvroField.logicalType[Long](LogicalTypes.timeMicros()) { microsFromMidnight =>
-        joda.LocalTime.fromMillisOfDay(microsFromMidnight / 1000)
-      } { time =>
-        // from LossyTimeMicrosConversion
-        1000L * time.millisOfDay().get()
-      }
+      AvroField.logicalType[Long](LogicalTypes.timeMicros())(microsToJodaLocalTime)(
+        microsFromJodaLocalTime
+      )
   }
 
   object millis {
     implicit val afTimestampMillis: AvroField[Instant] =
-      AvroField.logicalType[Long](LogicalTypes.timestampMillis()) { millisFromEpoch =>
-        Instant.ofEpochMilli(millisFromEpoch)
-      } { timestamp =>
-        timestamp.toEpochMilli
-      }
+      AvroField.logicalType[Long](LogicalTypes.timestampMillis())(millisToInstant)(
+        millisFromInstant
+      )
 
     implicit val afTimeMillis: AvroField[LocalTime] =
-      AvroField.logicalType[Int](LogicalTypes.timeMillis()) { millisFromMidnight =>
-        LocalTime.ofNanoOfDay(TimeUnit.MILLISECONDS.toNanos(millisFromMidnight.toLong))
-      } { time =>
-        TimeUnit.NANOSECONDS.toMillis(time.toNanoOfDay).toInt
-      }
+      AvroField.logicalType[Int](LogicalTypes.timeMillis())(millisToLocalTime)(millisFromLocalTime)
 
     // `LogicalTypes.localTimestampMillis` is Avro 1.10.0+
     implicit val afLocalTimestampMillis: AvroField[LocalDateTime] =
-      AvroField.logicalType[Long](new LogicalType("local-timestamp-millis")) { millisFromEpoch =>
-        val instant = Instant.ofEpochMilli(millisFromEpoch)
-        LocalDateTime.ofInstant(instant, ZoneOffset.UTC)
-      } { timestamp =>
-        val instant = timestamp.toInstant(ZoneOffset.UTC)
-        instant.toEpochMilli
-      }
+      AvroField.logicalType[Long](new LogicalType("local-timestamp-millis"))(millisToLocalDateTime)(
+        millisFromLocalDateTime
+      )
 
     // avro 1.8 uses joda-time
     implicit val afJodaTimestampMillis: AvroField[joda.DateTime] =
-      AvroField.logicalType[Long](LogicalTypes.timestampMillis()) { millisFromEpoch =>
-        new joda.DateTime(millisFromEpoch, joda.DateTimeZone.UTC)
-      } { timestamp =>
-        timestamp.getMillis
-      }
+      AvroField.logicalType[Long](LogicalTypes.timestampMillis())(millisToJodaDateTime)(
+        millisFromJodaDateTime
+      )
 
     implicit val afJodaTimeMillis: AvroField[joda.LocalTime] =
-      AvroField.logicalType[Int](LogicalTypes.timeMillis()) { millisFromMidnight =>
-        joda.LocalTime.fromMillisOfDay(millisFromMidnight.toLong)
-      } { time =>
-        time.millisOfDay().get()
-      }
+      AvroField.logicalType[Int](LogicalTypes.timeMillis())(millisToJodaLocalTime)(
+        millisFromJodaLocalTime
+      )
   }
 
   object bigquery {
diff --git a/beam/src/main/scala/magnolify/beam/BeamSchemaType.scala b/beam/src/main/scala/magnolify/beam/BeamSchemaType.scala
new file mode 100644
index 000000000..1c9c4033a
--- /dev/null
+++ b/beam/src/main/scala/magnolify/beam/BeamSchemaType.scala
@@ -0,0 +1,342 @@
+/*
+ * 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
+
+import magnolia1.*
+import magnolify.shared.*
+import org.apache.beam.sdk.schemas.Schema
+import org.apache.beam.sdk.schemas.Schema.{FieldType, LogicalType}
+import org.apache.beam.sdk.values.Row
+import org.joda.time as joda
+import com.google.protobuf.ByteString
+import magnolify.shims.FactoryCompat
+import org.apache.beam.sdk.schemas.logicaltypes as logicaltypes
+
+import java.nio.ByteBuffer
+import java.time.LocalDate
+import java.{time as jt, util as ju}
+import scala.annotation.implicitNotFound
+import scala.collection.concurrent
+import scala.jdk.CollectionConverters.*
+
+// https://beam.apache.org/documentation/programming-guide/#schema-definition
+sealed trait BeamSchemaType[T] extends Converter[T, Row, Row] {
+  val schema: Schema
+  def apply(r: Row): T = from(r)
+  def apply(t: T): Row = to(t)
+}
+
+object BeamSchemaType {
+  implicit def apply[T: BeamSchemaField]: BeamSchemaType[T] =
+    BeamSchemaType[T](CaseMapper.identity)
+
+  def apply[T](cm: CaseMapper)(implicit f: BeamSchemaField[T]): BeamSchemaType[T] = {
+    f match {
+      case r: BeamSchemaField.Record[_] =>
+        r.schema(cm) // fail fast on bad annotations
+        new BeamSchemaType[T] {
+          private val caseMapper: CaseMapper = cm
+          @transient override lazy val schema: Schema = r.schema(caseMapper)
+
+          override def from(v: Row): T = r.from(v)(caseMapper)
+          override def to(v: T): Row = r.to(v)(caseMapper)
+        }
+      case _ =>
+        throw new IllegalArgumentException(
+          s"BeamSchemaType can only be created from Record. Got $f"
+        )
+    }
+  }
+}
+
+sealed trait BeamSchemaField[T] extends Serializable {
+  type FromT
+  type ToT
+  def fieldType(cm: CaseMapper): FieldType
+  def from(v: FromT)(cm: CaseMapper): T
+  def to(v: T)(cm: CaseMapper): ToT
+  def fromAny(v: Any)(cm: CaseMapper): T = from(v.asInstanceOf[FromT])(cm)
+}
+
+object BeamSchemaField {
+  sealed trait Aux[T, From, To] extends BeamSchemaField[T] {
+    override type FromT = From
+    override type ToT = To
+  }
+
+  private[magnolify] def aux[T, From, To](
+    ft: CaseMapper => FieldType
+  )(fromFn: From => T)(toFn: T => To): BeamSchemaField[T] =
+    new Aux[T, From, To] {
+      override def fieldType(cm: CaseMapper): FieldType = ft(cm)
+      override def from(v: FromT)(cm: CaseMapper): T = fromFn(v)
+      override def to(v: T)(cm: CaseMapper): ToT = toFn(v)
+    }
+
+//  private[magnolify] def aux2[T, Repr](fieldTypeFn: CaseMapper => FieldType)(fromFn: Repr => T)(
+//    toFn: T => Repr
+//  ): BeamSchemaField[T] =
+//    aux[T, Repr, Repr](fieldTypeFn)(fromFn)(toFn)
+
+  private[magnolify] def id[T](ft: CaseMapper => FieldType): BeamSchemaField[T] =
+    aux[T, T, T](ft)(identity)(identity)
+
+  private[magnolify] def logicalId[T](ft: CaseMapper => FieldType): BeamSchemaField[T] = id(ft)
+
+//  private[magnolify] def logicalId[T](ft: CaseMapper => FieldType): BeamSchemaLogicalField[T] =
+//    new BeamSchemaLogicalField[T] {
+//      type FromT = T
+//      type ToT = T
+//      override def fieldType(cm: CaseMapper): FieldType = ft(cm)
+//      override def from(v: FromT)(cm: CaseMapper): T = v
+//      override def to(v: T)(cm: CaseMapper): ToT = v
+//    }
+
+  def from[T]: FromWord[T] = new FromWord[T]
+
+  class FromWord[T] {
+    def apply[U](f: T => U)(g: U => T)(implicit bsf: BeamSchemaField[T]): BeamSchemaField[U] =
+      new Aux[U, bsf.FromT, bsf.ToT] {
+        override def fieldType(cm: CaseMapper): FieldType = bsf.fieldType(cm)
+        override def from(v: FromT)(cm: CaseMapper): U = f(bsf.from(v)(cm))
+        override def to(v: U)(cm: CaseMapper): ToT = bsf.to(g(v))(cm)
+      }
+  }
+
+//  def logicalType[T]: LogicalWord[T] = new LogicalWord[T]
+//
+//  class LogicalWord[T] {
+//    def apply[From, To](lt: LogicalType[From, ?], nullable: Boolean = false): BeamSchemaLogicalField[T] = {
+//      new BeamSchemaLogicalField[T] {
+//        type FromT = From
+//        type ToT = To
+//        override def logicalType: LogicalType[From, To] = lt
+//        override def fieldType(cm: CaseMapper): FieldType =
+//          FieldType.logicalType(logicalType).withNullable(nullable)
+//        override def from(v: From)(cm: CaseMapper): T =
+//          throw new UnsupportedOperationException("Do not call from() on logical types")
+//        override def to(v: T)(cm: CaseMapper): To =
+//          throw new UnsupportedOperationException("Do not call to() on logical types")
+//      }
+//    }
+//  }
+
+  sealed trait Record[T] extends Aux[T, Row, Row] {
+    @transient private lazy val schemaCache: concurrent.Map[ju.UUID, Schema] =
+      concurrent.TrieMap.empty
+    protected def buildSchema(cm: CaseMapper): Schema
+    def schema(cm: CaseMapper): Schema = schemaCache.getOrElseUpdate(cm.uuid, buildSchema(cm))
+  }
+
+  // ////////////////////////////////////////////////
+
+  type Typeclass[T] = BeamSchemaField[T]
+  implicit def gen[T]: BeamSchemaField[T] = macro Magnolia.gen[T]
+
+  @implicitNotFound("Cannot derive BeamSchemaField for sealed trait") // TODO does this make sense?
+  private sealed trait Dispatchable[T]
+  def split[T: Dispatchable](sealedTrait: SealedTrait[Typeclass, T]): BeamSchemaField[T] = ???
+
+  def join[T](caseClass: CaseClass[Typeclass, T]): BeamSchemaField[T] = {
+    if (caseClass.isValueClass) {
+      // FIXME
+//      val p = caseClass.parameters.head
+//      val tc = p.typeclass
+//      new BeamSchemaField[T] {
+//        override type FromT = tc.FromT
+//        override type ToT = tc.ToT
+//        // override protected def buildSchema(cm: CaseMapper): Schema = tc.buildSchema(cm)
+//        override def from(v: FromT)(cm: CaseMapper): T = caseClass.construct(_ => tc.fromAny(v)(cm))
+//        override def to(v: T)(cm: CaseMapper): ToT = tc.to(p.dereference(v))(cm)
+//      }
+      ???
+    } else {
+      new Record[T] {
+        override def fieldType(cm: CaseMapper): FieldType = FieldType.row(schema(cm))
+
+        override protected def buildSchema(cm: CaseMapper): Schema =
+          caseClass.parameters
+            .foldLeft(Schema.builder()) { case (s, p) =>
+              s.addField(p.label, p.typeclass.fieldType(cm))
+            }
+            .build()
+
+        override def from(v: Row)(cm: CaseMapper): T =
+          caseClass.construct(p => p.typeclass.fromAny(v.getValue[Any](p.index))(cm))
+
+        override def to(v: T)(cm: CaseMapper): Row = {
+          val zero: Either[Row.Builder, Row.FieldValueBuilder] = Left(Row.withSchema(schema(cm)))
+          val eitherBuilder = caseClass.parameters
+            .foldLeft(zero) { (eitherBuilder, p) =>
+              val value = p.typeclass.to(p.dereference(v))(cm)
+              eitherBuilder match {
+                case Left(rowBuilder)    => Right(rowBuilder.withFieldValue(p.index, value))
+                case Right(fieldBuilder) => Right(fieldBuilder.withFieldValue(p.index, value))
+              }
+            }
+          eitherBuilder match {
+            case Left(rb)  => rb.build()
+            case Right(fb) => fb.build()
+          }
+        }
+      }
+    }
+  }
+
+  // BYTE	An 8-bit signed value
+  implicit val bsfByte: BeamSchemaField[Byte] = id[Byte](_ => FieldType.BYTE)
+  implicit val bsfChar: BeamSchemaField[Char] = from[Byte](_.toChar)(_.toByte)
+  // INT16	A 16-bit signed value
+  implicit val bsfShort: BeamSchemaField[Short] = id[Short](_ => FieldType.INT16)
+  // INT32	A 32-bit signed value
+  implicit val bsfInt: BeamSchemaField[Int] = id[Int](_ => FieldType.INT32)
+  // INT64	A 64-bit signed value
+  implicit val bsfLong: BeamSchemaField[Long] = id[Long](_ => FieldType.INT64)
+  // FLOAT	A 32-bit IEEE 754 floating point number
+  implicit val bsfFloat: BeamSchemaField[Float] = id[Float](_ => FieldType.FLOAT)
+  // DOUBLE	A 64-bit IEEE 754 floating point number
+  implicit val bsfDouble: BeamSchemaField[Double] = id[Double](_ => FieldType.DOUBLE)
+  // STRING	A string
+  implicit val bsfString: BeamSchemaField[String] = id[String](_ => FieldType.STRING)
+  implicit val bsfCharSeq: BeamSchemaField[CharSequence] =
+    from[String](_.asInstanceOf[CharSequence])(_.toString)
+  // BOOLEAN	A boolean value
+  implicit val bsfBoolean: BeamSchemaField[Boolean] = id[Boolean](_ => FieldType.BOOLEAN)
+  // BYTES	A raw byte array
+  implicit val bsfByteArray: BeamSchemaField[Array[Byte]] = id[Array[Byte]](_ => FieldType.BYTES)
+  implicit val bsfByteBuffer: BeamSchemaField[ByteBuffer] =
+    from[Array[Byte]](x => ByteBuffer.wrap(x))(_.array())
+  implicit val bsfByteString: BeamSchemaField[ByteString] =
+    from[Array[Byte]](x => ByteString.copyFrom(x))(_.toByteArray)
+  // DECIMAL	An arbitrary-precision decimal type
+  implicit val bsfDecimal: BeamSchemaField[BigDecimal] =
+    aux[BigDecimal, java.math.BigDecimal, java.math.BigDecimal](_ => FieldType.DECIMAL)(
+      BigDecimal.apply
+    )(_.bigDecimal)
+
+  implicit val bsfUUID: BeamSchemaField[ju.UUID] =
+    logicalId[ju.UUID](_ => FieldType.logicalType(new logicaltypes.UuidLogicalType))
+//    new BeamSchemaLogicalField[ju.UUID] {
+//      type FromT = ju.UUID
+//      type ToT = ju.UUID
+//      val logicalType = new logicaltypes.UuidLogicalType
+//      override def fieldType(cm: CaseMapper): FieldType = FieldType.logicalType(logicalType)
+//      override def from(v: ju.UUID)(cm: CaseMapper): ju.UUID = v
+//      override def to(v: ju.UUID)(cm: CaseMapper): ju.UUID = v
+//    }
+
+//  implicit val bsfLocalDate: BeamSchemaField[jt.LocalDate] =
+//    from[Long](LocalDate.ofEpochDay)(_.toEpochDay)
+  implicit val bsfLocalDate: BeamSchemaField[jt.LocalDate] =
+    logicalId[jt.LocalDate](_ => FieldType.logicalType(new logicaltypes.Date))
+  private lazy val EpochJodaDate = new joda.LocalDate(1970, 1, 1)
+  implicit val bsfJodaLocalDate: BeamSchemaField[joda.LocalDate] =
+    from[Int](daysFromEpoch => EpochJodaDate.plusDays(daysFromEpoch))(d =>
+      joda.Days.daysBetween(EpochJodaDate, d).getDays
+    )
+
+  implicit def bsfEnum[T](implicit et: EnumType[T], lp: shapeless.LowPriority): BeamSchemaField[T] =
+    new BeamSchemaField[T] {
+      type FromT = logicaltypes.EnumerationType.Value
+      type ToT = logicaltypes.EnumerationType.Value
+
+      @transient private lazy val enumTypeCache: concurrent.Map[ju.UUID, EnumType[T]] =
+        concurrent.TrieMap.empty
+      @transient private lazy val beamEnumTypeCache
+        : concurrent.Map[ju.UUID, logicaltypes.EnumerationType] =
+        concurrent.TrieMap.empty
+
+      private def enumType(cm: CaseMapper): EnumType[T] =
+        enumTypeCache.getOrElseUpdate(cm.uuid, et.map(cm))
+      private def beamEnumType(cm: CaseMapper): logicaltypes.EnumerationType =
+        beamEnumTypeCache.getOrElseUpdate(
+          cm.uuid,
+          logicaltypes.EnumerationType.create(enumType(cm).values.asJava)
+        )
+      override def fieldType(cm: CaseMapper): FieldType = FieldType.logicalType(beamEnumType(cm))
+      override def to(v: T)(cm: CaseMapper): ToT = beamEnumType(cm).valueOf(enumType(cm).to(v))
+      override def from(v: FromT)(cm: CaseMapper): T =
+        enumType(cm).from(beamEnumType(cm).toString(v))
+    }
+
+  implicit def bsfMap[K, V](implicit
+    bsfK: BeamSchemaField[K],
+    bsfV: BeamSchemaField[V]
+  ): BeamSchemaField[Map[K, V]] =
+    new Aux[Map[K, V], ju.Map[bsfK.FromT, bsfV.FromT], ju.Map[bsfK.ToT, bsfV.ToT]] {
+      override def fieldType(cm: CaseMapper): FieldType =
+        FieldType.map(bsfK.fieldType(cm), bsfV.fieldType(cm))
+      // TODO need to convert/not convert logical types
+      override def from(v: ju.Map[bsfK.FromT, bsfV.FromT])(cm: CaseMapper): Map[K, V] =
+        v.asScala.map { case (k, v) => bsfK.from(k)(cm) -> bsfV.from(v)(cm) }.toMap
+      override def to(v: Map[K, V])(cm: CaseMapper): ju.Map[bsfK.ToT, bsfV.ToT] =
+        v.map { case (k, v) => bsfK.to(k)(cm) -> bsfV.to(v)(cm) }.asJava
+    }
+
+  implicit def bsfIterable[T, C[_]](implicit
+    f: BeamSchemaField[T],
+    ti: C[T] => Iterable[T],
+    fc: FactoryCompat[T, C[T]]
+  ): BeamSchemaField[C[T]] = {
+//    f match {
+//      case l: BeamSchemaLogicalField[T] =>
+//        new BeamSchemaLogicalField[C[T]] {
+//          type FromT = ju.List[l.FromT]
+//          type ToT = ju.List[l.ToT]
+//          override def fieldType(cm: CaseMapper): FieldType = FieldType.iterable(l.fieldType(cm))
+//          override def to(v: C[T])(cm: CaseMapper): ToT =
+//            v.iterator.map(l.to(_)(cm)).toList.asJava
+//          override def from(v: FromT)(cm: CaseMapper): C[T] =
+//            fc.fromSpecific(v.asScala.iterator.map(p => l.from(p)(cm)))
+//        }
+//      case _: BeamSchemaField[_] =>
+    new Aux[C[T], ju.List[f.FromT], ju.List[f.ToT]] {
+      override def from(v: ju.List[f.FromT])(cm: CaseMapper): C[T] =
+        fc.fromSpecific(v.asScala.iterator.map(p => f.from(p)(cm)))
+      override def to(v: C[T])(cm: CaseMapper): ju.List[f.ToT] =
+        v.iterator.map(f.to(_)(cm)).toList.asJava
+      override def fieldType(cm: CaseMapper): FieldType = FieldType.iterable(f.fieldType(cm))
+    }
+//    }
+  }
+
+  implicit def bsfOption[T](implicit f: BeamSchemaField[T]): BeamSchemaField[Option[T]] = {
+//    f match {
+//      case l: BeamSchemaLogicalField[T] =>
+//        new BeamSchemaLogicalField[Option[T]] {
+//          type FromT = l.FromT
+//          type ToT = l.ToT
+//          override def fieldType(cm: CaseMapper): FieldType = l.fieldType(cm).withNullable(true)
+//          override def to(v: Option[T])(cm: CaseMapper): ToT = v match {
+//            case None        => null.asInstanceOf[ToT]
+//            case Some(value) => l.to(value)(cm)
+//          }
+//          override def from(v: FromT)(cm: CaseMapper): Option[T] = Option(v).map(l.from(_)(cm))
+//        }
+//      case _: BeamSchemaField[T] =>
+    new Aux[Option[T], f.FromT, f.ToT] {
+      override def from(v: f.FromT)(cm: CaseMapper): Option[T] =
+        if (v == null) None else Some(f.from(v)(cm))
+      override def to(v: Option[T])(cm: CaseMapper): f.ToT = v match {
+        case None    => null.asInstanceOf[f.ToT]
+        case Some(x) => f.to(x)(cm)
+      }
+      override def fieldType(cm: CaseMapper): FieldType = f.fieldType(cm).withNullable(true)
+    }
+//    }
+  }
+}
diff --git a/beam/src/main/scala/magnolify/beam/logical/package.scala b/beam/src/main/scala/magnolify/beam/logical/package.scala
new file mode 100644
index 000000000..6475a2fdc
--- /dev/null
+++ b/beam/src/main/scala/magnolify/beam/logical/package.scala
@@ -0,0 +1,92 @@
+/*
+ * 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
+
+import org.joda.time as joda
+import java.time as jt
+
+package object logical {
+  import magnolify.shared.Time._
+
+  object millis {
+    // joda
+    // DATETIME	A timestamp represented as milliseconds since the epoch
+    // joda.DateTime only has millisecond resolution
+    implicit val bsfJodaDateTimeMillis: BeamSchemaField[joda.DateTime] =
+      BeamSchemaField.from[Long](millisToJodaDateTime)(millisFromJodaDateTime)
+    // DATETIME	A timestamp represented as milliseconds since the epoch
+    implicit val bsfJodaInstantMillis: BeamSchemaField[joda.Instant] =
+      BeamSchemaField.from[Long](millisToJodaInstant)(millisFromJodaInstant)
+    implicit val bsfJodaDurationMillis: BeamSchemaField[joda.Duration] =
+      BeamSchemaField.from[Long](millisToJodaDuration)(millisFromJodaDuration)
+    implicit val bsfJodaLocalTimeMillis: BeamSchemaField[joda.LocalTime] =
+      BeamSchemaField.from[Int](millisToJodaLocalTime)(millisFromJodaLocalTime)
+    // java
+    implicit val bsfInstantMillis: BeamSchemaField[jt.Instant] =
+      BeamSchemaField.from[Long](millisToInstant)(millisFromInstant)
+    implicit val bsLocalTimeMillis: BeamSchemaField[jt.LocalTime] =
+      BeamSchemaField.from[Int](millisToLocalTime)(millisFromLocalTime)
+    implicit val bsfLocalDateTimeMillis: BeamSchemaField[jt.LocalDateTime] =
+      BeamSchemaField.from[Long](millisToLocalDateTime)(millisFromLocalDateTime)
+    implicit val bsfDurationMillis: BeamSchemaField[jt.Duration] =
+      BeamSchemaField.from[Long](millisToDuration)(millisFromDuration)
+  }
+
+  object micros {
+    // joda.DateTime only has millisecond resolution, so excess precision is discarded
+    implicit val bsfJodaDateTimeMicros: BeamSchemaField[joda.DateTime] =
+      BeamSchemaField.from[Long](microsToJodaDateTime)(microsFromJodaDateTime)
+    // joda.Instant has millisecond precision, excess precision discarded
+    implicit val bsfJodaInstantMicros: BeamSchemaField[joda.Instant] =
+      BeamSchemaField.from[Long](microsToJodaInstant)(microsFromJodaInstant)
+    // joda.Duration has millisecond precision, excess precision discarded
+    implicit val bsfJodaDurationMicros: BeamSchemaField[joda.Duration] =
+      BeamSchemaField.from[Long](microsToJodaDuration)(microsFromJodaDuration)
+    implicit val bsfJodaLocalTimeMicros: BeamSchemaField[joda.LocalTime] =
+      BeamSchemaField.from[Long](microsToJodaLocalTime)(microsFromJodaLocalTime)
+    // java
+    implicit val bsfInstantMicros: BeamSchemaField[jt.Instant] =
+      BeamSchemaField.from[Long](microsToInstant)(microsFromInstant)
+    implicit val bsLocalTimeMicros: BeamSchemaField[jt.LocalTime] =
+      BeamSchemaField.from[Long](microsToLocalTime)(microsFromLocalTime)
+    implicit val bsfLocalDateTimeMicros: BeamSchemaField[jt.LocalDateTime] =
+      BeamSchemaField.from[Long](microsToLocalDateTime)(microsFromLocalDateTime)
+    implicit val bsfDurationMicros: BeamSchemaField[jt.Duration] =
+      BeamSchemaField.from[Long](microsToDuration)(microsFromDuration)
+  }
+
+  object nanos {
+    // joda.DateTime only has millisecond resolution
+    implicit val bsfJodaDateTimeNanos: BeamSchemaField[joda.DateTime] =
+      BeamSchemaField.from[Long](nanosToJodaDateTime)(nanosFromJodaDateTime)
+    implicit val bsfJodaInstantNanos: BeamSchemaField[joda.Instant] =
+      BeamSchemaField.from[Long](nanosToJodaInstant)(nanosFromJodaInstant)
+    implicit val bsfJodaDurationNanos: BeamSchemaField[joda.Duration] =
+      BeamSchemaField.from[Long](nanosToJodaDuration)(nanosFromJodaDuration)
+    implicit val bsfJodaLocalTimeNanos: BeamSchemaField[joda.LocalTime] =
+      BeamSchemaField.from[Long](nanosToJodaLocalTime)(nanosFromJodaLocalTime)
+    // java
+    implicit val bsfInstantNanos: BeamSchemaField[jt.Instant] =
+      BeamSchemaField.from[Long](nanosToInstant)(nanosFromInstant)
+    implicit val bsLocalTimeNanos: BeamSchemaField[jt.LocalTime] =
+      BeamSchemaField.from[Long](nanosToLocalTime)(nanosFromLocalTime)
+    implicit val bsfLocalDateTimeNanos: BeamSchemaField[jt.LocalDateTime] =
+      BeamSchemaField.from[Long](nanosToLocalDateTime)(nanosFromLocalDateTime)
+    implicit val bsfDurationNanos: BeamSchemaField[jt.Duration] =
+      BeamSchemaField.from[Long](nanosToDuration)(nanosFromDuration)
+  }
+}
diff --git a/beam/src/main/scala/magnolify/beam/unsafe/package.scala b/beam/src/main/scala/magnolify/beam/unsafe/package.scala
new file mode 100644
index 000000000..9379200cf
--- /dev/null
+++ b/beam/src/main/scala/magnolify/beam/unsafe/package.scala
@@ -0,0 +1,8 @@
+package magnolify.beam
+
+import magnolify.shared._
+
+package object unsafe {
+  implicit def afUnsafeEnum[T: EnumType]: BeamSchemaField[UnsafeEnum[T]] =
+    BeamSchemaField.from[String](UnsafeEnum.from[T])(UnsafeEnum.to[T])
+}
diff --git a/beam/src/test/scala/magnolify/beam/BeamSchemaTypeSuite.scala b/beam/src/test/scala/magnolify/beam/BeamSchemaTypeSuite.scala
new file mode 100644
index 000000000..c15180c04
--- /dev/null
+++ b/beam/src/test/scala/magnolify/beam/BeamSchemaTypeSuite.scala
@@ -0,0 +1,133 @@
+/*
+ * 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
+
+import cats.*
+import magnolify.cats.auto.*
+import magnolify.cats.TestEq.*
+import magnolify.scalacheck.auto.*
+import magnolify.scalacheck.TestArbitrary.*
+import magnolify.test.MagnolifySuite
+import magnolify.test.Simple.*
+import org.joda.time as joda
+import org.scalacheck.{Arbitrary, Gen, Prop}
+
+import java.time.{Instant, LocalDate, LocalDateTime, LocalTime}
+import java.util.UUID
+import scala.reflect.ClassTag
+
+class BeamSchemaTypeSuite extends MagnolifySuite {
+  private def test[T: Arbitrary: ClassTag](implicit
+    bst: BeamSchemaType[T],
+    eq: Eq[T]
+  ): Unit = {
+    // Ensure serializable even after evaluation of `schema`
+    bst.schema: Unit
+    ensureSerializable(bst)
+
+    property(className[T]) {
+      Prop.forAll { (t: T) =>
+        val converted = bst.apply(t)
+        val roundtripped = bst.apply(converted)
+        Prop.all(eq.eqv(t, roundtripped))
+      }
+    }
+  }
+
+  test[Integers]
+  test[Floats]
+  test[Required]
+  test[Nullable]
+  test[Repeated]
+  test[Nested]
+  test[Collections]
+  test[MoreCollections]
+
+  test[Maps]
+  test[Logical]
+
+  {
+    import magnolify.beam.unsafe._
+    import magnolify.shared.TestEnumType._
+    test[Enums]
+    test[UnsafeEnums]
+  }
+
+  implicit val arbBigDecimal: Arbitrary[BigDecimal] =
+    Arbitrary(Gen.chooseNum(0, Int.MaxValue).map(BigDecimal(_)))
+  test[Decimal]
+
+  {
+    import magnolify.beam.logical.millis._
+    test[Time]
+    test[Joda]
+  }
+
+//  {
+//    //   FIXME need special Eq instances that are lossy
+//    import magnolify.beam.logical.micros._
+//    test[Time]
+//    test[Joda]
+//  }
+//
+//  {
+////   FIXME need special Eq instances that are lossy
+//    import magnolify.beam.logical.nanos._
+//    test[Time]
+//    test[Joda]
+//  }
+
+//  {
+//    implicit val bst: BeamSchemaType[LowerCamel] =
+//      BeamSchemaType[LowerCamel](CaseMapper(_.toUpperCase))
+//    test[LowerCamel]
+//
+//    test("LowerCamel mapping") {
+//      val schema = bst.schema
+//      // FIXME
+//    }
+//  }
+
+}
+
+case class Decimal(bd: BigDecimal, bdo: Option[BigDecimal])
+case class Logical(
+  u: UUID,
+  uo: Option[UUID],
+  ul: List[UUID],
+  ulo: List[Option[UUID]]
+)
+
+case class Time(
+  i: Instant,
+  d: LocalDate,
+  dt: LocalDateTime,
+  t: LocalTime
+)
+case class Joda(
+  i: joda.Instant,
+  dt: joda.DateTime,
+  lt: joda.LocalTime,
+  d: joda.Duration
+)
+case class Maps(
+  ms: Map[String, String],
+  mi: Map[Int, Int],
+  mso: Map[Option[String], Option[String]],
+  ml: Map[UUID, UUID],
+  mlo: Map[Option[UUID], Option[UUID]]
+)
diff --git a/build.sbt b/build.sbt
index f601f343e..6e271ffba 100644
--- a/build.sbt
+++ b/build.sbt
@@ -24,6 +24,7 @@ val magnoliaScala3Version = "1.3.7"
 
 val algebirdVersion = "0.13.10"
 val avroVersion = Option(sys.props("avro.version")).getOrElse("1.11.3")
+val beamVersion = "2.57.0"
 val bigqueryVersion = "v2-rev20240229-2.0.0"
 val bigtableVersion = "2.41.0"
 val catsVersion = "2.10.0"
@@ -128,7 +129,7 @@ ThisBuild / crossScalaVersions := Seq(scala3, scala213, scala212)
 ThisBuild / githubWorkflowTargetBranches := Seq("main")
 ThisBuild / githubWorkflowJavaVersions := Seq(java17, java11)
 ThisBuild / tlJdkRelease := Some(8)
-ThisBuild / tlFatalWarnings := true
+ThisBuild / tlFatalWarnings := false
 ThisBuild / tlCiHeaderCheck := true
 ThisBuild / tlCiScalafmtCheck := true
 ThisBuild / tlCiDocCheck := true
@@ -325,6 +326,7 @@ lazy val root = tlCrossRootProject
   )
   .aggregate(
     avro,
+    beam,
     bigquery,
     bigtable,
     bom,
@@ -488,6 +490,25 @@ lazy val avro = project
     )
   )
 
+lazy val beam = project
+  .in(file("beam"))
+  .dependsOn(
+    shared,
+    cats % "test->test",
+    scalacheck % "test->test",
+    test % "test->test"
+  )
+  .settings(
+    commonSettings,
+    protobufSettings,
+    moduleName := "magnolify-beam",
+    description := "Magnolia add-on for Apache Beam",
+    libraryDependencies ++= Seq(
+      "org.apache.beam" % "beam-sdks-java-core" % beamVersion % Provided,
+      "com.google.protobuf" % "protobuf-java" % protobufVersion % ProtobufConfig,
+    )
+  )
+
 lazy val bigquery = project
   .in(file("bigquery"))
   .dependsOn(
diff --git a/cats/src/test/scala/magnolify/cats/TestEq.scala b/cats/src/test/scala/magnolify/cats/TestEq.scala
index 152cccbd8..22ecb0233 100644
--- a/cats/src/test/scala/magnolify/cats/TestEq.scala
+++ b/cats/src/test/scala/magnolify/cats/TestEq.scala
@@ -60,6 +60,8 @@ object TestEq {
   implicit val eqJodaDateTime: Eq[joda.DateTime] = Eq.fromUniversalEquals
   implicit val eqJodaLocalTime: Eq[joda.LocalTime] = Eq.fromUniversalEquals
   implicit val eqJodaLocalDateTime: Eq[joda.LocalDateTime] = Eq.fromUniversalEquals
+  implicit val eqJodaDuration: Eq[joda.Duration] = Eq.fromUniversalEquals
+  implicit val eqJodaInstant: Eq[joda.Instant] = Eq.fromUniversalEquals
 
   // enum
   implicit lazy val eqJavaEnum: Eq[JavaEnums.Color] = Eq.fromUniversalEquals
diff --git a/parquet/src/main/scala/magnolify/parquet/logical/package.scala b/parquet/src/main/scala/magnolify/parquet/logical/package.scala
index 9223c4342..2883349e8 100644
--- a/parquet/src/main/scala/magnolify/parquet/logical/package.scala
+++ b/parquet/src/main/scala/magnolify/parquet/logical/package.scala
@@ -23,84 +23,69 @@ import org.apache.parquet.schema.LogicalTypeAnnotation
 import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
 
 package object logical {
+  import magnolify.shared.Time._
   // TIME (millis i32, micros i64, nanos, i64), UTC true/false
   // TIMESTAMP (millis, micros, nanos), UTC true/false
 
-  object millis {
-    private val unit = TimeUnit.MILLIS
+  private trait TimeTypes {
+    protected def unit: TimeUnit
+    protected def ts(adjusted: Boolean): LogicalTypeAnnotation =
+      LogicalTypeAnnotation.timestampType(adjusted, unit)
+    protected def time(adjusted: Boolean): LogicalTypeAnnotation =
+      LogicalTypeAnnotation.timeType(adjusted, unit)
+  }
+
+  object millis extends TimeTypes {
+    protected val unit = TimeUnit.MILLIS
 
     // TIMESTAMP
     implicit val pfTimestampMillis: Primitive[Instant] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(true, unit))(
-        Instant.ofEpochMilli
-      )(_.toEpochMilli)
+      ParquetField.logicalType[Long](ts(true))(millisToInstant)(millisFromInstant)
     implicit val pfLocalDateTimeMillis: Primitive[LocalDateTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(false, unit))(ms =>
-        LocalDateTime.ofInstant(Instant.ofEpochMilli(ms), ZoneOffset.UTC)
-      )(
-        _.toInstant(ZoneOffset.UTC).toEpochMilli
-      )
+      ParquetField.logicalType[Long](ts(false))(millisToLocalDateTime)(millisFromLocalDateTime)
 
     // TIME
     implicit val pfOffsetTimeMillis: Primitive[OffsetTime] =
-      ParquetField.logicalType[Int](LogicalTypeAnnotation.timeType(true, unit))(ms =>
+      ParquetField.logicalType[Int](time(true))(ms =>
         LocalTime.ofNanoOfDay(ms * 1000000L).atOffset(ZoneOffset.UTC)
       )(t => (t.toLocalTime.toNanoOfDay / 1000000).toInt)
     implicit val pfLocalTimeMillis: Primitive[LocalTime] =
-      ParquetField.logicalType[Int](LogicalTypeAnnotation.timeType(false, unit))(ms =>
-        LocalTime.ofNanoOfDay(ms * 1000000L)
-      )(t => (t.toNanoOfDay / 1000000).toInt)
+      ParquetField.logicalType[Int](time(false))(millisToLocalTime)(millisFromLocalTime)
   }
 
-  object micros {
-    private val unit = TimeUnit.MICROS
+  object micros extends TimeTypes {
+    override protected val unit = TimeUnit.MICROS
 
     // TIMESTAMP
-    implicit val pfTimestampMillis: Primitive[Instant] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(true, unit))(us =>
-        Instant.ofEpochMilli(us / 1000)
-      )(_.toEpochMilli * 1000)
-    implicit val pfLocalDateTimeMillis: Primitive[LocalDateTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(false, unit))(us =>
-        LocalDateTime.ofInstant(Instant.ofEpochMilli(us / 1000), ZoneOffset.UTC)
-      )(
-        _.toInstant(ZoneOffset.UTC).toEpochMilli * 1000
-      )
+    implicit val pfTimestampMicros: Primitive[Instant] =
+      ParquetField.logicalType[Long](ts(true))(microsToInstant)(microsFromInstant)
+    implicit val pfLocalDateTimeMicros: Primitive[LocalDateTime] =
+      ParquetField.logicalType[Long](ts(false))(microsToLocalDateTime)(microsFromLocalDateTime)
 
     // TIME
     implicit val pfOffsetTimeMicros: Primitive[OffsetTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timeType(true, unit))(us =>
+      ParquetField.logicalType[Long](time(true))(us =>
         LocalTime.ofNanoOfDay(us * 1000).atOffset(ZoneOffset.UTC)
       )(_.toLocalTime.toNanoOfDay / 1000)
     implicit val pfLocalTimeMicros: Primitive[LocalTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timeType(false, unit))(us =>
-        LocalTime.ofNanoOfDay(us * 1000)
-      )(_.toNanoOfDay / 1000)
+      ParquetField.logicalType[Long](time(false))(microsToLocalTime)(microsFromLocalTime)
   }
 
-  object nanos {
-    private val unit = TimeUnit.NANOS
+  object nanos extends TimeTypes {
+    override protected val unit = TimeUnit.NANOS
 
     // TIMESTAMP
     implicit val pfTimestampMillis: Primitive[Instant] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(true, unit))(ns =>
-        Instant.ofEpochMilli(ns / 1000000)
-      )(_.toEpochMilli * 1000000)
+      ParquetField.logicalType[Long](ts(true))(nanosToInstant)(nanosFromInstant)
     implicit val pfLocalDateTimeMillis: Primitive[LocalDateTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timestampType(false, unit))(ns =>
-        LocalDateTime.ofInstant(Instant.ofEpochMilli(ns / 1000000), ZoneOffset.UTC)
-      )(
-        _.toInstant(ZoneOffset.UTC).toEpochMilli * 1000000
-      )
+      ParquetField.logicalType[Long](ts(false))(nanosToLocalDateTime)(nanosFromLocalDateTime)
 
     // TIME
     implicit val pfOffsetTimeNanos: Primitive[OffsetTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timeType(true, unit))(ns =>
+      ParquetField.logicalType[Long](time(true))(ns =>
         LocalTime.ofNanoOfDay(ns).atOffset(ZoneOffset.UTC)
       )(_.toLocalTime.toNanoOfDay)
     implicit val pfLocalTimeNanos: Primitive[LocalTime] =
-      ParquetField.logicalType[Long](LogicalTypeAnnotation.timeType(false, unit))(
-        LocalTime.ofNanoOfDay
-      )(_.toNanoOfDay)
+      ParquetField.logicalType[Long](time(false))(nanosToLocalTime)(nanosFromLocalTime)
   }
 }
diff --git a/scalacheck/src/test/scala/magnolify/scalacheck/TestArbitrary.scala b/scalacheck/src/test/scala/magnolify/scalacheck/TestArbitrary.scala
index c0d9de75e..e3da13639 100644
--- a/scalacheck/src/test/scala/magnolify/scalacheck/TestArbitrary.scala
+++ b/scalacheck/src/test/scala/magnolify/scalacheck/TestArbitrary.scala
@@ -79,6 +79,10 @@ object TestArbitrary {
       joda.LocalDateTime.parse(ldt.toString)
     }
   }
+  implicit val arbJodaDuration: Arbitrary[joda.Duration] =
+    Arbitrary(Gen.posNum[Long].map(joda.Duration.millis))
+  implicit val arbJodaInstant: Arbitrary[joda.Instant] =
+    Arbitrary(Gen.posNum[Long].map(l => new joda.Instant(l)))
 
   // enum
   implicit lazy val arbJavaEnum: Arbitrary[JavaEnums.Color] =
diff --git a/shared/src/main/scala/magnolify/shared/Time.scala b/shared/src/main/scala/magnolify/shared/Time.scala
new file mode 100644
index 000000000..8eb6f177f
--- /dev/null
+++ b/shared/src/main/scala/magnolify/shared/Time.scala
@@ -0,0 +1,134 @@
+package magnolify.shared
+
+import org.joda.time as joda
+import java.time.{Duration, Instant, LocalDateTime, LocalTime, ZoneOffset}
+import java.util.concurrent.TimeUnit
+
+object Time {
+//  @inline def microsToMillis(micros: Long): Long = TimeUnit.MICROSECONDS.toMillis(micros)
+//  @inline def millisToMicros(millis: Long): Long = TimeUnit.MILLISECONDS.toMicros(millis)
+
+  // millis /////////////////////////////////////////////////////
+  @inline def millisToInstant(millisFromEpoch: Long): Instant =
+    Instant.ofEpochMilli(millisFromEpoch)
+  @inline def millisFromInstant(instant: Instant): Long = instant.toEpochMilli
+  @inline def millisToJodaInstant(millisFromEpoch: Long): joda.Instant =
+    new joda.Instant(millisFromEpoch)
+  @inline def millisFromJodaInstant(instant: joda.Instant): Long = instant.getMillis
+
+  @inline def millisToLocalTime(millisFromMidnight: Int): LocalTime =
+    LocalTime.ofNanoOfDay(TimeUnit.MILLISECONDS.toNanos(millisFromMidnight.toLong))
+  @inline def millisFromLocalTime(lt: LocalTime): Int =
+    TimeUnit.NANOSECONDS.toMillis(lt.toNanoOfDay).toInt
+  @inline def millisToJodaLocalTime(millisFromMidnight: Int): joda.LocalTime =
+    joda.LocalTime.fromMillisOfDay(millisFromMidnight.toLong)
+  @inline def millisFromJodaLocalTime(lt: joda.LocalTime): Int = lt.millisOfDay().get()
+
+  @inline def millisToJodaDateTime(millisFromEpoch: Long): joda.DateTime =
+    new joda.DateTime(millisFromEpoch, joda.DateTimeZone.UTC)
+  @inline def millisFromJodaDateTime(dt: joda.DateTime): Long = dt.getMillis
+
+  @inline def millisToLocalDateTime(millisFromEpoch: Long): LocalDateTime =
+    LocalDateTime.ofInstant(millisToInstant(millisFromEpoch), ZoneOffset.UTC)
+  @inline def millisFromLocalDateTime(ldt: LocalDateTime): Long =
+    millisFromInstant(ldt.toInstant(ZoneOffset.UTC))
+  @inline def millisToJodaLocalDateTime(millisFromEpoch: Long): joda.LocalDateTime =
+    new joda.LocalDateTime(millisFromEpoch, joda.DateTimeZone.UTC)
+  @inline def millisFromJodaLocalDateTime(ldt: joda.LocalDateTime): Long =
+    ldt.toDateTime(joda.DateTimeZone.UTC).getMillis
+
+  @inline def millisToDuration(millis: Long): Duration = Duration.ofMillis(millis)
+  @inline def millisFromDuration(d: Duration): Long =
+    TimeUnit.SECONDS.toMillis(d.getSeconds) + TimeUnit.NANOSECONDS.toMillis(d.getNano)
+  @inline def millisToJodaDuration(millis: Long): joda.Duration = joda.Duration.millis(millis)
+  @inline def millisFromJodaDuration(d: joda.Duration): Long = d.getMillis
+
+  // micros /////////////////////////////////////////////////////
+  @inline def microsToInstant(microsFromEpoch: Long): Instant = {
+    val epochSeconds = TimeUnit.MICROSECONDS.toSeconds(microsFromEpoch)
+    val nanoAdjustment = TimeUnit.MICROSECONDS.toNanos(microsFromEpoch % 1_000_000L)
+    Instant.ofEpochSecond(epochSeconds, nanoAdjustment)
+  }
+  @inline def microsFromInstant(instant: Instant): Long = {
+    val seconds = instant.getEpochSecond
+    val nanos = instant.getNano
+    if (seconds < 0 && nanos > 0) {
+      val micros = Math.multiplyExact(seconds + 1, 1000000L)
+      val adjustment = (nanos / 1000L) - 1000000
+      Math.addExact(micros, adjustment)
+    } else {
+      val micros = Math.multiplyExact(seconds, 1000000L)
+      Math.addExact(micros, nanos / 1000L)
+    }
+  }
+  @inline def microsToJodaInstant(microsFromEpoch: Long): joda.Instant =
+    new joda.Instant(TimeUnit.MICROSECONDS.toMillis(microsFromEpoch), joda.DateTimeZone.UTC)
+  @inline def microsFromJodaInstant(instant: joda.Instant): Long =
+    TimeUnit.MILLISECONDS.toMicros(instant.getMillis)
+
+  @inline def microsToJodaDateTime(microsFromEpoch: Long): joda.DateTime =
+    new joda.DateTime(TimeUnit.MICROSECONDS.toMillis(microsFromEpoch), joda.DateTimeZone.UTC)
+  @inline def microsFromJodaDateTime(dt: joda.DateTime): Long =
+    TimeUnit.MILLISECONDS.toMicros(dt.getMillis)
+
+  @inline def microsToLocalTime(microsFromMidnight: Long): LocalTime =
+    LocalTime.ofNanoOfDay(TimeUnit.MICROSECONDS.toNanos(microsFromMidnight))
+  @inline def microsFromLocalTime(lt: LocalTime): Long =
+    TimeUnit.NANOSECONDS.toMicros(lt.toNanoOfDay)
+  @inline def microsToJodaLocalTime(microsFromMidnight: Long): joda.LocalTime =
+    joda.LocalTime.fromMillisOfDay(TimeUnit.MICROSECONDS.toMillis(microsFromMidnight))
+  @inline def microsFromJodaLocalTime(lt: joda.LocalTime): Long =
+    TimeUnit.MILLISECONDS.toMicros(lt.millisOfDay().get())
+
+  @inline def microsToLocalDateTime(microsFromEpoch: Long): LocalDateTime =
+    LocalDateTime.ofInstant(microsToInstant(microsFromEpoch), ZoneOffset.UTC)
+  @inline def microsFromLocalDateTime(ldt: LocalDateTime): Long =
+    microsFromInstant(ldt.toInstant(ZoneOffset.UTC))
+
+  @inline def microsToDuration(micros: Long): Duration =
+    Duration.ofMillis(TimeUnit.MICROSECONDS.toMillis(micros))
+  @inline def microsFromDuration(d: Duration): Long =
+    TimeUnit.SECONDS.toMicros(d.getSeconds) + TimeUnit.NANOSECONDS.toMicros(d.getNano)
+  @inline def microsToJodaDuration(micros: Long): joda.Duration =
+    joda.Duration.millis(TimeUnit.MICROSECONDS.toMillis(micros))
+  @inline def microsFromJodaDuration(d: joda.Duration): Long =
+    TimeUnit.MILLISECONDS.toMicros(d.getMillis)
+
+  // nanos /////////////////////////////////////////////////////
+  // Long does not technically have enough range for Instant
+  @inline def nanosToInstant(epochNanos: Long): Instant =
+    Instant.ofEpochSecond(TimeUnit.NANOSECONDS.toSeconds(epochNanos), epochNanos % 1_000_000_000L)
+  @inline def nanosFromInstant(instant: Instant): Long =
+    TimeUnit.MILLISECONDS.toNanos(instant.toEpochMilli) + instant.getNano
+  @inline def nanosToJodaInstant(nanosFromEpoch: Long): joda.Instant =
+    new joda.Instant(TimeUnit.NANOSECONDS.toMillis(nanosFromEpoch), joda.DateTimeZone.UTC)
+  @inline def nanosFromJodaInstant(instant: joda.Instant): Long =
+    TimeUnit.MILLISECONDS.toNanos(instant.getMillis)
+
+  @inline def nanosToJodaDateTime(nanosFromEpoch: Long): joda.DateTime =
+    new joda.DateTime(TimeUnit.NANOSECONDS.toMillis(nanosFromEpoch), joda.DateTimeZone.UTC)
+  @inline def nanosFromJodaDateTime(dt: joda.DateTime): Long =
+    TimeUnit.MILLISECONDS.toNanos(dt.getMillis)
+
+  @inline def nanosToLocalTime(nanosFromMidnight: Long): LocalTime =
+    LocalTime.ofNanoOfDay(nanosFromMidnight)
+  @inline def nanosFromLocalTime(lt: LocalTime): Long = lt.toNanoOfDay
+  @inline def nanosToJodaLocalTime(nanosFromMidnight: Long): joda.LocalTime =
+    joda.LocalTime.fromMillisOfDay(TimeUnit.NANOSECONDS.toMillis(nanosFromMidnight))
+  @inline def nanosFromJodaLocalTime(lt: joda.LocalTime): Long =
+    TimeUnit.MILLISECONDS.toNanos(lt.millisOfDay().get())
+
+  @inline def nanosToLocalDateTime(nanosFromEpoch: Long): LocalDateTime =
+    LocalDateTime.ofInstant(nanosToInstant(nanosFromEpoch), ZoneOffset.UTC)
+  @inline def nanosFromLocalDateTime(ldt: LocalDateTime): Long =
+    nanosFromInstant(ldt.toInstant(ZoneOffset.UTC))
+
+  @inline def nanosToDuration(nanos: Long): Duration =
+    Duration.ofNanos(nanos)
+  @inline def nanosFromDuration(d: Duration): Long =
+    TimeUnit.SECONDS.toNanos(d.getSeconds) + d.getNano
+  @inline def nanosToJodaDuration(nanos: Long): joda.Duration =
+    joda.Duration.millis(TimeUnit.NANOSECONDS.toMillis(nanos))
+  @inline def nanosFromJodaDuration(d: joda.Duration): Long =
+    TimeUnit.MILLISECONDS.toNanos(d.getMillis)
+}