Hi Sai,
Thanks for the careful review, all points are valid.
1. Composite PKs — chunking splits on the first PK column only, which can skew
chunks (or make them impossible to bound) for multi-column keys. This is now
called out as a known limitation in the Table chunking section; range-based
chunking over a composite key is left as future work.
2. Column order — good catch, Collectors.toSet() produced a hash-ordered
projection while the row extraction is positional, so discovered-column reads
could disagree with the query. The implementation now preserves the database's
ordinal order (LinkedHashSet built from columnPosition), and an explicit user
projection is used verbatim in the given order. For consistency I also added
the missing withColumnNames(...) to DatabaseSplitterEnumeratorBuilder (the
table-mode builder already had it).
3. Annotations — the FLIP and code are now aligned. The dialect-extension
surface (ConnectionProvider, AbstractConnectionProvider,
ConnectionOptions/ConnectionOptionsBuilder, and the
Table/TableId/TableColumn/TableBounds domain types) is @PublicEvolving; pure
plumbing (ConnectionDataSource, StatementPreparer, ResultSetMapper) stays
@Internal and the FLIP now says so explicitly, class by class.
4. Postgres version — right, exported snapshots via pg_export_snapshot()
require ≥ 9.2; corrected throughout the FLIP.
5. HikariCP — deliberate for now: AbstractConnectionProvider uses HikariCP at
compile scope, unshaded, assuming it on the user classpath (same posture as
Flink's JDBC sink). The FLIP now states this explicitly and lists
shading/relocating as a follow-up before the first stable release.
Best
Joao Boto
On 2026/09/29 06:52:03 Krishna Sai wrote:
> Hi João,
>
> Thanks for the FLIP. +1 on the direction, this is a real gap today.
>
> I built PR #209 and ran the Postgres parts against PostgreSQL 16.2 with pgjdbc
> 42.7.13. The snapshot part works the way you describe it. What I
> checked by hand:
> the importer has to be REPEATABLE READ (a READ COMMITTED importer is
> rejected, and
> so is a SERIALIZABLE one, because the exporter is only RR), SET TRANSACTION
> SNAPSHOT has to be the first statement in the transaction, rows committed
> after
> the export are not visible, the snapshot id dies with the exporting
> transaction,
> and DDL under the snapshot fails loudly instead of returning wrong data.
>
> Five questions on the FLIP:
>
> 1. Composite primary keys. The FLIP says keyset chunking has no skew
> assumption.
> That holds for a single-column PK, but chunking only uses the first PK
> column, and
> the boundary query moves forward with col > lowerBound. So rows that share the
> same leading value can never end up in different splits. I tried it:
> events(tenant_id, id) with 3 tenants and 1000 rows each, chunkSize 10, gives 4
> splits of about 1000 rows. With one tenant and 3000 rows it gives one split
> for
> the whole table, logged as "Table ... has only one row, single split
> generated".
> Nothing is lost or duplicated, and a single-column PK chunks exactly as
> documented. But (tenant_id, id) is a common key shape, and today it quietly
> ends
> up with little or no parallelism. Can the FLIP say what users should
> expect here?
> Checking the leading column's cardinality may be better than only warning.
>
> 2. Column order when no columns are given. The names go through a
> Collectors.toSet(), so the SELECT list comes out in hash order. Reading is by
> position: RowResultExtractor, and RowDataResultExtractor does
> resultSet.getObject(pos + 1) against the declared RowType. For
> orders(id, customer_id, order_date, amount, status, notes) I get
> SELECT "order_date","amount","notes","id","customer_id","status" FROM
> ... I raise
> it here rather than on the PR because DatabaseSplitterEnumeratorBuilder has no
> column option, so database mode always hits this and users cannot avoid it.
> getTableColumns() already returns the columns in ORDINAL_POSITION order, so
> keeping that order would be enough.
>
> 3. Annotations. The FLIP says all new types are @PublicEvolving and lists
> AbstractConnectionProvider, ConnectionDataSource, StatementPreparer and
> ResultSetMapper. In the PR all four are @Internal. PostgresConnectionProvider
> is
> @PublicEvolving and extends the @Internal AbstractConnectionProvider,
> and that is
> the class a new dialect has to extend. Which of these do you want to commit
> to?
>
> 4. Postgres version. pg_export_snapshot() and SET TRANSACTION SNAPSHOT came in
> 9.2, not 9.1 ("Allow importing and exporting of transaction snapshots"
> in the 9.2
> release notes).
>
> 5. HikariCP. It is declared with no scope in flink-connector-jdbc-core and
> flink-connector-jdbc-postgres, so it resolves as compile, not runtime.
> And neither
> built jar contains any com.zaxxer classes, so nothing shades it today.
> It ends up
> as a new dependency on the user classpath. Since the dependency is part of the
> vote, I would reword that line.
>
> I will put two smaller implementation notes on the PR instead.
>
> Best,
> Sai Krishna
>
> On Tue, Sep 29, 2026 at 2:38 AM Joao Boto <[email protected]> wrote:
> >
> > Hi all,
> >
> > I'd like to start a discussion on FLINK-40108 [1], for which I've drafted a
> > FLIP introducing snapshot split enumerators for the JDBC connector's
> > DataStream source.
> >
> > Today, making a parallel JDBC read correct and balanced is entirely the
> > user's job: hand-written chunking SQL and parameter providers in the
> > DataStream API, and a numeric partition column with explicit bounds in the
> > Table/SQL API. There is also no cross-split consistency, each reader opens
> > its own transaction, so rows can move between chunk boundaries while a scan
> > is running, and enumeration progress is not checkpointed, so a failure
> > re-runs the scan from scratch.
> >
> > The FLIP proposes two SplitterEnumerator [2] implementations on JdbcSource,
> > plus the provider SPI they need:
> > 1. TableSplitterEnumerator — reads one table as primary-key ranged chunks
> > derived automatically from database metadata (keyset pagination over the PK
> > index, no user SQL, bounded enumerator memory).
> > 2. DatabaseSplitterEnumerator — discovers all tables of a schema (or an
> > allowlist) and emits chunked splits across them from a single source.
> > 3. Global snapshot consistency — all splits of one enumeration run read the
> > same database snapshot, implemented for PostgreSQL via pg_export_snapshot()
> > / SET TRANSACTION SNAPSHOT, with a dialect-neutral, capability-gated SPI in
> > core so other databases can follow. Unlike physical-id (ctid/ROWID)
> > partitioning, this remains consistent under concurrent DML.
> > 4. Resumable enumeration — per-table progress is checkpointed and
> > confirmed-split staging prevents split loss; a restored run resumes at its
> > cursor instead of re-enumerating, with at-least-once split delivery.
> >
> > Existing behavior is unchanged: everything is additive (new classes and
> > default interface methods), with a serializer version bump (v0 checkpoints
> > remain restorable) and one new dependency (HikariCP) that motivated this
> > FLIP per the dependency-change policy [3].
> >
> > Full design: [4]
> > Implementation (mostly complete, happy to align details with the
> > discussion): https://github.com/apache/flink-connector-jdbc/pull/209
> >
> > FLIP-602 [5] addresses partitioning strategy for the Table/SQL scan; this
> > FLIP is complementary and covers metadata-driven discovery, snapshot
> > consistency, and resumability at the DataStream level.
> >
> > Feedback and questions are very welcome.
> >
> > Thanks,
> > João Boto
> >
> > 1 https://issues.apache.org/jira/browse/FLINK-40108
> > 2 https://issues.apache.org/jira/browse/FLINK-38733
> > 3 https://github.com/apache/flink-connector-jdbc/commit/de12859
> > 4
> > https://cwiki.apache.org/confluence/spaces/FLINK/pages/451975192/FLIP-615+Snapshot+Splitters+for+Table+and+Database+Ingestion
> > 5
> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-602%3A+Extended+JDBC+Scan+Partitioning+Options+for+Efficient+Data+Ingestion
>
