java/feathub-connectors/flink-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisLookupSourceITCase.java
/*
 * Copyright 2022 The FeatHub Authors
 *
 * Licensed 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
 *
 *     https://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.streaming.connectors.redis;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import org.apache.commons.collections.IteratorUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.Jedis;

import java.util.List;

import static org.assertj.core.api.Assertions.assertThat;

/** Tests redis lookup source. */
public class RedisLookupSourceITCase extends RedisITCaseBase {

    private StreamTableEnvironment tEnv;

    private Schema schema;

    private Jedis jedis;

    @Before
    public void setUp() {
        jedis = new Jedis(REDIS_HOST, redisPort.getPort());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        tEnv = StreamTableEnvironment.create(env);

        DataStream<Row> stream = env.fromElements(Row.of("0"), Row.of("1"), Row.of("2"));

        Schema scanSourceSchema =
                Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .columnByExpression("proctime", "PROCTIME()")
                        .build();
        tEnv.createTemporaryView("scan_source", stream, scanSourceSchema);

        schema =
                Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .column("f1", DataTypes.STRING())
                        .build();

        jedis.set("test_namespace:0:f1", "a");
        jedis.set("test_namespace:1:f1", "b");
    }

    @Test
    public void testRedisLookup() {
        TableDescriptor descriptor =
                TableDescriptor.forConnector("redis")
                        .schema(schema)
                        .option("host", REDIS_HOST)
                        .option("port", Integer.toString(redisPort.getPort()))
                        .option("dbNum", "0")
                        .option("namespace", "test_namespace")
                        .option("keyFields", "f0")
                        .build();

        tEnv.createTemporaryTable("lookup_source", descriptor);

        Row[] expectedResult =
                new Row[] {Row.of("0", "0", "a"), Row.of("1", "1", "b"), Row.of("2", "2", null)};

        List<Row> actualResult =
                IteratorUtils.toList(
                        tEnv.executeSql(
                                        "SELECT scan_source.f0, lookup_source.f0 AS f0_2, lookup_source.f1 "
                                                + "FROM scan_source LEFT JOIN lookup_source "
                                                + "FOR SYSTEM_TIME AS OF scan_source.proctime "
                                                + "ON scan_source.f0 = lookup_source.f0;")
                                .collect());
        assertThat(actualResult).containsExactly(expectedResult);
    }

    @After
    public void tearDown() {
        for (String key : jedis.keys("*")) {
            jedis.del(key);
        }

        if (jedis != null) {
            jedis.close();
        }
    }
}
