From 3112223987e4cfe7620722479717e046300d0aa1 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Thu, 28 May 2026 19:50:06 -0700 Subject: [PATCH] Fix cdc_order_by with multiple fields. PR #6245 caused a regression for `cdc_order_by` which prevented parsing of order by clauses with multiple fields. That was because we didn't use the right DF function to parse it. This corrects the mistakes and adds a series of tests. Signed-off-by: Gerd Zellweger --- crates/adapterlib/src/utils/datafusion.rs | 117 +++++++++ .../src/integrated/delta_table/input.rs | 52 +++- .../src/integrated/delta_table/test.rs | 246 ++++++++++++++++++ .../platform/test_delta_input_catchup.py | 111 +++++++- 4 files changed, 515 insertions(+), 11 deletions(-) diff --git a/crates/adapterlib/src/utils/datafusion.rs b/crates/adapterlib/src/utils/datafusion.rs index 7c2c70ef24a..7cb9302a7bc 100644 --- a/crates/adapterlib/src/utils/datafusion.rs +++ b/crates/adapterlib/src/utils/datafusion.rs @@ -10,6 +10,7 @@ use datafusion::logical_expr::sqlparser::parser::ParserError; use datafusion::prelude::{SQLOptions, SessionConfig, SessionContext}; use datafusion::sql::sqlparser::dialect::GenericDialect; use datafusion::sql::sqlparser::parser::Parser; +use datafusion::sql::sqlparser::tokenizer::Token; use feldera_types::config::PipelineConfig; use feldera_types::constants::DATAFUSION_TEMP_DIR; use feldera_types::program_schema::{ColumnType, Field, Relation, SqlType}; @@ -294,6 +295,20 @@ pub fn validate_sql_expression(expr: &str) -> Result<(), ParserError> { Ok(()) } +/// Validate the body of an ORDER BY clause (e.g. "ts asc, lsn desc"). +/// +/// Unlike [`validate_sql_expression`], this accepts the comma-separated, +/// ASC/DESC/NULLS annotated key list a real ORDER BY allows, and +/// requires the whole string to parse so a malformed clause fails here rather +/// than silently dropping every key after the first. +pub fn validate_sql_order_by(order_by: &str) -> Result<(), ParserError> { + let mut parser = Parser::new(&GenericDialect).try_with_sql(order_by)?; + parser.parse_comma_separated(Parser::parse_order_by_expr)?; + parser.expect_token(&Token::EOF)?; + + Ok(()) +} + /// Convert a value of the timestamp column returned by a SQL query into a valid /// SQL expression. pub fn timestamp_to_sql_expression(column_type: &ColumnType, expr: &str) -> String { @@ -664,4 +679,106 @@ mod tests { assert_eq!(min_pool_mb_for_adhoc_sort(2), 135); assert_eq!(min_pool_mb_for_adhoc_sort(8), 537); } + + /// Make sure random shapes for `filter`, `cdc_delete_filter`, and `cdc_order_by` + /// are parsed correctly by our connector. + #[test] + fn cdc_connector_expr_shapes_validate() { + use super::{validate_sql_expression, validate_sql_order_by}; + + // Set as `filter` or `cdc_delete_filter`; validated as a scalar predicate. + const FILTER_SHAPES: &[&str] = &[ + "0=0", + "0=0 AND (a = 's0' AND b = 's1')", + "0=0 AND (a = 's0')", + "0=0 AND (a IN ('s0'))", + "0=0 AND (a IN ('s0','s1'))", + "0=0 AND (a IN (1,2) OR a IS NULL)", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b = false)", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b IN ('s0'))", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b IN ('s0','s1'))", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b IS NOT NULL)", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b IS NULL AND c IS NULL)", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b IS NULL)", + "0=0 AND (a IN (1,2) OR a IS NULL) AND (b NOT IN ('s0','s1') AND c IS NOT NULL)", + "0=0 AND (a IN('s0','s1'))", + "0=0 AND (a IS NOT NULL AND b IS NOT NULL)", + "0=0 AND a = false", + "0=0 AND a = false AND (b = 's0' AND c = 's1')", + "0=0 AND a = false AND (b = 's0')", + "0=0 AND a = false AND (b IN ('s0'))", + "0=0 AND a = false AND (b IN ('s0','s1'))", + "0=0 AND a = false AND (b IN (1,2) OR b IS NULL)", + "0=0 AND a = false AND (b IN (1,2) OR b IS NULL) AND (c = false)", + "0=0 AND a = false AND (b IN (1,2) OR b IS NULL) AND (c IS NOT NULL)", + "0=0 AND a = false AND (b IS NOT NULL AND c IS NOT NULL)", + "0=0 AND a = false AND b is null", + "0=0 AND a = false AND b is null AND (c = 's0')", + "0=0 AND a = false AND b is null AND (c IN ('s0','s1'))", + "0=0 AND a = false AND b is null AND (c IN (1,2) OR c IS NULL)", + "0=0 AND a = false AND b is null AND (c IN (1,2) OR c IS NULL) AND (d NOT IN ('s0','s1') AND e IS NOT NULL)", + "a > 0", + "a >= 0 AND a <= 9", + "a <> 's0'", + "a != 's0'", + "a BETWEEN 0 AND 9", + "a LIKE 's0'", + "a IS NULL OR b IS NOT NULL", + "NOT (a = false)", + "lower(a) = 's0'", + "cast(a AS bigint) = 0", + "a + b > 0", + "coalesce(a, b) = 's0'", + "a > timestamp '2020-01-02 03:04:05'", + "a = 's0''s1'", + ]; + const CDC_DELETE_FILTER_SHAPES: &[&str] = &[ + "a = true", + "a = true OR b is not null", + "a = true AND b = false", + "a IN ('s0','s1')", + "a IS NOT NULL", + "NOT a", + ]; + const CDC_ORDER_BY_SHAPES: &[&str] = &[ + "a", + "a, b", + "a asc, b asc", + "a ASC", + "a desc", + "a ASC, b DESC", + "a NULLS FIRST", + "a ASC NULLS LAST", + "a DESC NULLS FIRST", + "a asc nulls last, b desc nulls first", + "a asc, b desc, c asc nulls last", + "a + b asc", + "a % 2 asc, b desc", + "lower(a) asc", + "abs(a) desc, b asc", + "cast(a AS bigint) asc", + "coalesce(a, b) asc, c desc", + "case when a then 0 else 1 end desc", + // Quoted identifier containing a space. + "\"a b\" asc", + ]; + + let mut failures = Vec::new(); + for expr in FILTER_SHAPES.iter().chain(CDC_DELETE_FILTER_SHAPES) { + if let Err(e) = validate_sql_expression(expr) { + failures.push(format!("predicate '{expr}' failed: {e}")); + } + } + for order_by in CDC_ORDER_BY_SHAPES { + if let Err(e) = validate_sql_order_by(order_by) { + failures.push(format!("cdc_order_by '{order_by}' failed: {e}")); + } + } + + assert!( + failures.is_empty(), + "validation failures:\n{}", + failures.join("\n") + ); + } } diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index cba01abd0b5..dd6a925218a 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -18,7 +18,11 @@ use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner}; use dbsp::circuit::tokio::TOKIO; use deltalake::datafusion::dataframe::DataFrame; use deltalake::datafusion::execution::context::SQLOptions; +use deltalake::datafusion::logical_expr::SortExpr; use deltalake::datafusion::prelude::SessionContext; +use deltalake::datafusion::sql::sqlparser::dialect::GenericDialect; +use deltalake::datafusion::sql::sqlparser::parser::Parser; +use deltalake::datafusion::sql::sqlparser::tokenizer::Token; use deltalake::kernel::Action; use deltalake::logstore::{self, IORuntime}; use deltalake::table::builder::ensure_table_uri; @@ -28,7 +32,8 @@ use feldera_adapterlib::metrics::{ConnectorMetrics, ValueType}; use feldera_adapterlib::transport::{InputQueueEntry, Resume, Watermark, parse_resume_info}; use feldera_adapterlib::utils::datafusion::{ array_to_string, create_session_context_with, execute_query_collect, execute_singleton_query, - timestamp_to_sql_expression, validate_sql_expression, validate_timestamp_column, + timestamp_to_sql_expression, validate_sql_expression, validate_sql_order_by, + validate_timestamp_column, }; use feldera_storage::tokio::TOKIO_DEDICATED_IO; use feldera_types::adapter_stats::ConnectorHealth; @@ -199,14 +204,49 @@ pub(super) fn build_cdc_dataframe( } }; - let order_by_expr = result_df - .parse_sql_expr(order_by) - .map_err(|e| anyhow!("invalid 'cdc_order_by' expression '{order_by}': {e}"))?; - result_df.sort_by(vec![order_by_expr]).map_err(|e| { + let sort_exprs = parse_cdc_order_by(&result_df, order_by)?; + result_df.sort(sort_exprs).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error applying 'cdc_order_by': {e}") }) } +/// Parse the `cdc_order_by` clause into DataFusion sort expressions resolved +/// against df's schema. +/// +/// `cdc_order_by` is the body of an ORDER BY clause: a comma-separated list +/// of keys, each optionally annotated with ASC / DESC and NULLS +/// FIRST / NULLS LAST. +pub(super) fn parse_cdc_order_by(df: &DataFrame, order_by: &str) -> AnyResult> { + let mut parser = Parser::new(&GenericDialect) + .try_with_sql(order_by) + .map_err(|e| anyhow!("invalid 'cdc_order_by' expression '{order_by}': {e}"))?; + let order_exprs = parser + .parse_comma_separated(Parser::parse_order_by_expr) + .map_err(|e| anyhow!("invalid 'cdc_order_by' expression '{order_by}': {e}"))?; + if parser.peek_token().token != Token::EOF { + bail!( + "invalid 'cdc_order_by' expression '{order_by}': unexpected trailing input near '{}'", + parser.peek_token() + ); + } + + order_exprs + .into_iter() + .map(|order_expr| { + let key = df + .parse_sql_expr(&order_expr.expr.to_string()) + .map_err(|e| anyhow!("invalid 'cdc_order_by' expression '{order_by}': {e}"))?; + // Match DataFusion's SQL planner defaults: a missing direction is + // ASC, and missing NULLS follows nulls_max (last for ASC). + // - https://datafusion.apache.org/user-guide/sql/select.html#order-by-clause + // - https://datafusion.apache.org/user-guide/configs.html + let asc = order_expr.options.asc.unwrap_or(true); + let nulls_first = order_expr.options.nulls_first.unwrap_or(!asc); + Ok(key.sort(asc, nulls_first)) + }) + .collect() +} + /// Integrated input connector that reads from a delta table. pub struct DeltaTableInputEndpoint { endpoint_name: String, @@ -1985,7 +2025,7 @@ impl DeltaTableInputEndpointInner { /// Validate the cdc_order_by expression. fn validate_cdc_order_by(&self) -> Result<(), ControllerError> { if let Some(order_by) = &self.config.cdc_order_by { - validate_sql_expression(order_by).map_err(|e| { + validate_sql_order_by(order_by).map_err(|e| { ControllerError::invalid_transport_configuration( &self.endpoint_name, &format!("error parsing 'cdc_order_by' expression '{order_by}': {e}"), diff --git a/crates/adapters/src/integrated/delta_table/test.rs b/crates/adapters/src/integrated/delta_table/test.rs index a3508578961..5b81c5391ba 100644 --- a/crates/adapters/src/integrated/delta_table/test.rs +++ b/crates/adapters/src/integrated/delta_table/test.rs @@ -2378,6 +2378,157 @@ async fn delta_table_cdc_rewrite_test() { read_pipeline.stop().unwrap(); } +/// A multi-key `cdc_order_by`: "__feldera_ts asc, lsn asc` +/// +/// One transaction deletes the old value and inserts a new one for the same id, +/// and the connector applies both, leaving the updated row. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn delta_table_cdc_multi_key_order_by_test() { + use crate::test::TestStruct; + use arrow::array::{ + Array, BooleanArray, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray, + }; + use arrow::datatypes::{ + DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema, TimeUnit, + }; + use deltalake::kernel::{DataType as KernelDataType, PrimitiveType, StructField}; + + init_logging(); + + let expect = |id: u32, label: &str| TestStruct { + id, + b: false, + i: None, + s: label.to_string(), + }; + + // TestStruct columns + CDC bookkeeping columns. `lsn` is the order + // tiebreaker; it is not a TestStruct field, so it never reaches the table. + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int64, false), + ArrowField::new("b", ArrowDataType::Boolean, false), + ArrowField::new("s", ArrowDataType::Utf8, false), + ArrowField::new("__feldera_op", ArrowDataType::Utf8, false), + ArrowField::new( + "__feldera_ts", + ArrowDataType::Timestamp(TimeUnit::Microsecond, None), + false, + ), + ArrowField::new("lsn", ArrowDataType::Int64, false), + ])); + let struct_fields = vec![ + StructField::new("id", KernelDataType::Primitive(PrimitiveType::Long), false), + StructField::new( + "b", + KernelDataType::Primitive(PrimitiveType::Boolean), + false, + ), + StructField::new("s", KernelDataType::Primitive(PrimitiveType::String), false), + StructField::new( + "__feldera_op", + KernelDataType::Primitive(PrimitiveType::String), + false, + ), + StructField::new( + "__feldera_ts", + KernelDataType::Primitive(PrimitiveType::TimestampNtz), + false, + ), + StructField::new("lsn", KernelDataType::Primitive(PrimitiveType::Long), false), + ]; + + // Each tuple is (id, op, ts, lsn, s). + let make_batch = |rows: &[(i64, &str, i64, i64, &str)]| -> RecordBatch { + RecordBatch::try_new( + arrow_schema.clone(), + vec![ + Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.0))) as Arc, + Arc::new(rows.iter().map(|_| Some(false)).collect::()), + Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.4))), + Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1))), + Arc::new(TimestampMicrosecondArray::from_iter_values( + rows.iter().map(|r| r.2), + )), + Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.3))), + ], + ) + .unwrap() + }; + + async fn append(delta: DeltaTable, batch: RecordBatch) -> DeltaTable { + delta + .write(vec![batch]) + .with_save_mode(SaveMode::Append) + .await + .unwrap() + } + + let table_dir = TempDir::new().unwrap(); + let table_uri = table_dir.path().display().to_string(); + let mut delta = create_table(&table_uri, &HashMap::new(), &struct_fields).await; + + let storage_dir = TempDir::new().unwrap(); + let read_pipeline = { + let table_uri = table_uri.clone(); + let storage_dir = storage_dir.path().to_path_buf(); + tokio::task::spawn_blocking(move || { + let pipeline_config: PipelineConfig = serde_json::from_value(json!({ + "name": "test", + "workers": 4, + "storage_config": { "path": storage_dir }, + "inputs": { + "test_input1": { + "stream": "test_input1", + "transport": { + "name": "delta_table_input", + "config": { + "uri": table_uri, + "mode": "cdc", + "filter": "id % 2 = 0", + "cdc_delete_filter": "__feldera_op = 'd'", + "cdc_order_by": "__feldera_ts asc, lsn asc", + } + } + } + } + })) + .unwrap(); + Controller::with_test_config( + move |workers| { + Ok(test_circuit::( + workers, + &TestStruct::schema(), + &[Some("output")], + )) + }, + &pipeline_config, + Box::new(move |e, _| panic!("cdc multi-key order_by test: {e}")), + ) + .unwrap() + }) + .await + .unwrap() + }; + read_pipeline.start(); + let output = SqlIdentifier::from("test_output1"); + + // Seed id 0 (kept by `id % 2 = 0`). + delta = append(delta, make_batch(&[(0, "i", 1_000, 0, "orig")])).await; + wait_for_records_materialized(&read_pipeline, &output, &[expect(0, "orig")]).await; + + // One transaction updates id 0: delete the old value, insert the new one. + // The multi-key `cdc_order_by` must parse for this transaction to run. + delta = append( + delta, + make_batch(&[(0, "d", 2_000, 1, "orig"), (0, "i", 2_000, 2, "updated")]), + ) + .await; + wait_for_records_materialized(&read_pipeline, &output, &[expect(0, "updated")]).await; + + let _ = delta; + read_pipeline.stop().unwrap(); +} + /// `cdc_delete_filter` is parsed against the snapshot schema at /// connector startup, so an expression that references a column that /// does not exist on the Delta table must fail before any row flows @@ -2603,6 +2754,101 @@ async fn build_cdc_dataframe_plan_structure() { ); } +/// A `cdc_order_by` with several keys and explicit ASC/DESC directions +/// (a real ORDER BY clause body) must build a multi-key Sort. Guards a +/// regression where the clause was parsed as a single scalar expression. +#[tokio::test] +async fn build_cdc_dataframe_multi_key_order_by() { + use crate::integrated::delta_table::input::build_cdc_dataframe; + use arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField, TimeUnit}; + use datafusion::datasource::MemTable; + use feldera_adapterlib::utils::datafusion::validate_sql_order_by; + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int64, false), + ArrowField::new( + "__feldera_ts", + ArrowDataType::Timestamp(TimeUnit::Microsecond, None), + false, + ), + ])); + + let ctx = SessionContext::new(); + ctx.register_table( + "cdc_adds", + Arc::new(MemTable::try_new(arrow_schema.clone(), vec![vec![]]).unwrap()), + ) + .unwrap(); + + let df = build_cdc_dataframe( + ctx.table("cdc_adds").await.unwrap(), + None, + None, + "__feldera_ts asc, id desc", + "unit-test", + ) + .unwrap(); + let plan = format!("{}", df.logical_plan().display_indent()); + + // One Sort node carrying both keys with their directions. + assert!( + plan.contains("__feldera_ts ASC") && plan.contains("id DESC"), + "expected both sort keys with directions; plan was:\n{plan}" + ); + + // The validator accepts the same multi-key clause and rejects garbage. + assert!(validate_sql_order_by("__feldera_ts asc, id desc").is_ok()); + assert!(validate_sql_order_by("ts asc !! lsn").is_err()); +} + +/// `parse_cdc_order_by` applies DataFusion's SQL planner defaults: a missing +/// direction is `ASC`, a missing `NULLS` placement is nulls-last for `ASC` and +/// nulls-first for `DESC` (the `nulls_max` convention), and explicit `ASC` / +/// `DESC` / `NULLS FIRST` / `NULLS LAST` override those defaults. +#[tokio::test] +async fn parse_cdc_order_by_defaults() { + use crate::integrated::delta_table::input::parse_cdc_order_by; + use arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField}; + use datafusion::datasource::MemTable; + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("a", ArrowDataType::Int64, false), + ArrowField::new("b", ArrowDataType::Int64, false), + ])); + let ctx = SessionContext::new(); + ctx.register_table( + "t", + Arc::new(MemTable::try_new(arrow_schema.clone(), vec![vec![]]).unwrap()), + ) + .unwrap(); + let df = ctx.table("t").await.unwrap(); + + // (clause, expected [(asc, nulls_first), ...] per key). + let cases: &[(&str, &[(bool, bool)])] = &[ + // No direction => ASC; ASC default nulls placement is last. + ("a", &[(true, false)]), + ("a asc", &[(true, false)]), + // DESC default nulls placement is first. + ("a desc", &[(false, true)]), + // Explicit NULLS overrides the default for either direction. + ("a nulls first", &[(true, true)]), + ("a asc nulls first", &[(true, true)]), + ("a desc nulls last", &[(false, false)]), + // Each key carries its own direction and default. + ("a asc, b desc", &[(true, false), (false, true)]), + ]; + + for (clause, expected) in cases { + let sort_exprs = parse_cdc_order_by(&df, clause).unwrap(); + let got: Vec<(bool, bool)> = sort_exprs.iter().map(|s| (s.asc, s.nulls_first)).collect(); + assert_eq!( + got.as_slice(), + *expected, + "clause '{clause}': expected {expected:?}, got {got:?}" + ); + } +} + /// Pin the exact set of arrow types that the `EXCEPT ALL` cancellation /// step in `do_process_cdc_transaction` cannot handle. /// diff --git a/python/tests/platform/test_delta_input_catchup.py b/python/tests/platform/test_delta_input_catchup.py index 815b5fc822f..da366073326 100644 --- a/python/tests/platform/test_delta_input_catchup.py +++ b/python/tests/platform/test_delta_input_catchup.py @@ -11,9 +11,8 @@ import json import re -from http import HTTPStatus - from datetime import datetime, timezone +from http import HTTPStatus import pyarrow as pa import pytest @@ -91,7 +90,7 @@ def _connector_config( config.update( { "cdc_delete_filter": "__feldera_op = 'd'", - "cdc_order_by": "__feldera_ts", + "cdc_order_by": "__feldera_ts asc, lsn asc", } ) connector = { @@ -175,10 +174,15 @@ def _materialized_row_count(pipeline) -> int: pa.field("id", pa.int64()), pa.field("__feldera_op", pa.string()), pa.field("__feldera_ts", pa.timestamp("us")), + pa.field("lsn", pa.int64()), ] ) +def _cdc_ts(ts_us: int) -> datetime: + return datetime.fromtimestamp(ts_us / 1_000_000, tz=timezone.utc) + + def _append_cdc_insert( loc: DeltaTestLocation, row_id: int, @@ -188,11 +192,18 @@ def _append_cdc_insert( ) -> None: from deltalake import write_deltalake - ts = datetime.fromtimestamp(ts_us / 1_000_000, tz=timezone.utc) + # `lsn` mirrors `__feldera_ts` here; the catchup test never ties on ts. write_deltalake( loc.uri, pa.Table.from_pylist( - [{"id": row_id, "__feldera_op": "i", "__feldera_ts": ts}], + [ + { + "id": row_id, + "__feldera_op": "i", + "__feldera_ts": _cdc_ts(ts_us), + "lsn": ts_us, + } + ], schema=_CDC_SCHEMA, ), mode=mode, @@ -386,5 +397,95 @@ def test_delta_input_catchup_cdc(pipeline_name): loc.cleanup() +_CDC_VAL_SCHEMA = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field("val", pa.int64()), + pa.field("__feldera_op", pa.string()), + pa.field("__feldera_ts", pa.timestamp("us")), + pa.field("lsn", pa.int64()), + ] +) + + +@enterprise_only +def test_delta_input_cdc_multi_key_order_by(pipeline_name): + """A multi-key `"cdc_order_by": "__feldera_ts asc, lsn asc"` parses and + honors the second key as a tiebreaker. + + The table has a primary key, so two inserts of the same key in one + transaction collapse to a last-writer-wins upsert. Both carry the same + `__feldera_ts`; only the `lsn asc` tiebreak makes the larger `lsn` + the last writer. The losing row is written to the file first, so a + single-key sort would let it win instead. + """ + from deltalake import write_deltalake + + loc = DeltaTestLocation.create(pipeline_name) + + def append(rows: list[dict], *, mode: str = "append") -> None: + write_deltalake( + loc.uri, + pa.Table.from_pylist( + [ + { + "id": r["id"], + "val": r["val"], + "__feldera_op": "i", + "__feldera_ts": _cdc_ts(r["ts_us"]), + "lsn": r["lsn"], + } + for r in rows + ], + schema=_CDC_VAL_SCHEMA, + ), + mode=mode, + schema_mode="merge", + storage_options=_writer_storage_options(loc), + ) + + try: + # CDC mode skips pre-existing commits; seed (not ingested) and follow. + append([{"id": 0, "val": 0, "ts_us": 1_000, "lsn": 0}], mode="overwrite") + + connectors = _connector_config(loc, mode="cdc").replace("'", "''") + sql = ( + f"CREATE TABLE {TABLE} (id BIGINT NOT NULL PRIMARY KEY, val BIGINT) " + f"WITH ('materialized' = 'true', 'connectors' = '{connectors}');" + ) + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=sql, + runtime_config=RuntimeConfig( + workers=1, + hosts=FELDERA_TEST_NUM_HOSTS, + logging="debug", + ), + ).create_or_replace() + pipeline.start() + _wait_for_completed_version(pipeline, _delta_version(loc)) + assert _materialized_row_count(pipeline) == 0 + + # One transaction, two upserts of id 0 at the same `__feldera_ts`. The + # winner (lsn 2, val 200) is written to the file first, so a sort that + # ignored `lsn` would leave the loser (lsn 1) last and let val 100 win. + append( + [ + {"id": 0, "val": 200, "ts_us": 2_000, "lsn": 2}, + {"id": 0, "val": 100, "ts_us": 2_000, "lsn": 1}, + ] + ) + _wait_for_completed_version(pipeline, _delta_version(loc)) + rows = list(pipeline.query(f"SELECT val FROM {TABLE} WHERE id = 0")) + assert rows and rows[0]["val"] == 200, ( + f"lsn asc must make the lsn=2 upsert the last writer; got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup() + + if __name__ == "__main__": pytest.main([__file__, "-v"])