svetakvsundhar commented on code in PR #34393:
URL: https://github.com/apache/beam/pull/34393#discussion_r2009269646
##########
sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOTest.java:
##########
@@ -515,6 +515,110 @@ public void testIncompatibleSchemaThrowsError() {
assertThrows(PipelineExecutionException.class, () ->
pipeline.run().waitUntilFinish());
}
+ @Test
+ public void testReadRowsPartitions() {
+ PCollection<Row> rows =
+ pipeline.apply(
+ JdbcIO.readRowsWithPartitions()
+ .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
+ .withTable(READ_TABLE_NAME)
+ .withNumPartitions(1)
+ .withPartitionColumn("id")
+ .withLowerBound(0L)
+ .withUpperBound(1000L));
+ PAssert.thatSingleton(rows.apply("Count All",
Count.globally())).isEqualTo(1000L);
+ pipeline.run();
+ }
+
+ @Test
+ public void testReadRowsPartitionsWithExplicitSchema() {
+ Schema customSchema =
+ Schema.of(
+ Schema.Field.of("CUSTOMER_NAME",
Schema.FieldType.STRING).withNullable(true),
+ Schema.Field.of("CUSTOMER_ID",
Schema.FieldType.INT32).withNullable(true));
+ PCollection<Row> rows =
+ pipeline.apply(
+ JdbcIO.readRowsWithPartitions()
+ .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
+ .withTable(String.format("(select name,id from %s) as subq",
READ_TABLE_NAME))
+ .withNumPartitions(5)
+ .withPartitionColumn("id")
+ .withLowerBound(0L)
+ .withUpperBound(1000L)
+ .withRowOutput()
+ .withSchema(customSchema));
+ assertEquals(customSchema, rows.getSchema());
+ PAssert.thatSingleton(rows.apply("Count All",
Count.globally())).isEqualTo(1000L);
+ pipeline.run();
+ }
+
+ @Test
+ public void testReadRowsPartitionsBySubqery() {
+ PCollection<Row> rows =
+ pipeline.apply(
+ JdbcIO.readRowsWithPartitions()
+ .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
+ .withTable(String.format("(select * from %s) as subq",
READ_TABLE_NAME))
+ .withNumPartitions(10)
+ .withPartitionColumn("id")
+ .withLowerBound(0L)
+ .withUpperBound(1000L));
+ PAssert.thatSingleton(rows.apply("Count All",
Count.globally())).isEqualTo(1000L);
+ pipeline.run();
+ }
+
+ @Test
+ public void testReadRowsPartitionsIfNumPartitionsIsZero() {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("numPartitions can not be less than 1");
+ pipeline.apply(
+ JdbcIO.readRowsWithPartitions()
+ .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
+ .withTable(READ_TABLE_NAME)
+ .withNumPartitions(0)
+ .withPartitionColumn("id")
+ .withLowerBound(0L)
+ .withUpperBound(1000L));
+ pipeline.run();
+ }
+
+ @Test
+ public void testReadRowsPartitionsLowerBoundIsMoreThanUpperBound() {
Review Comment:
Can we also the following tests:
1) test a null partition column
2) null table name
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
