Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions modules/core/src/main/scala/vulcan/AvroException.scala
Original file line number Diff line number Diff line change
Expand Up @@ -381,12 +381,14 @@ object AvroException {
)

/**
* Exception raised when adding a logical type to a
* schema with a type which has no logical types.
* Exception raised when adding a logical type to a schema
* that already has a logical type, or that has a type
* for which the logical type is invalid.
*/
final class UnsupportedLogicalType private[vulcan] (logicalType: LogicalType, schema: Schema)
extends AvroException(
s"Unable to add logical type ${logicalType.name} to ${SchemaNames.describe(schema)}"
s"Unable to add logical type ${logicalType.name} to ${SchemaNames.describe(schema)}" +
schema.props("logicalType").flatMap(_.asString).fold("")(name => s" with logical type $name")
)

/**
Expand Down
41 changes: 28 additions & 13 deletions modules/core/src/main/scala/vulcan/Codec.scala
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,17 @@ sealed abstract class Codec[A] {
final def exmap[B](f: A => Either[AvroException, B])(g: B => Either[AvroException, A]): Codec[B] =
new Codec.Exmap(this, f, g)

/**
* Returns a new [[Codec]] with the logical type added to its schema.
*
* The schema is an error if the logical type is invalid for the type
* of the schema, or if the schema already has a logical type.
*
* Decoding requires writer schemas to have the same logical type.
*/
final def withLogicalType(logicalType: LogicalType): Codec[A] =
new Codec.LogicalTypeCodec(this, logicalType)

/** Attempts to encode the specified value. */
private[vulcan] def encode(a: A): Either[AvroException, AvroValue]

Expand Down Expand Up @@ -208,9 +219,6 @@ sealed abstract class Codec[A] {
private[vulcan] final def withTypeName(typeName: String): Codec[A] =
new Codec.WithTypeName(this, typeName)

private[vulcan] final def withLogicalType(logicalType: LogicalType): Codec[A] =
new Codec.LogicalTypeCodec(this, logicalType)

/** Encodes the value as JSON for defaults in schemas, where unions are without the type. */
private[vulcan] final def encodeForSchema(a: A): Either[AvroException, Json] =
for {
Expand Down Expand Up @@ -1699,16 +1707,23 @@ object Codec {
loop()
}

private def addLogicalType(schema: Schema, logicalType: LogicalType): Either[AvroException, Schema] =
schema match {
case int: Schema.Int => Right(int.withLogicalType(logicalType))
case long: Schema.Long => Right(long.withLogicalType(logicalType))
case bytes: Schema.Bytes => Right(bytes.withLogicalType(logicalType))
case string: Schema.String => Right(string.withLogicalType(logicalType))
case fixed: Schema.Fixed => Right(fixed.withLogicalType(logicalType))
case _ =>
Left(new UnsupportedLogicalType(logicalType, schema))
}
/** The schema with the logical type, unless it has a logical type or the logical type is invalid. */
private def addLogicalType(schema: Schema, logicalType: LogicalType): Either[AvroException, Schema] = {
val added: Option[Schema] =
schema match {
case _ if schema.props.contains("logicalType") => None
case int: Schema.Int => Some(int.withLogicalType(logicalType))
case long: Schema.Long => Some(long.withLogicalType(logicalType))
case bytes: Schema.Bytes => Some(bytes.withLogicalType(logicalType))
case string: Schema.String => Some(string.withLogicalType(logicalType))
case fixed: Schema.Fixed => Some(fixed.withLogicalType(logicalType))
case _ => None
}

added
.filter(_.logicalType.contains(logicalType))
.toRight(new UnsupportedLogicalType(logicalType, schema))
}

/** Decodes the decimal using the scale, and checks that it does not exceed the precision. */
private def decodeDecimal(
Expand Down
75 changes: 75 additions & 0 deletions tests/shared/src/test/scala/vulcan/CodecSpec.scala
Original file line number Diff line number Diff line change
Expand Up @@ -3271,4 +3271,79 @@ object CodecSpec extends BaseSuite {
)
}
}

locally {
case class Nanos(value: Long)

object Nanos {
implicit val nanosCodec: Codec[Nanos] =
Codec.long.imap(Nanos(_))(_.value).withLogicalType(LogicalType.timestampNanos)
}

pureTest("Codec withLogicalType schema should add the logical type") {
expectSchema[Nanos] {
"""{"type":"long","logicalType":"timestamp-nanos"}"""
}
}

pureTest("Codec withLogicalType decode should decode values written with the logical type") {
expectDecode[Nanos](
Codec[Long].encode(1L),
Right(Schema.Long.withLogicalType(LogicalType.timestampNanos)),
Nanos(1L)
)
}

pureTest("Codec withLogicalType decode should error if the writer schema has no logical type") {
expectDecodeError[Nanos](
Codec[Long].encode(1L),
Right(Schema.Long),
"Got unexpected missing logical type"
)
}
}

locally {
case class Id(value: ByteVector)

object Id {
implicit val idCodec: Codec[Id] =
Codec
.fixed[Id]("Id", "com.example", 16, _.value, bytes => Right(Id(bytes)))
.withLogicalType(LogicalType.uuid)
}

pureTest("Codec withLogicalType schema should add the logical type to fixed") {
expectSchema[Id] {
"""{"type":"fixed","name":"Id","namespace":"com.example","size":16,"logicalType":"uuid"}"""
}
}

pureTest("Codec withLogicalType should roundtrip fixed with the logical type") {
expectRoundtrip(Id(ByteVector.fill(16L)(1)))
}
}

pureTest("Codec withLogicalType schema should error if the logical type is invalid for the type") {
implicit val codec: Codec[String] =
Codec.string.withLogicalType(LogicalType.timestampMillis)

expectSchemaError[String]("Unable to add logical type timestamp-millis to string")
}

pureTest("Codec withLogicalType schema should error if the schema already has a logical type") {
implicit val codec: Codec[Instant] =
Codec.instant.withLogicalType(LogicalType.timestampMicros)

expectSchemaError[Instant](
"Unable to add logical type timestamp-micros to long with logical type timestamp-millis"
)
}

pureTest("Codec withLogicalType schema should error if the type has no logical types") {
implicit val codec: Codec[Boolean] =
Codec.boolean.withLogicalType(LogicalType.uuid)

expectSchemaError[Boolean]("Unable to add logical type uuid to boolean")
}
}
Loading