fatmanverse opened a new issue, #12799:
URL: https://github.com/apache/seatunnel/issues/12799
### Search before asking
- [x] I had searched in the issues and found no similar issues. Related:
#12598 (source side, fixed by #12597).
### What happened
The Jdbc sink silently drops rows when its key has a NULL value.
When `primary_keys` is not configured and the upstream table has no primary
key, `JdbcSinkFactory#createSink` uses the first UNIQUE KEY of the upstream
table as the sink `primary_keys`. A unique key column can be nullable, and the
database allows any number of rows with NULL in it.
With a key, the sink writes through `BufferReducedBatchStatementExecutor`.
It buffers rows in a `LinkedHashMap` keyed by the extracted key row, so that
only the last change per key is written in a batch. `SeaTunnelRow#equals`
compares the fields with `Arrays.deepEquals`, so all rows whose key is NULL are
equal and collapse into one entry. Only the last of them is written; the others
are dropped before any SQL is executed. The job finishes successfully and the
read and write metrics count every row.
The database itself handles these rows correctly: NULL never equals NULL in
SQL, so `INSERT ... ON CONFLICT (code)` does not conflict on a NULL `code`, and
the target unique index accepts all of them. Only the in-memory buffer merges
them.
### How to reproduce (verified on `dev` 42d2aab)
MySQL source table without a primary key, with a nullable unique key, 30
rows of which 10 have `code = NULL`:
```sql
CREATE TABLE nullable_uk (
id INT NOT NULL,
code INT DEFAULT NULL,
name VARCHAR(32),
UNIQUE KEY uk_code (code)
);
```
```hocon
source {
Jdbc {
url = "jdbc:mysql://127.0.0.1:3306/st_src"
driver = "com.mysql.cj.jdbc.Driver"
table_list = [{ table_path = "st_src.nullable_uk" }]
# username / password
}
}
sink {
Jdbc {
url = "jdbc:postgresql://127.0.0.1:5432/postgres"
driver = "org.postgresql.Driver"
generate_sink_sql = true
database = "postgres"
table = "public.${table_name}"
schema_save_mode = "RECREATE_SCHEMA"
data_save_mode = "DROP_DATA"
# username / password, no primary_keys
}
}
```
Result: the job reports `SourceReceivedCount = 30` and `SinkWriteCount =
30`, but the target table has **21 rows**. Only 1 of the 10 `code IS NULL` rows
is left. The sink created `UNIQUE INDEX ... (code)` and writes with:
```
INSERT INTO "postgres"."public"."nullable_uk" ("id", "code", "name") VALUES
(?, ?, ?)
ON CONFLICT ("code") DO UPDATE SET "id"=EXCLUDED."id", "name"=EXCLUDED."name"
```
The same happens with an explicitly configured `primary_keys` that contains
a nullable column.
### Expected behavior
All 30 rows are written. Rows whose key contains NULL must not be merged
with each other in the sink buffer, which matches how the database treats NULL
in a unique key.
### SeaTunnel Version
dev (3.0.1-SNAPSHOT, 42d2aab)
### Zeta or Flink or Spark Version
Zeta
### Java or Scala Version
JDK 11
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
--
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]
