ryerraguntla commented on code in PR #3568:
URL: https://github.com/apache/iggy/pull/3568#discussion_r3603736966
##########
core/connectors/sources/mysql_source/README.md:
##########
@@ -0,0 +1,424 @@
+# MySQL Source Connector
+
+The MySQL source connector fetches data from MySQL databases and streams it to
Iggy topics. It supports incremental table polling with flexible payload
extraction.
+
+> **Note:** Only polling mode is available. Binlog CDC support is planned for
a future release.
+
+## Features
+
+- **Table Polling**: Incrementally fetch data from MySQL tables using a
tracking column
+- **Flexible Payload Extraction**: Extract BLOB, TEXT, or JSON columns
directly as payload
+- **Custom Queries**: Use custom SQL queries with parameter substitution
+- **Delete After Read**: Automatically delete rows after processing
+- **Mark as Processed**: Mark rows as processed using a boolean column
+- **Multiple Tables**: Monitor multiple tables simultaneously
+- **Batch Processing**: Fetch data in configurable batch sizes
+- **Offset Tracking**: Resume incremental polling from the last processed
tracking value (see [Delivery semantics](#delivery-semantics))
+
+## Configuration
+
+```toml
+[plugin_config]
+connection_string = "mysql://user:pass@localhost:3306/database"
+tables = ["users", "orders"]
+poll_interval = "1s"
+batch_size = 1000
+tracking_column = "id"
+initial_offset = "0"
+max_connections = 10
+snake_case_columns = false
+include_metadata = true
+
+# Payload extraction (optional)
+payload_column = "payload"
+payload_format = "bytea"
+
+# Delete/mark processed (optional)
+delete_after_read = false
+processed_column = "is_processed"
+primary_key_column = "id"
+
+# Custom query (optional)
+custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT
$limit"
+```
+
+## Configuration Options
+
+| Option | Type | Default | Description |
+| ------ | ---- | ------- | ----------- |
+| `connection_string` | string | required | MySQL connection string
(`mysql://user:pass@host:3306/db`) |
+| `tables` | array | required | List of tables to monitor |
+| `poll_interval` | string | `10s` | How often to poll (e.g., `1s`, `5m`) |
+| `batch_size` | u32 | `1000` | Max rows per poll |
+| `tracking_column` | string | `id` | Column for incremental polling; must be
unique and monotonically increasing (see [Tracking Column
Requirements](#tracking-column-requirements)) |
+| `initial_offset` | string | none | Starting value for tracking column |
+| `max_connections` | u32 | `10` | Max database connections |
+| `snake_case_columns` | bool | `false` | Convert column names to snake_case |
+| `include_metadata` | bool | `true` | Wrap results with metadata envelope |
+| `payload_column` | string | none | Column to extract directly as payload |
+| `payload_format` | string | `json` (invalid if `payload_column` is set —
`payload_format` is required in that case) | Format of payload_column: `bytea`,
`text`, or `json_direct` |
+| `delete_after_read` | bool | `false` | Delete rows after reading |
+| `processed_column` | string | none | Boolean column to mark as processed |
+| `primary_key_column` | string | tracking_column | PK for delete/mark
operations |
+| `custom_query` | string | none | Custom SQL with parameter substitution |
+| `verbose_logging` | bool | `false` | Log at info level instead of debug |
+| `max_retries` | u32 | `3` | Retries after the initial attempt for transient
errors (3 = 4 total attempts) |
+| `retry_delay` | string | `1s` | Base delay between retries (e.g., `500ms`,
`2s`) |
+
+## Tracking Column Requirements
+
+Polling is **insert-only**. Each poll runs roughly `SELECT ... WHERE
tracking_column > last_offset ORDER BY tracking_column ASC LIMIT batch_size`
and stores the largest tracking value it saw as the next offset. For this to be
lossless the **`tracking_column` must be unique and monotonically increasing**
— an auto-increment primary key is the canonical choice.
+
+The requirement is not cosmetic: if more than `batch_size` rows share the same
tracking value, only the first `batch_size` are read and the next poll's `>`
filter skips the rest of that value permanently. The same skip applies to
`processed_column` and `delete_after_read`, because the tracking filter runs
before those.
+
+For this reason a mutable, low-resolution column such as `updated_at` is
**not** a safe tracking column: many rows commonly share the same second, and
rows updated in place are never re-read. Capturing updates to existing rows is
out of scope for polling.
+
+## Output Modes
+
+### JSON Mode (Default)
+
+When `payload_column` is not set, each row is wrapped in a `DatabaseRecord`
JSON structure:
+
+```json
+{
+ "table_name": "users",
+ "operation_type": "SELECT",
+ "timestamp": "2024-01-15T10:30:00Z",
+ "data": {
+ "id": 123,
+ "name": "John Doe",
+ "email": "[email protected]"
+ },
+ "old_data": null
+}
+```
+
+The stream config should use `schema = "json"`.
+
+When `include_metadata = false`, the envelope is omitted and the row columns
are serialized directly:
+
+```json
+{
+ "id": 123,
+ "name": "John Doe",
+ "email": "[email protected]"
+}
+```
+
+### Payload Column Extraction
+
+When `payload_column` is set, the connector extracts that column directly as
the Iggy message payload. The `payload_format` option determines how the column
is read:
+
+| Format | Column Type | Schema | Description |
+| ------ | ----------- | ------ | ----------- |
+| `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 |
+
+## Payload Format Examples
+
+### BLOB (Raw Bytes)
+
+Extract raw bytes from a BLOB column:
+
+```sql
+CREATE TABLE message_queue (
+ id INT AUTO_INCREMENT PRIMARY KEY,
+ payload BLOB NOT NULL
+);
+```
+
+```toml
+[[streams]]
+stream = "messages"
+topic = "queue"
+schema = "raw"
+batch_length = 100
+
+[plugin_config]
+tables = ["message_queue"]
+tracking_column = "id"
+payload_column = "payload"
+payload_format = "bytea"
+```
+
+### TEXT
+
+Extract text from a TEXT column:
+
+```sql
+CREATE TABLE logs (
+ id INT AUTO_INCREMENT PRIMARY KEY,
+ message TEXT NOT NULL
+);
+```
+
+```toml
+[[streams]]
+stream = "logs"
+topic = "app_logs"
+schema = "text"
+batch_length = 100
+
+[plugin_config]
+tables = ["logs"]
+tracking_column = "id"
+payload_column = "message"
+payload_format = "text"
+```
+
+### JSON (Direct)
+
+Extract a JSON column directly as JSON payload (without `DatabaseRecord`
wrapper):
+
+```sql
+CREATE TABLE events (
+ id INT AUTO_INCREMENT PRIMARY KEY,
+ data JSON NOT NULL
+);
+```
+
+```toml
+[[streams]]
+stream = "events"
+topic = "user_events"
+schema = "json"
+batch_length = 100
+
+[plugin_config]
+tables = ["events"]
+tracking_column = "id"
+payload_column = "data"
+payload_format = "json_direct"
+```
+
+## Custom Query Parameters
+
+When using `custom_query`, these placeholders are available:
+
+| Placeholder | Replaced With |
+| ----------- | ------------- |
+| `$table` | Current table name |
+| `$offset` | Last processed offset (or `initial_offset`) |
+| `$limit` | `batch_size` value |
+| `$now` | Current UTC timestamp (RFC3339) |
+| `$now_unix` | Current Unix timestamp (seconds) |
+
+If the query uses `$offset`, it must be ordered by the tracking column in
ascending order (`ORDER BY <tracking_column>` — MySQL sorts ascending by
default, so `ASC` doesn't need to be spelled out). The connector takes the
tracking value of the *last row returned* as the next `$offset`; without
ascending order, that's not guaranteed to be the max, so rows can be skipped or
re-emitted on the next poll.
+
+The column you order by is your effective tracking column, so the same
[Tracking Column Requirements](#tracking-column-requirements) apply: it must be
unique and monotonically increasing, or rows sharing a value can be skipped
across a batch boundary.
+
+Example:
+
+```sql
+SELECT * FROM $table
+WHERE id > $offset
+ AND (scheduled_at IS NULL OR scheduled_at <= '$now')
+ORDER BY id
+LIMIT $limit
+```
+
+## Delete After Read / Mark as Processed
+
+### Delete After Read
+
+Deletes rows from the source table after successful processing. At-most-once —
see [Delivery semantics](#delivery-semantics) below.
+
+```toml
+[plugin_config]
+delete_after_read = true
+primary_key_column = "id"
+```
+
+### Mark as Processed
+
+Updates a boolean column instead of deleting. At-most-once — see [Delivery
semantics](#delivery-semantics) below.
+
+```toml
+[plugin_config]
+processed_column = "is_processed"
+primary_key_column = "id"
+```
+
+Your table needs the boolean column:
+
+```sql
+ALTER TABLE users ADD COLUMN is_processed BOOLEAN DEFAULT false;
+```
+
+When `processed_column` is set, the connector automatically adds a `WHERE
is_processed = FALSE` filter to the polling query, so only unprocessed rows are
fetched.
+
+### 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.
Review 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
--
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]
