docs: avro: document workers parameter

abhizer · abhizer · commit 03b51447c33b · 2026-02-09T20:28:51.000Z
Adds documentation for the `workers` parameter on the Avro format
configuration. Default: 4.

Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/crates/adapterlib/src/catalog.rs b/crates/adapterlib/src/catalog.rs
@@ -283,7 +283,7 @@ pub trait SerBatchReader: 'static + Send + Sync {
 
     fn keys_factory(&self) -> &'static dyn Factory<DynVec<DynData>>;
 
-    fn data_factory(&self) -> &'static dyn Factory<DynData>;
+    fn key_factory(&self) -> &'static dyn Factory<DynData>;
 
     fn sample_keys(&self, sample_size: usize, sample: &mut DynVec<DynData>);
 
@@ -407,14 +407,14 @@ impl SplitCursorBuilder {
 
             // Clone the actual key the cursor landed on.
             cursor.get_key().map(|s| {
-                let mut key = batch.data_factory().default_box();
+                let mut key = batch.key_factory().default_box();
                 s.clone_to(key.as_mut());
                 key
             })
         }?;
 
         let end_key = end_bound.map(|e| {
-            let mut key = batch.data_factory().default_box();
+            let mut key = batch.key_factory().default_box();
             e.clone_to(key.as_mut());
             key
         });
diff --git a/crates/adapters/benches/avro_encoder.rs b/crates/adapters/benches/avro_encoder.rs
@@ -1,6 +1,6 @@
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use dbsp::OrdIndexedZSet;
 use dbsp::utils::Tup2;
-use dbsp::{OrdIndexedZSet, OrdZSet};
 use dbsp_adapters::format::avro::output::AvroEncoder;
 use dbsp_adapters::static_compile::seroutput::SerBatchImpl;
 use dbsp_adapters::{Encoder, OutputConsumer, SerBatch};
@@ -66,19 +66,6 @@ impl BenchTestStruct {
             properties: BTreeMap::new(),
         }
     }
-
-    fn avro_schema() -> &'static str {
-        r#"{
-            "type": "record",
-            "name": "BenchTestStruct",
-            "fields": [
-                { "name": "id", "type": "long" },
-                { "name": "b", "type": "boolean" },
-                { "name": "i", "type": ["null", "long"] },
-                { "name": "s", "type": "string" }
-            ]
-        }"#
-    }
 }
 
 #[derive(
@@ -212,15 +199,9 @@ fn build_indexed_batch(data: &[BenchTestStruct]) -> Arc<dyn SerBatch> {
     ))
 }
 
-fn build_plain_batch(data: &[BenchTestStruct]) -> Arc<dyn SerBatch> {
-    let tuples: Vec<_> = data.iter().map(|v| Tup2(v.clone(), 1i64)).collect();
-    let zset = OrdZSet::from_keys((), tuples);
-    Arc::new(<SerBatchImpl<_, BenchTestStruct, ()>>::new(zset))
-}
-
-fn create_indexed_encoder(workers: usize) -> AvroEncoder {
+fn create_indexed_encoder(threads: usize) -> AvroEncoder {
     let config = AvroEncoderConfig {
-        workers,
+        threads,
         update_format: AvroUpdateFormat::Raw,
         skip_schema_id: true,
         ..Default::default()
@@ -237,26 +218,6 @@ fn create_indexed_encoder(workers: usize) -> AvroEncoder {
     .unwrap()
 }
 
-fn create_plain_encoder(workers: usize) -> AvroEncoder {
-    let config = AvroEncoderConfig {
-        workers,
-        schema: Some(BenchTestStruct::avro_schema().to_string()),
-        update_format: AvroUpdateFormat::Raw,
-        skip_schema_id: true,
-        ..Default::default()
-    };
-    let consumer = BenchOutputConsumer::new();
-    AvroEncoder::create(
-        "bench_endpoint",
-        &None,
-        &BenchTestStruct::relation_schema(),
-        Box::new(consumer),
-        config,
-        None,
-    )
-    .unwrap()
-}
-
 // ---------------------------------------------------------------------------
 // Benchmarks
 // ---------------------------------------------------------------------------
diff --git a/crates/adapters/src/format/avro/output.rs b/crates/adapters/src/format/avro/output.rs
@@ -106,10 +106,10 @@ impl OutputFormat for AvroOutputFormat {
             _ => None,
         };
 
-        if avro_config.workers == 0 {
+        if avro_config.threads == 0 {
             return Err(ControllerError::invalid_encoder_configuration(
                 endpoint_name,
-                "the 'workers' configuration must be greater than zero",
+                "the 'threads' configuration must be greater than zero",
             ));
         }
 
@@ -394,8 +394,8 @@ pub struct AvroEncoder {
     /// The thread pool for parallel encoding.
     pool: Option<ThreadPool>,
 
-    /// The number of workers to run in parallel.
-    workers: usize,
+    /// The number of worker threads to run in parallel.
+    threads: usize,
 }
 
 /// `true` - this config will create messages with key and value components.
@@ -715,7 +715,7 @@ Consider defining an index with `CREATE INDEX` and setting `index` field in conn
             cdc_field: config.cdc_field,
             value_avro_schema_with_cdc,
             pool: None,
-            workers: config.workers,
+            threads: config.threads,
         })
     }
 
@@ -812,12 +812,12 @@ Consider defining an index with `CREATE INDEX` and setting `index` field in conn
     fn encode_indexed(&mut self, batch: Arc<dyn SerBatchReader>) -> AnyResult<()> {
         let pool = self
             .pool
-            .get_or_insert_with(|| ThreadPool::new(self.workers));
+            .get_or_insert_with(|| ThreadPool::new(self.threads));
 
-        let (tx, rx) = crossbeam::channel::bounded(self.workers);
+        let (tx, rx) = crossbeam::channel::bounded(self.threads);
 
         let mut bounds = batch.keys_factory().default_box();
-        batch.partition_keys(self.workers, &mut *bounds);
+        batch.partition_keys(self.threads, &mut *bounds);
 
         for i in 0..=bounds.len() {
             let Some(cursor_builder) =
diff --git a/crates/adapters/src/static_compile/seroutput.rs b/crates/adapters/src/static_compile/seroutput.rs
@@ -476,7 +476,7 @@ where
         self.batch.inner().factories().keys_factory()
     }
 
-    fn data_factory(&self) -> &'static dyn Factory<DynData> {
+    fn key_factory(&self) -> &'static dyn Factory<DynData> {
         self.batch.inner().factories().key_factory()
     }
 
diff --git a/crates/feldera-types/src/format/avro.rs b/crates/feldera-types/src/format/avro.rs
@@ -250,13 +250,13 @@ pub struct AvroEncoderConfig {
     #[serde(flatten)]
     pub registry_config: AvroSchemaRegistryConfig,
 
-    /// The number of workers to use during encoding.
+    /// The number of threads to use during encoding.
     ///
     /// Avro encoder supports encoding multiple records in parallel. This configuration specifies
-    /// the number of workers to run in parallel.
+    /// the number of threads to run in parallel.
     /// Default: 4
-    #[serde(default = "default_encoder_workers")]
-    pub workers: usize,
+    #[serde(default = "default_encoder_threads")]
+    pub threads: usize,
 }
 
 impl Default for AvroEncoderConfig {
@@ -270,11 +270,11 @@ impl Default for AvroEncoderConfig {
             subject_name_strategy: Default::default(),
             skip_schema_id: Default::default(),
             registry_config: Default::default(),
-            workers: default_encoder_workers(),
+            threads: default_encoder_threads(),
         }
     }
 }
 
