MartijnVisser commented on code in PR #220:
URL:
https://github.com/apache/flink-connector-jdbc/pull/220#discussion_r4079751511
##########
docs/data/jdbc.yml:
##########
@@ -16,7 +16,23 @@
# limitations under the License.
################################################################################
-version: 3.3.0-SNAPSHOT
+version: 4.0-SNAPSHOT
Review Comment:
Without `flink_compatibility` the download table renders "There is no
connector (yet) available" on stable docs. Also 4.1.0 is released, so
4.0-SNAPSHOT isn't right for main.
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -72,15 +81,79 @@ JdbcSource source = JdbcSourceBuilder.builder()
.setResultSetFetchSize(...)
.setConnectionProvider(...)
.build();
+```
+
+`setSplitter` describes the query and how it is divided into splits, and is
the only required way to
+define what the source reads. The older `setSql` /
`setJdbcParameterValuesProvider` pair is
+deprecated in favour of it; see [Deprecated query API](#deprecated-query-api).
Setting both a
+splitter and a query fails at `build()` time.
+### SplitterEnumerator
+
+A `SplitterEnumerator` produces the `JdbcSourceSplit`s that the readers
execute, and decides whether
+the source is bounded or continuously unbounded.
+
+#### PreparedSplitterEnumerator
+
+`PreparedSplitterEnumerator` builds bounded splits from a query template. With
no parameters the
+query becomes a single split:
+
+```java
+PreparedSplitterEnumerator.of("select * from books");
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+To read in parallel, provide a parameterized query template (i.e. a valid
+[JDBC prepared
statement](https://docs.oracle.com/en/java/javase/11/docs/api/java.sql/java/sql/PreparedStatement.html))
+together with the binding values. One split is generated per row of the
parameter array:
+
+```java
+String query = "select * from books where author = ?";
+Serializable[][] queryParameters = new String[2][1];
+queryParameters[0] = new String[]{"Kumar"};
+queryParameters[1] = new String[]{"Tan Ah Teck"};
+
+PreparedSplitterEnumerator.of(query, queryParameters);
```
-{{< /tab >}}
-{{< /tabs >}}
+
+For a numeric range there are convenience overloads that generate the
parameter pairs for you. The
+template must take a lower and an upper bound, and the range is divided either
into splits of a given
+size or into a given number of splits:
+
+```java
+// splits of at most 1000 values each
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 1000L);
+
+// exactly 10 splits
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 10);
+```
+
+Note that the two overloads differ only in whether the last argument is a
`long` (batch size) or an
+`int` (number of batches). The same can be expressed explicitly with
+`PreparedSplitterNumericParameters`:
+
+```java
+PreparedSplitterEnumerator.of(
+ "select * from books where id between ? and ?",
+ new PreparedSplitterNumericParameters(1L,
10_000L).withBatchSize(1000L));
Review Comment:
`PreparedSplitterNumericParameters` is `@Internal`, so I don't think we
should point users at it.
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -72,15 +81,79 @@ JdbcSource source = JdbcSourceBuilder.builder()
.setResultSetFetchSize(...)
.setConnectionProvider(...)
.build();
+```
+
+`setSplitter` describes the query and how it is divided into splits, and is
the only required way to
+define what the source reads. The older `setSql` /
`setJdbcParameterValuesProvider` pair is
+deprecated in favour of it; see [Deprecated query API](#deprecated-query-api).
Setting both a
+splitter and a query fails at `build()` time.
+### SplitterEnumerator
+
+A `SplitterEnumerator` produces the `JdbcSourceSplit`s that the readers
execute, and decides whether
+the source is bounded or continuously unbounded.
+
+#### PreparedSplitterEnumerator
+
+`PreparedSplitterEnumerator` builds bounded splits from a query template. With
no parameters the
+query becomes a single split:
Review Comment:
On main this gives zero splits, `SqlSplitterEnumerator` returns an empty
list for an empty parameter array. #243 fixes that, so this is correct once
that is merged.
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -198,59 +218,86 @@ public class JdbcSourceExample {
JdbcSource<Book> jdbcSource =
JdbcSource.<Book>builder()
.setTypeInformation(TypeInformation.of(Book.class))
- .setSql("select * from testing_table where id < ?")
- .setDBUrl(...)
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from books where id < ?",
new Serializable[][] {{1001L}}))
+ .setDBUrl(...)
.setDriverName(...)
.setResultExtractor(resultSet ->
new Book(
resultSet.getLong("id"),
resultSet.getString("title")))
.build();
env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(),
"TestSource")
- .addSink(new DiscardingSink());
+ .sinkTo(new DiscardingSink<>());
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+### Deprecated query API
+
+Before `SplitterEnumerator` was introduced, the query was defined with
`setSql` and the splits with a
+`JdbcParameterValuesProvider`. Both methods are deprecated and are equivalent
to a
+`PreparedSplitterEnumerator`:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where id >= ? and id <= ?")
+ .setJdbcParameterValuesProvider(
+ new JdbcGenericParameterValuesProvider(
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+
+// current
+JdbcSource.<TestEntry>builder()
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from testing_table where id >= ? and id <=
?",
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+```
+
+On this deprecated path, continuous unbounded reads are configured with
Review Comment:
#242 also deprecates `setContinuousUnBoundingSettings` and
`setOptionalSqlSplitEnumeratorState`. And `SlideTimingSplitterEnumerator` has
no discovery interval, so it isn't a like-for-like replacement. Please reflect
both.
##########
docs/content/docs/connectors/table/jdbc.md:
##########
@@ -447,9 +471,21 @@ As there is no standard syntax for upsert, the following
table describes the dat
WHEN NOT MATCHED THEN INSERT (..) <br>
VALUES (..)</td>
</tr>
+ <tr>
+ <td>CrateDB</td>
+ <td>INSERT .. ON CONFLICT .. DO UPDATE SET ..</td>
+ </tr>
+ <tr>
+ <td>OceanBase</td>
+ <td>Depends on <code>'compatible-mode'</code>: the MySQL grammar
in <code>'mysql'</code>
+ mode, the Oracle grammar in <code>'oracle'</code> mode.</td>
+ </tr>
</tbody>
</table>
+Trino does not support upsert. A Trino table always operates in append mode,
even when a primary key
Review Comment:
Trino has no primary keys, but the sink uses the one in the Flink DDL. With
that set, `JdbcOutputFormatBuilder` does an exists check plus UPDATE or INSERT
and emits deletes, so it's not append mode. Whether that works depends on the
Trino connector supporting UPDATE and DELETE, and no enabled test covers it.
##########
docs/content/docs/connectors/table/jdbc.md:
##########
@@ -475,14 +511,39 @@ Please refer to [Dependencies](#dependencies) section for
how to setup a JDBC co
The JDBC catalog supports the following options:
- `name`: required, name of the catalog.
-- `default-database`: required, default database to connect to.
- `username`: required, username of database account.
- `password`: required, password of the account.
-- `base-url`: required (should not contain the database name)
+- `base-url`: required
- for Postgres Catalog this should be `"jdbc:postgresql://<ip>:<port>"`
- for MySQL Catalog this should be `"jdbc:mysql://<ip>:<port>"`
- for OceanBase Catalog this should be `jdbc:oceanbase://<ip>:<port>`
-- `compatible-mode`: optional, the compatible mode of database.
+- `default-database`: optional, default database to connect to.
+- `compatible-mode`: optional, the compatible mode of database. Only OceanBase
supports this option,
+ see [Connector Options](#connector-options).
+
+The database the catalog connects to has to be identifiable from `base-url`,
`default-database`, or
+both. Any one of the following is valid:
+
+- `default-database` is set and `base-url` carries no database name, for
example
+ `'base-url' = 'jdbc:postgresql://localhost:5432'` with `'default-database' =
'mydb'`
+- `base-url` carries the database name and `default-database` is omitted, for
example
+ `'base-url' = 'jdbc:postgresql://localhost:5432/mydb'`
+- both are set and the database names match
+
+If the two disagree, or if neither carries a database name, catalog creation
fails.
+
+`base-url` may also carry arbitrary driver options as query parameters, which
are passed through to
Review Comment:
When `base-url` has a query string, the database name has to be in
`base-url` itself, `AbstractJdbcCatalog` rejects it otherwise. Can you add that?
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -198,59 +218,86 @@ public class JdbcSourceExample {
JdbcSource<Book> jdbcSource =
JdbcSource.<Book>builder()
.setTypeInformation(TypeInformation.of(Book.class))
- .setSql("select * from testing_table where id < ?")
- .setDBUrl(...)
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from books where id < ?",
new Serializable[][] {{1001L}}))
+ .setDBUrl(...)
.setDriverName(...)
.setResultExtractor(resultSet ->
new Book(
resultSet.getLong("id"),
resultSet.getString("title")))
.build();
env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(),
"TestSource")
- .addSink(new DiscardingSink());
+ .sinkTo(new DiscardingSink<>());
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+### Deprecated query API
+
+Before `SplitterEnumerator` was introduced, the query was defined with
`setSql` and the splits with a
+`JdbcParameterValuesProvider`. Both methods are deprecated and are equivalent
to a
+`PreparedSplitterEnumerator`:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where id >= ? and id <= ?")
+ .setJdbcParameterValuesProvider(
+ new JdbcGenericParameterValuesProvider(
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+
+// current
+JdbcSource.<TestEntry>builder()
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from testing_table where id >= ? and id <=
?",
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+```
+
+On this deprecated path, continuous unbounded reads are configured with
+`setContinuousUnBoundingSettings` and a `JdbcSlideTimingParameterProvider`,
which must be set
+together:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where ts >= ? and ts < ?")
+ .setContinuousUnBoundingSettings(
+ new ContinuousUnBoundingSettings(Duration.ofMillis(10L),
Duration.ofSeconds(1L)))
+ .setJdbcParameterValuesProvider(
+ new JdbcSlideTimingParameterProvider(0L, 1000L, 1000L, 100L))
Review Comment:
This throws, the provider requires `startMills > 0`. It came over from the
old page, but can you fix it while you're at it?
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -198,59 +218,86 @@ public class JdbcSourceExample {
JdbcSource<Book> jdbcSource =
JdbcSource.<Book>builder()
.setTypeInformation(TypeInformation.of(Book.class))
- .setSql("select * from testing_table where id < ?")
- .setDBUrl(...)
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from books where id < ?",
new Serializable[][] {{1001L}}))
+ .setDBUrl(...)
.setDriverName(...)
.setResultExtractor(resultSet ->
new Book(
resultSet.getLong("id"),
resultSet.getString("title")))
.build();
env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(),
"TestSource")
- .addSink(new DiscardingSink());
+ .sinkTo(new DiscardingSink<>());
Review Comment:
Can you add the import? Only the `sink.v2` `DiscardingSink` compiles with
`sinkTo`.
##########
docs/content/docs/connectors/table/jdbc.md:
##########
@@ -38,8 +38,29 @@ The JDBC sink operate in upsert mode for exchange
UPDATE/DELETE messages with th
Dependencies
------------
+Since version 4.0 the JDBC connector is no longer published as a single
artifact. It is split into
+`flink-connector-jdbc-core`, which contains the shared runtime, and one
artifact per supported
+database, which contains the dialect and the catalog for that database. You
need the artifact of the
+database you are connecting to; it pulls in `flink-connector-jdbc-core`
transitively.
+
+| Database | Connector artifact |
+|:-----------|:----------------------------------|
+| MySQL | `flink-connector-jdbc-mysql` |
+| Oracle | `flink-connector-jdbc-oracle` |
+| PostgreSQL | `flink-connector-jdbc-postgres` |
+| SQL Server | `flink-connector-jdbc-sqlserver` |
+| CrateDB | `flink-connector-jdbc-cratedb` |
+| Db2 | `flink-connector-jdbc-db2` |
+| Trino | `flink-connector-jdbc-trino` |
+| OceanBase | `flink-connector-jdbc-oceanbase` |
+| Derby | `flink-connector-jdbc-core` |
+
{{< sql_connector_download_table "jdbc" >}}
+None of these artifacts is a self-contained SQL uber jar. When using the SQL
Client or a session
+cluster, put both `flink-connector-jdbc-core` and the artifact for your
database into the `lib/`
Review Comment:
CrateDB also needs the postgres jar, and OceanBase needs both the mysql and
oracle jars, since those dialects build on them. Might make sense to say so
here.
##########
docs/content/docs/connectors/datastream/jdbc.md:
##########
@@ -266,30 +313,20 @@ It then repeatedly calls a user-provided function to
update that prepared statem
(preparedStatement, someRecord) -> { ... update here the preparedStatement
with values from someRecord ... }
```
+The two are passed together to `withQueryStatement`. A `JdbcQueryStatement`
can also be supplied
+directly if the query itself has to be derived from the record.
Review Comment:
`query()` doesn't take the record and the writer prepares the statement
once, so the query can't be derived from the record.
##########
docs/content/docs/connectors/table/jdbc.md:
##########
@@ -1030,4 +1105,7 @@ Flink supports connect to several databases which uses
dialect like MySQL, Oracl
</tbody>
</table>
+PostgreSQL `JSON` and `JSONB` columns are read and written as their textual
representation. `UUID`
Review Comment:
Reads work, but writes go through `setString`, which Postgres rejects for
`jsonb` and `uuid` columns unless the URL sets `stringtype=unspecified`. I'd
either document that or drop "written".
--
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]
