JdbcIO { public static <T> Read<T> read() { return new AutoValue_JdbcIO_Read.Builder<T>() .setFetchSize(DEFAULT_FETCH_SIZE) .setOutputParallelization(true) .build(); } private  JdbcIO(); static Read<T> read(); @Experimental(Kind.SCHEMAS) static ReadRows readRows(); static ReadAll<ParameterT, OutputT> readAll(); static Write<T> write(); static WriteVoid<T> writeVoid();  }
@Test public void testRead() { PCollection<TestRow> rows = pipeline.apply( JdbcIO.<TestRow>read() .withFetchSize(12) .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION) .withQuery("select name,id from " + READ_TABLE_NAME) .withRowMapper(new JdbcTestHelper.CreateTestRowOfNameAndId()) .withCoder(SerializableCoder.of(TestRow.class))); PAssert.thatSingleton(rows.apply("Count All", Count.globally())) .isEqualTo((long) EXPECTED_ROW_COUNT); Iterable<TestRow> expectedValues = TestRow.getExpectedValues(0, EXPECTED_ROW_COUNT); PAssert.that(rows).containsInAnyOrder(expectedValues); pipeline.run(); }
@Test public void testReadWithSingleStringParameter() { PCollection<TestRow> rows = pipeline.apply( JdbcIO.<TestRow>read() .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION) .withQuery(String.format("select name,id from %s where name = ?", READ_TABLE_NAME)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1))) .withRowMapper(new JdbcTestHelper.CreateTestRowOfNameAndId()) .withCoder(SerializableCoder.of(TestRow.class))); PAssert.thatSingleton(rows.apply("Count All", Count.globally())).isEqualTo(1L); Iterable<TestRow> expectedValues = Collections.singletonList(TestRow.fromSeed(1)); PAssert.that(rows).containsInAnyOrder(expectedValues); pipeline.run(); }
@Test public void testReadWithSchema() { SerializableFunction<Void, DataSource> dataSourceProvider = ignored -> DATA_SOURCE; JdbcIO.RowMapper<RowWithSchema> rowMapper = rs -> new RowWithSchema(rs.getString("NAME"), rs.getInt("ID")); pipeline.getSchemaRegistry().registerJavaBean(RowWithSchema.class); PCollection<RowWithSchema> rows = pipeline.apply( JdbcIO.<RowWithSchema>read() .withDataSourceProviderFn(dataSourceProvider) .withQuery(String.format("select name,id from %s where name = ?", READ_TABLE_NAME)) .withRowMapper(rowMapper) .withCoder(SerializableCoder.of(RowWithSchema.class)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1)))); Schema expectedSchema = Schema.of( Schema.Field.of("name", Schema.FieldType.STRING), Schema.Field.of("id", Schema.FieldType.INT32)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("name", "id")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues("Testval1", 1).build())); pipeline.run(); }
@Test public void testRead() { PCollection<TestRow> rows = pipeline.apply( JdbcIO.<TestRow>read() .withFetchSize(12) .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(dataSource)) .withQuery("select name,id from " + readTableName) .withRowMapper(new JdbcTestHelper.CreateTestRowOfNameAndId()) .withCoder(SerializableCoder.of(TestRow.class))); PAssert.thatSingleton(rows.apply("Count All", Count.globally())) .isEqualTo((long) EXPECTED_ROW_COUNT); Iterable<TestRow> expectedValues = TestRow.getExpectedValues(0, EXPECTED_ROW_COUNT); PAssert.that(rows).containsInAnyOrder(expectedValues); pipeline.run(); }
@Test public void testReadWithSingleStringParameter() { PCollection<TestRow> rows = pipeline.apply( JdbcIO.<TestRow>read() .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(dataSource)) .withQuery(String.format("select name,id from %s where name = ?", readTableName)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1))) .withRowMapper(new JdbcTestHelper.CreateTestRowOfNameAndId()) .withCoder(SerializableCoder.of(TestRow.class))); PAssert.thatSingleton(rows.apply("Count All", Count.globally())).isEqualTo(1L); Iterable<TestRow> expectedValues = Collections.singletonList(TestRow.fromSeed(1)); PAssert.that(rows).containsInAnyOrder(expectedValues); pipeline.run(); }
@Test public void testReadWithSchema() { SerializableFunction<Void, DataSource> dataSourceProvider = ignored -> dataSource; JdbcIO.RowMapper<RowWithSchema> rowMapper = rs -> new RowWithSchema(rs.getString("NAME"), rs.getInt("ID")); pipeline.getSchemaRegistry().registerJavaBean(RowWithSchema.class); PCollection<RowWithSchema> rows = pipeline.apply( JdbcIO.<RowWithSchema>read() .withDataSourceProviderFn(dataSourceProvider) .withQuery(String.format("select name,id from %s where name = ?", readTableName)) .withRowMapper(rowMapper) .withCoder(SerializableCoder.of(RowWithSchema.class)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1)))); Schema expectedSchema = Schema.of( Schema.Field.of("name", Schema.FieldType.STRING), Schema.Field.of("id", Schema.FieldType.INT32)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("name", "id")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues("Testval1", 1).build())); pipeline.run(); }