From 87bf4401ff81fc255ca3b3f2ade18daed0fc7ea1 Mon Sep 17 00:00:00 2001
From: Song Huang <xiasongh@gmail.com>
Date: Thu, 23 Jul 2026 22:53:42 -0400
Subject: [PATCH] adapters: serialize Avro timestamptz as micros

---
 crates/adapters/src/format/avro/serializer.rs | 31 ++++++++++++++++++-
 .../src/serde_with_context/serde_config.rs    |  5 +++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/crates/adapters/src/format/avro/serializer.rs b/crates/adapters/src/format/avro/serializer.rs
index 6a2601b32c6..5c09fa047be 100644
--- a/crates/adapters/src/format/avro/serializer.rs
+++ b/crates/adapters/src/format/avro/serializer.rs
@@ -19,6 +19,7 @@ use super::schema::{OptionalField, schema_unwrap_optional};
 pub fn avro_ser_config() -> SqlSerdeConfig {
     SqlSerdeConfig::default()
         .with_timestamp_format(TimestampFormat::MicrosSinceEpoch)
+        .with_timestamp_tz_format(TimestampFormat::MicrosSinceEpoch)
         .with_time_format(TimeFormat::Micros)
         .with_date_format(DateFormat::DaysSinceEpoch)
         .with_decimal_format(DecimalFormat::I128)
@@ -809,7 +810,7 @@ mod test {
         Decimal, Schema as AvroSchema, from_avro_datum, to_avro_datum, types::Value as AvroValue,
     };
     use dbsp::algebra::{F32, F64};
-    use feldera_sqllib::{Date, SqlDecimal, Timestamp};
+    use feldera_sqllib::{Date, SqlDecimal, Timestamp, TimestampTz};
     use feldera_types::{serde_with_context::SerializeWithContext, serialize_table_record};
     use num_bigint::BigInt;
     use serde::Serialize;
@@ -898,6 +899,22 @@ mod test {
       ]
     }"#;
 
+    struct TestTimestampTz {
+        ts: TimestampTz,
+    }
+
+    serialize_table_record!(TestTimestampTz[1] {
+        ts["ts"]: TimestampTz
+    });
+
+    const SCHEMA_TIMESTAMP_TZ: &str = r#"{
+      "type": "record",
+      "name": "TimestampTz",
+      "fields": [
+        { "name": "ts", "type": { "type": "long", "logicalType": "timestamp-micros" } }
+      ]
+    }"#;
+
     macro_rules! serializer_test {
         ($schema: expr_2021, $record: ident, $avro: ident) => {
             let schema = serde_json::Value::from_str($schema).unwrap();
@@ -918,6 +935,18 @@ mod test {
         };
     }
 
+    #[test]
+    fn test_avro_serializer_timestamp_tz() {
+        let micros = 1_713_597_703_000_123;
+        let record = TestTimestampTz {
+            ts: TimestampTz::from_microseconds(micros),
+        };
+        let expected =
+            AvroValue::Record(vec![("ts".to_string(), AvroValue::TimestampMicros(micros))]);
+
+        serializer_test!(SCHEMA_TIMESTAMP_TZ, record, expected);
+    }
+
     #[test]
     fn test_avro_serializer() {
         let record1_1: TestStruct1 = TestStruct1 {
diff --git a/crates/feldera-types/src/serde_with_context/serde_config.rs b/crates/feldera-types/src/serde_with_context/serde_config.rs
index a9dbe279cf5..db71c202ba8 100644
--- a/crates/feldera-types/src/serde_with_context/serde_config.rs
+++ b/crates/feldera-types/src/serde_with_context/serde_config.rs
@@ -158,6 +158,11 @@ impl SqlSerdeConfig {
         self
     }
 
+    pub fn with_timestamp_tz_format(mut self, timestamp_tz_format: TimestampFormat) -> Self {
+        self.timestamp_tz_format = timestamp_tz_format;
+        self
+    }
+
     pub fn with_decimal_format(mut self, decimal_format: DecimalFormat) -> Self {
         self.decimal_format = decimal_format;
         self