mmodzelewski commented on code in PR #3640:
URL: https://github.com/apache/iggy/pull/3640#discussion_r3586375899
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value>
{
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String,
serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') &&
value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
Review Comment:
Quoted column names keep their literal quotes as JSON keys. `test_decoding`
applies `quote_identifier` to column names, so a mixed-case or keyword column
(`createdAt`, `user`) arrives as `"createdAt"[timestamp with time zone]:'...'`
and this parser emits the JSON key `\"createdAt\"` with the double-quote
characters included. Table names already go through `unquote_pg_identifier`;
column names should get the same treatment, otherwise downstream consumers
looking up the plain column name miss. Worth a unit test with a quoted column
name fixture.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -303,10 +301,6 @@ impl PostgresSource {
}
async fn setup_cdc(&self) -> Result<(), Error> {
Review Comment:
Not on this line, but related to CDC startup: `open()` validates `mode` but
not `cdc_backend`. A typo like `cdc_backend = "built-in"` passes startup, then
every `poll_cdc` call returns an error that the SDK loop logs and swallows,
leaving a Running connector that produces nothing. That is the same
silent-death failure mode this PR fixes for the slot mismatch. Validating the
backend in the `"cdc"` arm of `open()`, where the slot mismatch already fails
loudly, would close it.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value>
{
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String,
serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') &&
value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
+
+ let mut depth = 0;
+ let mut type_end = name_end;
+ loop {
+ match bytes.get(type_end) {
+ Some(b'[') => depth += 1,
+ Some(b']') => {
+ depth -= 1;
+ if depth == 0 {
+ break;
+ }
+ }
+ Some(_) => {}
+ None => return result,
+ }
+ type_end += 1;
+ }
+ if bytes.get(type_end + 1) != Some(&b':') {
+ break;
+ }
- let value = if let Ok(num) = cleaned_value.parse::<i64>() {
- serde_json::Value::Number(serde_json::Number::from(num))
- } else if let Ok(float) = cleaned_value.parse::<f64>() {
- serde_json::Value::Number(
-
serde_json::Number::from_f64(float).unwrap_or(serde_json::Number::from(0)),
- )
- } else if cleaned_value.eq_ignore_ascii_case("true") {
- serde_json::Value::Bool(true)
- } else if cleaned_value.eq_ignore_ascii_case("false") {
- serde_json::Value::Bool(false)
- } else {
- serde_json::Value::String(cleaned_value.to_string())
- };
+ let (value, next_pos) = parse_column_value(data, type_end + 2);
+ result.insert(column_name.to_string(), value);
+ pos = next_pos;
+ }
+
+ result
+}
+
+fn parse_column_value(data: &str, start: usize) -> (serde_json::Value, usize) {
Review Comment:
The `unchanged-toast-datum` sentinel leaks into records as a real value.
When an UPDATE does not touch a TOASTed column (large text/jsonb),
`test_decoding` emits the bare token `unchanged-toast-datum`, and the parser
stores `data["col"] = "unchanged-toast-datum"`, which is indistinguishable from
a legitimate string. Consumers will silently ingest the sentinel as the column
value. Suggest handling it explicitly in `parse_bare_scalar` (map to null, or
omit the key) and adding a fixture for it.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -728,99 +762,59 @@ impl PostgresSource {
&self,
data: &str,
capture_ops: &[&str],
+ captured_tables: Option<&[String]>,
) -> Option<DatabaseRecord> {
if data.starts_with("BEGIN") || data.starts_with("COMMIT") {
return None;
}
- if data.starts_with("INSERT:") && capture_ops.contains(&"INSERT") {
- return self.parse_insert_message(data);
- }
-
- if data.starts_with("UPDATE:") && capture_ops.contains(&"UPDATE") {
- return self.parse_update_message(data);
- }
+ let rest = data.strip_prefix("table ")?;
+ let (qualified_table, rest) = rest.split_once(": ")?;
+ let (operation, rest) = rest.split_once(": ")?;
- if data.starts_with("DELETE:") && capture_ops.contains(&"DELETE") {
- return self.parse_delete_message(data);
- }
-
- None
- }
-
- fn parse_insert_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ if !matches!(operation, "INSERT" | "UPDATE" | "DELETE") ||
!capture_ops.contains(&operation)
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "INSERT".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_update_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ let table_name = unquote_pg_identifier(
+ qualified_table
+ .rsplit('.')
+ .next()
+ .unwrap_or(qualified_table),
+ );
+ let unquoted_qualified_table = qualified_table
+ .split('.')
+ .map(unquote_pg_identifier)
+ .collect::<Vec<_>>()
+ .join(".");
+
+ // test_decoding has no server-side table filter, so Postgres already
+ // sent us every table's changes - config.tables scoping happens here.
+ if let Some(tables) = captured_tables
+ && !tables.iter().any(|t| {
+ if t.contains('.') {
+ t == &unquoted_qualified_table
+ } else {
+ t == &table_name
+ }
+ })
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "UPDATE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_delete_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
- {
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "DELETE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
- }
- None
+ let rest = rest
Review Comment:
The old-key columns are parsed past and discarded here, so
`DatabaseRecord.old_data` is always `None` even though the struct has the field
and the data is sitting in the row. For a PK-changing UPDATE (and for DELETE
under `REPLICA IDENTITY FULL`) consumers only see the new tuple and cannot
learn which row it replaced, which breaks e.g. keyed upsert/delete downstream.
Issue #3582 lists unpopulated `old_data` as well. Suggest running
`parse_record_columns` over the old-key section and populating `old_data`
instead of dropping it.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -41,9 +41,7 @@ primary_key_column = "id"
# Custom query (optional)
custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT
$limit"
-# CDC options (optional)
-enable_wal_cdc = false
-publication_name = "iggy_publication"
+# CDC options (only used when mode = "cdc")
replication_slot = "iggy_slot"
Review Comment:
Two operational hazards of the new slot handling deserve README coverage:
1. Slot sharing: `setup_cdc` accepts any pre-existing `test_decoding` slot,
so two CDC connectors against the same database with the default
`replication_slot = "iggy_slot"` silently share one slot.
`pg_logical_slot_get_changes` consumes on read, so each connector steals a
subset of the other's changes. The README should state that every CDC connector
needs a unique slot name.
2. WAL retention: a slot orphaned after a connector is decommissioned
retains WAL indefinitely (default `max_slot_wal_keep_size = -1`) until the disk
fills. The README should document dropping the slot (`SELECT
pg_drop_replication_slot('iggy_slot')`) as part of decommissioning.
##########
core/integration/tests/connectors/postgres/postgres_source_cdc.rs:
##########
@@ -0,0 +1,399 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{POLL_ATTEMPTS, POLL_INTERVAL_MS};
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{PostgresOps, PostgresSourceCdcFixture,
PostgresSourceOps};
+use iggy::prelude::IggyClient;
+use iggy_common::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use iggy_connector_sdk::api::{ConnectorStatus, SourceInfoResponse};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use reqwest::Client;
+use serde::Deserialize;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const API_KEY: &str = "test-api-key";
+const SOURCE_KEY: &str = "postgres";
+const DEFAULT_SLOT: &str = "iggy_slot";
+
+#[derive(Debug, Deserialize)]
+struct CdcRecord {
+ table_name: String,
+ operation_type: String,
+ data: serde_json::Value,
+}
+
+async fn poll_cdc_records(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ consumer_id: &Identifier,
+ want: usize,
+) -> Vec<CdcRecord> {
+ let mut received = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ stream_id,
+ topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(record) =
serde_json::from_slice::<CdcRecord>(&msg.payload) {
+ received.push(record);
+ }
+ }
+ if received.len() >= want {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ received
+}
+
+// End-to-end CDC coverage against a real wal_level=logical container:
+// INSERT, UPDATE, PK-changing UPDATE, DELETE, a rolled-back transaction
+// (must produce nothing), an untracked table (must be filtered out), a
+// 25-row sustained batch, and a final INSERT proving the slot/connector
+// are still healthy after all of the above.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_captures_insert_update_delete(
+ harness: &TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ const BATCH_SIZE: usize = 25;
+
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+ fixture.create_untracked_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_test_consumer".try_into().unwrap();
+
+ // INSERT: basic capture, full column fidelity
(id/name/count/amount/active).
+ let [msg] = create_test_messages(1).try_into().unwrap();
+ fixture
+ .insert_row(
+ &pool,
+ msg.id as i32,
+ &msg.name,
+ msg.count as i32,
+ msg.amount,
+ msg.active,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 INSERT message");
+ assert_eq!(received[0].table_name, fixture.table_name());
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+ assert_eq!(received[0].data["count"], serde_json::json!(msg.count));
+ assert_eq!(received[0].data["amount"].as_f64(), Some(msg.amount));
+ assert_eq!(received[0].data["active"], serde_json::json!(msg.active));
+
+ // UPDATE: non-key column change.
+ let updated_count = msg.count + 1;
+ fixture
+ .update_row(&pool, msg.id as i32, updated_count as i32)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["count"], serde_json::json!(updated_count));
+
+ // UPDATE: primary key change - test_decoding emits "old-key: ...
new-tuple: ...",
+ // the parser must report the new-tuple's id, not the old one.
+ let new_id = msg.id as i32 + 1000;
+ fixture
+ .update_primary_key(&pool, msg.id as i32, new_id)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message for PK change");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(
+ received[0].data["id"],
+ serde_json::json!(new_id),
+ "PK-changing UPDATE must report the new-tuple id, not the old-key one"
+ );
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+
+ // DELETE: only replica-identity columns (here, just id) are present.
+ fixture.delete_row(&pool, new_id).await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 DELETE message");
+ assert_eq!(received[0].operation_type, "DELETE");
+ assert_eq!(received[0].data["id"], serde_json::json!(new_id));
+
+ // Negative cases, verified together with the batch below: a rolled-back
+ // transaction must never reach test_decoding, and a table outside
+ // config.tables must be dropped by the client-side filter.
+ fixture
+ .insert_row_rolled_back(&pool, 9000, "should_not_appear")
+ .await;
+ fixture.insert_untracked_row(&pool, "not_configured").await;
+
+ // Sustained batch: BATCH_SIZE separate INSERT rows, asserting order and
+ // full data fidelity hold under a longer run, and that neither the
+ // rolled-back row nor the untracked-table row leaked into the output.
+ let batch = create_test_messages(BATCH_SIZE);
+ for row in &batch {
+ fixture
+ .insert_row(
+ &pool,
+ row.id as i32,
+ &row.name,
+ row.count as i32,
+ row.amount,
+ row.active,
+ row.timestamp,
+ )
+ .await;
+ }
+
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, BATCH_SIZE).await;
+ assert_eq!(
+ received.len(),
+ BATCH_SIZE,
+ "Expected exactly the {BATCH_SIZE} batch of INSERT rows - the
rolled-back row and the \
+ untracked-table row must not appear"
+ );
+ for (i, record) in received.iter().enumerate() {
+ assert_eq!(record.operation_type, "INSERT");
+ assert_eq!(record.table_name, fixture.table_name());
+ assert_eq!(
+ record.data["id"],
+ serde_json::json!(batch[i].id),
+ "Out-of-order or missing row at index {i}"
+ );
+ assert_eq!(record.data["name"], serde_json::json!(batch[i].name));
+ }
+
+ // Final sanity check: the slot/connector must still be healthy after the
+ // earlier PK-changing UPDATE and the sustained batch, not left in some
+ // degraded state that silently swallows further changes.
+ let final_id = new_id + BATCH_SIZE as i32 + 1;
+ fixture
+ .insert_row(
+ &pool,
+ final_id,
+ "after_pk_change",
+ 1,
+ 1.0,
+ true,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected the final INSERT after the PK change to still be captured"
+ );
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(final_id));
+ assert_eq!(
+ received[0].data["name"],
+ serde_json::json!("after_pk_change")
+ );
+
+ pool.close().await;
+}
+
+async fn wait_for_source_status(
+ http: &Client,
+ api_url: &str,
+ expected: ConnectorStatus,
+) -> SourceInfoResponse {
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(resp) = http
+ .get(format!("{api_url}/sources/{SOURCE_KEY}"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ && let Ok(info) = resp.json::<SourceInfoResponse>().await
+ && info.status == expected
+ {
+ return info;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ panic!("Source connector did not reach {expected:?} status in time");
+}
+
+// Two restart scenarios against one container, run in sequence:
+// 1. Simulates upgrading from the old broken version, which left a slot
+// behind created with the pgoutput plugin. setup_cdc must refuse to
+// silently reuse a mismatched slot (which used to fail forever on
+// every poll instead) - it should fail loudly at startup, and recover
+// cleanly once the operator drops the bad slot and restarts.
+// 2. Changes written while the connector is down (the slot retains WAL
+// regardless of consumer state) - not the at-least-once crash window
+// where the slot has already been consumed but send/state-persist
+// hasn't happened yet. That gap remains open until the slot-peek/LSN
+// work lands.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_recovers_from_slot_mismatch_and_restart(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_restart_consumer".try_into().unwrap();
+
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let http = Client::new();
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ sqlx::query("SELECT pg_drop_replication_slot($1)")
Review Comment:
Flake risk: this first `pg_drop_replication_slot` races the still-running
connector, which calls `pg_logical_slot_get_changes` every 50ms and briefly
holds the slot on each call. If the drop lands inside that window, Postgres
raises `ERROR 55006: replication slot "iggy_slot" is active for PID ...` and
the `expect` panics. Rough odds are a few percent per run, which is enough to
flake CI. Wrapping the drop in a short retry loop fixes it. The second drop
below is fine since the failed restart leaves no poller running.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
