From 43dfd8875f2955f500ffb5d654d0c73afb7da72c Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Thu, 26 Feb 2026 15:27:11 +0000 Subject: [PATCH 01/11] Support avro date values --- lib/codecs/src/decoding/format/avro.rs | 5 +---- lib/codecs/tests/avro.rs | 2 +- lib/codecs/tests/bin/generate-avro-fixtures.rs | 2 +- lib/codecs/tests/data/avro/generated/array.avsc | 2 +- lib/codecs/tests/data/avro/generated/date.avro | 1 + lib/codecs/tests/data/avro/generated/date.avsc | 1 + .../tests/data/avro/generated/local-timestamp_micros.avsc | 2 +- .../tests/data/avro/generated/local-timestamp_millis.avsc | 2 +- lib/codecs/tests/data/avro/generated/map.avsc | 2 +- lib/codecs/tests/data/avro/generated/time_micros.avsc | 2 +- lib/codecs/tests/data/avro/generated/timestamp_micros.avsc | 2 +- lib/codecs/tests/data/avro/generated/timestamp_millis.avsc | 2 +- lib/codecs/tests/data/avro/generated/uuid.avsc | 2 +- 13 files changed, 13 insertions(+), 14 deletions(-) create mode 100644 lib/codecs/tests/data/avro/generated/date.avro create mode 100644 lib/codecs/tests/data/avro/generated/date.avsc diff --git a/lib/codecs/src/decoding/format/avro.rs b/lib/codecs/src/decoding/format/avro.rs index 73a703b72ca38..bf7779d70481f 100644 --- a/lib/codecs/src/decoding/format/avro.rs +++ b/lib/codecs/src/decoding/format/avro.rs @@ -91,7 +91,6 @@ impl From<&AvroDeserializerOptions> for AvroSerializerOptions { pub struct AvroDeserializerOptions { /// The Avro schema definition. /// **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - /// * `Date` /// * `Decimal` /// * `Duration` /// * `Fixed` @@ -190,9 +189,7 @@ pub fn try_from(value: AvroValue) -> vector_common::Result { } AvroValue::Boolean(boolean) => Ok(VrlValue::from(boolean)), AvroValue::Bytes(bytes) => Ok(VrlValue::from(bytes)), - AvroValue::Date(_) => Err(vector_common::Error::from( - "AvroValue::Date is not supported", - )), + AvroValue::Date(days) => Ok(VrlValue::from(days)), AvroValue::Decimal(_) => Err(vector_common::Error::from( "AvroValue::Decimal is not supported", )), diff --git a/lib/codecs/tests/avro.rs b/lib/codecs/tests/avro.rs index fff274706d7d9..4f3b57d653ffb 100644 --- a/lib/codecs/tests/avro.rs +++ b/lib/codecs/tests/avro.rs @@ -22,7 +22,7 @@ use vector_core::{config::LogNamespace, event::Event}; #[case(false)] fn roundtrip_avro_fixtures( #[files("tests/data/avro/generated/*.avro")] - #[exclude(".*(date|fixed|time_millis).avro")] + #[exclude(".*(fixed|time_millis).avro")] path: PathBuf, #[case] reserialize: bool, ) { diff --git a/lib/codecs/tests/bin/generate-avro-fixtures.rs b/lib/codecs/tests/bin/generate-avro-fixtures.rs index 29f22ef0991c0..a8efdb4809db7 100644 --- a/lib/codecs/tests/bin/generate-avro-fixtures.rs +++ b/lib/codecs/tests/bin/generate-avro-fixtures.rs @@ -280,7 +280,6 @@ fn generate_avro_test_case_record() -> Result<()> { generate_test_case(schema, value, "record") } -#[allow(unused)] fn generate_avro_test_case_date() -> Result<()> { let schema = r#" { @@ -488,6 +487,7 @@ fn main() -> Result<()> { generate_avro_test_case_array()?; generate_avro_test_case_boolean()?; generate_avro_test_case_bytes()?; + generate_avro_test_case_date()?; generate_avro_test_case_double()?; generate_avro_test_case_enum()?; generate_avro_test_case_float()?; diff --git a/lib/codecs/tests/data/avro/generated/array.avsc b/lib/codecs/tests/data/avro/generated/array.avsc index 41f098e5deafe..e029b80030371 100644 --- a/lib/codecs/tests/data/avro/generated/array.avsc +++ b/lib/codecs/tests/data/avro/generated/array.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"array_field","type":{"type":"array","items":"string"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"array_field","type":{"type":"array","items":"string"},"items":"string"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/date.avro b/lib/codecs/tests/data/avro/generated/date.avro new file mode 100644 index 0000000000000..b067210aeac93 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/date.avro @@ -0,0 +1 @@ +ü² \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/date.avsc b/lib/codecs/tests/data/avro/generated/date.avsc new file mode 100644 index 0000000000000..ffd6356594a1e --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/date.avsc @@ -0,0 +1 @@ +{"name":"test","type":"record","fields":[{"name":"date_field","type":{"type":"int"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc b/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc index 35d5441f13078..1ebf1a0dee09c 100644 --- a/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"local_timestamp_micros_field","type":{"type":"long","logicalType":"local-timestamp-micros"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"local_timestamp_micros_field","type":{"type":"long"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc b/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc index 51ec12c1e477e..8446bc0c66e9d 100644 --- a/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"local_timestamp_millis_field","type":{"type":"long","logicalType":"local-timestamp-millis"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"local_timestamp_millis_field","type":{"type":"long"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/map.avsc b/lib/codecs/tests/data/avro/generated/map.avsc index cc3e2ce9003dd..fe3cda2c18265 100644 --- a/lib/codecs/tests/data/avro/generated/map.avsc +++ b/lib/codecs/tests/data/avro/generated/map.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"map_field","type":{"type":"map","values":"long"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"map_field","type":{"type":"map","values":"long"},"values":"long"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/time_micros.avsc b/lib/codecs/tests/data/avro/generated/time_micros.avsc index 6624415706d0e..1bc70cfbfea69 100644 --- a/lib/codecs/tests/data/avro/generated/time_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/time_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"time_micros_field","type":{"type":"long","logicalType":"time-micros"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"time_micros_field","type":{"type":"long"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc b/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc index b466968145c9f..1932246b2fb80 100644 --- a/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"timestamp_micros_field","type":{"type":"long","logicalType":"timestamp-micros"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"timestamp_micros_field","type":{"type":"long"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc b/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc index eea96da274baf..b26dcecd231dd 100644 --- a/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc +++ b/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"timestamp_millis_field","type":{"type":"long","logicalType":"timestamp-millis"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"timestamp_millis_field","type":{"type":"long"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/uuid.avsc b/lib/codecs/tests/data/avro/generated/uuid.avsc index ca1b0bf400de1..deafef0c1bb77 100644 --- a/lib/codecs/tests/data/avro/generated/uuid.avsc +++ b/lib/codecs/tests/data/avro/generated/uuid.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"uuid_field","type":{"type":"string","logicalType":"uuid"}}]} \ No newline at end of file +{"name":"test","type":"record","fields":[{"name":"uuid_field","type":{"type":"string"}}]} \ No newline at end of file From 8f2d0fde86ffcf2773da138ac17156abacba933f Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Thu, 26 Feb 2026 16:27:59 +0000 Subject: [PATCH 02/11] Add changelog fragment --- changelog.d/24773_support_avro_date_values.fix.md | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 changelog.d/24773_support_avro_date_values.fix.md diff --git a/changelog.d/24773_support_avro_date_values.fix.md b/changelog.d/24773_support_avro_date_values.fix.md new file mode 100644 index 0000000000000..36b375a10fc3d --- /dev/null +++ b/changelog.d/24773_support_avro_date_values.fix.md @@ -0,0 +1,3 @@ +The Avro codec now supports decoding Avro Date values. + +authors: omwbennett From 6ea80e76ae630013040dc4e7fc31e960e6574fc9 Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Tue, 3 Mar 2026 14:59:24 +0000 Subject: [PATCH 03/11] Run make generate-component-docs --- .../reference/components/sinks/generated/websocket_server.cue | 1 - website/cue/reference/components/sources/generated/amqp.cue | 1 - .../components/sources/generated/aws_kinesis_firehose.cue | 1 - website/cue/reference/components/sources/generated/aws_s3.cue | 1 - website/cue/reference/components/sources/generated/aws_sqs.cue | 1 - .../cue/reference/components/sources/generated/datadog_agent.cue | 1 - website/cue/reference/components/sources/generated/demo_logs.cue | 1 - website/cue/reference/components/sources/generated/exec.cue | 1 - .../reference/components/sources/generated/file_descriptor.cue | 1 - .../cue/reference/components/sources/generated/gcp_pubsub.cue | 1 - .../cue/reference/components/sources/generated/heroku_logs.cue | 1 - website/cue/reference/components/sources/generated/http.cue | 1 - .../cue/reference/components/sources/generated/http_client.cue | 1 - .../cue/reference/components/sources/generated/http_server.cue | 1 - website/cue/reference/components/sources/generated/kafka.cue | 1 - website/cue/reference/components/sources/generated/mqtt.cue | 1 - website/cue/reference/components/sources/generated/nats.cue | 1 - website/cue/reference/components/sources/generated/pulsar.cue | 1 - website/cue/reference/components/sources/generated/redis.cue | 1 - website/cue/reference/components/sources/generated/socket.cue | 1 - website/cue/reference/components/sources/generated/stdin.cue | 1 - website/cue/reference/components/sources/generated/websocket.cue | 1 - 22 files changed, 22 deletions(-) diff --git a/website/cue/reference/components/sinks/generated/websocket_server.cue b/website/cue/reference/components/sinks/generated/websocket_server.cue index 586b2bc4f4383..7df8b9be2eeac 100644 --- a/website/cue/reference/components/sinks/generated/websocket_server.cue +++ b/website/cue/reference/components/sinks/generated/websocket_server.cue @@ -594,7 +594,6 @@ generated: components: sinks: websocket_server: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/amqp.cue b/website/cue/reference/components/sources/generated/amqp.cue index 1108e0eef13ab..3b58bc542b30b 100644 --- a/website/cue/reference/components/sources/generated/amqp.cue +++ b/website/cue/reference/components/sources/generated/amqp.cue @@ -62,7 +62,6 @@ generated: components: sources: amqp: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue b/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue index d8d6fa2806e1f..8807aeffba320 100644 --- a/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue +++ b/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue @@ -82,7 +82,6 @@ generated: components: sources: aws_kinesis_firehose: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/aws_s3.cue b/website/cue/reference/components/sources/generated/aws_s3.cue index 7909847e25921..0fc4763c51bed 100644 --- a/website/cue/reference/components/sources/generated/aws_s3.cue +++ b/website/cue/reference/components/sources/generated/aws_s3.cue @@ -180,7 +180,6 @@ generated: components: sources: aws_s3: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/aws_sqs.cue b/website/cue/reference/components/sources/generated/aws_sqs.cue index 5d350cac9a901..c5d113713dfc2 100644 --- a/website/cue/reference/components/sources/generated/aws_sqs.cue +++ b/website/cue/reference/components/sources/generated/aws_sqs.cue @@ -175,7 +175,6 @@ generated: components: sources: aws_sqs: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/datadog_agent.cue b/website/cue/reference/components/sources/generated/datadog_agent.cue index f9eb4fded5e14..6059924b5ed22 100644 --- a/website/cue/reference/components/sources/generated/datadog_agent.cue +++ b/website/cue/reference/components/sources/generated/datadog_agent.cue @@ -47,7 +47,6 @@ generated: components: sources: datadog_agent: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/demo_logs.cue b/website/cue/reference/components/sources/generated/demo_logs.cue index cc4dca331b68b..c0f94f4b1eceb 100644 --- a/website/cue/reference/components/sources/generated/demo_logs.cue +++ b/website/cue/reference/components/sources/generated/demo_logs.cue @@ -26,7 +26,6 @@ generated: components: sources: demo_logs: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/exec.cue b/website/cue/reference/components/sources/generated/exec.cue index b72470346e5dd..62acd72bbdfdd 100644 --- a/website/cue/reference/components/sources/generated/exec.cue +++ b/website/cue/reference/components/sources/generated/exec.cue @@ -27,7 +27,6 @@ generated: components: sources: exec: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/file_descriptor.cue b/website/cue/reference/components/sources/generated/file_descriptor.cue index daa12cce7cea7..06bc07eba6e46 100644 --- a/website/cue/reference/components/sources/generated/file_descriptor.cue +++ b/website/cue/reference/components/sources/generated/file_descriptor.cue @@ -17,7 +17,6 @@ generated: components: sources: file_descriptor: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue index af645bf7384b3..dc9a9f8319f9c 100644 --- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue +++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue @@ -93,7 +93,6 @@ generated: components: sources: gcp_pubsub: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/heroku_logs.cue b/website/cue/reference/components/sources/generated/heroku_logs.cue index 2eb2141d26444..1f383b3d18c49 100644 --- a/website/cue/reference/components/sources/generated/heroku_logs.cue +++ b/website/cue/reference/components/sources/generated/heroku_logs.cue @@ -90,7 +90,6 @@ generated: components: sources: heroku_logs: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/http.cue b/website/cue/reference/components/sources/generated/http.cue index 9024dcde6d039..076ea8a3a2816 100644 --- a/website/cue/reference/components/sources/generated/http.cue +++ b/website/cue/reference/components/sources/generated/http.cue @@ -98,7 +98,6 @@ generated: components: sources: http: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/http_client.cue b/website/cue/reference/components/sources/generated/http_client.cue index 527a7b7eae9c0..9f38084556c88 100644 --- a/website/cue/reference/components/sources/generated/http_client.cue +++ b/website/cue/reference/components/sources/generated/http_client.cue @@ -228,7 +228,6 @@ generated: components: sources: http_client: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/http_server.cue b/website/cue/reference/components/sources/generated/http_server.cue index 9f2fd3b59393f..685c8f0781987 100644 --- a/website/cue/reference/components/sources/generated/http_server.cue +++ b/website/cue/reference/components/sources/generated/http_server.cue @@ -98,7 +98,6 @@ generated: components: sources: http_server: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/kafka.cue b/website/cue/reference/components/sources/generated/kafka.cue index 0979aa41c30a0..5e472a5aa1ac8 100644 --- a/website/cue/reference/components/sources/generated/kafka.cue +++ b/website/cue/reference/components/sources/generated/kafka.cue @@ -71,7 +71,6 @@ generated: components: sources: kafka: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/mqtt.cue b/website/cue/reference/components/sources/generated/mqtt.cue index 4a699a4e6951a..5dddd7141790a 100644 --- a/website/cue/reference/components/sources/generated/mqtt.cue +++ b/website/cue/reference/components/sources/generated/mqtt.cue @@ -22,7 +22,6 @@ generated: components: sources: mqtt: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/nats.cue b/website/cue/reference/components/sources/generated/nats.cue index bf29e808bdf80..dce7eb1dc9dca 100644 --- a/website/cue/reference/components/sources/generated/nats.cue +++ b/website/cue/reference/components/sources/generated/nats.cue @@ -114,7 +114,6 @@ generated: components: sources: nats: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/pulsar.cue b/website/cue/reference/components/sources/generated/pulsar.cue index a4f831bd62191..fa16e8f93c710 100644 --- a/website/cue/reference/components/sources/generated/pulsar.cue +++ b/website/cue/reference/components/sources/generated/pulsar.cue @@ -120,7 +120,6 @@ generated: components: sources: pulsar: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/redis.cue b/website/cue/reference/components/sources/generated/redis.cue index a37029b9d5893..80d40e9c89f4f 100644 --- a/website/cue/reference/components/sources/generated/redis.cue +++ b/website/cue/reference/components/sources/generated/redis.cue @@ -32,7 +32,6 @@ generated: components: sources: redis: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/socket.cue b/website/cue/reference/components/sources/generated/socket.cue index b0b399c3ae2d0..9f7e37cb836f3 100644 --- a/website/cue/reference/components/sources/generated/socket.cue +++ b/website/cue/reference/components/sources/generated/socket.cue @@ -34,7 +34,6 @@ generated: components: sources: socket: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/stdin.cue b/website/cue/reference/components/sources/generated/stdin.cue index 3eef2fc198037..a7472a3511662 100644 --- a/website/cue/reference/components/sources/generated/stdin.cue +++ b/website/cue/reference/components/sources/generated/stdin.cue @@ -17,7 +17,6 @@ generated: components: sources: stdin: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` diff --git a/website/cue/reference/components/sources/generated/websocket.cue b/website/cue/reference/components/sources/generated/websocket.cue index bce2e24609cfb..cb39facf06008 100644 --- a/website/cue/reference/components/sources/generated/websocket.cue +++ b/website/cue/reference/components/sources/generated/websocket.cue @@ -204,7 +204,6 @@ generated: components: sources: websocket: configuration: { description: """ The Avro schema definition. **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: - * `Date` * `Decimal` * `Duration` * `Fixed` From b41b736d7b373ba45b6968611006418a8c6f64f3 Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Mon, 13 Apr 2026 15:42:27 +0100 Subject: [PATCH 04/11] Fix avro fixture generation --- lib/codecs/tests/bin/generate-avro-fixtures.rs | 2 +- lib/codecs/tests/data/avro/generated/array.avsc | 2 +- lib/codecs/tests/data/avro/generated/boolean.avsc | 2 +- lib/codecs/tests/data/avro/generated/bytes.avsc | 2 +- lib/codecs/tests/data/avro/generated/date.avsc | 2 +- lib/codecs/tests/data/avro/generated/double.avsc | 2 +- lib/codecs/tests/data/avro/generated/enum.avsc | 2 +- lib/codecs/tests/data/avro/generated/float.avsc | 2 +- lib/codecs/tests/data/avro/generated/int.avsc | 2 +- .../tests/data/avro/generated/local-timestamp_micros.avsc | 2 +- .../tests/data/avro/generated/local-timestamp_millis.avsc | 2 +- lib/codecs/tests/data/avro/generated/long.avsc | 2 +- lib/codecs/tests/data/avro/generated/map.avsc | 2 +- lib/codecs/tests/data/avro/generated/record.avsc | 2 +- lib/codecs/tests/data/avro/generated/string.avsc | 2 +- lib/codecs/tests/data/avro/generated/time_micros.avsc | 2 +- lib/codecs/tests/data/avro/generated/timestamp_micros.avsc | 2 +- lib/codecs/tests/data/avro/generated/timestamp_millis.avsc | 2 +- lib/codecs/tests/data/avro/generated/union.avsc | 2 +- lib/codecs/tests/data/avro/generated/uuid.avsc | 2 +- 20 files changed, 20 insertions(+), 20 deletions(-) diff --git a/lib/codecs/tests/bin/generate-avro-fixtures.rs b/lib/codecs/tests/bin/generate-avro-fixtures.rs index a8efdb4809db7..549df090ff627 100644 --- a/lib/codecs/tests/bin/generate-avro-fixtures.rs +++ b/lib/codecs/tests/bin/generate-avro-fixtures.rs @@ -475,7 +475,7 @@ fn generate_test_case_from_value(schema: &str, value: Value, filename: &str) -> let mut schema_file = File::create(format!("{FIXTURES_PATH}/{filename}.avsc"))?; let mut avro_file = File::create(format!("{FIXTURES_PATH}/{filename}.avro"))?; - schema_file.write_all(schema.canonical_form().as_bytes())?; + schema_file.write_all(serde_json::to_string(&schema)?.as_bytes())?; avro_file.write_all(&bytes)?; Ok(()) } diff --git a/lib/codecs/tests/data/avro/generated/array.avsc b/lib/codecs/tests/data/avro/generated/array.avsc index e029b80030371..3ce68e7b38060 100644 --- a/lib/codecs/tests/data/avro/generated/array.avsc +++ b/lib/codecs/tests/data/avro/generated/array.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"array_field","type":{"type":"array","items":"string"},"items":"string"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"array_field","type":{"type":"array","items":"string"},"items":"string"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/boolean.avsc b/lib/codecs/tests/data/avro/generated/boolean.avsc index aa04929ca9db4..d0a8839a6a2f1 100644 --- a/lib/codecs/tests/data/avro/generated/boolean.avsc +++ b/lib/codecs/tests/data/avro/generated/boolean.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"bool_field","type":"boolean"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"bool_field","type":"boolean","default":false}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/bytes.avsc b/lib/codecs/tests/data/avro/generated/bytes.avsc index 2d8d0acefc72a..c34b6ff57321c 100644 --- a/lib/codecs/tests/data/avro/generated/bytes.avsc +++ b/lib/codecs/tests/data/avro/generated/bytes.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"bytes_field","type":"bytes"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"bytes_field","type":"bytes"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/date.avsc b/lib/codecs/tests/data/avro/generated/date.avsc index ffd6356594a1e..c7a70afe9be1a 100644 --- a/lib/codecs/tests/data/avro/generated/date.avsc +++ b/lib/codecs/tests/data/avro/generated/date.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"date_field","type":{"type":"int"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"date_field","type":{"type":"int","logicalType":"date"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/double.avsc b/lib/codecs/tests/data/avro/generated/double.avsc index 60dfa5b2aab17..7efdd059d60b8 100644 --- a/lib/codecs/tests/data/avro/generated/double.avsc +++ b/lib/codecs/tests/data/avro/generated/double.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"double_field","type":"double"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"double_field","type":"double","default":0}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/enum.avsc b/lib/codecs/tests/data/avro/generated/enum.avsc index 5d7468754da98..59d5d0e34dad2 100644 --- a/lib/codecs/tests/data/avro/generated/enum.avsc +++ b/lib/codecs/tests/data/avro/generated/enum.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"enum_field","type":{"name":"enum_field","type":"enum","symbols":["Spades","Hearts","Diamonds","Clubs"]}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"enum_field","type":{"type":"enum","name":"enum_field","symbols":["Spades","Hearts","Diamonds","Clubs"]}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/float.avsc b/lib/codecs/tests/data/avro/generated/float.avsc index 9a7f836ddfe0d..aa0d287e4dfc3 100644 --- a/lib/codecs/tests/data/avro/generated/float.avsc +++ b/lib/codecs/tests/data/avro/generated/float.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"float_field","type":"float"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"float_field","type":"float","default":0}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/int.avsc b/lib/codecs/tests/data/avro/generated/int.avsc index af009f13394f2..134feda0ea323 100644 --- a/lib/codecs/tests/data/avro/generated/int.avsc +++ b/lib/codecs/tests/data/avro/generated/int.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"int_field","type":"int"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"int_field","type":"int","default":0}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc b/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc index 1ebf1a0dee09c..01d0dcf5eca96 100644 --- a/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"local_timestamp_micros_field","type":{"type":"long"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"local_timestamp_micros_field","type":{"type":"long","logicalType":"local-timestamp-micros"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc b/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc index 8446bc0c66e9d..39722937a5d13 100644 --- a/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_millis.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"local_timestamp_millis_field","type":{"type":"long"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"local_timestamp_millis_field","type":{"type":"long","logicalType":"local-timestamp-millis"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/long.avsc b/lib/codecs/tests/data/avro/generated/long.avsc index e4a052e4eacef..4976e3d40a8fb 100644 --- a/lib/codecs/tests/data/avro/generated/long.avsc +++ b/lib/codecs/tests/data/avro/generated/long.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"long_field","type":"long"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"long_field","type":"long","default":0}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/map.avsc b/lib/codecs/tests/data/avro/generated/map.avsc index fe3cda2c18265..69441d98fa538 100644 --- a/lib/codecs/tests/data/avro/generated/map.avsc +++ b/lib/codecs/tests/data/avro/generated/map.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"map_field","type":{"type":"map","values":"long"},"values":"long"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"map_field","type":{"type":"map","values":"long","default":{}},"default":{},"values":"long"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/record.avsc b/lib/codecs/tests/data/avro/generated/record.avsc index 04632d003a183..e644a5f586fe9 100644 --- a/lib/codecs/tests/data/avro/generated/record.avsc +++ b/lib/codecs/tests/data/avro/generated/record.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/string.avsc b/lib/codecs/tests/data/avro/generated/string.avsc index b6efaad1e7e11..d26a0c1f81e07 100644 --- a/lib/codecs/tests/data/avro/generated/string.avsc +++ b/lib/codecs/tests/data/avro/generated/string.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"string_field","type":"string"}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"string_field","type":"string"}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/time_micros.avsc b/lib/codecs/tests/data/avro/generated/time_micros.avsc index 1bc70cfbfea69..455d3e9cd2ada 100644 --- a/lib/codecs/tests/data/avro/generated/time_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/time_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"time_micros_field","type":{"type":"long"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"time_micros_field","type":{"type":"long","logicalType":"time-micros"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc b/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc index 1932246b2fb80..d51fdda84588b 100644 --- a/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc +++ b/lib/codecs/tests/data/avro/generated/timestamp_micros.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"timestamp_micros_field","type":{"type":"long"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"timestamp_micros_field","type":{"type":"long","logicalType":"timestamp-micros"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc b/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc index b26dcecd231dd..5a42c53182c7d 100644 --- a/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc +++ b/lib/codecs/tests/data/avro/generated/timestamp_millis.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"timestamp_millis_field","type":{"type":"long"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"timestamp_millis_field","type":{"type":"long","logicalType":"timestamp-millis"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/union.avsc b/lib/codecs/tests/data/avro/generated/union.avsc index b67afd5112a6b..bd31d552c42f2 100644 --- a/lib/codecs/tests/data/avro/generated/union.avsc +++ b/lib/codecs/tests/data/avro/generated/union.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"union_field","type":["string","int"]}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"union_field","type":["string","int"]}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/uuid.avsc b/lib/codecs/tests/data/avro/generated/uuid.avsc index deafef0c1bb77..d20f63c4d503a 100644 --- a/lib/codecs/tests/data/avro/generated/uuid.avsc +++ b/lib/codecs/tests/data/avro/generated/uuid.avsc @@ -1 +1 @@ -{"name":"test","type":"record","fields":[{"name":"uuid_field","type":{"type":"string"}}]} \ No newline at end of file +{"type":"record","name":"test","fields":[{"name":"uuid_field","type":{"type":"string","logicalType":"uuid"}}]} \ No newline at end of file From 490291ae9815ab31b41deb71bea6094356272853 Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Wed, 22 Apr 2026 15:07:44 +0100 Subject: [PATCH 05/11] Make avro encoder aware of schemas --- lib/codecs/Cargo.toml | 2 +- lib/codecs/src/decoding/format/avro.rs | 2 +- lib/codecs/src/encoding/format/avro.rs | 174 ++++++++++++++++++++++++- 3 files changed, 173 insertions(+), 5 deletions(-) diff --git a/lib/codecs/Cargo.toml b/lib/codecs/Cargo.toml index 0c24d81913846..7f42e3258e304 100644 --- a/lib/codecs/Cargo.toml +++ b/lib/codecs/Cargo.toml @@ -63,6 +63,7 @@ vector-config-macros = { path = "../vector-config-macros", default-features = fa vector-core = { path = "../vector-core", default-features = false, features = ["vrl"] } vector-vrl-functions.workspace = true toml = { version = "0.9.8", optional = true } +uuid.workspace = true [dev-dependencies] criterion.workspace = true @@ -74,7 +75,6 @@ similar-asserts = "1.7.0" vector-core = { path = "../vector-core", default-features = false, features = ["vrl", "test"] } rstest = "0.26.1" tracing-test = "0.2.6" -uuid.workspace = true vrl.workspace = true [features] diff --git a/lib/codecs/src/decoding/format/avro.rs b/lib/codecs/src/decoding/format/avro.rs index bf7779d70481f..8c46e17f87068 100644 --- a/lib/codecs/src/decoding/format/avro.rs +++ b/lib/codecs/src/decoding/format/avro.rs @@ -188,7 +188,7 @@ pub fn try_from(value: AvroValue) -> vector_common::Result { Ok(VrlValue::Array(vector)) } AvroValue::Boolean(boolean) => Ok(VrlValue::from(boolean)), - AvroValue::Bytes(bytes) => Ok(VrlValue::from(bytes)), + AvroValue::Bytes(bytes) => Ok(VrlValue::Bytes(Bytes::from(bytes))), AvroValue::Date(days) => Ok(VrlValue::from(days)), AvroValue::Decimal(_) => Err(vector_common::Error::from( "AvroValue::Decimal is not supported", diff --git a/lib/codecs/src/encoding/format/avro.rs b/lib/codecs/src/encoding/format/avro.rs index 304dcc2e7cc90..297b0b39a491f 100644 --- a/lib/codecs/src/encoding/format/avro.rs +++ b/lib/codecs/src/encoding/format/avro.rs @@ -1,11 +1,176 @@ use bytes::{BufMut, BytesMut}; use serde::{Deserialize, Serialize}; use tokio_util::codec::Encoder; +use uuid::Uuid; use vector_config::configurable_component; use vector_core::{config::DataType, event::Event, schema}; use crate::encoding::BuildError; +type VrlValue = vrl::value::Value; +type AvroValue = apache_avro::types::Value; + +/// Converts a VRL [`Value`](VrlValue) to an [`apache_avro::types::Value`] using the provided +/// schema to resolve ambiguous types (e.g. `Integer` -> `Int` vs `Date`). +pub(crate) fn to_avro( + value: &VrlValue, + schema: &apache_avro::Schema, + names: &apache_avro::schema::NamesRef<'_>, +) -> vector_common::Result { + use apache_avro::Schema; + match (value, schema) { + (VrlValue::Null, Schema::Null) => Ok(AvroValue::Null), + + (VrlValue::Boolean(b), Schema::Boolean) => Ok(AvroValue::Boolean(*b)), + + (VrlValue::Integer(i), Schema::Int) => i32::try_from(*i) + .map(AvroValue::Int) + .map_err(|_| vector_common::Error::from(format!("Integer {i} overflows Avro int (i32)"))), + (VrlValue::Integer(i), Schema::Date) => i32::try_from(*i) + .map(AvroValue::Date) + .map_err(|_| vector_common::Error::from(format!("Integer {i} overflows Avro date (i32)"))), + (VrlValue::Integer(i), Schema::TimeMillis) => i32::try_from(*i) + .map(AvroValue::TimeMillis) + .map_err(|_| { + vector_common::Error::from(format!("Integer {i} overflows Avro time-millis (i32)")) + }), + + (VrlValue::Integer(i), Schema::Long) => Ok(AvroValue::Long(*i)), + (VrlValue::Integer(i), Schema::TimeMicros) => Ok(AvroValue::TimeMicros(*i)), + (VrlValue::Integer(i), Schema::TimestampMillis) => Ok(AvroValue::TimestampMillis(*i)), + (VrlValue::Integer(i), Schema::TimestampMicros) => Ok(AvroValue::TimestampMicros(*i)), + (VrlValue::Integer(i), Schema::LocalTimestampMillis) => { + Ok(AvroValue::LocalTimestampMillis(*i)) + } + (VrlValue::Integer(i), Schema::LocalTimestampMicros) => { + Ok(AvroValue::LocalTimestampMicros(*i)) + } + + (VrlValue::Float(f), Schema::Float) => Ok(AvroValue::Float(f.into_inner() as f32)), + (VrlValue::Float(f), Schema::Double) => Ok(AvroValue::Double(f.into_inner())), + + (VrlValue::Bytes(b), Schema::Bytes) => Ok(AvroValue::Bytes(b.to_vec())), + (VrlValue::Bytes(b), Schema::String) => String::from_utf8(b.to_vec()) + .map(AvroValue::String) + .map_err(|e| { + vector_common::Error::from(format!("Invalid UTF-8 in string field: {e}")) + }), + (VrlValue::Regex(b), Schema::String) => Ok(AvroValue::String(b.as_str().to_owned())), + + (VrlValue::Bytes(b), Schema::Uuid) => { + let s = String::from_utf8(b.to_vec()).map_err(|e| { + vector_common::Error::from(format!("Invalid UTF-8 in UUID field: {e}")) + })?; + Uuid::parse_str(&s) + .map(AvroValue::Uuid) + .map_err(|e| vector_common::Error::from(format!("Invalid UUID: {e}"))) + } + + (VrlValue::Bytes(b), Schema::Enum(enum_schema)) => { + let s = String::from_utf8(b.to_vec()).map_err(|e| { + vector_common::Error::from(format!("Invalid UTF-8 in enum field: {e}")) + })?; + let index = enum_schema + .symbols + .iter() + .position(|sym| sym == &s) + .ok_or_else(|| vector_common::Error::from(format!("Unknown enum symbol: {s}")))?; + Ok(AvroValue::Enum(index as u32, s)) + } + + (VrlValue::Timestamp(ts), Schema::TimestampMillis) => { + Ok(AvroValue::TimestampMillis(ts.timestamp_millis())) + } + (VrlValue::Timestamp(ts), Schema::TimestampMicros) => { + Ok(AvroValue::TimestampMicros(ts.timestamp_micros())) + } + (VrlValue::Timestamp(ts), Schema::LocalTimestampMillis) => { + Ok(AvroValue::LocalTimestampMillis(ts.timestamp_millis())) + } + (VrlValue::Timestamp(ts), Schema::LocalTimestampMicros) => { + Ok(AvroValue::LocalTimestampMicros(ts.timestamp_micros())) + } + (VrlValue::Timestamp(ts), Schema::Long) => Ok(AvroValue::Long(ts.timestamp_millis())), + (VrlValue::Timestamp(ts), Schema::String) => Ok(AvroValue::String( + ts.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true), + )), + + (v, Schema::Ref { name }) => { + let resolved = names.get(name).ok_or_else(|| { + vector_common::Error::from(format!("Unknown schema ref: {}", name.fullname(None))) + })?; + to_avro(v, resolved, names) + } + + (VrlValue::Array(items), Schema::Array(array_schema)) => items + .iter() + .map(|item| to_avro(item, &array_schema.items, names)) + .collect::, _>>() + .map(AvroValue::Array), + + (VrlValue::Object(map), Schema::Map(map_schema)) => map + .iter() + .map(|(k, v)| to_avro(v, &map_schema.types, names).map(|av| (k.to_string(), av))) + .collect::, _>>() + .map(|items| AvroValue::Map(items.into_iter().collect())), + + (VrlValue::Object(map), Schema::Record(record_schema)) => { + let fields = record_schema + .fields + .iter() + .map(|field| { + let av = match map.get(field.name.as_str()) { + Some(v) => to_avro(v, &field.schema, names)?, + None => match &field.default { + Some(json_default) => { + AvroValue::from(json_default.clone()).resolve(&field.schema)? + } + None => { + return Err(vector_common::Error::from(format!( + "Missing record field: {}", + field.name + ))) + } + }, + }; + Ok((field.name.clone(), av)) + }) + .collect::, _>>()?; + Ok(AvroValue::Record(fields)) + } + + (v, Schema::Union(union_schema)) => { + // Prefer null variant for Null values, otherwise try each variant in order + if matches!(v, VrlValue::Null) + && let Some(idx) = union_schema + .variants() + .iter() + .position(|s| matches!(s, Schema::Null)) + { + return Ok(AvroValue::Union(idx as u32, Box::new(AvroValue::Null))); + } + for (idx, variant_schema) in union_schema.variants().iter().enumerate() { + if matches!(variant_schema, Schema::Null) { + continue; + } + if let Ok(av) = to_avro(v, variant_schema, names) { + return Ok(AvroValue::Union(idx as u32, Box::new(av))); + } + } + Err(vector_common::Error::from(format!( + "No matching union variant for value of kind {}", + v.kind_str() + ))) + } + + (v, s) => Err(vector_common::Error::from(format!( + "Cannot convert VRL {} to Avro schema {:?}", + v.kind_str(), + s + ))), + } +} + /// Config used to build a `AvroSerializer`. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct AvroSerializerConfig { @@ -70,9 +235,12 @@ impl Encoder for AvroSerializer { fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> { let log = event.into_log(); - let value = apache_avro::to_value(log)?; - let value = value.resolve(&self.schema)?; - let bytes = apache_avro::to_avro_datum(&self.schema, value)?; + let (value, _metadata) = log.into_parts(); + let resolved = apache_avro::schema::ResolvedSchema::try_from(&self.schema) + .map_err(|e| vector_common::Error::from(format!("Failed resolving Avro schema: {e}")))?; + let names = resolved.get_names(); + let avro_value = to_avro(&value, &self.schema, names)?; + let bytes = apache_avro::to_avro_datum(&self.schema, avro_value)?; buffer.put_slice(&bytes); Ok(()) } From 8fb61ef673855a7efa496a6ec736c9ca9a75f15e Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Wed, 22 Apr 2026 17:36:47 +0100 Subject: [PATCH 06/11] Support Avro Fixed, TimeMillis, TimestampNanos, LocalTimestampNanos types --- lib/codecs/src/decoding/format/avro.rs | 18 ++------ lib/codecs/src/encoding/format/avro.rs | 31 +++++++++++++ lib/codecs/tests/avro.rs | 1 - .../tests/bin/generate-avro-fixtures.rs | 46 ++++++++++++++++++- .../tests/data/avro/generated/fixed.avro | 1 + .../tests/data/avro/generated/fixed.avsc | 1 + .../avro/generated/local-timestamp_nanos.avro | 1 + .../avro/generated/local-timestamp_nanos.avsc | 1 + .../data/avro/generated/time_millis.avro | 1 + .../data/avro/generated/time_millis.avsc | 1 + .../data/avro/generated/timestamp_nanos.avro | 1 + .../data/avro/generated/timestamp_nanos.avsc | 1 + .../sinks/generated/websocket_server.cue | 2 - .../components/sources/generated/amqp.cue | 2 - .../generated/aws_kinesis_firehose.cue | 2 - .../components/sources/generated/aws_s3.cue | 2 - .../components/sources/generated/aws_sqs.cue | 2 - .../sources/generated/datadog_agent.cue | 2 - .../sources/generated/demo_logs.cue | 2 - .../components/sources/generated/exec.cue | 2 - .../sources/generated/file_descriptor.cue | 2 - .../sources/generated/gcp_pubsub.cue | 2 - .../sources/generated/heroku_logs.cue | 2 - .../components/sources/generated/http.cue | 2 - .../sources/generated/http_client.cue | 2 - .../sources/generated/http_server.cue | 2 - .../components/sources/generated/kafka.cue | 2 - .../components/sources/generated/mqtt.cue | 2 - .../components/sources/generated/nats.cue | 2 - .../components/sources/generated/pulsar.cue | 2 - .../components/sources/generated/redis.cue | 2 - .../components/sources/generated/socket.cue | 2 - .../components/sources/generated/stdin.cue | 2 - .../sources/generated/websocket.cue | 2 - 34 files changed, 87 insertions(+), 61 deletions(-) create mode 100644 lib/codecs/tests/data/avro/generated/fixed.avro create mode 100644 lib/codecs/tests/data/avro/generated/fixed.avsc create mode 100644 lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avro create mode 100644 lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avsc create mode 100644 lib/codecs/tests/data/avro/generated/time_millis.avro create mode 100644 lib/codecs/tests/data/avro/generated/time_millis.avsc create mode 100644 lib/codecs/tests/data/avro/generated/timestamp_nanos.avro create mode 100644 lib/codecs/tests/data/avro/generated/timestamp_nanos.avsc diff --git a/lib/codecs/src/decoding/format/avro.rs b/lib/codecs/src/decoding/format/avro.rs index 8c46e17f87068..f49c9c86b0691 100644 --- a/lib/codecs/src/decoding/format/avro.rs +++ b/lib/codecs/src/decoding/format/avro.rs @@ -93,8 +93,6 @@ pub struct AvroDeserializerOptions { /// **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: /// * `Decimal` /// * `Duration` - /// * `Fixed` - /// * `TimeMillis` #[configurable(metadata( docs::examples = r#"{ "type": "record", "name": "log", "fields": [{ "name": "message", "type": "string" }] }"#, docs::additional_props_description = r#"Supports most avro data types, unsupported data types includes @@ -198,9 +196,7 @@ pub fn try_from(value: AvroValue) -> vector_common::Result { "AvroValue::Duration is not supported", )), AvroValue::Enum(_, string) => Ok(VrlValue::from(string)), - AvroValue::Fixed(_, _) => Err(vector_common::Error::from( - "AvroValue::Fixed is not supported", - )), + AvroValue::Fixed(_, bytes) => Ok(VrlValue::Bytes(Bytes::from(bytes))), AvroValue::Float(float) => Ok(VrlValue::from_f64_or_zero(float as f64)), AvroValue::Int(int) => Ok(VrlValue::from(int)), AvroValue::Long(long) => Ok(VrlValue::from(long)), @@ -217,9 +213,7 @@ pub fn try_from(value: AvroValue) -> vector_common::Result { .map(|v| VrlValue::Object(v.into_iter().collect())), AvroValue::String(string) => Ok(VrlValue::from(string)), AvroValue::TimeMicros(time_micros) => Ok(VrlValue::from(time_micros)), - AvroValue::TimeMillis(_) => Err(vector_common::Error::from( - "AvroValue::TimeMillis is not supported", - )), + AvroValue::TimeMillis(time_millis) => Ok(VrlValue::from(time_millis)), AvroValue::TimestampMicros(ts_micros) => Ok(VrlValue::from(ts_micros)), AvroValue::TimestampMillis(ts_millis) => Ok(VrlValue::from(ts_millis)), AvroValue::Union(_, v) => try_from(*v), @@ -229,12 +223,8 @@ pub fn try_from(value: AvroValue) -> vector_common::Result { AvroValue::BigDecimal(_) => Err(vector_common::Error::from( "AvroValue::BigDecimal is not supported", )), - AvroValue::TimestampNanos(_) => Err(vector_common::Error::from( - "AvroValue::TimestampNanos is not supported", - )), - AvroValue::LocalTimestampNanos(_) => Err(vector_common::Error::from( - "AvroValue::LocalTimestampNanos is not supported", - )), + AvroValue::TimestampNanos(ts_nanos) => Ok(VrlValue::from(ts_nanos)), + AvroValue::LocalTimestampNanos(ts_nanos) => Ok(VrlValue::from(ts_nanos)), } } diff --git a/lib/codecs/src/encoding/format/avro.rs b/lib/codecs/src/encoding/format/avro.rs index 297b0b39a491f..373093dbb99d4 100644 --- a/lib/codecs/src/encoding/format/avro.rs +++ b/lib/codecs/src/encoding/format/avro.rs @@ -39,16 +39,31 @@ pub(crate) fn to_avro( (VrlValue::Integer(i), Schema::TimeMicros) => Ok(AvroValue::TimeMicros(*i)), (VrlValue::Integer(i), Schema::TimestampMillis) => Ok(AvroValue::TimestampMillis(*i)), (VrlValue::Integer(i), Schema::TimestampMicros) => Ok(AvroValue::TimestampMicros(*i)), + (VrlValue::Integer(i), Schema::TimestampNanos) => Ok(AvroValue::TimestampNanos(*i)), (VrlValue::Integer(i), Schema::LocalTimestampMillis) => { Ok(AvroValue::LocalTimestampMillis(*i)) } (VrlValue::Integer(i), Schema::LocalTimestampMicros) => { Ok(AvroValue::LocalTimestampMicros(*i)) } + (VrlValue::Integer(i), Schema::LocalTimestampNanos) => { + Ok(AvroValue::LocalTimestampNanos(*i)) + } (VrlValue::Float(f), Schema::Float) => Ok(AvroValue::Float(f.into_inner() as f32)), (VrlValue::Float(f), Schema::Double) => Ok(AvroValue::Double(f.into_inner())), + (VrlValue::Bytes(b), Schema::Fixed(fixed_schema)) => { + let bytes = b.to_vec(); + if bytes.len() != fixed_schema.size { + return Err(vector_common::Error::from(format!( + "Bytes length {} does not match fixed schema size {}", + bytes.len(), + fixed_schema.size + ))); + } + Ok(AvroValue::Fixed(fixed_schema.size, bytes)) + } (VrlValue::Bytes(b), Schema::Bytes) => Ok(AvroValue::Bytes(b.to_vec())), (VrlValue::Bytes(b), Schema::String) => String::from_utf8(b.to_vec()) .map(AvroValue::String) @@ -90,6 +105,22 @@ pub(crate) fn to_avro( (VrlValue::Timestamp(ts), Schema::LocalTimestampMicros) => { Ok(AvroValue::LocalTimestampMicros(ts.timestamp_micros())) } + (VrlValue::Timestamp(ts), Schema::TimestampNanos) => ts + .timestamp_nanos_opt() + .map(AvroValue::TimestampNanos) + .ok_or_else(|| { + vector_common::Error::from(format!( + "Timestamp {ts} is out of range for timestamp-nanos" + )) + }), + (VrlValue::Timestamp(ts), Schema::LocalTimestampNanos) => ts + .timestamp_nanos_opt() + .map(AvroValue::LocalTimestampNanos) + .ok_or_else(|| { + vector_common::Error::from(format!( + "Timestamp {ts} is out of range for local-timestamp-nanos" + )) + }), (VrlValue::Timestamp(ts), Schema::Long) => Ok(AvroValue::Long(ts.timestamp_millis())), (VrlValue::Timestamp(ts), Schema::String) => Ok(AvroValue::String( ts.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true), diff --git a/lib/codecs/tests/avro.rs b/lib/codecs/tests/avro.rs index 4f3b57d653ffb..b40391228418f 100644 --- a/lib/codecs/tests/avro.rs +++ b/lib/codecs/tests/avro.rs @@ -22,7 +22,6 @@ use vector_core::{config::LogNamespace, event::Event}; #[case(false)] fn roundtrip_avro_fixtures( #[files("tests/data/avro/generated/*.avro")] - #[exclude(".*(fixed|time_millis).avro")] path: PathBuf, #[case] reserialize: bool, ) { diff --git a/lib/codecs/tests/bin/generate-avro-fixtures.rs b/lib/codecs/tests/bin/generate-avro-fixtures.rs index 549df090ff627..9dafe1a62cf55 100644 --- a/lib/codecs/tests/bin/generate-avro-fixtures.rs +++ b/lib/codecs/tests/bin/generate-avro-fixtures.rs @@ -142,7 +142,6 @@ fn generate_avro_test_case_string() -> Result<()> { generate_test_case(schema, value, "string") } -#[allow(unused)] fn generate_avro_test_case_fixed() -> Result<()> { let schema = r#" { @@ -319,7 +318,6 @@ fn generate_avro_test_case_decimal_var() -> Result<()> { generate_test_case_from_value(schema, record, "decimal_var") } -#[allow(unused)] fn generate_avro_test_case_time_millis() -> Result<()> { let schema = r#" { @@ -440,6 +438,46 @@ fn generate_avro_test_case_local_timestamp_micros() -> Result<()> { generate_test_case(schema, value, "local-timestamp_micros") } +fn generate_avro_test_case_timestamp_nanos() -> Result<()> { + let schema = r#" + { + "type": "record", + "name": "test", + "fields": [ + {"name": "timestamp_nanos_field", "type": "long", "logicalType": "timestamp-nanos"} + ] + } + "#; + #[derive(Debug, Serialize, Deserialize, Clone)] + struct Test { + timestamp_nanos_field: i64, + } + let value = Test { + timestamp_nanos_field: 1697445291056567890i64, + }; + generate_test_case(schema, value, "timestamp_nanos") +} + +fn generate_avro_test_case_local_timestamp_nanos() -> Result<()> { + let schema = r#" + { + "type": "record", + "name": "test", + "fields": [ + {"name": "local_timestamp_nanos_field", "type": "long", "logicalType": "local-timestamp-nanos"} + ] + } + "#; + #[derive(Debug, Serialize, Deserialize, Clone)] + struct Test { + local_timestamp_nanos_field: i64, + } + let value = Test { + local_timestamp_nanos_field: 1697445291056567890i64, + }; + generate_test_case(schema, value, "local-timestamp_nanos") +} + fn generate_avro_test_case_uuid() -> Result<()> { let schema = r#" { @@ -490,6 +528,7 @@ fn main() -> Result<()> { generate_avro_test_case_date()?; generate_avro_test_case_double()?; generate_avro_test_case_enum()?; + generate_avro_test_case_fixed()?; generate_avro_test_case_float()?; generate_avro_test_case_int()?; generate_avro_test_case_long()?; @@ -497,10 +536,13 @@ fn main() -> Result<()> { generate_avro_test_case_record()?; generate_avro_test_case_string()?; generate_avro_test_case_time_micros()?; + generate_avro_test_case_time_millis()?; generate_avro_test_case_timestamp_micros()?; generate_avro_test_case_timestamp_millis()?; + generate_avro_test_case_timestamp_nanos()?; generate_avro_test_case_local_timestamp_micros()?; generate_avro_test_case_local_timestamp_millis()?; + generate_avro_test_case_local_timestamp_nanos()?; generate_avro_test_case_union()?; generate_avro_test_case_uuid()?; Ok(()) diff --git a/lib/codecs/tests/data/avro/generated/fixed.avro b/lib/codecs/tests/data/avro/generated/fixed.avro new file mode 100644 index 0000000000000..50fd96c66abba --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/fixed.avro @@ -0,0 +1 @@ +1019181716151413 \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/fixed.avsc b/lib/codecs/tests/data/avro/generated/fixed.avsc new file mode 100644 index 0000000000000..ee03066a96081 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/fixed.avsc @@ -0,0 +1 @@ +{"type":"record","name":"test","fields":[{"name":"fixed_field","type":{"type":"fixed","name":"fixed_field","size":16}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avro b/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avro new file mode 100644 index 0000000000000..bb475c1f66117 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avro @@ -0,0 +1 @@ +¤Ù°ñõßÄŽ/ \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avsc b/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avsc new file mode 100644 index 0000000000000..72fc1dc5ea873 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/local-timestamp_nanos.avsc @@ -0,0 +1 @@ +{"type":"record","name":"test","fields":[{"name":"local_timestamp_nanos_field","type":{"type":"long","logicalType":"local-timestamp-nanos"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/time_millis.avro b/lib/codecs/tests/data/avro/generated/time_millis.avro new file mode 100644 index 0000000000000..c1aee88444369 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/time_millis.avro @@ -0,0 +1 @@ +¶¡†9 \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/time_millis.avsc b/lib/codecs/tests/data/avro/generated/time_millis.avsc new file mode 100644 index 0000000000000..9a281db2c6e99 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/time_millis.avsc @@ -0,0 +1 @@ +{"type":"record","name":"test","fields":[{"name":"time_millis_field","type":{"type":"int","logicalType":"time-millis"}}]} \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_nanos.avro b/lib/codecs/tests/data/avro/generated/timestamp_nanos.avro new file mode 100644 index 0000000000000..bb475c1f66117 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/timestamp_nanos.avro @@ -0,0 +1 @@ +¤Ù°ñõßÄŽ/ \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/timestamp_nanos.avsc b/lib/codecs/tests/data/avro/generated/timestamp_nanos.avsc new file mode 100644 index 0000000000000..e03634505d251 --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/timestamp_nanos.avsc @@ -0,0 +1 @@ +{"type":"record","name":"test","fields":[{"name":"timestamp_nanos_field","type":{"type":"long","logicalType":"timestamp-nanos"}}]} \ No newline at end of file diff --git a/website/cue/reference/components/sinks/generated/websocket_server.cue b/website/cue/reference/components/sinks/generated/websocket_server.cue index 7df8b9be2eeac..cd08af2e36fac 100644 --- a/website/cue/reference/components/sinks/generated/websocket_server.cue +++ b/website/cue/reference/components/sinks/generated/websocket_server.cue @@ -596,8 +596,6 @@ generated: components: sinks: websocket_server: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/amqp.cue b/website/cue/reference/components/sources/generated/amqp.cue index 3b58bc542b30b..f76bb291329fc 100644 --- a/website/cue/reference/components/sources/generated/amqp.cue +++ b/website/cue/reference/components/sources/generated/amqp.cue @@ -64,8 +64,6 @@ generated: components: sources: amqp: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue b/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue index 8807aeffba320..e96ecbe7be460 100644 --- a/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue +++ b/website/cue/reference/components/sources/generated/aws_kinesis_firehose.cue @@ -84,8 +84,6 @@ generated: components: sources: aws_kinesis_firehose: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/aws_s3.cue b/website/cue/reference/components/sources/generated/aws_s3.cue index 0fc4763c51bed..d1ae1aeb650ae 100644 --- a/website/cue/reference/components/sources/generated/aws_s3.cue +++ b/website/cue/reference/components/sources/generated/aws_s3.cue @@ -182,8 +182,6 @@ generated: components: sources: aws_s3: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/aws_sqs.cue b/website/cue/reference/components/sources/generated/aws_sqs.cue index c5d113713dfc2..d9d1de9b62f0c 100644 --- a/website/cue/reference/components/sources/generated/aws_sqs.cue +++ b/website/cue/reference/components/sources/generated/aws_sqs.cue @@ -177,8 +177,6 @@ generated: components: sources: aws_sqs: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/datadog_agent.cue b/website/cue/reference/components/sources/generated/datadog_agent.cue index 6059924b5ed22..1e376589b4be0 100644 --- a/website/cue/reference/components/sources/generated/datadog_agent.cue +++ b/website/cue/reference/components/sources/generated/datadog_agent.cue @@ -49,8 +49,6 @@ generated: components: sources: datadog_agent: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/demo_logs.cue b/website/cue/reference/components/sources/generated/demo_logs.cue index c0f94f4b1eceb..257dd839c1888 100644 --- a/website/cue/reference/components/sources/generated/demo_logs.cue +++ b/website/cue/reference/components/sources/generated/demo_logs.cue @@ -28,8 +28,6 @@ generated: components: sources: demo_logs: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/exec.cue b/website/cue/reference/components/sources/generated/exec.cue index 62acd72bbdfdd..14edb358885f8 100644 --- a/website/cue/reference/components/sources/generated/exec.cue +++ b/website/cue/reference/components/sources/generated/exec.cue @@ -29,8 +29,6 @@ generated: components: sources: exec: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/file_descriptor.cue b/website/cue/reference/components/sources/generated/file_descriptor.cue index 06bc07eba6e46..594f9ccf50ed9 100644 --- a/website/cue/reference/components/sources/generated/file_descriptor.cue +++ b/website/cue/reference/components/sources/generated/file_descriptor.cue @@ -19,8 +19,6 @@ generated: components: sources: file_descriptor: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue index dc9a9f8319f9c..fd9651c27ba2d 100644 --- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue +++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue @@ -95,8 +95,6 @@ generated: components: sources: gcp_pubsub: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/heroku_logs.cue b/website/cue/reference/components/sources/generated/heroku_logs.cue index 1f383b3d18c49..161edaa2e4095 100644 --- a/website/cue/reference/components/sources/generated/heroku_logs.cue +++ b/website/cue/reference/components/sources/generated/heroku_logs.cue @@ -92,8 +92,6 @@ generated: components: sources: heroku_logs: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/http.cue b/website/cue/reference/components/sources/generated/http.cue index 076ea8a3a2816..70b87500a84c0 100644 --- a/website/cue/reference/components/sources/generated/http.cue +++ b/website/cue/reference/components/sources/generated/http.cue @@ -100,8 +100,6 @@ generated: components: sources: http: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/http_client.cue b/website/cue/reference/components/sources/generated/http_client.cue index 9f38084556c88..e163ce952ee90 100644 --- a/website/cue/reference/components/sources/generated/http_client.cue +++ b/website/cue/reference/components/sources/generated/http_client.cue @@ -230,8 +230,6 @@ generated: components: sources: http_client: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/http_server.cue b/website/cue/reference/components/sources/generated/http_server.cue index 685c8f0781987..7d602b46dd972 100644 --- a/website/cue/reference/components/sources/generated/http_server.cue +++ b/website/cue/reference/components/sources/generated/http_server.cue @@ -100,8 +100,6 @@ generated: components: sources: http_server: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/kafka.cue b/website/cue/reference/components/sources/generated/kafka.cue index 5e472a5aa1ac8..c19941a7d8dbe 100644 --- a/website/cue/reference/components/sources/generated/kafka.cue +++ b/website/cue/reference/components/sources/generated/kafka.cue @@ -73,8 +73,6 @@ generated: components: sources: kafka: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/mqtt.cue b/website/cue/reference/components/sources/generated/mqtt.cue index 5dddd7141790a..d011d3b963d42 100644 --- a/website/cue/reference/components/sources/generated/mqtt.cue +++ b/website/cue/reference/components/sources/generated/mqtt.cue @@ -24,8 +24,6 @@ generated: components: sources: mqtt: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/nats.cue b/website/cue/reference/components/sources/generated/nats.cue index dce7eb1dc9dca..6be275885c20a 100644 --- a/website/cue/reference/components/sources/generated/nats.cue +++ b/website/cue/reference/components/sources/generated/nats.cue @@ -116,8 +116,6 @@ generated: components: sources: nats: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/pulsar.cue b/website/cue/reference/components/sources/generated/pulsar.cue index fa16e8f93c710..821dc7836f473 100644 --- a/website/cue/reference/components/sources/generated/pulsar.cue +++ b/website/cue/reference/components/sources/generated/pulsar.cue @@ -122,8 +122,6 @@ generated: components: sources: pulsar: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/redis.cue b/website/cue/reference/components/sources/generated/redis.cue index 80d40e9c89f4f..d9783bc963a8e 100644 --- a/website/cue/reference/components/sources/generated/redis.cue +++ b/website/cue/reference/components/sources/generated/redis.cue @@ -34,8 +34,6 @@ generated: components: sources: redis: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/socket.cue b/website/cue/reference/components/sources/generated/socket.cue index 9f7e37cb836f3..3768bb33cd460 100644 --- a/website/cue/reference/components/sources/generated/socket.cue +++ b/website/cue/reference/components/sources/generated/socket.cue @@ -36,8 +36,6 @@ generated: components: sources: socket: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/stdin.cue b/website/cue/reference/components/sources/generated/stdin.cue index a7472a3511662..0a8c4a60c549e 100644 --- a/website/cue/reference/components/sources/generated/stdin.cue +++ b/website/cue/reference/components/sources/generated/stdin.cue @@ -19,8 +19,6 @@ generated: components: sources: stdin: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] diff --git a/website/cue/reference/components/sources/generated/websocket.cue b/website/cue/reference/components/sources/generated/websocket.cue index cb39facf06008..c1d715625b7be 100644 --- a/website/cue/reference/components/sources/generated/websocket.cue +++ b/website/cue/reference/components/sources/generated/websocket.cue @@ -206,8 +206,6 @@ generated: components: sources: websocket: configuration: { **Note**: The following [`apache_avro::types::Value`] variants are *not* supported: * `Decimal` * `Duration` - * `Fixed` - * `TimeMillis` """ required: true type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] From 53a9071685b4e48f26a9b6856b77ee80923bbf07 Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Wed, 22 Apr 2026 17:36:51 +0100 Subject: [PATCH 07/11] Update changelog --- changelog.d/24773_support_avro_date_values.fix.md | 3 --- changelog.d/24773_support_more_avro_types.fix.md | 3 +++ 2 files changed, 3 insertions(+), 3 deletions(-) delete mode 100644 changelog.d/24773_support_avro_date_values.fix.md create mode 100644 changelog.d/24773_support_more_avro_types.fix.md diff --git a/changelog.d/24773_support_avro_date_values.fix.md b/changelog.d/24773_support_avro_date_values.fix.md deleted file mode 100644 index 36b375a10fc3d..0000000000000 --- a/changelog.d/24773_support_avro_date_values.fix.md +++ /dev/null @@ -1,3 +0,0 @@ -The Avro codec now supports decoding Avro Date values. - -authors: omwbennett diff --git a/changelog.d/24773_support_more_avro_types.fix.md b/changelog.d/24773_support_more_avro_types.fix.md new file mode 100644 index 0000000000000..47d44a6bee32b --- /dev/null +++ b/changelog.d/24773_support_more_avro_types.fix.md @@ -0,0 +1,3 @@ +The Avro codec now supports encoding and decoding Date, Fixed, TimeMillis, TimestampNanos, LocalTimestampNanos values. + +authors: omwbennett From 20815d0f76db7541d3dd1516b09c840044124b4e Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Thu, 23 Apr 2026 09:59:13 +0100 Subject: [PATCH 08/11] Add conversion of vrl int to avro float/double --- lib/codecs/src/encoding/format/avro.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/codecs/src/encoding/format/avro.rs b/lib/codecs/src/encoding/format/avro.rs index 373093dbb99d4..aff293c84abec 100644 --- a/lib/codecs/src/encoding/format/avro.rs +++ b/lib/codecs/src/encoding/format/avro.rs @@ -50,6 +50,8 @@ pub(crate) fn to_avro( Ok(AvroValue::LocalTimestampNanos(*i)) } + (VrlValue::Integer(i), Schema::Float) => Ok(AvroValue::Float(*i as f32)), + (VrlValue::Integer(i), Schema::Double) => Ok(AvroValue::Double(*i as f64)), (VrlValue::Float(f), Schema::Float) => Ok(AvroValue::Float(f.into_inner() as f32)), (VrlValue::Float(f), Schema::Double) => Ok(AvroValue::Double(f.into_inner())), From 05045a98cee88f6c58294345f135880cbefcad67 Mon Sep 17 00:00:00 2001 From: Oliver Bennett Date: Sun, 26 Apr 2026 10:33:44 +0100 Subject: [PATCH 09/11] Refactor to_avro to coerce_logical_types --- lib/codecs/src/encoding/format/avro.rs | 421 ++++++++++++++----------- lib/codecs/tests/avro.rs | 3 +- 2 files changed, 240 insertions(+), 184 deletions(-) diff --git a/lib/codecs/src/encoding/format/avro.rs b/lib/codecs/src/encoding/format/avro.rs index aff293c84abec..1c7daee66f7b8 100644 --- a/lib/codecs/src/encoding/format/avro.rs +++ b/lib/codecs/src/encoding/format/avro.rs @@ -1,206 +1,111 @@ use bytes::{BufMut, BytesMut}; use serde::{Deserialize, Serialize}; use tokio_util::codec::Encoder; -use uuid::Uuid; use vector_config::configurable_component; use vector_core::{config::DataType, event::Event, schema}; use crate::encoding::BuildError; -type VrlValue = vrl::value::Value; type AvroValue = apache_avro::types::Value; -/// Converts a VRL [`Value`](VrlValue) to an [`apache_avro::types::Value`] using the provided -/// schema to resolve ambiguous types (e.g. `Integer` -> `Int` vs `Date`). -pub(crate) fn to_avro( - value: &VrlValue, +/// `apache_avro::to_value` may serialize VRL values into Avro types which later +/// cannot be resolved against certain Avro types +/// (e.g. VRL integer (i64) -> Avro `Long` which cannot be resolved to Avro `Date`). +/// `coerce_logical_types` does a recursive pre-pass to fix such cases. +fn coerce_logical_types( + value: AvroValue, schema: &apache_avro::Schema, - names: &apache_avro::schema::NamesRef<'_>, ) -> vector_common::Result { use apache_avro::Schema; match (value, schema) { - (VrlValue::Null, Schema::Null) => Ok(AvroValue::Null), - - (VrlValue::Boolean(b), Schema::Boolean) => Ok(AvroValue::Boolean(*b)), - - (VrlValue::Integer(i), Schema::Int) => i32::try_from(*i) - .map(AvroValue::Int) - .map_err(|_| vector_common::Error::from(format!("Integer {i} overflows Avro int (i32)"))), - (VrlValue::Integer(i), Schema::Date) => i32::try_from(*i) - .map(AvroValue::Date) - .map_err(|_| vector_common::Error::from(format!("Integer {i} overflows Avro date (i32)"))), - (VrlValue::Integer(i), Schema::TimeMillis) => i32::try_from(*i) - .map(AvroValue::TimeMillis) - .map_err(|_| { - vector_common::Error::from(format!("Integer {i} overflows Avro time-millis (i32)")) - }), - - (VrlValue::Integer(i), Schema::Long) => Ok(AvroValue::Long(*i)), - (VrlValue::Integer(i), Schema::TimeMicros) => Ok(AvroValue::TimeMicros(*i)), - (VrlValue::Integer(i), Schema::TimestampMillis) => Ok(AvroValue::TimestampMillis(*i)), - (VrlValue::Integer(i), Schema::TimestampMicros) => Ok(AvroValue::TimestampMicros(*i)), - (VrlValue::Integer(i), Schema::TimestampNanos) => Ok(AvroValue::TimestampNanos(*i)), - (VrlValue::Integer(i), Schema::LocalTimestampMillis) => { - Ok(AvroValue::LocalTimestampMillis(*i)) - } - (VrlValue::Integer(i), Schema::LocalTimestampMicros) => { - Ok(AvroValue::LocalTimestampMicros(*i)) - } - (VrlValue::Integer(i), Schema::LocalTimestampNanos) => { - Ok(AvroValue::LocalTimestampNanos(*i)) - } - - (VrlValue::Integer(i), Schema::Float) => Ok(AvroValue::Float(*i as f32)), - (VrlValue::Integer(i), Schema::Double) => Ok(AvroValue::Double(*i as f64)), - (VrlValue::Float(f), Schema::Float) => Ok(AvroValue::Float(f.into_inner() as f32)), - (VrlValue::Float(f), Schema::Double) => Ok(AvroValue::Double(f.into_inner())), - - (VrlValue::Bytes(b), Schema::Fixed(fixed_schema)) => { - let bytes = b.to_vec(); - if bytes.len() != fixed_schema.size { - return Err(vector_common::Error::from(format!( - "Bytes length {} does not match fixed schema size {}", - bytes.len(), - fixed_schema.size - ))); - } - Ok(AvroValue::Fixed(fixed_schema.size, bytes)) - } - (VrlValue::Bytes(b), Schema::Bytes) => Ok(AvroValue::Bytes(b.to_vec())), - (VrlValue::Bytes(b), Schema::String) => String::from_utf8(b.to_vec()) - .map(AvroValue::String) - .map_err(|e| { - vector_common::Error::from(format!("Invalid UTF-8 in string field: {e}")) - }), - (VrlValue::Regex(b), Schema::String) => Ok(AvroValue::String(b.as_str().to_owned())), - - (VrlValue::Bytes(b), Schema::Uuid) => { - let s = String::from_utf8(b.to_vec()).map_err(|e| { - vector_common::Error::from(format!("Invalid UTF-8 in UUID field: {e}")) - })?; - Uuid::parse_str(&s) - .map(AvroValue::Uuid) - .map_err(|e| vector_common::Error::from(format!("Invalid UUID: {e}"))) - } - - (VrlValue::Bytes(b), Schema::Enum(enum_schema)) => { - let s = String::from_utf8(b.to_vec()).map_err(|e| { - vector_common::Error::from(format!("Invalid UTF-8 in enum field: {e}")) - })?; - let index = enum_schema - .symbols - .iter() - .position(|sym| sym == &s) - .ok_or_else(|| vector_common::Error::from(format!("Unknown enum symbol: {s}")))?; - Ok(AvroValue::Enum(index as u32, s)) - } - - (VrlValue::Timestamp(ts), Schema::TimestampMillis) => { - Ok(AvroValue::TimestampMillis(ts.timestamp_millis())) - } - (VrlValue::Timestamp(ts), Schema::TimestampMicros) => { - Ok(AvroValue::TimestampMicros(ts.timestamp_micros())) - } - (VrlValue::Timestamp(ts), Schema::LocalTimestampMillis) => { - Ok(AvroValue::LocalTimestampMillis(ts.timestamp_millis())) - } - (VrlValue::Timestamp(ts), Schema::LocalTimestampMicros) => { - Ok(AvroValue::LocalTimestampMicros(ts.timestamp_micros())) - } - (VrlValue::Timestamp(ts), Schema::TimestampNanos) => ts - .timestamp_nanos_opt() - .map(AvroValue::TimestampNanos) - .ok_or_else(|| { + (AvroValue::Long(days), Schema::Date) => { + i32::try_from(days).map(AvroValue::Date).map_err(|_| { vector_common::Error::from(format!( - "Timestamp {ts} is out of range for timestamp-nanos" + "Avro date value {days} is out of range for i32" )) - }), - (VrlValue::Timestamp(ts), Schema::LocalTimestampNanos) => ts - .timestamp_nanos_opt() - .map(AvroValue::LocalTimestampNanos) - .ok_or_else(|| { + }) + } + (AvroValue::Long(millis), Schema::TimeMillis) => i32::try_from(millis) + .map(AvroValue::TimeMillis) + .map_err(|_| { vector_common::Error::from(format!( - "Timestamp {ts} is out of range for local-timestamp-nanos" + "Avro time-millis value {millis} is out of range for i32" )) }), - (VrlValue::Timestamp(ts), Schema::Long) => Ok(AvroValue::Long(ts.timestamp_millis())), - (VrlValue::Timestamp(ts), Schema::String) => Ok(AvroValue::String( - ts.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true), - )), - - (v, Schema::Ref { name }) => { - let resolved = names.get(name).ok_or_else(|| { - vector_common::Error::from(format!("Unknown schema ref: {}", name.fullname(None))) - })?; - to_avro(v, resolved, names) - } - - (VrlValue::Array(items), Schema::Array(array_schema)) => items - .iter() - .map(|item| to_avro(item, &array_schema.items, names)) - .collect::, _>>() - .map(AvroValue::Array), - - (VrlValue::Object(map), Schema::Map(map_schema)) => map - .iter() - .map(|(k, v)| to_avro(v, &map_schema.types, names).map(|av| (k.to_string(), av))) - .collect::, _>>() - .map(|items| AvroValue::Map(items.into_iter().collect())), - - (VrlValue::Object(map), Schema::Record(record_schema)) => { - let fields = record_schema - .fields - .iter() - .map(|field| { - let av = match map.get(field.name.as_str()) { - Some(v) => to_avro(v, &field.schema, names)?, - None => match &field.default { - Some(json_default) => { - AvroValue::from(json_default.clone()).resolve(&field.schema)? - } - None => { - return Err(vector_common::Error::from(format!( - "Missing record field: {}", - field.name - ))) - } - }, + (AvroValue::Record(fields), Schema::Record(record_schema)) => { + let fields = fields + .into_iter() + .map(|(name, value)| { + let value = match record_schema.lookup.get(&name) { + Some(index) => { + let field_schema = &record_schema.fields[*index].schema; + coerce_logical_types(value, field_schema)? + } + None => value, }; - Ok((field.name.clone(), av)) + Ok((name, value)) }) - .collect::, _>>()?; + .collect::>>()?; Ok(AvroValue::Record(fields)) } - - (v, Schema::Union(union_schema)) => { - // Prefer null variant for Null values, otherwise try each variant in order - if matches!(v, VrlValue::Null) - && let Some(idx) = union_schema - .variants() - .iter() - .position(|s| matches!(s, Schema::Null)) - { - return Ok(AvroValue::Union(idx as u32, Box::new(AvroValue::Null))); + (AvroValue::Map(entries), Schema::Record(record_schema)) => { + let entries = entries + .into_iter() + .map(|(name, value)| { + let value = match record_schema.lookup.get(&name) { + Some(index) => { + let field_schema = &record_schema.fields[*index].schema; + coerce_logical_types(value, field_schema)? + } + None => value, + }; + Ok((name, value)) + }) + .collect::>()?; + Ok(AvroValue::Map(entries)) + } + (AvroValue::Array(items), Schema::Array(array_schema)) => items + .into_iter() + .map(|item| coerce_logical_types(item, &array_schema.items)) + .collect::, _>>() + .map(AvroValue::Array), + (AvroValue::Map(entries), Schema::Map(map_schema)) => entries + .into_iter() + .map(|(key, value)| { + coerce_logical_types(value, &map_schema.types).map(|value| (key, value)) + }) + .collect::>() + .map(AvroValue::Map), + (AvroValue::Union(index, value), Schema::Union(union_schema)) => { + let schema = union_schema + .variants() + .get(index as usize) + .unwrap_or(schema); + coerce_logical_types(*value, schema) + .map(|value| AvroValue::Union(index, Box::new(value))) + } + (value, Schema::Union(union_schema)) => { + if let Ok(resolved) = value.clone().resolve(schema) { + return Ok(resolved); } - for (idx, variant_schema) in union_schema.variants().iter().enumerate() { - if matches!(variant_schema, Schema::Null) { - continue; - } - if let Ok(av) = to_avro(v, variant_schema, names) { - return Ok(AvroValue::Union(idx as u32, Box::new(av))); + + let mut last_err = None; + for (index, variant) in union_schema.variants().iter().enumerate() { + match coerce_logical_types(value.clone(), variant) { + Ok(coerced) if coerced.clone().resolve(variant).is_ok() => { + return Ok(AvroValue::Union(index as u32, Box::new(coerced))); + } + Ok(_) => {} + Err(err) => last_err = Some(err), } } - Err(vector_common::Error::from(format!( - "No matching union variant for value of kind {}", - v.kind_str() - ))) - } - (v, s) => Err(vector_common::Error::from(format!( - "Cannot convert VRL {} to Avro schema {:?}", - v.kind_str(), - s - ))), + match last_err { + Some(err) => Err(err), + None => Ok(value), + } + } + (value, _) => Ok(value), } } @@ -268,12 +173,10 @@ impl Encoder for AvroSerializer { fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> { let log = event.into_log(); - let (value, _metadata) = log.into_parts(); - let resolved = apache_avro::schema::ResolvedSchema::try_from(&self.schema) - .map_err(|e| vector_common::Error::from(format!("Failed resolving Avro schema: {e}")))?; - let names = resolved.get_names(); - let avro_value = to_avro(&value, &self.schema, names)?; - let bytes = apache_avro::to_avro_datum(&self.schema, avro_value)?; + let value = apache_avro::to_value(log)?; + let value = coerce_logical_types(value, &self.schema)?; + let value = value.resolve(&self.schema)?; + let bytes = apache_avro::to_avro_datum(&self.schema, value)?; buffer.put_slice(&bytes); Ok(()) } @@ -314,4 +217,158 @@ mod tests { assert_eq!(bytes.freeze(), b"\0\x06bar".as_slice()); } + + #[test] + fn coerce_date_fields_recursively() { + let schema = apache_avro::Schema::parse_str(indoc! {r#" + { + "type": "record", + "name": "Outer", + "fields": [ + { + "name": "direct_date", + "type": {"type": "int", "logicalType": "date"} + }, + { + "name": "inner", + "type": { + "type": "record", + "name": "Inner", + "fields": [ + { + "name": "date", + "type": {"type": "int", "logicalType": "date"} + } + ] + } + }, + { + "name": "record_as_map", + "type": { + "type": "record", + "name": "MapBackedInner", + "fields": [ + { + "name": "date", + "type": {"type": "int", "logicalType": "date"} + } + ] + } + }, + { + "name": "date_array", + "type": { + "type": "array", + "items": {"type": "int", "logicalType": "date"} + } + }, + { + "name": "date_map", + "type": { + "type": "map", + "values": {"type": "int", "logicalType": "date"} + } + }, + { + "name": "union_date", + "type": ["null", {"type": "int", "logicalType": "date"}] + }, + { + "name": "fallback_union_date", + "type": [ + "null", + {"type": "int", "logicalType": "date"}, + "long" + ] + }, + { + "name": "logical_only_union_date", + "type": [ + "null", + {"type": "int", "logicalType": "date"} + ] + } + ] + } + "#}) + .unwrap(); + let value = AvroValue::Record(vec![ + ("direct_date".to_owned(), AvroValue::Long(20_000)), + ( + "inner".to_owned(), + AvroValue::Record(vec![("date".to_owned(), AvroValue::Long(20_001))]), + ), + ( + "record_as_map".to_owned(), + AvroValue::Map( + [("date".to_owned(), AvroValue::Long(20_002))] + .into_iter() + .collect(), + ), + ), + ( + "date_array".to_owned(), + AvroValue::Array(vec![AvroValue::Long(20_003), AvroValue::Long(20_004)]), + ), + ( + "date_map".to_owned(), + AvroValue::Map( + [ + ("first".to_owned(), AvroValue::Long(20_005)), + ("second".to_owned(), AvroValue::Long(20_006)), + ] + .into_iter() + .collect(), + ), + ), + ("union_date".to_owned(), AvroValue::Long(20_007)), + ("fallback_union_date".to_owned(), AvroValue::Long(20_009)), + ( + "logical_only_union_date".to_owned(), + AvroValue::Long(20_008), + ), + ]); + + let value = coerce_logical_types(value, &schema).unwrap(); + let value = value.resolve(&schema).unwrap(); + + assert!(matches!( + value, + AvroValue::Record(fields) if { + matches!(fields[0].1, AvroValue::Date(20_000)) + && matches!( + &fields[1].1, + AvroValue::Record(inner) if matches!(inner[0].1, AvroValue::Date(20_001)) + ) + && matches!( + &fields[2].1, + AvroValue::Record(inner) if matches!(inner[0].1, AvroValue::Date(20_002)) + ) + && matches!( + &fields[3].1, + AvroValue::Array(items) + if matches!(items.as_slice(), [AvroValue::Date(20_003), AvroValue::Date(20_004)]) + ) + && matches!( + &fields[4].1, + AvroValue::Map(entries) + if matches!(entries.get("first"), Some(AvroValue::Date(20_005))) + && matches!(entries.get("second"), Some(AvroValue::Date(20_006))) + ) + && matches!( + &fields[5].1, + AvroValue::Union(1, value) if matches!(value.as_ref(), AvroValue::Date(20_007)) + ) + && matches!( + &fields[6].1, + AvroValue::Union(2, value) + if matches!(value.as_ref(), AvroValue::Long(20_009)) + ) + && matches!( + &fields[7].1, + AvroValue::Union(1, value) if matches!(value.as_ref(), AvroValue::Date(20_008)) + ) + } + )); + } } diff --git a/lib/codecs/tests/avro.rs b/lib/codecs/tests/avro.rs index b40391228418f..b56c15dbea61b 100644 --- a/lib/codecs/tests/avro.rs +++ b/lib/codecs/tests/avro.rs @@ -21,8 +21,7 @@ use vector_core::{config::LogNamespace, event::Event}; #[case(true)] #[case(false)] fn roundtrip_avro_fixtures( - #[files("tests/data/avro/generated/*.avro")] - path: PathBuf, + #[files("tests/data/avro/generated/*.avro")] path: PathBuf, #[case] reserialize: bool, ) { let schema_path = path.as_path().with_extension("avsc"); From 88a117b73e3d609fd586fcb7da6f13ff9d3229be Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Mon, 27 Jul 2026 14:02:19 -0400 Subject: [PATCH 10/11] fix(codecs): coerce logical types through schema refs --- lib/codecs/src/encoding/format/avro.rs | 62 ++++++++++++++++++++++---- 1 file changed, 54 insertions(+), 8 deletions(-) diff --git a/lib/codecs/src/encoding/format/avro.rs b/lib/codecs/src/encoding/format/avro.rs index 1c7daee66f7b8..6ccf15e498164 100644 --- a/lib/codecs/src/encoding/format/avro.rs +++ b/lib/codecs/src/encoding/format/avro.rs @@ -15,6 +15,7 @@ type AvroValue = apache_avro::types::Value; fn coerce_logical_types( value: AvroValue, schema: &apache_avro::Schema, + names: &apache_avro::schema::NamesRef<'_>, ) -> vector_common::Result { use apache_avro::Schema; match (value, schema) { @@ -32,6 +33,12 @@ fn coerce_logical_types( "Avro time-millis value {millis} is out of range for i32" )) }), + (value, Schema::Ref { name }) => { + let schema = names.get(name).ok_or_else(|| { + vector_common::Error::from(format!("Unknown schema ref: {}", name.fullname(None))) + })?; + coerce_logical_types(value, schema, names) + } (AvroValue::Record(fields), Schema::Record(record_schema)) => { let fields = fields .into_iter() @@ -39,7 +46,7 @@ fn coerce_logical_types( let value = match record_schema.lookup.get(&name) { Some(index) => { let field_schema = &record_schema.fields[*index].schema; - coerce_logical_types(value, field_schema)? + coerce_logical_types(value, field_schema, names)? } None => value, }; @@ -55,7 +62,7 @@ fn coerce_logical_types( let value = match record_schema.lookup.get(&name) { Some(index) => { let field_schema = &record_schema.fields[*index].schema; - coerce_logical_types(value, field_schema)? + coerce_logical_types(value, field_schema, names)? } None => value, }; @@ -66,13 +73,13 @@ fn coerce_logical_types( } (AvroValue::Array(items), Schema::Array(array_schema)) => items .into_iter() - .map(|item| coerce_logical_types(item, &array_schema.items)) + .map(|item| coerce_logical_types(item, &array_schema.items, names)) .collect::, _>>() .map(AvroValue::Array), (AvroValue::Map(entries), Schema::Map(map_schema)) => entries .into_iter() .map(|(key, value)| { - coerce_logical_types(value, &map_schema.types).map(|value| (key, value)) + coerce_logical_types(value, &map_schema.types, names).map(|value| (key, value)) }) .collect::>() .map(AvroValue::Map), @@ -81,7 +88,7 @@ fn coerce_logical_types( .variants() .get(index as usize) .unwrap_or(schema); - coerce_logical_types(*value, schema) + coerce_logical_types(*value, schema, names) .map(|value| AvroValue::Union(index, Box::new(value))) } (value, Schema::Union(union_schema)) => { @@ -91,7 +98,7 @@ fn coerce_logical_types( let mut last_err = None; for (index, variant) in union_schema.variants().iter().enumerate() { - match coerce_logical_types(value.clone(), variant) { + match coerce_logical_types(value.clone(), variant, names) { Ok(coerced) if coerced.clone().resolve(variant).is_ok() => { return Ok(AvroValue::Union(index as u32, Box::new(coerced))); } @@ -174,7 +181,11 @@ impl Encoder for AvroSerializer { fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> { let log = event.into_log(); let value = apache_avro::to_value(log)?; - let value = coerce_logical_types(value, &self.schema)?; + let resolved = + apache_avro::schema::ResolvedSchema::try_from(&self.schema).map_err(|error| { + vector_common::Error::from(format!("Failed resolving Avro schema: {error}")) + })?; + let value = coerce_logical_types(value, &self.schema, resolved.get_names())?; let value = value.resolve(&self.schema)?; let bytes = apache_avro::to_avro_datum(&self.schema, value)?; buffer.put_slice(&bytes); @@ -329,7 +340,8 @@ mod tests { ), ]); - let value = coerce_logical_types(value, &schema).unwrap(); + let resolved = apache_avro::schema::ResolvedSchema::try_from(&schema).unwrap(); + let value = coerce_logical_types(value, &schema, resolved.get_names()).unwrap(); let value = value.resolve(&schema).unwrap(); assert!(matches!( @@ -371,4 +383,38 @@ mod tests { } )); } + + #[test] + fn coerce_date_through_named_record_reference() { + let schema = apache_avro::Schema::parse_str(indoc! {r#" + { + "type": "record", + "name": "Outer", + "fields": [ + { + "name": "definition", + "type": { + "type": "record", + "name": "Inner", + "fields": [{ + "name": "date", + "type": {"type": "int", "logicalType": "date"} + }] + } + }, + {"name": "reference", "type": "Inner"} + ] + } + "#}) + .unwrap(); + let inner = || AvroValue::Record(vec![("date".to_owned(), AvroValue::Long(20_000))]); + let value = AvroValue::Record(vec![ + ("definition".to_owned(), inner()), + ("reference".to_owned(), inner()), + ]); + + let resolved = apache_avro::schema::ResolvedSchema::try_from(&schema).unwrap(); + let value = coerce_logical_types(value, &schema, resolved.get_names()).unwrap(); + value.resolve(&schema).unwrap(); + } } From 63ff8274c0fcdfb99c9a3771dd615e95e2226d94 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Mon, 27 Jul 2026 14:04:18 -0400 Subject: [PATCH 11/11] test(codecs): add named Avro record reference fixture --- .../tests/bin/generate-avro-fixtures.rs | 30 +++++++++++++++++++ .../named_record_reference_date.avro | 1 + .../named_record_reference_date.avsc | 1 + 3 files changed, 32 insertions(+) create mode 100644 lib/codecs/tests/data/avro/generated/named_record_reference_date.avro create mode 100644 lib/codecs/tests/data/avro/generated/named_record_reference_date.avsc diff --git a/lib/codecs/tests/bin/generate-avro-fixtures.rs b/lib/codecs/tests/bin/generate-avro-fixtures.rs index 9dafe1a62cf55..52430ce75b7d5 100644 --- a/lib/codecs/tests/bin/generate-avro-fixtures.rs +++ b/lib/codecs/tests/bin/generate-avro-fixtures.rs @@ -297,6 +297,35 @@ fn generate_avro_test_case_date() -> Result<()> { generate_test_case(schema, value, "date") } +fn generate_avro_test_case_named_record_reference_date() -> Result<()> { + let schema = r#" + { + "type": "record", + "name": "Outer", + "fields": [ + { + "name": "definition", + "type": { + "type": "record", + "name": "Inner", + "fields": [{ + "name": "date", + "type": {"type": "int", "logicalType": "date"} + }] + } + }, + {"name": "reference", "type": "Inner"} + ] + } + "#; + let inner = |date| Value::Record(vec![("date".into(), Value::Date(date))]); + let value = Value::Record(vec![ + ("definition".into(), inner(20_000)), + ("reference".into(), inner(20_001)), + ]); + generate_test_case_from_value(schema, value, "named_record_reference_date") +} + #[allow(unused)] fn generate_avro_test_case_decimal_var() -> Result<()> { let schema = r#" @@ -526,6 +555,7 @@ fn main() -> Result<()> { generate_avro_test_case_boolean()?; generate_avro_test_case_bytes()?; generate_avro_test_case_date()?; + generate_avro_test_case_named_record_reference_date()?; generate_avro_test_case_double()?; generate_avro_test_case_enum()?; generate_avro_test_case_fixed()?; diff --git a/lib/codecs/tests/data/avro/generated/named_record_reference_date.avro b/lib/codecs/tests/data/avro/generated/named_record_reference_date.avro new file mode 100644 index 0000000000000..0fab28d24933d --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/named_record_reference_date.avro @@ -0,0 +1 @@ +À¸¸ \ No newline at end of file diff --git a/lib/codecs/tests/data/avro/generated/named_record_reference_date.avsc b/lib/codecs/tests/data/avro/generated/named_record_reference_date.avsc new file mode 100644 index 0000000000000..52826b675e3bd --- /dev/null +++ b/lib/codecs/tests/data/avro/generated/named_record_reference_date.avsc @@ -0,0 +1 @@ +{"type":"record","name":"Outer","fields":[{"name":"definition","type":{"type":"record","name":"Inner","fields":[{"name":"date","type":{"type":"int","logicalType":"date"}}]}},{"name":"reference","type":"Inner"}]} \ No newline at end of file