|
20 | 20 | import org.apache.flink.api.common.JobID; |
21 | 21 | import org.apache.flink.api.common.RuntimeExecutionMode; |
22 | 22 | import org.apache.flink.api.common.time.Deadline; |
| 23 | +import org.apache.flink.api.common.typeinfo.Types; |
23 | 24 | import org.apache.flink.configuration.Configuration; |
24 | 25 | import org.apache.flink.configuration.RestartStrategyOptions; |
25 | 26 | import org.apache.flink.core.execution.JobClient; |
26 | 27 | import org.apache.flink.runtime.minicluster.RpcServiceSharing; |
27 | 28 | import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; |
| 29 | +import org.apache.flink.streaming.api.datastream.DataStream; |
28 | 30 | import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; |
| 31 | +import org.apache.flink.table.api.DataTypes; |
| 32 | +import org.apache.flink.table.api.Schema; |
29 | 33 | import org.apache.flink.table.api.TableResult; |
30 | 34 | import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; |
31 | 35 | import org.apache.flink.test.util.MiniClusterWithClientResource; |
| 36 | +import org.apache.flink.types.Row; |
| 37 | +import org.apache.flink.types.RowKind; |
32 | 38 | import org.apache.flink.util.StringUtils; |
33 | 39 |
|
34 | 40 | import com.fasterxml.jackson.databind.ObjectMapper; |
|
44 | 50 | import org.apache.doris.flink.table.DorisConfigOptions; |
45 | 51 | import org.apache.doris.flink.utils.MockSource; |
46 | 52 | import org.junit.Assert; |
| 53 | +import org.junit.Ignore; |
47 | 54 | import org.junit.Rule; |
48 | 55 | import org.junit.Test; |
49 | 56 | import org.junit.runner.RunWith; |
@@ -80,6 +87,7 @@ public class DorisSinkITCase extends AbstractITCaseService { |
80 | 87 | static final String TABLE_CSV_JM = "tbl_csv_jm"; |
81 | 88 | static final String TABLE_CSV_TM = "tbl_csv_tm"; |
82 | 89 | static final String TABLE_UNICODE_COLUMN = "tbl_unicode_column"; |
| 90 | + static final String TABLE_FLEXIBLE_COLUMN = "tbl_flexible_column"; |
83 | 91 |
|
84 | 92 | private final boolean batchMode; |
85 | 93 |
|
@@ -819,4 +827,112 @@ public void testSinkUnicodeColumn() throws Exception { |
819 | 827 | "select `名称`,`年龄` from %s.%s order by 1", DATABASE, TABLE_UNICODE_COLUMN); |
820 | 828 | ContainerUtils.checkResult(getDorisQueryConnection(), LOG, expected, query, 2); |
821 | 829 | } |
| 830 | + |
| 831 | + // todo: only test for doris3.1+ |
| 832 | + @Ignore |
| 833 | + @Test |
| 834 | + public void testFlexibleColumnUpdate() throws Exception { |
| 835 | + initializeFlexibleColumnTable(TABLE_FLEXIBLE_COLUMN); |
| 836 | + |
| 837 | + // pre-insert full rows with all columns into Doris |
| 838 | + ContainerUtils.executeSQLStatement( |
| 839 | + getDorisQueryConnection(), |
| 840 | + LOG, |
| 841 | + String.format( |
| 842 | + "INSERT INTO %s.%s VALUES(1,'doris',18,100),(2,'flink',20,200)", |
| 843 | + DATABASE, TABLE_FLEXIBLE_COLUMN)); |
| 844 | + |
| 845 | + String query = |
| 846 | + String.format( |
| 847 | + "select id,name,age,score from %s.%s order by 1", |
| 848 | + DATABASE, TABLE_FLEXIBLE_COLUMN); |
| 849 | + ContainerUtils.checkResult( |
| 850 | + getDorisQueryConnection(), |
| 851 | + LOG, |
| 852 | + Arrays.asList("1,doris,18,100", "2,flink,20,200"), |
| 853 | + query, |
| 854 | + 4); |
| 855 | + |
| 856 | + // Build a changelog source: |
| 857 | + // INSERT (1, 25) -> update age of id=1 to 25, name/score remain unchanged |
| 858 | + // DELETE (2, 20) -> delete the row with id=2 |
| 859 | + final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); |
| 860 | + env.setParallelism(DEFAULT_PARALLELISM); |
| 861 | + final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); |
| 862 | + |
| 863 | + DataStream<Row> changelogStream = |
| 864 | + env.fromCollection( |
| 865 | + Arrays.asList( |
| 866 | + Row.ofKind(RowKind.INSERT, 1, 25), |
| 867 | + Row.ofKind(RowKind.DELETE, 2, 20)), |
| 868 | + Types.ROW_NAMED(new String[] {"id", "age"}, Types.INT, Types.INT)); |
| 869 | + |
| 870 | + Schema sourceSchema = |
| 871 | + Schema.newBuilder() |
| 872 | + .column("id", DataTypes.INT().notNull()) |
| 873 | + .column("age", DataTypes.INT()) |
| 874 | + .primaryKey("id") |
| 875 | + .build(); |
| 876 | + tEnv.createTemporaryView( |
| 877 | + "source_view", tEnv.fromChangelogStream(changelogStream, sourceSchema)); |
| 878 | + |
| 879 | + // Doris sink declares only 'id' and 'age'; connector will NOT auto-build columns header |
| 880 | + // and NOT add hidden_columns header because unique_key_update_mode=UPDATE_FLEXIBLE_COLUMNS |
| 881 | + String sinkDDL = |
| 882 | + String.format( |
| 883 | + "CREATE TABLE doris_flex_sink (" |
| 884 | + + " id INT," |
| 885 | + + " age INT" |
| 886 | + + ") WITH (" |
| 887 | + + " 'connector' = '" |
| 888 | + + DorisConfigOptions.IDENTIFIER |
| 889 | + + "'," |
| 890 | + + " 'fenodes' = '%s'," |
| 891 | + + " 'table.identifier' = '%s'," |
| 892 | + + " 'username' = '%s'," |
| 893 | + + " 'password' = '%s'," |
| 894 | + + " 'sink.enable.batch-mode' = '%s'," |
| 895 | + + " 'sink.label-prefix' = '" |
| 896 | + + UUID.randomUUID() |
| 897 | + + "'," |
| 898 | + + " 'sink.properties.format' = 'json'," |
| 899 | + + " 'sink.properties.read_json_by_line' = 'true'," |
| 900 | + + " 'sink.properties.unique_key_update_mode' = 'UPDATE_FLEXIBLE_COLUMNS'" |
| 901 | + + ")", |
| 902 | + getFenodes(), |
| 903 | + DATABASE + "." + TABLE_FLEXIBLE_COLUMN, |
| 904 | + getDorisUsername(), |
| 905 | + getDorisPassword(), |
| 906 | + batchMode); |
| 907 | + tEnv.executeSql(sinkDDL); |
| 908 | + tEnv.executeSql("INSERT INTO doris_flex_sink SELECT id, age FROM source_view"); |
| 909 | + |
| 910 | + Thread.sleep(10000); |
| 911 | + // id=1: age updated to 25, name/score remain unchanged |
| 912 | + // id=2: deleted |
| 913 | + ContainerUtils.checkResult( |
| 914 | + getDorisQueryConnection(), LOG, Arrays.asList("1,doris,25,100"), query, 4); |
| 915 | + } |
| 916 | + |
| 917 | + private void initializeFlexibleColumnTable(String table) { |
| 918 | + ContainerUtils.executeSQLStatement( |
| 919 | + getDorisQueryConnection(), |
| 920 | + LOG, |
| 921 | + String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE), |
| 922 | + String.format("DROP TABLE IF EXISTS %s.%s", DATABASE, table), |
| 923 | + String.format( |
| 924 | + "CREATE TABLE %s.%s (\n" |
| 925 | + + " `id` int,\n" |
| 926 | + + " `name` varchar(256),\n" |
| 927 | + + " `age` int,\n" |
| 928 | + + " `score` int\n" |
| 929 | + + ") UNIQUE KEY(`id`)\n" |
| 930 | + + "DISTRIBUTED BY HASH(`id`) BUCKETS 1\n" |
| 931 | + + "PROPERTIES (\n" |
| 932 | + + " \"replication_num\" = \"1\",\n" |
| 933 | + + " \"enable_unique_key_merge_on_write\" = \"true\",\n" |
| 934 | + + " \"enable_unique_key_skip_bitmap_column\" = \"true\"\n" |
| 935 | + + ")", |
| 936 | + DATABASE, table)); |
| 937 | + } |
822 | 938 | } |
0 commit comments