Skip to main content

feldera_types/format/
avro.rs

1use serde::{Deserialize, Serialize};
2use std::{collections::HashMap, fmt::Display};
3use utoipa::ToSchema;
4
5/// Supported Avro data change event formats.
6#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq, ToSchema, Default)]
7pub enum AvroUpdateFormat {
8    /// Raw encoding.
9    ///
10    /// Each message in the stream represents a single-record update: an insert, upsert, or delete.
11    ///
12    /// ### Input Connectors
13    /// Raw Avro encoding can be used for insert and upsert operations, but not deletes.
14    /// - The message value contains the record to be inserted or updated.
15    /// - The message key and headers are ignored.
16    ///
17    /// ### Output Connectors
18    /// The raw format supports inserts, upserts, and deletes.
19    /// - The message value contains the record to be inserted or deleted.
20    /// - The operation type is specified in the `op` message header field, which can be
21    ///   `insert`, `update`, or `delete`.
22    /// - The message key can optionally store the primary key (see the `key_mode` property).
23    #[serde(rename = "raw")]
24    #[default]
25    Raw,
26
27    /// Debezium data change event format.
28    ///
29    /// ### Temporal types
30    ///
31    /// Debezium encodes temporal columns with a `connect.name` annotation
32    /// rather than a native Avro logical type. The input connector recognizes
33    /// these annotations and converts each value into the matching Feldera
34    /// column type:
35    ///
36    /// | `connect.name`                          | wire encoding                     | Feldera column |
37    /// |-----------------------------------------|-----------------------------------|----------------|
38    /// | `io.debezium.time.Date`                 | `int` days since epoch            | `DATE`         |
39    /// | `io.debezium.time.Time`                 | `int` milliseconds since midnight | `TIME`         |
40    /// | `io.debezium.time.MicroTime`            | `long` microseconds since midnight| `TIME`         |
41    /// | `io.debezium.time.NanoTime`             | `long` nanoseconds since midnight | `TIME`         |
42    /// | `io.debezium.time.ZonedTime`            | ISO-8601 `string`                 | `TIME`         |
43    /// | `io.debezium.time.Timestamp`            | `long` milliseconds since epoch   | `TIMESTAMP`    |
44    /// | `io.debezium.time.MicroTimestamp`       | `long` microseconds since epoch   | `TIMESTAMP`    |
45    /// | `io.debezium.time.NanoTimestamp`        | `long` nanoseconds since epoch    | `TIMESTAMP`    |
46    /// | `io.debezium.time.ZonedTimestamp`       | ISO-8601 `string`                 | `TIMESTAMP`    |
47    ///
48    /// Any timestamp type populates either a `TIMESTAMP` or a `TIMESTAMP WITH
49    /// TIME ZONE` column; the connector stores the same instant (microseconds
50    /// since the Unix epoch) in both cases.
51    ///
52    /// The `org.apache.kafka.connect.data.{Date,Time,Timestamp}` types emitted
53    /// with `time.precision.mode=connect` are also supported.
54    ///
55    /// `io.debezium.data.VariableScaleDecimal` (used for `NUMERIC`/`DECIMAL`
56    /// columns without a fixed scale, encoded as a `{scale, value}` record) is
57    /// parsed into a `DECIMAL` column.
58    #[serde(rename = "debezium")]
59    Debezium,
60
61    /// Confluent JDBC connector change event format.
62    #[serde(rename = "confluent_jdbc")]
63    ConfluentJdbc,
64}
65
66impl Display for AvroUpdateFormat {
67    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68        match self {
69            Self::Raw => f.write_str("raw"),
70            Self::Debezium => f.write_str("debezium"),
71            Self::ConfluentJdbc => f.write_str("confluent_jdbc"),
72        }
73    }
74}
75
76/// Schema registry configuration.
77#[derive(Clone, Serialize, Deserialize, Debug, Default, ToSchema)]
78pub struct AvroSchemaRegistryConfig {
79    /// List of schema registry URLs.
80    ///
81    /// * **Input connector**: When non-empty, the connector retrieves Avro
82    ///   message schemas from the registry.
83    ///
84    /// * **Output connector**: When non-empty, the connector will
85    ///   post the schema to the registry and embed the schema id returned
86    ///   by the registry in Avro messages.  Otherwise, schema id 0 is used.
87    #[serde(default)]
88    pub registry_urls: Vec<String>,
89
90    /// Custom headers that will be added to every call to the schema registry.
91    ///
92    /// This property is only applicable to output connectors.
93    ///
94    /// Requires `registry_urls` to be set.
95    #[serde(default)]
96    pub registry_headers: HashMap<String, String>,
97
98    /// Proxy that will be used to access the schema registry.
99    ///
100    /// Requires `registry_urls` to be set.
101    pub registry_proxy: Option<String>,
102
103    /// Timeout in seconds used to connect to the registry.
104    ///
105    /// Requires `registry_urls` to be set.
106    pub registry_timeout_secs: Option<u64>,
107
108    /// Username used to authenticate with the registry.
109    ///
110    /// Requires `registry_urls` to be set. This option is mutually exclusive with
111    /// token-based authentication (see `registry_authorization_token`).
112    pub registry_username: Option<String>,
113
114    /// Password used to authenticate with the registry.
115    ///
116    /// Requires `registry_urls` to be set.
117    pub registry_password: Option<String>,
118
119    /// Token used to authenticate with the registry.
120    ///
121    /// Requires `registry_urls` to be set. This option is mutually exclusive with
122    /// password-based authentication (see `registry_username` and `registry_password`).
123    pub registry_authorization_token: Option<String>,
124}
125
126/// Avro output format configuration.
127#[derive(Clone, Serialize, Deserialize, Debug, Default, ToSchema)]
128#[serde(deny_unknown_fields)]
129pub struct AvroParserConfig {
130    /// Format used to encode data change events in this stream.
131    ///
132    /// The default value is 'raw'.
133    #[serde(default)]
134    pub update_format: AvroUpdateFormat,
135
136    /// Avro schema used to encode all records in this stream, specified as a JSON-encoded string.
137    ///
138    /// When this property is set, the connector uses the provided schema instead of
139    /// retrieving the schema from the schema registry. This setting is mutually exclusive
140    /// with `registry_urls`.
141    pub schema: Option<String>,
142
143    /// `true` if serialized messages only contain raw data without the
144    /// header carrying schema ID.
145    ///
146    /// See <https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format>
147    ///
148    /// The default value is `false`.
149    #[serde(default)]
150    pub skip_schema_id: bool,
151
152    /// Schema registry configuration.
153    #[serde(flatten)]
154    pub registry_config: AvroSchemaRegistryConfig,
155}
156
157/// Subject name strategies used in registering key and value schemas
158/// with the schema registry.
159#[derive(Clone, Serialize, Deserialize, Debug, ToSchema)]
160pub enum SubjectNameStrategy {
161    /// The subject name is derived directly from the Kafka topic name.
162    ///
163    /// For update formats with both key and value components, use subject names
164    /// `{topic_name}-key` and `{topic_name}-value` for key and value schemas respectively.
165    /// For update formats without a key (e.g., `raw`), publish value schema
166    /// under the subject name `{topic_name}`.
167    ///
168    /// Only applicable when using Kafka as a transport.
169    #[serde(rename = "topic_name")]
170    TopicName,
171
172    /// The name of the SQL relation that the schema is derived from is used as the subject name:
173    /// * the SQL view name for the message value schema.
174    /// * the SQL index name for the message key schema.
175    #[serde(rename = "record_name")]
176    RecordName,
177
178    /// Combines both the topic name and the record name to form the subject.
179    ///
180    /// For update formats with both key and value components, use subject names
181    /// `{topic_name}-{record_name}-key` and `{topic_name}-{record_name}-value` for
182    /// key and value schemas respectively.
183    /// For update formats without a key (e.g., `raw`), publish value schema
184    /// under the subject name `{topic_name}-{record_name}`.
185    ///
186    /// `{record_name}` is the name of the SQL view or index that this connector
187    /// is attached to.
188    ///
189    /// Only applicable when using Kafka as a transport.
190    #[serde(rename = "topic_record_name")]
191    TopicRecordName,
192}
193
194/// Determines how the message key is generated when the Avro encoder is configured
195/// in the `raw` mode.
196#[derive(Clone, Serialize, Deserialize, Debug, ToSchema, PartialEq, Eq)]
197pub enum AvroEncoderKeyMode {
198    /// Produce messages without a key.
199    #[serde(rename = "none")]
200    None,
201
202    /// Uses the unique key columns of the view as the message key.
203    ///
204    /// This setting is supported when the output connector is configured with the `index` property.
205    /// It utilizes the values of the index columns specified in the associated `CREATE INDEX` statement
206    /// as the Avro message key.
207    ///
208    /// A separate Avro schema will be created and registered in the schema registry
209    /// for the key component of the message.
210    #[serde(rename = "key_fields")]
211    KeyFields,
212}
213
214/// Avro output format configuration.
215#[derive(Clone, Serialize, Deserialize, Debug, ToSchema)]
216#[serde(deny_unknown_fields)]
217pub struct AvroEncoderConfig {
218    /// Format used to encode data change events in this stream.
219    ///
220    /// The default value is `raw`.
221    #[serde(default)]
222    pub update_format: AvroUpdateFormat,
223
224    /// Determines how the message key is generated when the Avro encoder is configured
225    /// in the `raw` mode.
226    ///
227    /// The default is `key_fields` when the `index` property of the connector is configured and `none` otherwise.
228    pub key_mode: Option<AvroEncoderKeyMode>,
229
230    /// Avro schema used to encode output records.
231    ///
232    /// When specified, the encoder will use this schema; otherwise it will automatically
233    /// generate an Avro schema based on the SQL view definition.
234    ///
235    /// Specified as a string containing schema definition in JSON format.
236    /// This schema must match precisely the SQL view definition, modulo
237    /// nullability of columns.
238    pub schema: Option<String>,
239
240    /// Optional name of the field used for Change Data Capture (CDC) annotations.
241    ///
242    /// Use this setting with data sinks that expect operation type
243    /// (insert, delete, or update) encoded as a column in the Avro record, such
244    /// as the [Iceberg Sink Kafka Connector](https://docs.feldera.com/connectors/sinks/iceberg).
245    ///
246    /// When set (e.g., `"cdc_field": "op"`), the specified field will be added to each record
247    /// to indicate the type of change:
248    /// - `"I"` for insert operations
249    /// - `"U"` for upserts
250    /// - `"D"` for deletions
251    ///
252    /// If not set, CDC metadata will not be included in the records.
253    /// Only works with the `raw` update format.
254    pub cdc_field: Option<String>,
255
256    /// Avro namespace for the generated Avro schemas.
257    pub namespace: Option<String>,
258
259    /// Subject name strategy used to publish Avro schemas used by the connector
260    /// in the schema registry.
261    ///
262    /// When this property is not specified, the connector chooses subject name strategy automatically:
263    /// * `topic_name` for `confluent_jdbc` update format
264    /// * `record_name` for `raw` update format
265    pub subject_name_strategy: Option<SubjectNameStrategy>,
266
267    /// Set to `true` if serialized messages should only contain raw data
268    /// without the header carrying schema ID.
269    /// `False` by default.
270    ///
271    /// See <https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format>
272    #[serde(default)]
273    pub skip_schema_id: bool,
274
275    /// Schema registry configuration.
276    ///
277    /// When configured, the connector will push the Avro schema, whether it is specified as part of
278    /// connector configuration or generated automatically, to the schema registry and use the schema id
279    /// assigned by the registry in the
280    #[serde(flatten)]
281    pub registry_config: AvroSchemaRegistryConfig,
282
283    /// The number of threads to use during encoding.
284    ///
285    /// Avro encoder supports encoding multiple records in parallel. This configuration specifies
286    /// the number of threads to run in parallel.
287    /// Default: 4
288    #[serde(default = "default_encoder_threads")]
289    pub threads: usize,
290}
291
292impl Default for AvroEncoderConfig {
293    fn default() -> Self {
294        Self {
295            update_format: Default::default(),
296            key_mode: Default::default(),
297            schema: Default::default(),
298            cdc_field: Default::default(),
299            namespace: Default::default(),
300            subject_name_strategy: Default::default(),
301            skip_schema_id: Default::default(),
302            registry_config: Default::default(),
303            threads: default_encoder_threads(),
304        }
305    }
306}
307
308fn default_encoder_threads() -> usize {
309    4
310}