This is an automated email from the ASF dual-hosted git repository. mgrigorov pushed a commit to branch avro-3916-rust-timestamp-nanos in repository https://gitbox.apache.org/repos/asf/avro.git
commit b4e2cb16b81db7501a7d73d666759630f203568e Author: Martin Tzvetanov Grigorov <[email protected]> AuthorDate: Tue Oct 24 10:34:46 2023 +0300 AVRO-3916: [Rust] Add TimestampNanos types Signed-off-by: Martin Tzvetanov Grigorov <[email protected]> --- lang/rust/avro/README.md | 2 ++ lang/rust/avro/src/de.rs | 54 ++++++++++++++++++++++++++++++++++++++++---- lang/rust/avro/src/decode.rs | 2 ++ lang/rust/avro/src/encode.rs | 2 ++ lang/rust/avro/src/error.rs | 6 +++++ lang/rust/avro/src/lib.rs | 2 ++ lang/rust/avro/src/schema.rs | 18 +++++++++++++++ lang/rust/avro/src/types.rs | 54 ++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 136 insertions(+), 4 deletions(-) diff --git a/lang/rust/avro/README.md b/lang/rust/avro/README.md index a349847fa..bb3fd2b2e 100644 --- a/lang/rust/avro/README.md +++ b/lang/rust/avro/README.md @@ -538,8 +538,10 @@ fn main() -> Result<(), Error> { record.put("time_micros", Value::TimeMicros(3)); record.put("timestamp_millis", Value::TimestampMillis(4)); record.put("timestamp_micros", Value::TimestampMicros(5)); + record.put("timestamp_nanos", Value::TimestampNanos(6)); record.put("local_timestamp_millis", Value::LocalTimestampMillis(4)); record.put("local_timestamp_micros", Value::LocalTimestampMicros(5)); + record.put("local_timestamp_nanos", Value::LocalTimestampMicros(6)); record.put("duration", Duration::new(Months::new(6), Days::new(7), Millis::new(8))); writer.append(record)?; diff --git a/lang/rust/avro/src/de.rs b/lang/rust/avro/src/de.rs index aba2b541f..ceffadbd3 100644 --- a/lang/rust/avro/src/de.rs +++ b/lang/rust/avro/src/de.rs @@ -245,8 +245,10 @@ impl<'a, 'de> de::Deserializer<'de> for &'a Deserializer<'de> { | Value::TimeMicros(i) | Value::TimestampMillis(i) | Value::TimestampMicros(i) + | Value::TimestampNanos(i) | Value::LocalTimestampMillis(i) - | Value::LocalTimestampMicros(i) => visitor.visit_i64(*i), + | Value::LocalTimestampMicros(i) + | Value::LocalTimestampNanos(i) => visitor.visit_i64(*i), &Value::Float(f) => visitor.visit_f32(f), &Value::Double(d) => visitor.visit_f64(d), Value::Union(_i, u) => match **u { @@ -257,8 +259,10 @@ impl<'a, 'de> de::Deserializer<'de> for &'a Deserializer<'de> { | Value::TimeMicros(i) | Value::TimestampMillis(i) | Value::TimestampMicros(i) + | Value::TimestampNanos(i) | Value::LocalTimestampMillis(i) - | Value::LocalTimestampMicros(i) => visitor.visit_i64(i), + | Value::LocalTimestampMicros(i) + | Value::LocalTimestampNanos(i) => visitor.visit_i64(i), Value::Float(f) => visitor.visit_f32(f), Value::Double(d) => visitor.visit_f64(d), Value::Record(ref fields) => visitor.visit_map(RecordDeserializer::new(fields)), @@ -1080,7 +1084,16 @@ mod tests { fn test_timestamp_micros() -> TestResult { let raw_value = 1; let value = Value::TimestampMicros(raw_value); - let result = crate::from_value::<i64>(&value)?; + let result = from_value::<i64>(&value)?; + assert_eq!(result, raw_value); + Ok(()) + } + + #[test] + fn test_avro_3916_timestamp_nanos() -> TestResult { + let raw_value = 1; + let value = Value::TimestampNanos(raw_value); + let result = from_value::<i64>(&value)?; assert_eq!(result, raw_value); Ok(()) } @@ -1089,7 +1102,7 @@ mod tests { fn test_avro_3853_local_timestamp_millis() -> TestResult { let raw_value = 1; let value = Value::LocalTimestampMillis(raw_value); - let result = crate::from_value::<i64>(&value)?; + let result = from_value::<i64>(&value)?; assert_eq!(result, raw_value); Ok(()) } @@ -1103,6 +1116,15 @@ mod tests { Ok(()) } + #[test] + fn test_avro_3916_local_timestamp_nanos() -> TestResult { + let raw_value = 1; + let value = Value::LocalTimestampNanos(raw_value); + let result = crate::from_value::<i64>(&value)?; + assert_eq!(result, raw_value); + Ok(()) + } + #[test] fn test_from_value_uuid_str() -> TestResult { let raw_value = "9ec535ff-3e2a-45bd-91d3-0a01321b5a49"; @@ -1146,8 +1168,10 @@ mod tests { ("time_micros_a".to_string(), 123), ("timestamp_millis_b".to_string(), 234), ("timestamp_micros_c".to_string(), 345), + ("timestamp_nanos_d".to_string(), 345_001), ("local_timestamp_millis_d".to_string(), 678), ("local_timestamp_micros_e".to_string(), 789), + ("local_timestamp_nanos_f".to_string(), 345_002), ] .iter() .cloned() @@ -1164,12 +1188,18 @@ mod tests { key if key.starts_with("timestamp_micros_") => { (k.clone(), Value::TimestampMicros(*v)) } + key if key.starts_with("timestamp_nanos_") => { + (k.clone(), Value::TimestampNanos(*v)) + } key if key.starts_with("local_timestamp_millis_") => { (k.clone(), Value::LocalTimestampMillis(*v)) } key if key.starts_with("local_timestamp_micros_") => { (k.clone(), Value::LocalTimestampMicros(*v)) } + key if key.starts_with("local_timestamp_nanos_") => { + (k.clone(), Value::LocalTimestampNanos(*v)) + } _ => unreachable!("unexpected key: {:?}", k), }) .collect(); @@ -1219,6 +1249,14 @@ mod tests { "a_non_existing_timestamp_micros".to_string(), Value::Union(0, Box::new(Value::TimestampMicros(-345))), ), + ( + "a_timestamp_nanos".to_string(), + Value::Union(0, Box::new(Value::TimestampNanos(345))), + ), + ( + "a_non_existing_timestamp_nanos".to_string(), + Value::Union(0, Box::new(Value::TimestampNanos(-345))), + ), ( "a_local_timestamp_millis".to_string(), Value::Union(0, Box::new(Value::LocalTimestampMillis(678))), @@ -1235,6 +1273,14 @@ mod tests { "a_non_existing_local_timestamp_micros".to_string(), Value::Union(0, Box::new(Value::LocalTimestampMicros(-789))), ), + ( + "a_local_timestamp_nanos".to_string(), + Value::Union(0, Box::new(Value::LocalTimestampNanos(789))), + ), + ( + "a_non_existing_local_timestamp_nanos".to_string(), + Value::Union(0, Box::new(Value::LocalTimestampNanos(-789))), + ), ( "a_record".to_string(), Value::Union( diff --git a/lang/rust/avro/src/decode.rs b/lang/rust/avro/src/decode.rs index 8c50e7702..bf8477fb7 100644 --- a/lang/rust/avro/src/decode.rs +++ b/lang/rust/avro/src/decode.rs @@ -137,8 +137,10 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>( Schema::TimeMicros => zag_i64(reader).map(Value::TimeMicros), Schema::TimestampMillis => zag_i64(reader).map(Value::TimestampMillis), Schema::TimestampMicros => zag_i64(reader).map(Value::TimestampMicros), + Schema::TimestampNanos => zag_i64(reader).map(Value::TimestampNanos), Schema::LocalTimestampMillis => zag_i64(reader).map(Value::LocalTimestampMillis), Schema::LocalTimestampMicros => zag_i64(reader).map(Value::LocalTimestampMicros), + Schema::LocalTimestampNanos => zag_i64(reader).map(Value::LocalTimestampNanos), Schema::Duration => { let mut buf = [0u8; 12]; reader.read_exact(&mut buf).map_err(Error::ReadDuration)?; diff --git a/lang/rust/avro/src/encode.rs b/lang/rust/avro/src/encode.rs index 5dd087771..23f94664c 100644 --- a/lang/rust/avro/src/encode.rs +++ b/lang/rust/avro/src/encode.rs @@ -90,8 +90,10 @@ pub(crate) fn encode_internal<S: Borrow<Schema>>( Value::Long(i) | Value::TimestampMillis(i) | Value::TimestampMicros(i) + | Value::TimestampNanos(i) | Value::LocalTimestampMillis(i) | Value::LocalTimestampMicros(i) + | Value::LocalTimestampNanos(i) | Value::TimeMicros(i) => encode_long(*i, buffer), Value::Float(x) => buffer.extend_from_slice(&x.to_le_bytes()), Value::Double(x) => buffer.extend_from_slice(&x.to_le_bytes()), diff --git a/lang/rust/avro/src/error.rs b/lang/rust/avro/src/error.rs index 13cec65c7..36c8d94a9 100644 --- a/lang/rust/avro/src/error.rs +++ b/lang/rust/avro/src/error.rs @@ -154,12 +154,18 @@ pub enum Error { #[error("TimestampMicros expected, got {0:?}")] GetTimestampMicros(ValueKind), + #[error("TimestampNanos expected, got {0:?}")] + GetTimestampNanos(ValueKind), + #[error("LocalTimestampMillis expected, got {0:?}")] GetLocalTimestampMillis(ValueKind), #[error("LocalTimestampMicros expected, got {0:?}")] GetLocalTimestampMicros(ValueKind), + #[error("LocalTimestampNanos expected, got {0:?}")] + GetLocalTimestampNanos(ValueKind), + #[error("Null expected, got {0:?}")] GetNull(ValueKind), diff --git a/lang/rust/avro/src/lib.rs b/lang/rust/avro/src/lib.rs index a613d93c0..2d9d79a76 100644 --- a/lang/rust/avro/src/lib.rs +++ b/lang/rust/avro/src/lib.rs @@ -651,8 +651,10 @@ //! record.put("time_micros", Value::TimeMicros(3)); //! record.put("timestamp_millis", Value::TimestampMillis(4)); //! record.put("timestamp_micros", Value::TimestampMicros(5)); +//! record.put("timestamp_nanos", Value::TimestampNanos(6)); //! record.put("local_timestamp_millis", Value::LocalTimestampMillis(4)); //! record.put("local_timestamp_micros", Value::LocalTimestampMicros(5)); +//! record.put("local_timestamp_nanos", Value::LocalTimestampMicros(6)); //! record.put("duration", Duration::new(Months::new(6), Days::new(7), Millis::new(8))); //! //! writer.append(record)?; diff --git a/lang/rust/avro/src/schema.rs b/lang/rust/avro/src/schema.rs index f2487e316..90cdeb1b2 100644 --- a/lang/rust/avro/src/schema.rs +++ b/lang/rust/avro/src/schema.rs @@ -130,10 +130,14 @@ pub enum Schema { TimestampMillis, /// An instant in time represented as the number of microseconds after the UNIX epoch. TimestampMicros, + /// An instant in time represented as the number of nanoseconds after the UNIX epoch. + TimestampNanos, /// An instant in localtime represented as the number of milliseconds after the UNIX epoch. LocalTimestampMillis, /// An instant in local time represented as the number of microseconds after the UNIX epoch. LocalTimestampMicros, + /// An instant in local time represented as the number of nanoseconds after the UNIX epoch. + LocalTimestampNanos, /// An amount of time defined by a number of months, days and milliseconds. Duration, /// A reference to another schema. @@ -199,8 +203,10 @@ impl From<&types::Value> for SchemaKind { Value::TimeMicros(_) => Self::TimeMicros, Value::TimestampMillis(_) => Self::TimestampMillis, Value::TimestampMicros(_) => Self::TimestampMicros, + Value::TimestampNanos(_) => Self::TimestampNanos, Value::LocalTimestampMillis(_) => Self::LocalTimestampMillis, Value::LocalTimestampMicros(_) => Self::LocalTimestampMicros, + Value::LocalTimestampNanos(_) => Self::LocalTimestampNanos, Value::Duration { .. } => Self::Duration, } } @@ -1942,6 +1948,12 @@ impl Serialize for Schema { map.serialize_entry("logicalType", "timestamp-micros")?; map.end() } + Schema::TimestampNanos => { + let mut map = serializer.serialize_map(None)?; + map.serialize_entry("type", "long")?; + map.serialize_entry("logicalType", "timestamp-nanos")?; + map.end() + } Schema::LocalTimestampMillis => { let mut map = serializer.serialize_map(None)?; map.serialize_entry("type", "long")?; @@ -1954,6 +1966,12 @@ impl Serialize for Schema { map.serialize_entry("logicalType", "local-timestamp-micros")?; map.end() } + Schema::LocalTimestampNanos => { + let mut map = serializer.serialize_map(None)?; + map.serialize_entry("type", "long")?; + map.serialize_entry("logicalType", "local-timestamp-nanos")?; + map.end() + } Schema::Duration => { let mut map = serializer.serialize_map(None)?; diff --git a/lang/rust/avro/src/types.rs b/lang/rust/avro/src/types.rs index 49fb9e6a3..c3fde21e4 100644 --- a/lang/rust/avro/src/types.rs +++ b/lang/rust/avro/src/types.rs @@ -112,10 +112,14 @@ pub enum Value { TimestampMillis(i64), /// Timestamp in microseconds. TimestampMicros(i64), + /// Timestamp in nanoseconds. + TimestampNanos(i64), /// Local timestamp in milliseconds. LocalTimestampMillis(i64), /// Local timestamp in microseconds. LocalTimestampMicros(i64), + /// Local timestamp in nanoseconds. + LocalTimestampNanos(i64), /// Avro Duration. An amount of time defined by months, days and milliseconds. Duration(Duration), /// Universally unique identifier. @@ -340,8 +344,10 @@ impl TryFrom<Value> for JsonValue { Value::TimeMicros(t) => Ok(Self::Number(t.into())), Value::TimestampMillis(t) => Ok(Self::Number(t.into())), Value::TimestampMicros(t) => Ok(Self::Number(t.into())), + Value::TimestampNanos(t) => Ok(Self::Number(t.into())), Value::LocalTimestampMillis(t) => Ok(Self::Number(t.into())), Value::LocalTimestampMicros(t) => Ok(Self::Number(t.into())), + Value::LocalTimestampNanos(t) => Ok(Self::Number(t.into())), Value::Duration(d) => Ok(Self::Array( <[u8; 12]>::from(d).iter().map(|&v| v.into()).collect(), )), @@ -428,8 +434,10 @@ impl Value { (&Value::Long(_), &Schema::LocalTimestampMicros) => None, (&Value::TimestampMicros(_), &Schema::TimestampMicros) => None, (&Value::TimestampMillis(_), &Schema::TimestampMillis) => None, + (&Value::TimestampNanos(_), &Schema::TimestampNanos) => None, (&Value::LocalTimestampMicros(_), &Schema::LocalTimestampMicros) => None, (&Value::LocalTimestampMillis(_), &Schema::LocalTimestampMillis) => None, + (&Value::LocalTimestampNanos(_), &Schema::LocalTimestampNanos) => None, (&Value::TimeMicros(_), &Schema::TimeMicros) => None, (&Value::TimeMillis(_), &Schema::TimeMillis) => None, (&Value::Date(_), &Schema::Date) => None, @@ -689,8 +697,10 @@ impl Value { Schema::TimeMicros => self.resolve_time_micros(), Schema::TimestampMillis => self.resolve_timestamp_millis(), Schema::TimestampMicros => self.resolve_timestamp_micros(), + Schema::TimestampNanos => self.resolve_timestamp_nanos(), Schema::LocalTimestampMillis => self.resolve_local_timestamp_millis(), Schema::LocalTimestampMicros => self.resolve_local_timestamp_micros(), + Schema::LocalTimestampNanos => self.resolve_local_timestamp_nanos(), Schema::Duration => self.resolve_duration(), Schema::Uuid => self.resolve_uuid(), } @@ -814,6 +824,14 @@ impl Value { } } + fn resolve_timestamp_nanos(self) -> Result<Self, Error> { + match self { + Value::TimestampNanos(ts) | Value::Long(ts) => Ok(Value::TimestampNanos(ts)), + Value::Int(ts) => Ok(Value::TimestampNanos(i64::from(ts))), + other => Err(Error::GetTimestampNanos(other.into())), + } + } + fn resolve_local_timestamp_millis(self) -> Result<Self, Error> { match self { Value::LocalTimestampMillis(ts) | Value::Long(ts) => { @@ -834,6 +852,14 @@ impl Value { } } + fn resolve_local_timestamp_nanos(self) -> Result<Self, Error> { + match self { + Value::LocalTimestampNanos(ts) | Value::Long(ts) => Ok(Value::LocalTimestampNanos(ts)), + Value::Int(ts) => Ok(Value::LocalTimestampNanos(i64::from(ts))), + other => Err(Error::GetLocalTimestampNanos(other.into())), + } + } + fn resolve_null(self) -> Result<Self, Error> { match self { Value::Null => Ok(Value::Null), @@ -1738,6 +1764,16 @@ Field with name '"b"' is not a member of the map items"#, assert!(value.resolve(&Schema::TimestampMicros).is_err()); } + #[test] + fn test_avro_3914_resolve_timestamp_nanos() { + let value = Value::TimestampNanos(10); + assert!(value.clone().resolve(&Schema::TimestampNanos).is_ok()); + assert!(value.resolve(&Schema::Int).is_err()); + + let value = Value::Double(10.0); + assert!(value.resolve(&Schema::TimestampNanos).is_err()); + } + #[test] fn test_avro_3853_resolve_timestamp_millis() { let value = Value::LocalTimestampMillis(10); @@ -1758,6 +1794,16 @@ Field with name '"b"' is not a member of the map items"#, assert!(value.resolve(&Schema::LocalTimestampMicros).is_err()); } + #[test] + fn test_avro_3916_resolve_timestamp_nanos() { + let value = Value::LocalTimestampNanos(10); + assert!(value.clone().resolve(&Schema::LocalTimestampNanos).is_ok()); + assert!(value.resolve(&Schema::Int).is_err()); + + let value = Value::Double(10.0); + assert!(value.resolve(&Schema::LocalTimestampNanos).is_err()); + } + #[test] fn resolve_duration() { let value = Value::Duration(Duration::new( @@ -1963,6 +2009,10 @@ Field with name '"b"' is not a member of the map items"#, JsonValue::try_from(Value::TimestampMicros(1))?, JsonValue::Number(1.into()) ); + assert_eq!( + JsonValue::try_from(Value::TimestampNanos(1))?, + JsonValue::Number(1.into()) + ); assert_eq!( JsonValue::try_from(Value::LocalTimestampMillis(1))?, JsonValue::Number(1.into()) @@ -1971,6 +2021,10 @@ Field with name '"b"' is not a member of the map items"#, JsonValue::try_from(Value::LocalTimestampMicros(1))?, JsonValue::Number(1.into()) ); + assert_eq!( + JsonValue::try_from(Value::LocalTimestampNanos(1))?, + JsonValue::Number(1.into()) + ); assert_eq!( JsonValue::try_from(Value::Duration( [1u8, 2u8, 3u8, 4u8, 5u8, 6u8, 7u8, 8u8, 9u8, 10u8, 11u8, 12u8].into()
