tusharagrahari commented on code in PR #3568: URL: https://github.com/apache/iggy/pull/3568#discussion_r4186485769
########## core/connectors/sources/mysql_source/src/lib.rs: ########## @@ -0,0 +1,3317 @@ +// 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 async_trait::async_trait; +use base64::Engine; +use chrono::{NaiveDate, NaiveDateTime, NaiveTime}; +use humantime::Duration as HumanDuration; +use iggy_common::{DateTime, Utc}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use sqlx::mysql::{MySqlDatabaseError, MySqlRow}; +use sqlx::{Column, MySql, Pool, Row, TypeInfo, mysql::MySqlPoolOptions}; +use std::cmp::Ordering as CmpOrdering; +use std::collections::{HashMap, HashSet}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +source_connector!(MySqlSource); + +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; + +/// Fixed namespace so a given (table, key) always hashes to the same message id +/// across restarts. This is what makes a replayed row idempotent for downstream +/// dedup, so the value must never change once connectors are in the field. +const MESSAGE_ID_NAMESPACE: Uuid = Uuid::from_u128(0x8f3b1e6a4c9d4f2a9b7c0d1e2f3a4b5c); + +/// A disabled table is skipped before anything is logged, so without a reminder the +/// only evidence a connector went dark is the single line written when it happened. +/// At the default 10s poll interval this repeats it every ten minutes. +const DISABLED_TABLES_LOG_INTERVAL: u64 = 60; + +/// How many consecutive polls a table may end in `CursorRegression` before the repeat +/// stops reading as transient. Nothing is published in the meantime, so past this it +/// is a stall the operator has to see. +const CURSOR_REGRESSION_ERROR_THRESHOLD: u32 = 3; + +#[derive(Debug)] +pub struct MySqlSource { + pub id: u32, + pool: Option<Pool<MySql>>, + config: MySqlSourceConfig, + state: Mutex<State>, + verbose: bool, + retry_delay: Duration, + poll_interval: Duration, + last_batch_full: AtomicBool, + poisoned_tables: Mutex<HashSet<String>>, + tracking_kinds: Mutex<HashMap<String, OffsetKind>>, + polls: AtomicU64, + cursor_regressions: Mutex<HashMap<String, u32>>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MySqlSourceConfig { + #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")] + pub connection_string: SecretString, + pub tables: Vec<String>, + pub poll_interval: Option<String>, + pub batch_size: Option<u32>, + pub tracking_column: Option<String>, + pub initial_offset: Option<String>, + pub max_connections: Option<u32>, + pub custom_query: Option<String>, + pub snake_case_columns: Option<bool>, + pub include_metadata: Option<bool>, + pub delete_after_read: Option<bool>, + pub processed_column: Option<String>, + pub primary_key_column: Option<String>, + pub payload_column: Option<String>, + pub payload_format: Option<String>, + pub verbose_logging: Option<bool>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PayloadFormat { + #[default] + Json, + Bytea, + Text, + JsonDirect, +} + +struct ProcessedRow { + message: ProducedMessage, + tracking_value: Option<String>, + row_pk: Option<PkValue>, +} + +/// Why a table's batch could not be produced. +enum FetchError { + Ordering(String), + /// The batch opened below the cursor it was told to resume from, so the query's + /// filter and the cursor disagree about what "past the offset" means. Retried + /// rather than latched, because unlike a within-batch decrease it depends on the + /// rows as much as on the query: a `custom_query` whose WHERE admits rows the + /// cursor does not gate (an OR branch is the usual one) only fails while such a + /// row is there, so it can clear without a config change. + CursorRegression(String), + MissingTrackingColumn(String), + UnusableTrackingValue(String), + StaleOffsetKind(String), + Other(Error), +} + +impl From<Error> for FetchError { + fn from(error: Error) -> Self { + FetchError::Other(error) + } +} + +/// How MySQL ordered a tracking column, which offsets alone cannot reveal: they are +/// carried as strings, so an `INT` 10 and a `VARCHAR` "10" are indistinguishable by +/// content while the server orders them differently. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum OffsetKind { + Numeric, + Lexical, + Unknown, +} + +/// Maps an `information_schema.columns.data_type` to the order MySQL sorts it in. +/// `decimal` belongs with the numbers even though `extract_column_value` renders it +/// as a string to preserve precision. +fn offset_kind_for_data_type(data_type: &str) -> OffsetKind { + const NUMERIC: &[&str] = &[ + "tinyint", + "smallint", + "mediumint", + "int", + "integer", + "bigint", + "decimal", + "numeric", + "float", + "double", + "real", + "bit", + "year", + ]; + if NUMERIC + .iter() + .any(|numeric| data_type.eq_ignore_ascii_case(numeric)) + { + OffsetKind::Numeric + } else { + OffsetKind::Lexical + } +} + +/// Whether MySQL orders the column by its collation. The other lexically compared +/// types are the temporal ones, which render fixed-width and so compare the same way +/// as text no matter what the collation is. +fn is_collation_ordered(data_type: &str) -> bool { + const TEXT: &[&str] = &[ + "char", + "varchar", + "tinytext", + "text", + "mediumtext", + "longtext", + ]; + TEXT.iter().any(|text| data_type.eq_ignore_ascii_case(text)) +} + +fn offset_kind_for_type_name(type_name: &str) -> OffsetKind { + match type_name { + "TINYINT" | "SMALLINT" | "MEDIUMINT" | "INT" | "BIGINT" | "TINYINT UNSIGNED" + | "SMALLINT UNSIGNED" | "MEDIUMINT UNSIGNED" | "INT UNSIGNED" | "BIGINT UNSIGNED" + | "FLOAT" | "DOUBLE" | "DECIMAL" | "BIT" | "YEAR" => OffsetKind::Numeric, + "DATE" | "TIME" | "DATETIME" | "TIMESTAMP" | "CHAR" | "VARCHAR" | "TINYTEXT" | "TEXT" + | "MEDIUMTEXT" | "LONGTEXT" => OffsetKind::Lexical, + _ => OffsetKind::Unknown, + } +} + +fn unusable_tracking_type(data_type: &str) -> Option<&'static str> { + const BINARY: &[&str] = &[ + "binary", + "varbinary", + "tinyblob", + "blob", + "mediumblob", + "longblob", + ]; + + if data_type.eq_ignore_ascii_case("json") { + return Some("it yields no ordered scalar cursor"); + } + // Judged on `data_type` rather than on the `tinyint(1)` display width that spells + // BOOLEAN, because that width is not reliably reported: MySQL 8.0.19 and later drop + // it for unsigned columns while MariaDB keeps it. + if data_type.eq_ignore_ascii_case("tinyint") { + return Some( + "at most 256 distinct values cannot be unique per row, and at display width 1 MySQL \ + spells the type BOOLEAN, which reaches the connector as true/false with no ordered \ + scalar form", + ); + } + if BINARY + .iter() + .any(|binary| data_type.eq_ignore_ascii_case(binary)) + { + return Some( + "binary values are carried as base64 text, which does not order the way the raw \ + bytes do, and that text is what gets compared against the column", + ); + } + if data_type.eq_ignore_ascii_case("enum") || data_type.eq_ignore_ascii_case("set") { + return Some( + "MySQL orders these by declaration index while only their labels reach the \ + connector, and a bounded set of labels cannot be unique per row either", + ); + } + None +} + +/// What a `custom_query` may begin with. `TABLE` is MySQL 8.0.19's table statement, +/// and `(` opens a parenthesized query such as `(SELECT ...) UNION (SELECT ...)`. +const READ_STATEMENT_STARTS: &[&str] = &["SELECT", "WITH", "TABLE", "("]; + +/// The token that decides which statement a query is, skipping leading comments in +/// all three MySQL spellings. Returns a single character when the statement does not +/// open with a word, so a parenthesized query is still recognisable. +fn leading_statement_token(query: &str) -> Option<String> { + let mut rest = query.trim_start(); + loop { + rest = if let Some(after) = rest.strip_prefix("/*") { + after.split_once("*/")?.1.trim_start() + } else if rest.starts_with("--") || rest.starts_with('#') { + rest.split_once('\n')?.1.trim_start() + } else { + break; + }; + } + + let word: String = rest + .chars() + .take_while(|c| c.is_ascii_alphanumeric() || *c == '_') + .collect(); + if word.is_empty() { + return rest.chars().next().map(String::from); + } + Some(word.to_uppercase()) +} + +/// The order the tracking column sorts in according to the result set that just +/// came back, or `None` when the query did not project the column under that name. +fn observed_tracking_kind(row: &MySqlRow, tracking_column: &str) -> Option<OffsetKind> { + row.columns() + .iter() + .find(|column| column.name().eq_ignore_ascii_case(tracking_column)) + .map(|column| offset_kind_for_type_name(column.type_info().name())) + .filter(|kind| *kind != OffsetKind::Unknown) +} + +/// `Clone` so the passing case, which is every batch, walks the names without collecting. +fn ensure_tracking_column_projected<'a>( + projected: impl IntoIterator<Item = &'a str> + Clone, + tracking_column: &str, +) -> Result<(), FetchError> { + if projected + .clone() + .into_iter() + .any(|column| column.eq_ignore_ascii_case(tracking_column)) + { + return Ok(()); + } + Err(FetchError::MissingTrackingColumn(format!( + "the result set has no column '{tracking_column}', it returned [{}]. Every row's cursor \ + comes from that column, so the same rows would be re-published on every poll while the \ + stored offset never moved.", + projected.into_iter().collect::<Vec<_>>().join(", ") + ))) +} + +fn unusable_tracking_value(tracking_column: &str, pk: &str) -> FetchError { + FetchError::UnusableTrackingValue(format!( + "row (pk {pk}) has no usable value for tracking_column '{tracking_column}': it is NULL, or \ + of a type with no ordered scalar form (tinyint(1), JSON). The cursor is taken from that \ + value, so publishing the row would advance nothing and the same rows would be re-read on \ + every poll." + )) +} + +/// Verifies rows arrive non-decreasing in the tracking column, which is what makes +/// "cursor = last row's value" correct. The built polling query orders ascending by +/// construction, so this only ever fires on a `custom_query`, where the operator +/// owns the `ORDER BY`. +struct OrderingGuard<'a> { + tracking_column: &'a str, + kind: OffsetKind, + last: Option<String>, + /// True while `last` still holds the cursor carried in from the previous poll + /// rather than a value from this batch. Distinguishes the two violations, and + /// keeps an empty batch from re-persisting the offset it started with. + at_cursor: bool, +} + +enum OrderingViolation { + WithinBatch(String), + BelowCursor(String), +} + +impl<'a> OrderingGuard<'a> { + fn new(tracking_column: &'a str, kind: OffsetKind, cursor: Option<String>) -> Self { + OrderingGuard { + tracking_column, + kind, + at_cursor: cursor.is_some(), + last: cursor, + } + } + + fn accept(&mut self, value: String) -> Result<(), OrderingViolation> { + if let Some(previous) = &self.last + && offset_decreased(previous, &value, self.kind) + { + let column = self.tracking_column; + return Err(if self.at_cursor { + OrderingViolation::BelowCursor(format!( + "a row arrived below the offset the poll resumed from by tracking_column \ + '{column}': stored offset '{previous}', first row '{value}'. Taking the \ + cursor from the last row would move it backwards. Filter the query on \ + '{column}' with $offset so a batch cannot reach behind its own cursor." + )) + } else { + OrderingViolation::WithinBatch(format!( + "rows arrived out of order by tracking_column '{column}': '{previous}' \ + preceded '{value}'. The cursor is taken from the last row, so it would \ + move backwards and rows would be skipped permanently. Order the query \ + ascending by '{column}'." + )) + }); + } + self.last = Some(value); + self.at_cursor = false; + Ok(()) + } + + /// `None` when no row was accepted, so an empty batch leaves the stored offset + /// untouched rather than rewriting the cursor it was seeded with. + fn into_cursor(self) -> Option<String> { + if self.at_cursor { None } else { self.last } + } +} + +struct TableBatch { + table: String, + messages: Vec<ProducedMessage>, + processed_ids: Vec<PkValue>, + max_offset: Option<String>, +} + +impl PayloadFormat { + fn from_config(s: Option<&str>) -> Self { + match s.map(|s| s.to_lowercase()).as_deref() { + Some("bytea") | Some("raw") => PayloadFormat::Bytea, + Some("text") => PayloadFormat::Text, + Some("json_direct") | Some("jsonb") => PayloadFormat::JsonDirect, + _ => PayloadFormat::Json, + } + } +} + +/// A primary-key value in a form the mark/delete `WHERE pk IN (...)` can bind +/// faithfully. `Text` covers ints/strings/dates/decimals (MySQL implicitly +/// converts the bound string to the column type). `Bytes` carries raw binary +/// (BINARY/VARBINARY/BLOB) so a UUID stored as bytes matches the row instead of +/// being compared as its base64 text, which never matches. +enum PkValue { + Text(String), + Bytes(Vec<u8>), +} + +impl PkValue { + fn as_key(&self) -> String { + match self { + PkValue::Text(text) => text.clone(), + PkValue::Bytes(bytes) => base64::engine::general_purpose::STANDARD.encode(bytes), + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct DatabaseRecord { + pub table_name: String, + pub operation_type: String, + pub timestamp: DateTime<Utc>, + pub data: serde_json::Value, + pub old_data: Option<serde_json::Value>, +} +#[derive(Clone, Copy)] +struct RowProcessingConfig<'a> { + table: &'a str, + tracking_column: &'a str, + pk_column: &'a str, + payload_format: PayloadFormat, + payload_col: &'a str, + snake_case_columns: bool, + include_metadata: bool, +} + +#[derive(Debug, Serialize, Deserialize)] +struct State { + tracking_offsets: HashMap<String, String>, + processed_rows: u64, +} + +const CONNECTOR_NAME: &str = "MySQL source"; + +#[async_trait] +impl Source for MySqlSource { + async fn open(&mut self) -> Result<(), Error> { + info!( + "Opening MySQL source connector with ID: {}, Tables: {:?}", + self.id, self.config.tables + ); + + // Every poll iterates `tables`, so an empty list makes the connector a + // silent no-op rather than an obvious misconfiguration. + if self.config.tables.is_empty() { + return Err(Error::InitError( + "tables must not be empty. Even when custom_query hardcodes the table, list it \ + here: the table name supplies the offset key, the deterministic message ID, \ + and the table_name metadata field" + .to_string(), + )); + } + + // A zero batch would make every fetch look like a full batch, so `poll` + // would drop its pacing sleep and spin on `LIMIT 0` queries forever. + if self.config.batch_size == Some(0) { + return Err(Error::InitError( + "batch_size must be greater than 0; omit it to use the default of 1000".to_string(), + )); + } + + // Only when the operator owns the ORDER BY. The built query sorts ascending by + // construction, so a one-row batch is safe there and legitimate for a slow table. + if self.config.batch_size == Some(1) + && self + .config + .custom_query + .as_deref() + .is_some_and(|query| query.contains("$offset")) + { + return Err(Error::InitError( + "batch_size must be greater than 1 when custom_query filters on $offset: a \ + one-row batch offers no pair to compare, so the ordering check reduces to \ + the row against the offset it resumed from, and a query ordered descending \ + opens above that offset on every poll. The cursor would walk to the table's \ + maximum and skip every row below it with nothing reporting it" + .to_string(), + )); + } + + if self + .config + .tracking_column + .as_deref() + .is_none_or(str::is_empty) + { + let guidance = if self.config.custom_query.is_some() { + "the connector does not parse custom_query, so it has to be told which projected \ + column supplies the cursor: name the one the query orders by and filters with \ + $offset" + } else { + "it has to be unique and monotonically increasing under the order MySQL sorts it \ + in." + }; + return Err(Error::InitError(format!( + "tracking_column must be set: every row's cursor is taken from it, and {guidance}." + ))); + } + + if let Some(ref col) = self.config.payload_column + && !col.is_empty() + && PayloadFormat::from_config(self.config.payload_format.as_deref()) + == PayloadFormat::Json + { + return Err(Error::InitError( + "payload_format must be 'bytea', 'text', or 'json_direct' when payload_column is set" + .to_string(), + )); + } + + // custom_query never changes after startup, so validate it once here + if let Some(ref query) = self.config.custom_query { + self.validate_custom_query(query)?; + + if query.contains("$offset") && self.config.initial_offset.is_none() { + let state = self.state.lock().await; + let missing: Vec<&str> = self + .config + .tables + .iter() + .filter(|t| !state.tracking_offsets.contains_key(*t)) + .map(String::as_str) + .collect(); + if !missing.is_empty() { + return Err(Error::InitError(format!( + "custom_query uses $offset but initial_offset is not set and no stored offset exists for table(s): {}", + missing.join(", ") + ))); + } + } + } + + if self.config.delete_after_read.unwrap_or(false) && self.config.processed_column.is_some() + { + warn!( + "both delete_after_read and processed_column are set; delete_after_read takes precedence, so rows are deleted and processed_column only acts as a poll filter" + ); + } + + self.connect().await?; + *self.tracking_kinds.get_mut() = self.resolve_tracking_columns().await?; + + info!( + "MySQL source connector with ID: {} opened successfully", + self.id + ); + Ok(()) + } + + async fn poll(&self) -> Result<ProducedMessages, Error> { + // Skip the pacing delay while draining a backlog + if !self.last_batch_full.load(Ordering::Relaxed) { + tokio::time::sleep(self.poll_interval).await; + } + + let messages = self.poll_tables().await?; + + let state = self.state.lock().await; + if self.verbose { + info!( + "MySQL source connector ID: {} produced {} messages. Total processed: {}", + self.id, + messages.len(), + state.processed_rows + ); + } else { + debug!( + "MySQL source connector ID: {} produced {} messages. Total processed: {}", + self.id, + messages.len(), + state.processed_rows + ); + } + + let schema = match self.payload_format() { + PayloadFormat::Bytea => Schema::Raw, + PayloadFormat::Text => Schema::Text, + PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json, + }; + + // Idle cycles produce no messages and leave offsets/processed_rows + // untouched, so return None to let the runtime skip the state fsync. + let persisted_state = if messages.is_empty() { Review Comment: Fixed. Every state handed to the runtime now goes through serialize_pending_state, which maps a failed serialization to Err(Error::Serialization) like postgres_source, so the poll fails instead of returning state: None. -- 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]
