leonardBang commented on code in PR #153:
URL:
https://github.com/apache/flink-connector-jdbc/pull/153#discussion_r1944348435
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcFullTest.java:
##########
@@ -106,66 +92,66 @@ void testEnrichedClassCastException() {
}
}
- private void runTest(boolean exploitParallelism) throws Exception {
- ExecutionEnvironment environment =
ExecutionEnvironment.getExecutionEnvironment();
- JdbcInputFormat.JdbcInputFormatBuilder inputBuilder =
- JdbcInputFormat.buildJdbcInputFormat()
- .setDrivername(getMetadata().getDriverClass())
- .setDBUrl(getMetadata().getJdbcUrl())
- .setQuery(SELECT_ALL_BOOKS)
- .setRowTypeInfo(ROW_TYPE_INFO);
-
- if (exploitParallelism) {
- final int fetchSize = 1;
- final long min = TEST_DATA[0].id;
- final long max = TEST_DATA[TEST_DATA.length - fetchSize].id;
- // use a "splittable" query to exploit parallelism
- inputBuilder =
- inputBuilder
- .setQuery(SELECT_ALL_BOOKS_SPLIT_BY_ID)
- .setParametersProvider(
- new
JdbcNumericBetweenParametersProvider(min, max)
- .ofBatchSize(fetchSize));
- }
- DataSet<Row> source = environment.createInput(inputBuilder.finish());
-
- // NOTE: in this case (with Derby driver) setSqlTypes could be
skipped, but
- // some databases don't null values correctly when no column type was
specified
- // in PreparedStatement.setObject (see its javadoc for more details)
- org.apache.flink.connector.jdbc.JdbcConnectionOptions
connectionOptions =
- new org.apache.flink.connector.jdbc.JdbcConnectionOptions
- .JdbcConnectionOptionsBuilder()
- .withUrl(getMetadata().getJdbcUrl())
- .withDriverName(getMetadata().getDriverClass())
- .build();
-
- JdbcOutputFormat<Row, Row, ?> jdbcOutputFormat =
- new JdbcOutputFormat<>(
- new SimpleJdbcConnectionProvider(connectionOptions),
- JdbcExecutionOptions.defaults(),
- () ->
- createSimpleRowExecutor(
- String.format(INSERT_TEMPLATE,
OUTPUT_TABLE),
- new int[] {
- Types.INTEGER,
- Types.VARCHAR,
- Types.VARCHAR,
- Types.DOUBLE,
- Types.INTEGER
- }));
- source.output(new TestOutputFormat(jdbcOutputFormat));
- environment.execute();
-
- try (Connection dbConn =
DriverManager.getConnection(getMetadata().getJdbcUrl());
- PreparedStatement statement =
dbConn.prepareStatement(SELECT_ALL_NEWBOOKS);
- ResultSet resultSet = statement.executeQuery()) {
- int count = 0;
- while (resultSet.next()) {
- count++;
- }
- assertThat(count).isEqualTo(TEST_DATA.length);
- }
- }
+ // private void runTest(boolean exploitParallelism) throws Exception {
Review Comment:
？
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
