feat(connectors): add MySQL source polling connector#3568
feat(connectors): add MySQL source polling connector#3568tusharagrahari wants to merge 7 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3568 +/- ##
=============================================
- Coverage 73.47% 41.85% -31.63%
Complexity 937 937
=============================================
Files 1286 1284 -2
Lines 141300 123445 -17855
Branches 117133 99321 -17812
=============================================
- Hits 103827 51670 -52157
- Misses 34229 68975 +34746
+ Partials 3244 2800 -444
🚀 New features to boost your workflow:
|
atharvalade
left a comment
atharvalade
left a comment
There was a problem hiding this comment.
Hey @tusharagrahari, thanks for the contribution and for taking the time to put this together.
A couple of process things before we dig into review:
-
Issue approval: I can see you created #3445 yourself, which is great. However, per our CONTRIBUTING.md, new functionality needs maintainer approval on the issue before coding begins (a
good-first-issuelabel or an explicit comment). Connectors are also listed as a high-risk area that needs a design discussion in the issue first. We typically ask first-time contributors to start with something labeledgood-first-issueto get familiar with the repo's workflow and CI, then move on to bigger features like this. -
Failing CI: Could you take a look and get those green?
Not saying this won't get merged, the implementation looks good to start with. Just want to make sure we follow the process so there are no surprises down the line.
github-actions
Bot
added
S-waiting-on-author
|
Hi @atharvalade, thanks for the detailed feedback! On the process side — I did reach out on the Discord community before starting, where I got clearance to proceed with the connector work. I understand that might not substitute for a formal maintainer approval comment on the issue itself, and I'll make sure to get that explicitly documented going forward before picking up anything in high-risk areas. On the CI failures — on it, will get those green shortly. Also working through the inline review comments now. Will push the fixes once the CI is sorted. |
1c4d3d9 to
dd4ea5c
Compare
|
/ready |
github-actions
Bot
added
S-waiting-on-review
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
a couple of notes that don't anchor to the diff:
core/connectors/sources/README.mdsource table doesn't list mysql_source yet - worth adding a row.- the two delivery-semantics comments below (delete/mark before send, offset advance on send failure) share one root cause - there is no send ack from the runtime back to the connector - and postgres_source behaves the same way. the sink side of the exact same gap is already tracked in #2927 (consume() return value discarded) and #2928 (offsets committed before sink processing), and #2940 discusses the broader replay-safe progress contract. the source side deserves a sibling issue rather than a fix in this PR.
github-actions
Bot
added
S-waiting-on-author
|
@tusharagrahari - Overall looks good. Outside of the mysql source connector, two places needs to be updated for the newer connectors..
|
|
/author |
6591d6b to
631cc2e
Compare
|
/ready |
github-actions
Bot
added
S-waiting-on-review
|
|
||
| ### Delivery semantics | ||
|
|
||
| Both `delete_after_read` and `processed_column` mutate MySQL during the poll, before the batch is actually sent to Iggy. If the connector crashes or the send fails in that window, those rows are gone or marked processed anyway — they won't be retried. So both options are at-most-once, not at-least-once. |
There was a problem hiding this comment.
those rows are gone or marked processed anyway - will it be deleted (gone) or marked? It should be either one in each pass, could this be confirmed and update the documentation
| | ------ | ----------- | ------ | ----------- | | ||
| | `bytea` / `raw` | `BLOB`, `BINARY`, `VARBINARY` | `raw` | Raw bytes passthrough | | ||
| | `text` | `TEXT`, `VARCHAR` | `text` | UTF-8 text | | ||
| | `json_direct` / `jsonb` | `JSON` | `json` | JSON object serialized to bytes | |
There was a problem hiding this comment.
lib.rs:110-116 — README lists jsonbalias; code only acceptsjson_direct. Fix: add jsonb to from_config or drop from README.
| Some(self.extract_payload_column(row, i, config.payload_format)?); | ||
| } | ||
| if name == config.tracking_column || name == config.pk_column { | ||
| let as_string = value_as_string(&extract_column_value(row, i)?); |
There was a problem hiding this comment.
value_as_string only String|Number; BOOL/NULL/JSON tracking/PK ⇒ no offset/PK for mark. Fix: canonical scalar encoding or open-time schema probe.
| let quoted_table = quote_qualified_identifier(table)?; | ||
| let quoted_pk = quote_identifier(pk_column)?; | ||
|
|
||
| let ids_list = ids |
There was a problem hiding this comment.
BINARY PK base64 in JSON path; IN clause uses quoted string literal ⇒ mark/delete miss. Fix: hex/binary bind or typed PK encoder.
| || self.config.processed_column.is_some(); | ||
|
|
||
| for mut batch in batches { | ||
| if mark_or_delete && !batch.messages.is_empty() && batch.processed_ids.is_empty() { |
There was a problem hiding this comment.
mark/delete configured but empty processed_ids: warn then still emit + advance offset. Fix: fail batch when mark configured without scalar PK.
| .unwrap_or_default(); | ||
| let offset = format_offset_value(&offset_value); | ||
| let now = Utc::now(); | ||
|
|
There was a problem hiding this comment.
lib.rs:399 vs lib.rs:424 - custom_query $table unquoted; built-in path quotes. Fix: quote_qualified_identifier on $table (postgres same pattern).
| ) -> Result<String, Error> { | ||
| let quoted_table = quote_qualified_identifier(table)?; | ||
| let quoted_tracking = quote_identifier(tracking_column)?; | ||
|
|
There was a problem hiding this comment.
748-791,427 — hot path: per-row serde_json Map + SELECT *. Fix: column projection + direct serialize (v1 acceptable, matches postgres shape).
| [dependencies] | ||
| async-trait = { workspace = true } | ||
| base64 = { workspace = true } | ||
| chrono = { workspace = true } |
There was a problem hiding this comment.
dashmap dep unused in lib.rs. Validate and Fix: remove dep + machete ignore.
|
No major issues. Most of them are suggestions/documentation. Should be good to merge in the next iteration |
|
/author |
github-actions
Bot
added
S-waiting-on-author
4 participants
Which issue does this PR address?
Relates to #3445
Rationale
MySQL is a widely used relational database with no existing Iggy source connector. This adds incremental table polling so users can stream MySQL rows into Iggy topics without CDC infrastructure.
What changed?
Before this PR, there was no way to source data from MySQL into Iggy.
This adds a mysql_source connector plugin supporting incremental table polling via a configurable tracking column. Rows are streamed as JSON, raw bytes, or text depending on the configured payload format. Post-processing options (delete after read, mark as processed) and custom SQL queries with parameter substitution are also supported. CDC (binlog-based) is out of scope and planned as a follow-up.
Local Execution
AI Usage