This is an automated email from the ASF dual-hosted git repository.
kriskras99 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
The following commit(s) were added to refs/heads/main by this push:
new cc37970 feat!: Rework `SpecificSingleObjectWriter` (#445)
cc37970 is described below
commit cc379701f727a00caf461558b406da5515371093
Author: Kriskras99 <[email protected]>
AuthorDate: Tue Jan 27 20:33:59 2026 +0100
feat!: Rework `SpecificSingleObjectWriter` (#445)
* feat: Share logic between `ResolvedSchema` and `ResolvedOwnedSchema`
`ResolvedOwnedSchema` now uses a self-referential struct when resolving the
schema, so it does not need to clone all named schemas contained in it.
This also fixes a bug in `reader::Block::read_writer_schema` where it
unnecessarily parses the schemata twice.
* feat!: Rework `SpecificSingleObjectWriter`
It now resolves the schema and caches it. It also no longer uses
`GenericSingleObjectWriter` for the header and does it directly. This removes
the need for a buffer.
This is a breaking change:
- `SpecificSingleObjectWriter::with_capcity` has been removed.
- `write_avro_datum_ref` now also takes a `names: &NamesRef` argument
* fix: Add back `SpecificSingleObjectWriter::with_capacity` as a deprecated
function
---
Cargo.lock | 58 ++++++-
avro/Cargo.toml | 1 +
avro/examples/specific_single_object.rs | 2 +-
.../test_interop_single_object_encoding.rs | 2 +-
avro/src/reader.rs | 17 +-
avro/src/schema/resolve.rs | 191 ++++++++-------------
avro/src/writer.rs | 143 +++++++++++----
avro/tests/serde_human_readable_false.rs | 9 +-
avro/tests/serde_human_readable_true.rs | 9 +-
9 files changed, 254 insertions(+), 178 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 7bef0b5..3c2c74e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -26,6 +26,12 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "aliasable"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd"
+
[[package]]
name = "alloca"
version = "0.4.0"
@@ -65,6 +71,7 @@ dependencies = [
"md-5",
"miniz_oxide 0.9.0",
"num-bigint",
+ "ouroboros",
"paste",
"pretty_assertions",
"quad-rand",
@@ -594,6 +601,12 @@ version = "0.16.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+[[package]]
+name = "heck"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
+
[[package]]
name = "heck"
version = "0.5.0"
@@ -816,6 +829,30 @@ version = "11.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
+[[package]]
+name = "ouroboros"
+version = "0.18.5"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59"
+dependencies = [
+ "aliasable",
+ "ouroboros_macro",
+ "static_assertions",
+]
+
+[[package]]
+name = "ouroboros_macro"
+version = "0.18.5"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0"
+dependencies = [
+ "heck 0.4.1",
+ "proc-macro2",
+ "proc-macro2-diagnostics",
+ "quote",
+ "syn",
+]
+
[[package]]
name = "page_size"
version = "0.6.0"
@@ -888,6 +925,19 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "proc-macro2-diagnostics"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+ "version_check",
+ "yansi",
+]
+
[[package]]
name = "proptest"
version = "1.9.0"
@@ -1156,6 +1206,12 @@ version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "1b6b67fb9a61334225b5b790716f609cd58395f895b3fe8b328786812a40bc3b"
+[[package]]
+name = "static_assertions"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
+
[[package]]
name = "strsim"
version = "0.11.1"
@@ -1174,7 +1230,7 @@ version = "0.27.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7"
dependencies = [
- "heck",
+ "heck 0.5.0",
"proc-macro2",
"quote",
"syn",
diff --git a/avro/Cargo.toml b/avro/Cargo.toml
index 49c3c88..08fcb6a 100644
--- a/avro/Cargo.toml
+++ b/avro/Cargo.toml
@@ -62,6 +62,7 @@ digest = { default-features = false, version = "0.10.7",
features = ["core-api"]
miniz_oxide = { default-features = false, version = "0.9.0", features =
["with-alloc"] }
log = { workspace = true }
num-bigint = { default-features = false, version = "0.4.6", features = ["std",
"serde"] }
+ouroboros = { default-features = false, version = "0.18.5", features = ["std"]
}
regex-lite = { default-features = false, version = "0.1.8", features = ["std",
"string"] }
serde = { workspace = true }
serde_bytes = { workspace = true }
diff --git a/avro/examples/specific_single_object.rs
b/avro/examples/specific_single_object.rs
index fa756cf..347721a 100644
--- a/avro/examples/specific_single_object.rs
+++ b/avro/examples/specific_single_object.rs
@@ -32,7 +32,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
b: "foo".to_string(),
};
- let mut writer = SpecificSingleObjectWriter::<Test>::with_capacity(1024)?;
+ let writer = SpecificSingleObjectWriter::<Test>::new()?;
let reader = SpecificSingleObjectReader::<Test>::new()?;
for test in repeat_n(test, 2) {
diff --git a/avro/examples/test_interop_single_object_encoding.rs
b/avro/examples/test_interop_single_object_encoding.rs
index e65b35e..e16c221 100644
--- a/avro/examples/test_interop_single_object_encoding.rs
+++ b/avro/examples/test_interop_single_object_encoding.rs
@@ -65,7 +65,7 @@ fn main() -> Result<(), Box<dyn Error>> {
fn test_write(expected: &[u8]) {
let mut encoded: Vec<u8> = Vec::new();
-
apache_avro::SpecificSingleObjectWriter::<InteropMessage>::with_capacity(1024)
+ apache_avro::SpecificSingleObjectWriter::<InteropMessage>::new()
.expect("Resolving failed")
.write_value(InteropMessage, &mut encoded)
.expect("Encoding failed");
diff --git a/avro/src/reader.rs b/avro/src/reader.rs
index 967edb5..fed35fe 100644
--- a/avro/src/reader.rs
+++ b/avro/src/reader.rs
@@ -226,21 +226,20 @@ impl<'r, R: Read> Block<'r, R> {
})
.ok_or(Details::GetAvroSchemaFromMap)?;
if !self.schemata.is_empty() {
- let rs = ResolvedSchema::try_from(self.schemata.clone())?;
- let names: Names = rs
- .get_names()
- .iter()
- .map(|(name, schema)| (name.clone(), (*schema).clone()))
- .collect();
- self.writer_schema = Schema::parse_with_names(&json, names)?;
+ let mut names = HashMap::new();
resolve_names_with_schemata(
self.schemata.iter().copied(),
- &mut self.names_refs,
+ &mut names,
&None,
+ &HashMap::new(),
)?;
+ self.names_refs = names.into_iter().map(|(n, s)| (n,
s.clone())).collect();
+ self.writer_schema = Schema::parse_with_names(&json,
self.names_refs.clone())?;
} else {
self.writer_schema = Schema::parse(&json)?;
- resolve_names(&self.writer_schema, &mut self.names_refs, &None)?;
+ let mut names = HashMap::new();
+ resolve_names(&self.writer_schema, &mut names, &None,
&HashMap::new())?;
+ self.names_refs = names.into_iter().map(|(n, s)| (n,
s.clone())).collect();
}
Ok(())
}
diff --git a/avro/src/schema/resolve.rs b/avro/src/schema/resolve.rs
index e82b87d..61b2e98 100644
--- a/avro/src/schema/resolve.rs
+++ b/avro/src/schema/resolve.rs
@@ -17,11 +17,10 @@
use crate::error::Details;
use crate::schema::{
- DecimalSchema, EnumSchema, FixedSchema, InnerDecimalSchema, Names,
NamesRef, Namespace,
- RecordSchema, UnionSchema, UuidSchema,
+ DecimalSchema, EnumSchema, FixedSchema, InnerDecimalSchema, NamesRef,
Namespace, RecordSchema,
+ UnionSchema, UuidSchema,
};
use crate::{AvroResult, Error, Schema};
-use std::borrow::Borrow;
use std::collections::HashMap;
#[derive(Debug)]
@@ -31,8 +30,8 @@ pub struct ResolvedSchema<'s> {
}
impl<'s> ResolvedSchema<'s> {
- pub fn get_schemata(&self) -> Vec<&'s Schema> {
- self.schemata.clone()
+ pub fn get_schemata(&self) -> &[&'s Schema] {
+ &self.schemata
}
pub fn get_names(&self) -> &NamesRef<'s> {
@@ -52,12 +51,7 @@ impl<'s> ResolvedSchema<'s> {
/// These schemas will be resolved in order, so references to schemas
later in the
/// list is not supported.
pub fn new_with_schemata(schemata: Vec<&'s Schema>) -> AvroResult<Self> {
- let mut rs = ResolvedSchema {
- names_ref: HashMap::new(),
- schemata,
- };
- rs.resolve(rs.get_schemata(), &None, None)?;
- Ok(rs)
+ Self::new_with_known_schemata(schemata, &None, &HashMap::new())
}
/// Creates `ResolvedSchema` with some already known schemas.
@@ -68,83 +62,17 @@ impl<'s> ResolvedSchema<'s> {
enclosing_namespace: &Namespace,
known_schemata: &'n NamesRef<'n>,
) -> AvroResult<Self> {
- let names = HashMap::new();
- let mut rs = ResolvedSchema {
+ let mut names = HashMap::new();
+ resolve_names_with_schemata(
+ schemata_to_resolve.iter().copied(),
+ &mut names,
+ enclosing_namespace,
+ known_schemata,
+ )?;
+ Ok(ResolvedSchema {
names_ref: names,
schemata: schemata_to_resolve,
- };
- rs.resolve(rs.get_schemata(), enclosing_namespace,
Some(known_schemata))?;
- Ok(rs)
- }
-
- fn resolve<'n>(
- &mut self,
- schemata: Vec<&'s Schema>,
- enclosing_namespace: &Namespace,
- known_schemata: Option<&'n NamesRef<'n>>,
- ) -> AvroResult<()> {
- for schema in schemata {
- match schema {
- Schema::Array(schema) => {
- self.resolve(vec![&schema.items], enclosing_namespace,
known_schemata)?
- }
- Schema::Map(schema) => {
- self.resolve(vec![&schema.types], enclosing_namespace,
known_schemata)?
- }
- Schema::Union(UnionSchema { schemas, .. }) => {
- for schema in schemas {
- self.resolve(vec![schema], enclosing_namespace,
known_schemata)?
- }
- }
- Schema::Enum(EnumSchema { name, .. })
- | Schema::Fixed(FixedSchema { name, .. })
- | Schema::Uuid(UuidSchema::Fixed(FixedSchema { name, .. }))
- | Schema::Decimal(DecimalSchema {
- inner: InnerDecimalSchema::Fixed(FixedSchema { name, .. }),
- ..
- })
- | Schema::Duration(FixedSchema { name, .. }) => {
- let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- if self
- .names_ref
- .insert(fully_qualified_name.clone(), schema)
- .is_some()
- {
- return
Err(Details::AmbiguousSchemaDefinition(fully_qualified_name).into());
- }
- }
- Schema::Record(RecordSchema { name, fields, .. }) => {
- let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- if self
- .names_ref
- .insert(fully_qualified_name.clone(), schema)
- .is_some()
- {
- return
Err(Details::AmbiguousSchemaDefinition(fully_qualified_name).into());
- } else {
- let record_namespace = fully_qualified_name.namespace;
- for field in fields {
- self.resolve(vec![&field.schema],
&record_namespace, known_schemata)?
- }
- }
- }
- Schema::Ref { name } => {
- let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- // first search for reference in current schemata, then
look into external references.
- if !self.names_ref.contains_key(&fully_qualified_name) {
- let is_resolved_with_known_schemas = known_schemata
- .as_ref()
- .map(|names|
names.contains_key(&fully_qualified_name))
- .unwrap_or(false);
- if !is_resolved_with_known_schemas {
- return
Err(Details::SchemaResolutionError(fully_qualified_name).into());
- }
- }
- }
- _ => (),
- }
- }
- Ok(())
+ })
}
}
@@ -164,26 +92,43 @@ impl<'s> TryFrom<Vec<&'s Schema>> for ResolvedSchema<'s> {
}
}
-pub struct ResolvedOwnedSchema {
- names: Names,
+/// Implementation detail of [`ResolvedOwnedSchema`]
+///
+/// This struct is self-referencing. The references in `names` point to
`root_schema`.
+/// This allows resolving an owned schema without having to clone all the
named schemas.
+#[ouroboros::self_referencing]
+struct InnerResolvedOwnedSchema {
root_schema: Schema,
+ #[borrows(root_schema)]
+ #[covariant]
+ names: NamesRef<'this>,
+}
+
+/// A variant of [`ResolvedSchema`] that owns the schema
+pub struct ResolvedOwnedSchema {
+ inner: InnerResolvedOwnedSchema,
}
impl ResolvedOwnedSchema {
pub fn new(root_schema: Schema) -> AvroResult<Self> {
- let mut rs = ResolvedOwnedSchema {
- names: HashMap::new(),
- root_schema,
- };
- resolve_names(&rs.root_schema, &mut rs.names, &None)?;
- Ok(rs)
+ Ok(Self {
+ inner: InnerResolvedOwnedSchemaTryBuilder {
+ root_schema,
+ names_builder: |schema: &Schema| {
+ let mut names = HashMap::new();
+ resolve_names(schema, &mut names, &None, &HashMap::new())?;
+ Ok::<_, Error>(names)
+ },
+ }
+ .try_build()?,
+ })
}
pub fn get_root_schema(&self) -> &Schema {
- &self.root_schema
+ self.inner.borrow_root_schema()
}
- pub fn get_names(&self) -> &Names {
- &self.names
+ pub fn get_names(&self) -> &NamesRef<'_> {
+ self.inner.borrow_names()
}
}
@@ -195,17 +140,25 @@ impl TryFrom<Schema> for ResolvedOwnedSchema {
}
}
-pub fn resolve_names(
- schema: &Schema,
- names: &mut Names,
+/// Resolve all references in the schema, saving any named type found in
`names`
+///
+/// `known_schemata` will be used to resolve references but they won't be
added to `names`.
+pub fn resolve_names<'s, 'n>(
+ schema: &'s Schema,
+ names: &mut NamesRef<'s>,
enclosing_namespace: &Namespace,
+ known_schemata: &NamesRef<'n>,
) -> AvroResult<()> {
match schema {
- Schema::Array(schema) => resolve_names(&schema.items, names,
enclosing_namespace),
- Schema::Map(schema) => resolve_names(&schema.types, names,
enclosing_namespace),
+ Schema::Array(schema) => {
+ resolve_names(&schema.items, names, enclosing_namespace,
known_schemata)
+ }
+ Schema::Map(schema) => {
+ resolve_names(&schema.types, names, enclosing_namespace,
known_schemata)
+ }
Schema::Union(UnionSchema { schemas, .. }) => {
for schema in schemas {
- resolve_names(schema, names, enclosing_namespace)?
+ resolve_names(schema, names, enclosing_namespace,
known_schemata)?
}
Ok(())
}
@@ -218,10 +171,7 @@ pub fn resolve_names(
})
| Schema::Duration(FixedSchema { name, .. }) => {
let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- if names
- .insert(fully_qualified_name.clone(), schema.clone())
- .is_some()
- {
+ if names.insert(fully_qualified_name.clone(), schema).is_some() {
Err(Details::AmbiguousSchemaDefinition(fully_qualified_name).into())
} else {
Ok(())
@@ -229,37 +179,38 @@ pub fn resolve_names(
}
Schema::Record(RecordSchema { name, fields, .. }) => {
let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- if names
- .insert(fully_qualified_name.clone(), schema.clone())
- .is_some()
- {
+ if names.insert(fully_qualified_name.clone(), schema).is_some() {
Err(Details::AmbiguousSchemaDefinition(fully_qualified_name).into())
} else {
let record_namespace = fully_qualified_name.namespace;
for field in fields {
- resolve_names(&field.schema, names, &record_namespace)?
+ resolve_names(&field.schema, names, &record_namespace,
known_schemata)?
}
Ok(())
}
}
Schema::Ref { name } => {
let fully_qualified_name =
name.fully_qualified_name(enclosing_namespace);
- names
- .get(&fully_qualified_name)
- .map(|_| ())
- .ok_or_else(||
Details::SchemaResolutionError(fully_qualified_name).into())
+ if names.contains_key(&fully_qualified_name)
+ || known_schemata.contains_key(&fully_qualified_name)
+ {
+ Ok(())
+ } else {
+
Err(Details::SchemaResolutionError(fully_qualified_name).into())
+ }
}
_ => Ok(()),
}
}
-pub fn resolve_names_with_schemata(
- schemata: impl IntoIterator<Item = impl Borrow<Schema>>,
- names: &mut Names,
+pub fn resolve_names_with_schemata<'s, 'n>(
+ schemata: impl IntoIterator<Item = &'s Schema>,
+ names: &mut NamesRef<'s>,
enclosing_namespace: &Namespace,
+ known_schemata: &NamesRef<'n>,
) -> AvroResult<()> {
for schema in schemata {
- resolve_names(schema.borrow(), names, enclosing_namespace)?;
+ resolve_names(schema, names, enclosing_namespace, known_schemata)?;
}
Ok(())
}
diff --git a/avro/src/writer.rs b/avro/src/writer.rs
index 72f87d0..1dd3fbf 100644
--- a/avro/src/writer.rs
+++ b/avro/src/writer.rs
@@ -21,7 +21,7 @@ use crate::{
encode::{encode, encode_internal, encode_to_vec},
error::Details,
headers::{HeaderBuilder, RabinFingerprintHeader},
- schema::{Name, ResolvedOwnedSchema, ResolvedSchema, Schema},
+ schema::{NamesRef, ResolvedOwnedSchema, ResolvedSchema, Schema},
serde::{AvroSchema, ser_schema::SchemaAwareWriteSerializer},
types::Value,
};
@@ -594,8 +594,8 @@ pub struct SpecificSingleObjectWriter<T>
where
T: AvroSchema,
{
- inner: GenericSingleObjectWriter,
- schema: Schema,
+ resolved: ResolvedOwnedSchema,
+ header: Vec<u8>,
_model: PhantomData<T>,
}
@@ -603,25 +603,52 @@ impl<T> SpecificSingleObjectWriter<T>
where
T: AvroSchema,
{
- pub fn with_capacity(buffer_cap: usize) ->
AvroResult<SpecificSingleObjectWriter<T>> {
+ pub fn new() -> AvroResult<Self> {
let schema = T::get_schema();
- Ok(SpecificSingleObjectWriter {
- inner: GenericSingleObjectWriter::new_with_capacity(&schema,
buffer_cap)?,
- schema,
+ let header =
RabinFingerprintHeader::from_schema(&schema).build_header();
+ let resolved = ResolvedOwnedSchema::new(schema)?;
+ // We don't use Self::new_with_header_builder as that would mean
calling T::get_schema() twice
+ Ok(Self {
+ resolved,
+ header,
+ _model: PhantomData,
+ })
+ }
+
+ pub fn new_with_header_builder(header_builder: impl HeaderBuilder) ->
AvroResult<Self> {
+ let header = header_builder.build_header();
+ let resolved = ResolvedOwnedSchema::new(T::get_schema())?;
+ Ok(Self {
+ resolved,
+ header,
_model: PhantomData,
})
}
+
+ /// Deprecated. Use [`SpecificSingleObjectWriter::new`] instead.
+ #[deprecated(since = "0.22.0", note = "Use new() instead")]
+ pub fn with_capacity(_buffer_cap: usize) -> AvroResult<Self> {
+ Self::new()
+ }
}
impl<T> SpecificSingleObjectWriter<T>
where
T: AvroSchema + Into<Value>,
{
- /// Write the `Into<Value>` to the provided Write object. Returns a result
with the number
- /// of bytes written including the header
- pub fn write_value<W: Write>(&mut self, data: T, writer: &mut W) ->
AvroResult<usize> {
- let v: Value = data.into();
- self.inner.write_value_ref(&v, writer)
+ /// Write the value to the writer
+ ///
+ /// Returns the number of bytes written.
+ ///
+ /// Each call writes a complete single-object encoded message (header +
data),
+ /// making each message independently decodable.
+ pub fn write_value<W: Write>(&self, data: T, writer: &mut W) ->
AvroResult<usize> {
+ writer
+ .write_all(&self.header)
+ .map_err(Details::WriteBytes)?;
+ let value: Value = data.into();
+ let bytes = write_value_ref_owned_resolved(&self.resolved, &value,
writer)?;
+ Ok(bytes + self.header.len())
}
}
@@ -629,32 +656,34 @@ impl<T> SpecificSingleObjectWriter<T>
where
T: AvroSchema + Serialize,
{
- /// Write the referenced `Serialize` object to the provided `Write` object.
+ /// Write the object to the writer.
///
/// Returns the number of bytes written.
///
/// Each call writes a complete single-object encoded message (header +
data),
/// making each message independently decodable.
- pub fn write_ref<W: Write>(&mut self, data: &T, writer: &mut W) ->
AvroResult<usize> {
- // Always write the header for each message (single object encoding
requires
- // each message to be independently decodable)
+ pub fn write_ref<W: Write>(&self, data: &T, writer: &mut W) ->
AvroResult<usize> {
writer
- .write_all(self.inner.buffer.as_slice())
+ .write_all(&self.header)
.map_err(Details::WriteBytes)?;
- let bytes_written =
- self.inner.buffer.len() + write_avro_datum_ref(&self.schema, data,
writer)?;
+ let bytes = write_avro_datum_ref(
+ self.resolved.get_root_schema(),
+ self.resolved.get_names(),
+ data,
+ writer,
+ )?;
- Ok(bytes_written)
+ Ok(bytes + self.header.len())
}
- /// Write the Serialize object to the provided Write object.
+ /// Write the object to the writer.
///
/// Returns the number of bytes written.
///
/// Each call writes a complete single-object encoded message (header +
data),
/// making each message independently decodable.
- pub fn write<W: Write>(&mut self, data: T, writer: &mut W) ->
AvroResult<usize> {
+ pub fn write<W: Write>(&self, data: T, writer: &mut W) ->
AvroResult<usize> {
self.write_ref(&data, writer)
}
}
@@ -682,11 +711,11 @@ fn write_value_ref_resolved(
}
}
-fn write_value_ref_owned_resolved(
+fn write_value_ref_owned_resolved<W: Write>(
resolved_schema: &ResolvedOwnedSchema,
value: &Value,
- buffer: &mut Vec<u8>,
-) -> AvroResult<()> {
+ writer: &mut W,
+) -> AvroResult<usize> {
let root_schema = resolved_schema.get_root_schema();
if let Some(reason) = value.validate_internal(
root_schema,
@@ -705,9 +734,8 @@ fn write_value_ref_owned_resolved(
root_schema,
resolved_schema.get_names(),
&root_schema.namespace(),
- buffer,
- )?;
- Ok(())
+ writer,
+ )
}
/// Encode a compatible value (implementing the `ToAvro` trait) into Avro
format, also
@@ -730,13 +758,12 @@ pub fn to_avro_datum<T: Into<Value>>(schema: &Schema,
value: T) -> AvroResult<Ve
/// if you don't know what you are doing, instead.
pub fn write_avro_datum_ref<T: Serialize, W: Write>(
schema: &Schema,
+ names: &NamesRef,
data: &T,
writer: &mut W,
) -> AvroResult<usize> {
- let names: HashMap<Name, &Schema> = HashMap::new();
- let mut serializer = SchemaAwareWriteSerializer::new(writer, schema,
&names, None);
- let bytes_written = data.serialize(&mut serializer)?;
- Ok(bytes_written)
+ let mut serializer = SchemaAwareWriteSerializer::new(writer, schema,
names, None);
+ data.serialize(&mut serializer)
}
/// Encode a compatible value (implementing the `ToAvro` trait) into Avro
format, also
@@ -856,7 +883,7 @@ mod tests {
zig_i64(3, &mut expected)?;
expected.extend([b'f', b'o', b'o']);
- let bytes = write_avro_datum_ref(&schema, &data, &mut writer)?;
+ let bytes = write_avro_datum_ref(&schema, &HashMap::new(), &data, &mut
writer)?;
assert_eq!(bytes, expected.len());
assert_eq!(writer, expected);
@@ -1586,9 +1613,8 @@ mod tests {
1024,
)
.expect("Should resolve schema");
- let mut specific_writer =
-
SpecificSingleObjectWriter::<TestSingleObjectWriter>::with_capacity(1024)
- .expect("Resolved should pass");
+ let specific_writer =
SpecificSingleObjectWriter::<TestSingleObjectWriter>::new()
+ .expect("Resolved should pass");
specific_writer
.write_ref(&obj1, &mut buf1)
.expect("Serialization expected");
@@ -1734,4 +1760,49 @@ mod tests {
Ok(())
}
+
+ #[test]
+ fn avro_rs_439_specific_single_object_writer_ref() -> TestResult {
+ #[derive(Serialize)]
+ struct Recursive {
+ field: bool,
+ recurse: Option<Box<Recursive>>,
+ }
+
+ impl AvroSchema for Recursive {
+ fn get_schema() -> Schema {
+ Schema::parse_str(
+ r#"{
+ "name": "Recursive",
+ "type": "record",
+ "fields": [
+ { "name": "field", "type": "boolean" },
+ { "name": "recurse", "type": ["null", "Recursive"] }
+ ]
+ }"#,
+ )
+ .unwrap()
+ }
+ }
+
+ let mut buffer = Vec::new();
+ let writer = SpecificSingleObjectWriter::new()?;
+
+ writer.write(
+ Recursive {
+ field: true,
+ recurse: Some(Box::new(Recursive {
+ field: false,
+ recurse: None,
+ })),
+ },
+ &mut buffer,
+ )?;
+ assert_eq!(
+ buffer,
+ &[195, 1, 83, 223, 43, 26, 181, 179, 227, 224, 1, 2, 0, 0][..]
+ );
+
+ Ok(())
+ }
}
diff --git a/avro/tests/serde_human_readable_false.rs
b/avro/tests/serde_human_readable_false.rs
index 9e87f52..2f8be3f 100644
--- a/avro/tests/serde_human_readable_false.rs
+++ b/avro/tests/serde_human_readable_false.rs
@@ -56,8 +56,7 @@ fn avro_rs_53_uuid_with_fixed() -> TestResult {
// serialize the Uuid as Fixed
assert!(!apache_avro::util::set_serde_human_readable(false));
- let bytes = SpecificSingleObjectWriter::<Comment>::with_capacity(64)?
- .write_ref(&payload, &mut buffer)?;
+ let bytes =
SpecificSingleObjectWriter::<Comment>::new()?.write_ref(&payload, &mut buffer)?;
assert_eq!(bytes, 26);
Ok(())
@@ -77,7 +76,7 @@ fn avro_rs_440_uuid_string() -> TestResult {
let mut buffer = Vec::new();
assert!(!apache_avro::util::set_serde_human_readable(false));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
assert_eq!(
writer.write(uuid, &mut buffer).unwrap_err().to_string(),
"Failed to serialize value of type bytes using schema Uuid(String):
55e840e29b41d4a7164466554400. Cause: Expected a string, but got 16 bytes. Did
you mean to use `Schema::Uuid(UuidSchema::Fixed)` or
`utils::serde_set_human_readable(true)`?"
@@ -100,7 +99,7 @@ fn avro_rs_440_uuid_bytes() -> TestResult {
let mut buffer = Vec::new();
assert!(!apache_avro::util::set_serde_human_readable(false));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
writer.write(uuid, &mut buffer)?;
assert_eq!(
@@ -131,7 +130,7 @@ fn avro_rs_440_uuid_fixed() -> TestResult {
let mut buffer = Vec::new();
assert!(!apache_avro::util::set_serde_human_readable(false));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
writer.write(uuid, &mut buffer)?;
assert_eq!(
diff --git a/avro/tests/serde_human_readable_true.rs
b/avro/tests/serde_human_readable_true.rs
index 960f72f..182839e 100644
--- a/avro/tests/serde_human_readable_true.rs
+++ b/avro/tests/serde_human_readable_true.rs
@@ -55,8 +55,7 @@ fn avro_rs_53_uuid_with_string_true() -> TestResult {
// serialize the Uuid as String
assert!(apache_avro::util::set_serde_human_readable(true));
- let bytes = SpecificSingleObjectWriter::<Comment>::with_capacity(64)?
- .write_ref(&payload, &mut buffer)?;
+ let bytes =
SpecificSingleObjectWriter::<Comment>::new()?.write_ref(&payload, &mut buffer)?;
assert_eq!(bytes, 47);
Ok(())
@@ -76,7 +75,7 @@ fn avro_rs_440_uuid_string() -> TestResult {
let mut buffer = Vec::new();
assert!(apache_avro::util::set_serde_human_readable(true));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
writer.write(uuid, &mut buffer)?;
assert_eq!(
@@ -101,7 +100,7 @@ fn avro_rs_440_uuid_bytes() -> TestResult {
let mut buffer = Vec::new();
assert!(apache_avro::util::set_serde_human_readable(true));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
assert_eq!(
writer.write(uuid, &mut buffer).unwrap_err().to_string(),
"Failed to serialize value of type string using schema Uuid(Bytes):
550e8400-e29b-41d4-a716-446655440000. Cause: Expected bytes but got a string.
Did you mean to use `Schema::Uuid(UuidSchema::String)` or
`utils::serde_set_human_readable(false)`?"
@@ -127,7 +126,7 @@ fn avro_rs_440_uuid_fixed() -> TestResult {
let mut buffer = Vec::new();
assert!(apache_avro::util::set_serde_human_readable(true));
- let mut writer = SpecificSingleObjectWriter::with_capacity(64)?;
+ let writer = SpecificSingleObjectWriter::new()?;
assert_eq!(
writer.write(uuid, &mut buffer).unwrap_err().to_string(),
r#"Failed to serialize value of type string using schema
Uuid(Fixed(FixedSchema { name: Name { name: "uuid", namespace: None }, aliases:
None, doc: None, size: 16, default: None, attributes: {} })):
550e8400-e29b-41d4-a716-446655440000. Cause: Expected bytes but got a string.
Did you mean to use `Schema::Uuid(UuidSchema::String)` or
`utils::serde_set_human_readable(false)`?"#
