ryerraguntla commented on code in PR #3568:
URL: https://github.com/apache/iggy/pull/3568#discussion_r3603747074
##########
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 |
Review Comment:
lib.rs:110-116 — README lists jsonbalias; code only acceptsjson_direct.
**Fix**: add jsonb to from_config or drop from README.
--
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]