-fn default_encoder_workers() -> usize {
+fn default_encoder_threads() -> usize {
     4
 }
diff --git a/docs.feldera.com/docs/formats/avro.md b/docs.feldera.com/docs/formats/avro.md
@@ -112,6 +112,7 @@ either `registry_urls` or `schema` properties must be specified.
 | `registry_username`           | string          | | Username used to authenticate with the registry.Requires `registry_urls` to be set. This option is mutually exclusive with token-based authentication (see `registry_authorization_token`).|
 | `registry_password`           | string          | | Password used to authenticate with the registry. Requires `registry_urls` to be set.|
 | `registry_authorization_token`| string          | | Token used to authenticate with the registry. Requires `registry_urls` to be set. This option is mutually exclusive with password-based authentication (see `registry_username` and `registry_password`).|
+| `threads`                     | integer         | `4` | Number of parallel worker threads used to encode messages. **Only supported with [indexed outputs](/connectors/unique_keys).** |
 
 ### Examples
 

Original file line number	Diff line number	Diff line change
`@@ -476,7 +476,7 @@ where`
`476`	`476`	`self.batch.inner().factories().keys_factory()`
`477`	`477`	`}`
`478`	`478`
`479`		`- fn data_factory(&self) -> &'static dyn Factory<DynData> {`
	`479`	`+ fn key_factory(&self) -> &'static dyn Factory<DynData> {`
`480`	`480`	`self.batch.inner().factories().key_factory()`
`481`	`481`	`}`
`482`	`482`
Original file line number	Diff line number	Diff line change
`@@ -250,13 +250,13 @@ pub struct AvroEncoderConfig {`
`250`	`250`	`#[serde(flatten)]`
`251`	`251`	`pub registry_config: AvroSchemaRegistryConfig,`
`252`	`252`
`253`		`- /// The number of workers to use during encoding.`
	`253`	`+ /// The number of threads to use during encoding.`
`254`	`254`	`///`
`255`	`255`	`/// Avro encoder supports encoding multiple records in parallel. This configuration specifies`
`256`		`- /// the number of workers to run in parallel.`
	`256`	`+ /// the number of threads to run in parallel.`
`257`	`257`	`/// Default: 4`
`258`		`- #[serde(default = "default_encoder_workers")]`
`259`		`- pub workers: usize,`
	`258`	`+ #[serde(default = "default_encoder_threads")]`
	`259`	`+ pub threads: usize,`
`260`	`260`	`}`
`261`	`261`
`262`	`262`	`impl Default for AvroEncoderConfig {`
`@@ -270,11 +270,11 @@ impl Default for AvroEncoderConfig {`
`270`	`270`	`subject_name_strategy: Default::default(),`
`271`	`271`	`skip_schema_id: Default::default(),`
`272`	`272`	`registry_config: Default::default(),`
`273`		`- workers: default_encoder_workers(),`
	`273`	`+ threads: default_encoder_threads(),`
`274`	`274`	`}`
`275`	`275`	`}`
`276`	`276`	`}`
`277`	`277`
`278`		`-fn default_encoder_workers() -> usize {`
	`278`	`+fn default_encoder_threads() -> usize {`
`279`	`279`	`4`
`280`	`280`	`}`