[Test]: Qualify Flink compatibility with MatrixOne through the MySQL ecosystem

## Goal

Establish a reproducible compatibility test matrix for using MatrixOne through the Apache Flink / MySQL ecosystem. The initial list below is the **minimum required coverage**; the test owner is encouraged to extend it with additional versions, workloads, data types, failure modes, and long-running scenarios.

This is a test/qualification tracker, not a claim that every MySQL behavior is supported. Each concrete MatrixOne defect found during execution should be filed as a separate minimal bug issue and linked back here.

## Why this is needed

MatrixOne exposes the MySQL wire protocol and can be connected with MySQL Connector/J. Flink's JDBC connector can therefore use a `jdbc:mysql://:6001/` URL to access MatrixOne. However, real compatibility also depends on JDBC metadata, prepared statements, type conversion, Flink changelog/upsert semantics, retry behavior, checkpoint recovery, connection lifecycle, and the SQL emitted by the MySQL dialect.

There are also important direction-specific boundaries:

- **Supported path to qualify:** MySQL/TiDB/Kafka -> Flink/Flink CDC -> JDBC sink -> MatrixOne.
- **Supported path to qualify:** MatrixOne -> Flink JDBC batch scan or lookup source.
- **Negative/documentation boundary:** MatrixOne must not be assumed to be a Flink `mysql-cdc` source merely because it speaks the MySQL protocol. Flink MySQL CDC requires MySQL binlog/replication commands, which MatrixOne does not expose.

## Version matrix

Record all exact versions and commits in the result. At minimum, test:

1. MatrixOne current `main`.
2. The latest released MatrixOne version used by customers.
3. The Flink version currently used by MatrixOne's existing CDC regression.
4. One current stable Flink version, with the matching Flink JDBC connector.
5. MySQL Connector/J 8.x used by the selected Flink versions.
6. The selected Flink CDC connector version for MySQL CDC scenarios.
7. MySQL 8.0 as a control target/source wherever differential comparison is meaningful.

The test owner may add other commonly deployed Flink, Flink CDC, Connector/J, JDK, deployment, or MatrixOne versions.

## Required topologies

### T1: Flink JDBC -> MatrixOne

- Flink SQL/Table API append sink.
- Primary-key upsert sink.
- Batch scan from MatrixOne.
- MatrixOne as a JDBC lookup/dimension table.

### T2: MySQL 8.0 -> Flink MySQL CDC -> MatrixOne

- Initial snapshot followed by incremental INSERT/UPDATE/DELETE.
- Multiple tables and databases.
- DDL/schema-change behavior where supported by the selected pipeline.
- Restart from checkpoint/savepoint.

### T3: Kafka -> Flink -> MatrixOne

- Append-only JSON/CSV input.
- Keyed changelog/upsert input.
- Malformed records and sink-side constraint errors.

### T4: MatrixOne -> Flink JDBC

- Bounded scan, including partitioned parallel scan when applicable.
- Lookup joins with cache disabled and enabled.
- Concurrent reads while MatrixOne is being updated.

### T5: Negative boundary — MatrixOne as `mysql-cdc` source

- Confirm and document that the Flink MySQL CDC connector cannot use MatrixOne as a binlog source.
- Capture the exact failure and ensure documentation does not imply that MySQL protocol compatibility includes MySQL binlog/replication compatibility.
- Do not treat this expected limitation as a JDBC sink failure.

## Minimum test scenarios

### 1. Connectivity, authentication, and lifecycle

- Correct/incorrect host, port, user, password, database, and tenant-qualified account where applicable.
- Connection URL options, connection timeout, socket timeout, TLS on/off where supported.
- Repeated connect/close, stale connection validation, idle connection reuse, concurrent connection creation, and pool recovery after all existing connections are invalidated.
- Verify that a transient MatrixOne/network failure does not leave the client pool permanently at `total=0`.

### 2. JDBC and metadata compatibility

- `DatabaseMetaData`: catalogs/databases, tables, columns, primary keys, indexes, nullable flags, precision/scale, signed/unsigned metadata.
- Prepared statements with positional parameters, NULL parameters, batching, large values, and repeated execution.
- Identifier cases: upper/lower/mixed case, reserved words, spaces/special characters with quoting, qualified table names.
- Auto-commit, explicit commit/rollback, read-only setting, statement cancellation, query timeout, and fetch size where supported.
- Compare behavior and error classification with MySQL 8.0; explicitly record intentional MatrixOne deviations.

### 3. Type mapping and boundary values

Cover NULL, zero, ordinary values, min/max boundaries, overflow, invalid conversion, and round trips for:

- Signed/unsigned integer types, including `BIGINT UNSIGNED`.
- `DECIMAL`, including precision/scale supported by MatrixOne and values beyond MatrixOne's supported range.
- `FLOAT`/`DOUBLE`, NaN/Infinity handling where accepted, and comparison with an explicit tolerance.
- `BOOLEAN`/`TINYINT(1)` and `BIT` widths.
- `CHAR`, `VARCHAR`, `TEXT`, empty strings, trailing spaces, UTF-8/utf8mb4, emoji, and multi-byte boundary lengths.
- `BINARY`, `VARBINARY`, `BLOB`, zero bytes, and large payloads.
- `JSON`, including nested structures, numeric values, Unicode, and canonicalized comparison.
- `DATE`, `TIME`, `DATETIME`, `TIMESTAMP`, fractional seconds, timezone/session-timezone changes, daylight-saving boundaries where applicable, and zero/invalid dates.
- Defaults, generated values/auto increment, NULL/default interactions.
- MySQL-only or partially compatible types such as `ENUM`/`SET`: verify explicit mapping or an understandable failure rather than silent corruption.

