@Test
    public void testSinkWithHashPrefixMatch() {
        StreamExecutionEnvironment executionEnv =
                StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv =
                StreamTableEnvironment.create(executionEnv);

        executionEnv.setParallelism(1);

        DataStream<Row> source =
                executionEnv.fromCollection(
                        Arrays.asList(
                                Row.of("1", "r12", 1.2, 1),
                                Row.of("2", "r22", 2.2, 2),
                                Row.of("3", "r32", 3.2, 3)));
        tableEnv.registerDataStream("source", source, "aaa, bbb, ccc, ddd");

        tableEnv.executeSql("CREATE TABLE sink (" +
                "    key STRING," +
                "    aaa STRING," +
                "    bbb DOUBLE," +
                "    ccc BIGINT," +
                "    PRIMARY KEY (`key`) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'redis-inlong'," +
                "  'sink.batch-size' = '1'," +
                "  'format' = 'csv'," +
                "  'data-type' = 'HASH'," +
                "  'redis-mode' = 'standalone'," +
                "  'host' = 'localhost'," +
                "  'port' = '" + redisPort + "'," +
                "  'maxIdle' = '8'," +
                "  'minIdle' = '1'," +
                "  'maxTotal' = '2'," +
                "  'timeout' = '2000'" +
                ")");
        Jedis jedis = new Jedis("localhost", redisPort);
        assertNull(jedis.get("1"));
        assertNull(jedis.get("2"));
        assertNull(jedis.get("3"));

        String query = "INSERT INTO sink SELECT * FROM source";
        tableEnv.executeSql(query);

        await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
            assertEquals("1.2,1", jedis.hget("1", "r12"));
            assertEquals("2.2,2", jedis.hget("2", "r22"));
            assertEquals("3.2,3", jedis.hget("3", "r32"));
        });

    }
hget only support get data from the given one key
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/test/java/org/apache/inlong/sort/redis/RedisTableTest.java