StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
    StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(executionEnvironment);
    streamTableEnvironment.executeSql("CREATE TABLE source (\n" +
        "  id INT,\n" +
        "  name STRING,\n" +
        "  proc_time AS PROCTIME()" +
        ") WITH (\n" +
        "   'connector' = 'starrocks',\n" +
        " 'password' = '',\n" +
        " 'table-name' = 'source', \n" +
        "  'scan-url' = '', \n" +
        " 'username'='root',\n" +
        "   'jdbc-url' = '',\n" +
        "   'database-name' = 'test'\n" +
        ");");
    streamTableEnvironment.executeSql("CREATE TABLE dim (\n" +
        "  id INT,\n" +
        "  name STRING\n" +
        ") WITH (\n" +
        "   'connector' = 'starrocks',\n" +
        " 'password' = '',\n" +
        " 'table-name' = 'dim', \n" +
        "  'scan-url' = '', \n" +
        " 'username'='root',\n" +
        "   'jdbc-url' = 'jdbc:mysql:///',\n" +
        "   'database-name' = 'test'\n" +
        ");");

    streamTableEnvironment.executeSql("CREATE TABLE sink (\n" +
        "  id INT,\n" +
        "  name STRING\n" +
        ") WITH (\n" +
        "   'connector' = 'starrocks',\n" +
        " 'password' = '',\n" +
        " 'table-name' = 'sink', \n" +
        "  'load-url' = '', \n" +
        " 'username'='root',\n" +
        "   'jdbc-url' = 'jdbc:mysql:///',\n" +
        "   'database-name' = 'test'\n" +
        ");");

    TableResult tableResult = streamTableEnvironment.executeSql(
        "INSERT INTO sink\n" +
            "SELECT source.id, dim.name FROM source\n" +
            "LEFT JOIN\n" +
            "dim for SYSTEM_TIME as OF source.proc_time\n" +
            "on dim.name = source.name;");
    tableResult.print();
}