@Test public void testJDBCReadProperties() { Map<String, String> properties = getBasicProperties(); properties.put("connector.read.partition.column", "aaa"); properties.put("connector.read.partition.lower-bound", "-10"); properties.put("connector.read.partition.upper-bound", "100"); properties.put("connector.read.partition.num", "10"); properties.put("connector.read.fetch-size", "20"); final StreamTableSource<?> actual = TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); final JDBCOptions options = JDBCOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); final JDBCReadOptions readOptions = JDBCReadOptions.builder() .setPartitionColumnName("aaa") .setPartitionLowerBound(-10) .setPartitionUpperBound(100) .setNumPartitions(10) .setFetchSize(20) .build(); final TableSchema schema = TableSchema.builder() .field("aaa", DataTypes.INT()) .field("bbb", DataTypes.STRING()) .field("ccc", DataTypes.DOUBLE()) .build(); final JDBCTableSource expected = JDBCTableSource.builder() .setOptions(options) .setReadOptions(readOptions) .setSchema(schema) .build(); assertEquals(expected, actual); }
@Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return JDBCTableSource.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setReadOptions(getJDBCReadOptions(descriptorProperties)) .setLookupOptions(getJDBCLookupOptions(descriptorProperties)) .setSchema(descriptorProperties.getTableSchema(SCHEMA)) .build(); }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return JDBCTableSource.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setReadOptions(getJDBCReadOptions(descriptorProperties)) .setLookupOptions(getJDBCLookupOptions(descriptorProperties)) .setSchema(descriptorProperties.getTableSchema(SCHEMA)) .build(); } }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return JDBCTableSource.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setReadOptions(getJDBCReadOptions(descriptorProperties)) .setLookupOptions(getJDBCLookupOptions(descriptorProperties)) .setSchema(descriptorProperties.getTableSchema(SCHEMA)) .build(); }  }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return JDBCTableSource.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setReadOptions(getJDBCReadOptions(descriptorProperties)) .setLookupOptions(getJDBCLookupOptions(descriptorProperties)) .setSchema(descriptorProperties.getTableSchema(SCHEMA)) .build(); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties); }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); return JDBCTableSource.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setReadOptions(getJDBCReadOptions(descriptorProperties)) .setLookupOptions(getJDBCLookupOptions(descriptorProperties)) .setSchema(descriptorProperties.getTableSchema(SCHEMA)) .build(); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties);  }