@Test public void testJDBCWithFilter() { Map<String, String> properties = getBasicProperties(); properties.put("connector.driver", "org.apache.derby.jdbc.EmbeddedDriver"); properties.put("connector.username", "user"); properties.put("connector.password", "pass"); final TableSource<?> actual = ((JDBCTableSource) TableFactoryService .find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties)) .projectFields(new int[] {0, 2}); Map<String, DataType> projectedFields = ((FieldsDataType) actual.getProducedDataType()).getFieldDataTypes(); assertEquals(projectedFields.get("aaa"), DataTypes.INT()); assertNull(projectedFields.get("bbb")); assertEquals(projectedFields.get("ccc"), DataTypes.DOUBLE()); }
@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);  }