wuchong commented on a change in pull request #9029: [FLINK-13118][jdbc] Introduce JDBC table factory and bridge JDBC table source with streaming table source URL: https://github.com/apache/flink/pull/9029#discussion_r302360021
########## File path: flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCTableSourceSinkFactoryITCase.java ########## @@ -0,0 +1,268 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.java.io.jdbc; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.tuple.Tuple5; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.java.StreamTableEnvironment; +import org.apache.flink.table.descriptors.ConnectorDescriptorValidator; +import org.apache.flink.table.factories.StreamTableSinkFactory; +import org.apache.flink.table.factories.StreamTableSourceFactory; +import org.apache.flink.table.factories.TableFactoryService; +import org.apache.flink.table.runtime.utils.StreamITCase; +import org.apache.flink.table.sinks.StreamTableSink; +import org.apache.flink.table.sources.StreamTableSource; +import org.apache.flink.types.Row; + +import org.junit.Test; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Test for {@link JDBCTableSource} and {@link JDBCUpsertTableSink} created + * by {@link JDBCTableSourceSinkFactory}. + */ +public class JDBCTableSourceSinkFactoryITCase extends JDBCTestBase { + + @Test + public void testJDBCScanSource() throws Exception { + testJDBCScanSourceImpl(false); + } + + @Test + public void testJDBCScanSourceWithParallelism() throws Exception { + testJDBCScanSourceImpl(true); + } + + private void testJDBCScanSourceImpl(boolean useParallelism) throws Exception { + Map<String, String> properties = getBasicProperties(INPUT_TABLE); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + if (useParallelism) { + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_SCAN_PARTITION_COLUMN, "id"); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_SCAN_LOWER_BOUND, TEST_DATA[0].id.toString()); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_SCAN_UPPER_BOUND, TEST_DATA[TEST_DATA.length - 1].id.toString()); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_SCAN_NUM_PARTITIONS, "5"); + + env.setParallelism(TEST_DATA.length / 2 - 1); + } + + final StreamTableSource<?> source = TableFactoryService.find(StreamTableSourceFactory.class, properties) + .createStreamTableSource(properties); + + DataStream<Row> resultSet = tEnv.toAppendStream(tEnv.fromTableSource(source), Row.class); + resultSet.addSink(new StreamITCase.StringSink<>()); + env.execute(); + + List<String> expected = new ArrayList<>(); + for (TestEntry entry : TEST_DATA) { + expected.add(entry.id + "," + entry.title + "," + entry.author + "," + entry.price + "," + entry.qty); + } + StreamITCase.compareWithList(expected); + StreamITCase.clear(); + } + + @Test + public void testJDBCScanSourceWithProjection() throws Exception { + Map<String, String> properties = getBasicProperties(INPUT_TABLE); + + final StreamTableSource<?> source = TableFactoryService.find(StreamTableSourceFactory.class, properties) + .createStreamTableSource(properties); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + tEnv.registerTableSource("books", source); + + Table result = tEnv.sql("select id, title, price from books"); + DataStream<Row> resultSet = tEnv.toAppendStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink<>()); + env.execute(); + + List<String> expected = new ArrayList<>(); + for (TestEntry entry : TEST_DATA) { + expected.add(entry.id + "," + entry.title + "," + entry.price); + } + StreamITCase.compareWithList(expected); + StreamITCase.clear(); + } + + @Test + public void testJDBCLookupSource() throws Exception { + testJDBCLookupSourceImpl(false); + } + + @Test + public void testJDBCLookupSourceWithCache() throws Exception { + testJDBCLookupSourceImpl(true); + } + + private void testJDBCLookupSourceImpl(boolean useCache) throws Exception { + Map<String, String> properties = getBasicProperties(INPUT_TABLE); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + if (useCache) { + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_LOOKUP_CACHE_MAX_ENTRIES, "1000"); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_LOOKUP_CACHE_TTL, "1000s"); + } + + List<Tuple2<Integer, Integer>> lookupList = Arrays.asList( + new Tuple2<>(1001, 11), + new Tuple2<>(1002, 22), + new Tuple2<>(1003, 33), + new Tuple2<>(1005, 55), + new Tuple2<>(1008, 88), + new Tuple2<>(1005, 55), + new Tuple2<>(1003, 33), + new Tuple2<>(1002, 22), + new Tuple2<>(1001, 11) + ); + Table t = tEnv.fromDataStream(env.fromCollection(lookupList), "id, qty"); + tEnv.registerTable("T", t); + + final JDBCTableSource source = (JDBCTableSource) TableFactoryService.find(StreamTableSourceFactory.class, properties) + .createStreamTableSource(properties); + tEnv.registerFunction("jdbcLookup", source.getLookupFunction(new String[]{"id", "qty"})); + + Table result = tEnv.sql( + "select id, title, author, qty from T, lateral table(jdbcLookup(id, qty)) as S(l_id, title, author, price, l_qty)"); + DataStream<Row> resultSet = tEnv.toAppendStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink<>()); + env.execute(); + + List<String> expected = new ArrayList<>(); + for (Tuple2<Integer, Integer> tpl : lookupList) { + TestEntry entry = TEST_DATA[tpl.f0 - 1001]; + expected.add(entry.id + "," + entry.title + "," + entry.author + "," + entry.qty); + } + StreamITCase.compareWithList(expected); + StreamITCase.clear(); + } + + @Test + public void testJDBCAppendSink() throws Exception { + Map<String, String> properties = getBasicProperties(OUTPUT_TABLE); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + final StreamTableSink<?> sink = TableFactoryService.find(StreamTableSinkFactory.class, properties) + .createStreamTableSink(properties); + tEnv.registerTableSink("newbooks", sink); + + tEnv.sql("insert into newbooks values(1, 'title1', 'author1', cast(1.1 as double), 11)"); + tEnv.sql("insert into newbooks values(2, 'title2', 'author2', cast(2.2 as double), 22)"); + tEnv.sql("insert into newbooks values(3, 'title3', 'author3', cast(3.3 as double), 33)"); + env.execute(); + + JDBCUpsertOutputFormatTest.check(new Row[]{ + Row.of(1, "title1", "author1", 1.1, 11), + Row.of(2, "title2", "author2", 2.2, 22), + Row.of(3, "title3", "author3", 3.3, 33) + }, DB_URL, OUTPUT_TABLE, new String[]{"id", "title", "author", "price", "qty"}); + deleteFromDb(OUTPUT_TABLE); + } + + @Test + public void testJDBCUpsertSink() throws Exception { + Map<String, String> properties = getBasicProperties(OUTPUT_TABLE); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_WRITE_FLUSH_MAX_ENTRIES, "1"); + properties.put(JDBCTableSourceSinkFactory.CONNECTOR_WRITE_FLUSH_INTERVAL, "1s"); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + final StreamTableSink<?> sink = TableFactoryService.find(StreamTableSinkFactory.class, properties) + .createStreamTableSink(properties); + tEnv.registerTableSink("newbooks", sink); + + List<Tuple5<Integer, String, String, Double, Integer>> list = Arrays.asList( + new Tuple5<>(1, "title1", "author1", 1.1, 11), + new Tuple5<>(2, "title1", "author1", 2.2, 22), + new Tuple5<>(3, "title2", "author1", 3.3, 33), + new Tuple5<>(4, "title2", "author1", 4.4, 44), + new Tuple5<>(5, "title3", "author2", 1.1, 11), + new Tuple5<>(6, "title3", "author2", 2.2, 22), + new Tuple5<>(7, "title4", "author2", 3.3, 33), + new Tuple5<>(8, "title4", "author2", 4.4, 44) + ); + tEnv.registerTable( + "T", tEnv.fromDataStream(env.fromCollection(list), "id, title, author, price, qty")); + + // Multiple levels of aggregate functions will trigger the upsert + tEnv.sql( + "insert into newbooks select id, title, author, price, qty from(" + + "select T.id as id, T.title as title, T.author as author, T.price as price, T.qty as qty from(\n" + + " select author, min(T2.qty) as qty from(\n" + + " select T.title as title, T.author as author, T.qty as qty from(\n" + + " select title, author, min(T.price) as price from T group by title, author\n" + + " ) as T1, T where T.title = T1.title and T.author = T1.author and T.price = T1.price\n" + + " ) as T2 group by author\n" + + ") as T3, T where T.author = T3.author and T.qty = T3.qty" + + ") group by id, title, author, price, qty" + ); + env.execute(); + + JDBCUpsertOutputFormatTest.check(new Row[]{ + Row.of(1, "title1", "author1", 1.1, 11), + Row.of(5, "title3", "author2", 1.1, 11) + }, DB_URL, OUTPUT_TABLE, new String[]{"id", "title", "author", "price", "qty"}); + deleteFromDb(OUTPUT_TABLE); + } + + private Map<String, String> getBasicProperties(String tableName) { + Map<String, String> properties = new HashMap<>(); + + properties.put(ConnectorDescriptorValidator.CONNECTOR_TYPE, "jdbc"); Review comment: We suggest to hard code property keys to verify factory. See ElasticsearchTest. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