### 4. Flink changelog and DML semantics

- Append INSERT.
- UPDATE and DELETE propagation.
- Primary-key upsert with repeated/replayed rows.
- Composite primary keys and primary-key value changes.
- Primary-key conflict versus secondary `UNIQUE` conflict. MatrixOne's current `ON DUPLICATE KEY UPDATE` behavior is known to differ from MySQL for secondary unique-key conflicts; classify the expected deviation explicitly.
- Flink records containing `INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, and `DELETE` row kinds.
- Out-of-order updates for the same key and multiple updates within one checkpoint.
- Constraint violations, partial batch failure, retry behavior, and whether successful rows are duplicated or lost after retry.

### 5. Correctness validation

- For small datasets, compare every row after canonicalizing intentionally different representations.
- For large datasets, compare row count, per-column NULL count, min/max, grouped distributions, and stable chunked checksums; drill down to row-level differences on failure.
- Validate after the initial snapshot, after each DML phase, and after recovery.
- Include empty tables, one-row tables, tables without primary keys, composite-primary-key tables, and concurrent multi-table workloads.
- No test may pass only because source and sink row counts match; value-level validation is required.

### 6. Checkpoint, retry, and failure recovery

Inject failures during both idle and active writes:

- Restart/kill a Flink TaskManager or job and recover from the last checkpoint/savepoint.
- Restart MatrixOne or temporarily interrupt the network.
- Expire/close JDBC connections and verify pool repopulation.
- Trigger sink timeout, retry, partial batch failure, and backpressure.
- Repeat failure injection during INSERT, UPDATE, and DELETE workloads.

After every recovery, verify no silent data loss, no unexpected duplicates, correct deletes, and continued checkpoint progress. Do not equate an enabled checkpoint with end-to-end exactly-once: document the actual guarantee. For primary-key sinks, verify replay idempotence through upsert semantics.

### 7. Scale, concurrency, and stability

- Suggested dataset levels: 1K, 1M, and 10M+ rows when resources allow.
- Multiple batch sizes, sink flush intervals, retry counts, and Flink parallelism levels.
- One large table, many small tables, multiple databases, and mixed concurrent INSERT/UPDATE/DELETE.
- Run a sustained mixed workload for at least 60 minutes; the test owner may extend the duration.
- Record throughput, end-to-end lag, checkpoint duration/failure count, retries, connection-pool state, MatrixOne CPU/memory, and relevant server/client errors.
- Include a resource-pressure run to detect OOM, stuck synchronization, connection loss, unbounded retry, or severe throughput collapse.

### 8. Documentation/configuration drift

- Verify that the current MatrixOne Flink examples work with the documented dependencies and DDL.
- Verify that Connector/J artifact/class names and Flink connector versions match the selected runtime.
- Document tested and unsupported directions clearly, especially JDBC compatibility versus MySQL CDC/binlog compatibility.

## Historical regression candidates

At minimum, revisit the scenarios represented by these closed issues:

- #23877 — Flink CDC issue summary: type precision, boundary values, deletes, consistency, and large-volume failures.
- #23306 — composite primary key upsert failure.
- #24030 — OOM during 10 GB Flink CDC synchronization.
- #25927 — JDBC pool lost all connections (`total=0`) and the consumer crashed.
- #26213 — consumer ran without synchronizing any rows to MatrixOne.
- #26740 — `int64` value rejected as out of range.

If an old issue still reproduces, open a new focused bug with current versions and link both the historical issue and this tracker.

## Deliverables

- Reproducible scripts/configuration and seed data, stored in or linked from the appropriate MatrixOne test repository.
- A versioned pass/fail/known-deviation compatibility matrix.
- Machine-readable test output plus retained Flink, connector, and MatrixOne logs for failed cases.
- A short performance/stability report for the scale matrix.
- Separate minimal bug issues for every confirmed MatrixOne defect, linked in this issue.
- Documentation follow-ups for expected limitations or configuration changes.

## Acceptance criteria

- [ ] All required topologies and minimum scenarios have an owner and recorded result.
- [ ] Exact MO/Flink/Flink CDC/JDBC driver/JDK versions and configuration are recorded.
- [ ] MySQL 8.0 control results distinguish connector defects from MatrixOne compatibility defects where applicable.
- [ ] Small datasets are validated row by row; large datasets use repeatable value-level validation, not count only.
- [ ] Failure recovery has been tested and shows no silent loss/corruption; actual delivery guarantees are documented.
- [ ] Known incompatibilities are explicitly classified and documented.
- [ ] Confirmed failures have focused reproduction issues linked below.
- [ ] Test scripts and the compatibility result can be rerun by another engineer.

## Result template

```text
MO version/commit:
Deployment mode:
Flink version:
Flink JDBC connector:
Flink CDC connector:
MySQL Connector/J:
JDK/OS:
Topology and test case:
Configuration/parallelism/batch size:
Expected result:
Actual result:
Validation method/result:
Checkpoint/recovery result:
Performance/resource observations:
Logs/artifacts:
Linked bug/documentation issue:
```

## Linked findings

- Add focused issues here as testing progresses.