Skip to content

[connectors] Rename CsvParserConfig to CsvFormatConfig #6537

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mihaibudiu merged 1 commit into from
Jun 25, 2026
Merged

[connectors] Rename CsvParserConfig to CsvFormatConfig #6537

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions crates/adapterlib/src/catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ use dbsp::dynamic::{ClonableTrait, DynData, DynVec, Erase, Factory};
use dbsp::operator::StagedBuffers;
use dyn_clone::DynClone;
use feldera_sqllib::Variant;
use feldera_types::format::csv::CsvParserConfig;
use feldera_types::format::csv::CsvFormatConfig;
use feldera_types::format::json::JsonFlavor;
use feldera_types::program_schema::{Relation, SqlIdentifier};
use feldera_types::serde_with_context::SqlSerdeConfig;
Expand All @@ -43,7 +43,7 @@ pub enum RecordFormat {
// raw encoding of this column only. This is particularly useful for
// tables that store raw JSON or binary data to be parsed using SQL.
Json(JsonFlavor),
Csv(CsvParserConfig),
Csv(CsvFormatConfig),
Parquet(SqlSerdeConfig),
#[cfg(feature = "with-avro")]
Avro,
Expand Down
40 changes: 20 additions & 20 deletions crates/adapters/src/format/csv.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ use feldera_adapterlib::{ConnectorMetadata, catalog::SerCursorFlattened};
use feldera_sqllib::Variant;
use feldera_types::{
config::ConnectorConfig,
format::csv::{CsvEncoderConfig, CsvParserConfig},
format::csv::{CsvEncoderConfig, CsvFormatConfig},
};
use serde::Deserialize;
use serde_json::json;
Expand Down Expand Up @@ -45,7 +45,7 @@ fn csv_trim(t: &feldera_types::format::csv::CsvTrim) -> csv::Trim {
///
/// `has_headers` is always `false` — the adapter layer skips the header row
/// itself and must not pass it to the underlying CSV reader.
pub(crate) fn csv_reader_builder(config: &CsvParserConfig) -> csv::ReaderBuilder {
pub(crate) fn csv_reader_builder(config: &CsvFormatConfig) -> csv::ReaderBuilder {
let mut b = csv::ReaderBuilder::new();
b.has_headers(false)
.delimiter(config.delimiter().0)
Expand Down Expand Up @@ -75,7 +75,7 @@ impl InputFormat for CsvInputFormat {
_endpoint_name: &str,
_request: &HttpRequest,
) -> Result<Box<dyn ErasedSerialize>, ControllerError> {
Ok(Box::new(CsvParserConfig::default()))
Ok(Box::new(CsvFormatConfig::default()))
}

fn new_parser(
Expand All @@ -85,7 +85,7 @@ impl InputFormat for CsvInputFormat {
config: &serde_json::Value,
) -> Result<Box<dyn Parser>, ControllerError> {
let config = if config.is_null() { &json!({}) } else { config };
let config = CsvParserConfig::deserialize(config).map_err(|e| {
let config = CsvFormatConfig::deserialize(config).map_err(|e| {
ControllerError::parser_config_parse_error(
endpoint_name,
&e,
Expand Down Expand Up @@ -121,7 +121,7 @@ struct CsvParser {
}

impl CsvParser {
fn new(input_stream: Box<dyn DeCollectionStream>, config: &CsvParserConfig) -> Self {
fn new(input_stream: Box<dyn DeCollectionStream>, config: &CsvFormatConfig) -> Self {
Self {
input_stream,
last_event_number: 0,
Expand Down Expand Up @@ -404,13 +404,13 @@ impl Encoder for CsvEncoder {

let mut cursor = if self.input_is_indexed {
CursorWithPolarity::new(Box::new(SerCursorFlattened::new(batch.cursor(
RecordFormat::Csv(CsvParserConfig {
RecordFormat::Csv(CsvFormatConfig {
delimiter: self.config.delimiter,
..Default::default()
}),
)?)))
} else {
CursorWithPolarity::new(batch.cursor(RecordFormat::Csv(CsvParserConfig {
CursorWithPolarity::new(batch.cursor(RecordFormat::Csv(CsvFormatConfig {
delimiter: self.config.delimiter,
..Default::default()
}))?)
Expand Down Expand Up @@ -482,7 +482,7 @@ mod test {

use super::CsvSplitter;
use crate::format::Splitter;
use feldera_types::format::csv::{CsvParserConfig, CsvTrim};
use feldera_types::format::csv::{CsvFormatConfig, CsvTrim};

#[derive(Debug, Eq, PartialEq)]
#[allow(non_snake_case)]
Expand Down Expand Up @@ -630,7 +630,7 @@ true,bar,buzz"#;

/// Push `data` through a fully-configured `CsvParser` and return the
/// inserted `TwoStrings` records.
fn e2e_parse_csv(config: CsvParserConfig, data: &[u8]) -> Vec<TwoStrings> {
fn e2e_parse_csv(config: CsvFormatConfig, data: &[u8]) -> Vec<TwoStrings> {
let format_config = FormatConfig {
name: Cow::from("csv"),
config: serde_json::to_value(config).unwrap(),
Expand All @@ -655,7 +655,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_quote() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
quote: '\'',
..Default::default()
},
Expand All @@ -682,7 +682,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_escape() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
escape: Some('\\'),
double_quote: false,
..Default::default()
Expand All @@ -699,7 +699,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_double_quote() {
let rows = e2e_parse_csv(
CsvParserConfig::default(),
CsvFormatConfig::default(),
// "foo""bar" → foo"bar
b"\"foo\"\"bar\",baz\n",
);
Expand All @@ -714,7 +714,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_double_quote_disabled() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
double_quote: false,
..Default::default()
},
Expand All @@ -728,7 +728,7 @@ true,bar,buzz"#;
// `trim = None` (default): leading and trailing whitespace is preserved.
#[test]
fn csv_e2e_trim_none() {
let rows = e2e_parse_csv(CsvParserConfig::default(), b" foo , bar \n");
let rows = e2e_parse_csv(CsvFormatConfig::default(), b" foo , bar \n");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].first, " foo ");
assert_eq!(rows[0].second, " bar ");
Expand All @@ -738,7 +738,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_quoting_disabled() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
quoting: false,
..Default::default()
},
Expand All @@ -754,7 +754,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_comment() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
comment: Some('#'),
..Default::default()
},
Expand All @@ -770,7 +770,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_flexible_true() {
let rows = e2e_parse_csv(
CsvParserConfig::default(), // flexible=true
CsvFormatConfig::default(), // flexible=true
b"foo,bar\nqux,quux,extra\n",
);
assert_eq!(rows.len(), 2);
Expand All @@ -785,7 +785,7 @@ true,bar,buzz"#;
fn csv_e2e_flexible_false() {
let format_config = FormatConfig {
name: Cow::from("csv"),
config: serde_json::to_value(CsvParserConfig {
config: serde_json::to_value(CsvFormatConfig {
flexible: false,
..Default::default()
})
Expand All @@ -811,7 +811,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_trim_fields() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
trim: CsvTrim::Fields,
..Default::default()
},
Expand All @@ -826,7 +826,7 @@ true,bar,buzz"#;
#[test]
fn csv_e2e_headers() {
let rows = e2e_parse_csv(
CsvParserConfig {
CsvFormatConfig {
headers: true,
..Default::default()
},
Expand Down
6 changes: 3 additions & 3 deletions crates/adapters/src/static_compile/deinput.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ use erased_serde::Deserializer as ErasedDeserializer;
use feldera_adapterlib::catalog::AvroSchemaRefs;
use feldera_adapterlib::format::{BufferSize, flatten_nested};
use feldera_sqllib::Variant;
use feldera_types::format::csv::CsvParserConfig;
use feldera_types::format::csv::CsvFormatConfig;
use feldera_types::serde_with_context::{DeserializeWithContext, SqlSerdeConfig};
use serde_arrow::Deserializer as ArrowDeserializer;
use serde_json::de::SliceRead;
Expand Down Expand Up @@ -94,8 +94,8 @@ pub struct CsvDeserializerFromBytes {
config: SqlSerdeConfig,
}

impl DeserializerFromBytes<(SqlSerdeConfig, CsvParserConfig)> for CsvDeserializerFromBytes {
fn create((serde_config, csv_config): (SqlSerdeConfig, CsvParserConfig)) -> Self {
impl DeserializerFromBytes<(SqlSerdeConfig, CsvFormatConfig)> for CsvDeserializerFromBytes {
fn create((serde_config, csv_config): (SqlSerdeConfig, CsvFormatConfig)) -> Self {
CsvDeserializerFromBytes {
// `csv_reader_builder` sets `has_headers(false)` — the adapter
// layer handles header-row skipping itself.
Expand Down
4 changes: 2 additions & 2 deletions crates/adapters/src/static_compile/seroutput.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ use feldera_types::serde_with_context::serialize::{
SerializeFieldsWithContextWrapper, SerializeWithContextWrapper,
};
use feldera_types::{
format::csv::CsvParserConfig,
format::csv::CsvFormatConfig,
serde_with_context::{SerializeWithContext, SqlSerdeConfig},
};
use rand::thread_rng;
Expand Down Expand Up @@ -197,7 +197,7 @@ struct CsvSerializer {
}

impl CsvSerializer {
fn create(config: CsvParserConfig) -> Self {
fn create(config: CsvFormatConfig) -> Self {
Self {
writer: CsvWriterBuilder::new()
.has_headers(false)
Expand Down
24 changes: 21 additions & 3 deletions crates/feldera-types/src/format/csv.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,19 +20,25 @@ pub enum CsvTrim {

#[derive(Clone, Debug, Deserialize, Serialize, ToSchema, PartialEq)]
#[serde(default)]
pub struct CsvParserConfig {
pub struct CsvFormatConfig {
/// Field delimiter (default `','`).
///
/// Must be an ASCII character.
///
/// Used by: input and output.
pub delimiter: char,

/// Whether the input begins with a header line (which is skipped).
///
/// Used by: input only.
pub headers: bool,

/// The quote character (default `'"'`).
///
/// Must be an ASCII character. Set `quoting` to `false` to disable
/// quoting entirely.
///
/// Used by: input only.
pub quote: char,

/// The escape character for quoted fields (default: `None`).
Expand All @@ -44,44 +50,56 @@ pub struct CsvParserConfig {
/// escaping).
///
/// Must be an ASCII character.
///
/// Used by: input only.
pub escape: Option<char>,

/// Enable double-quote escaping (default `true`).
///
/// When `true`, a quote character inside a quoted field may be escaped by
/// doubling it. Setting this to `false` disables double-quote escaping
/// (an explicit `escape` character can still be used).
///
/// Used by: input only.
pub double_quote: bool,

/// Enable quoting (default `true`).
///
/// When `false`, the `quote` and `escape` characters have no special
/// meaning and every newline terminates a record regardless of context.
///
/// Used by: input only.
pub quoting: bool,

/// Comment character (default: `None`).
///
/// When set, lines whose first byte matches this character are treated as
/// comments and skipped entirely. Must be an ASCII character.
///
/// Used by: input only.
pub comment: Option<char>,

/// Allow records with a variable number of fields (default `true`).
///
/// When `true`, records that have fewer or more fields than expected are
/// accepted rather than treated as errors.
///
/// Used by: input only.
pub flexible: bool,

/// Whitespace trimming policy (default [`CsvTrim::None`]).
///
/// Used by: input only.
pub trim: CsvTrim,
}

impl CsvParserConfig {
impl CsvFormatConfig {
pub fn delimiter(&self) -> CsvDelimiter {
self.delimiter.into()
}
}

impl Default for CsvParserConfig {
impl Default for CsvFormatConfig {
fn default() -> Self {
Self {
delimiter: CsvDelimiter::default().0.into(),
Expand Down
4 changes: 2 additions & 2 deletions ...bsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/OtherTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,7 +436,7 @@ pub fn test() {
#[test]
pub fn test() {
use dbsp_adapters::{CircuitCatalog, RecordFormat};
use feldera_types::format::csv::CsvParserConfig;
use feldera_types::format::csv::CsvFormatConfig;

let (mut circuit, catalog) = circuit(CircuitConfig::with_workers(2))
.expect("Failed to build circuit");
Expand Down Expand Up @@ -470,7 +470,7 @@ pub fn test() {
// Read the produced output
let reader = adult.concat();
let mut cursor = reader
.cursor(RecordFormat::Csv(CsvParserConfig::default()))
.cursor(RecordFormat::Csv(CsvFormatConfig::default()))
.unwrap();
while cursor.key_valid() {
let mut w = cursor.weight();
Expand Down
4 changes: 2 additions & 2 deletions sql-to-dbsp-compiler/using.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,7 @@ We exercise this circuit by inserting data using a CSV format:
#[test]
pub fn test() {
use dbsp_adapters::{CircuitCatalog, RecordFormat};
use use feldera_types::format::csv::CsvParserConfig;
use feldera_types::format::csv::CsvFormatConfig;

let (mut circuit, catalog) = circuit(2)
.expect("Failed to build circuit");
Expand Down Expand Up @@ -430,7 +430,7 @@ pub fn test() {
// Read the produced output
let reader = adult.concat().consolidate();
let mut cursor = reader
.cursor(RecordFormat::Csv(CsvParserConfig::default()))
.cursor(RecordFormat::Csv(CsvFormatConfig::default()))
.unwrap();
while cursor.key_valid() {
let mut w = cursor.weight();
Expand Down