### Flink SQL Redis Query and Datagen Join Example

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Shows how to query data from a Redis table using Flink SQL with a specific scan key option and how to perform a left join with a datagen table. This example highlights the connector's ability to integrate with other Flink connectors and perform complex queries.

```sql
-- 查询  
  insert into redis_table select name,age + 1 from redis_table /*+ options('scan.key'='test') */
  
create table gen_table (age int , level int, proctime as procTime()) with ('connector'='datagen','fields.age.kind' = 'sequence', 
 'fields.age.start' = '2','fields.age.end' = '2','fields.level.kind' = 'sequence','fields.level.start' = '10','fields.level.end' = '10'); 

-- 关联查询 
insert into redis_table select 'test', j.age + 10 from gen_table s left join redis_table  for system_time as of proctime as j
on j.name = 'test'
```

--------------------------------

### Flink SQL Setup and Redis Sink

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Demonstrates setting up a Flink streaming environment and defining a Redis sink table using SQL DDL. It then inserts data into the Redis sink. Requires Flink and the Redis connector.

```java
EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);

String ddl = "create table sink_redis(username VARCHAR, passport VARCHAR) with ( 'connector'='redis', " +
              "'cluster-nodes'='10.11.80.147:7000,10.11.80.147:7001','redis- mode'='cluster','password'='******','command'='set')" ;

tEnv.executeSql(ddl);
String sql = " insert into sink_redis select * from (values ('test', 'test11'))";
TableResult tableResult = tEnv.executeSql(sql);
tableResult.getJobClient().get()
.getJobExecutionResult()
.get();
```

--------------------------------

### Flink DataStream API: Redis HSET Sink

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Provides a Java example demonstrating how to use the Flink DataStream API to write data to Redis using the HSET command. It configures Redis connection parameters, creates a `RedisSinkMapper` for HSET, sets up a `StreamExecutionEnvironment`, prepares data as `BinaryRowData`, and adds a `RedisSinkFunction` to the data stream. This example is suitable for streaming data processing pipelines.

```java
        Configuration configuration = new Configuration();
        configuration.setString(REDIS_MODE, REDIS_SINGLE);
        configuration.setString(REDIS_COMMAND, RedisCommand.HSET.name());
        configuration.setInteger(TTL, 10);

        RedisSinkMapper redisMapper = new RowRedisSinkMapper(RedisCommand.HSET, configuration);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        BinaryRowData binaryRowData = new BinaryRowData(3);
        BinaryRowWriter binaryRowWriter = new BinaryRowWriter(binaryRowData);
        binaryRowWriter.writeString(0, StringData.fromString("tom"));
        binaryRowWriter.writeString(1, StringData.fromString("math"));
        binaryRowWriter.writeString(2, StringData.fromString("152"));

        DataStream<BinaryRowData> dataStream = env.fromElements(binaryRowData, binaryRowData);

        List<String> columnNames = Arrays.asList("name", "subject", "scope");
        List<DataType> columnDataTypes =
                Arrays.asList(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING());
        ResolvedSchema resolvedSchema = ResolvedSchema.physical(columnNames, columnDataTypes);

        FlinkConfigBase conf = 
                new FlinkSingleConfig.Builder()
                        .setHost(REDIS_HOST)
                        .setPort(REDIS_PORT)
                        .setPassword(REDIS_PASSWORD)
                        .build();

        RedisSinkFunction redisSinkFunction =
                new RedisSinkFunction<>(conf, redisMapper, resolvedSchema, configuration);

        dataStream.addSink(redisSinkFunction).setParallelism(1);
        env.execute("RedisSinkTest");

```

--------------------------------

### Flink DataStream API: Redis SET for Cluster Write

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Illustrates a Java example for writing data to a Redis cluster using the Flink DataStream API with the SET command. This snippet focuses on setting up the Redis connection for a cluster environment and preparing data for insertion. It's a foundational example for streaming data to Redis clusters.

```java
/*
 * Example code path: src/test/java/org.apache.flink.streaming.connectors.redis.table.SQLInsertTest.java
 * Equivalent Redis command: set test test11
 */
// This is a placeholder for the actual Java code. 
// The provided text only indicates the file path and the conceptual Redis command.
// A full implementation would involve similar setup as the DataStream HSET example,
// but configured for a Redis cluster and potentially using a different RedisCommand enum value or mapper.

```

--------------------------------

### Flink Redis Connector: SET Command Setup

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Configures a Flink sink table to use the Redis SET command. It defines the table schema and specifies how to construct the Redis key and value from table columns, using '\01' as a separator.

```SQL
create table sink_redis(username VARCHAR, passport VARCHAR)  with ('command'='set') 
#key: username , value: username\01passport.
```

--------------------------------

### Flink Redis Sink Setup with SQL

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Demonstrates how to configure and use the Flink Redis Connector to create a Redis sink table using SQL DDL. It specifies connection details like cluster nodes, mode, password, and command, then inserts data into the sink.

```Java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);

String ddl = "create table sink_redis(username VARCHAR, passport VARCHAR) with ( 'connector'='redis', " +
              "'cluster-nodes'='10.11.80.147:7000,10.11.80.147:7001','redis- mode'='cluster','password'='******','command'='set')" ;

tEnv.executeSql(ddl);
String sql = " insert into sink_redis select * from (values ('test', 'test11'))";
TableResult tableResult = tEnv.executeSql(sql);
tableResult.getJobClient().get()
.getJobExecutionResult()
.get();
```

--------------------------------

### Flink Redis Connector: HSET Command Setup

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Configures a Flink sink table to use the Redis HSET command. It maps table columns to the Redis key, field, and value, utilizing '\01' as a delimiter for constructing the value.

```SQL
create table sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR)  with ('command'='hset') 
key: name, field:subject, value: name\01subject\01score.
```

--------------------------------

### Flink DataStream API Redis HSET Example

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Shows how to use the Flink DataStream API to write data to Redis using the HSET command. This involves configuring the Redis connection, creating a RedisSinkMapper, and setting up the RedisSinkFunction with data streams.

```Java
Configuration configuration = new Configuration();
configuration.setString(REDIS_MODE, REDIS_CLUSTER);
configuration.setString(REDIS_COMMAND, RedisCommand.HSET.name());

RedisSinkMapper redisMapper = (RedisSinkMapper)RedisHandlerServices
.findRedisHandler(RedisMapperHandler.class, configuration.toMap())
.createRedisMapper(configuration);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

GenericRowData genericRowData = new GenericRowData(3);
genericRowData.setField(0, "tom");
genericRowData.setField(1, "math");
genericRowData.setField(2, "152");
DataStream<GenericRowData> dataStream = env.fromElements(genericRowData, genericRowData);

RedisSinkOptions redisSinkOptions = 
        new RedisSinkOptions.Builder().setMaxRetryTimes(3).build();
FlinkConfigBase conf = 
        new FlinkSingleConfig.Builder() 
                .setHost(REDIS_HOST) 
                .setPort(REDIS_PORT) 
                .setPassword(REDIS_PASSWORD) 
                .build();

RedisSinkFunction redisSinkFunction = 
        new RedisSinkFunction<>(conf, redisMapper, redisSinkOptions, resolvedSchema);

dataStream.addSink(redisSinkFunction).setParallelism(1);
env.execute("RedisSinkTest");
```

--------------------------------

### Flink SQL: Redis HSET for Dimension Table Lookup

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Demonstrates using Flink SQL to query Redis as a dimension table with the HSET command. It involves creating a sink table for Redis HSET, inserting data into Redis, creating a dimension table for lookup, and then performing a join with a datagen source table. This example shows how to link data from a source table to Redis dimension data based on matching keys.

```sql
create table sink_redis(
  name varchar,
  level varchar,
  age varchar
) with (
  'connector'='redis',
  'host'='10.11.80.147',
  'port'='7001',
  'redis-mode'='single',
  'password'='******',
  'command'='hset'
);

-- 先在redis中插入数据,相当于redis命令： hset 3 3 100 --
insert into sink_redis select * from (values ('3', '3', '100'));
                
create table dim_table (
  name varchar,
  level varchar,
  age varchar
) with (
  'connector'='redis',
  'host'='10.11.80.147',
  'port'='7001',
  'redis-mode'='single',
  'password'='*****',
  'command'='hget',
  'maxIdle'='2',
  'minIdle'='1',
  'lookup.cache.max-rows'='10',
  'lookup.cache.ttl'='10',
  'max-retries'='3'
);
    
-- 随机生成10以内的数据作为数据源 --
-- 其中有一条数据会是： username = 3  level = 3, 会跟上面插入的数据关联 -- 
create table source_table (
  username varchar,
  level varchar,
  proctime as procTime()
) with (
  'connector'='datagen',
  'rows-per-second'='1',
  'fields.username.kind'='sequence',
  'fields.username.start'='1',
  'fields.username.end'='10',
  'fields.level.kind'='sequence',
  'fields.level.start'='1',
  'fields.level.end'='10'
);

create table sink_table(username varchar, level varchar,age varchar) with ('connector'='print');

insert into
	sink_table
select
	s.username,
	s.level,
	d.age
from
	source_table s
left join dim_table for system_time as of s.proctime as d on
	d.name = s.username
	and d.level = s.level;
-- username为3那一行会关联到redis内的值，输出为： 3,3,100

```

--------------------------------

### Flink TIME to Redis String Conversion

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Details the conversion of Flink TIME types. It explains how to represent time as a String (e.g., 'HH:MM:SS.mmm') and how to get the millisecond value from midnight.

```Java
Flink Type: TIME
Redis Converter: String
Conversion:
  - To Redis: String representation (e.g., '04:04:01.023')
  - From Redis: int (millisecond from 0'clock)
```

--------------------------------

### Flink SQL: Redis SET for Multi-Field Dimension Lookup

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Illustrates how to use Flink SQL with Redis's SET command to store and retrieve multiple fields within a single Redis value, using '01' as a delimiter. This is achieved by setting 'value.data.structure' to 'row'. The example shows creating a sink table, inserting delimited multi-field data, creating a join table configured to read this structure, and then performing a join with a source table to retrieve these fields.

```sql
-- 创建表
create table sink_redis(
  uid VARCHAR,
  score double,
  score2 double 
)
with (
  'connector' = 'redis',
  'host' = '10.11.69.176',
  'port' = '6379',
  'redis-mode' = 'single',
  'password' = '****',
  'command' = 'SET',
  'value.data.structure' = 'row'
);  -- 'value.data.structure'='row':整行内容保存至value并以'\01'分割
-- 写入测试数据，score、score2为需要被关联查询出的两个维度
insert into sink_redis select * from (values ('1', 10.3, 10.1));

-- 在redis中，value的值为: "1\x0110.3\x0110.1" --
-- 写入结束 --

-- create join table --
create table join_table with ('command'='get', 'value.data.structure'='row') like sink_redis

-- create result table --
create table result_table(uid VARCHAR, username VARCHAR, score double, score2 double) with ('connector'='print')

-- create source table --
create table source_table(uid VARCHAR, username VARCHAR, proc_time as procTime()) with ('connector'='datagen', 'fields.uid.kind'='sequence', 'fields.uid.start'='1', 'fields.uid.end'='2')

-- 关联查询维表，获得维表的多个字段值 --
insert
	into
	result_table
select
	s.uid,
	s.username,
	j.score, -- 来自维表
	j.score2 -- 来自维表
from
	source_table as s
join join_table for system_time as of s.proc_time as j on
	j.uid = s.uid
	
result:
2> +I[2, 1e0fe885a2990edd7f13dd0b81f923713182d5c559b21eff6bda3960cba8df27c69a3c0f26466efaface8976a2e16d9f68b3, null, null]
1> +I[1, 30182e00eca2bff6e00a2d5331e8857a087792918c4379155b635a3cf42a53a1b8f3be7feb00b0c63c556641423be5537476, 10.3, 10.1]

```

--------------------------------

### Flink SQL Multi-field Dimension Table Query (SET/GET)

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Illustrates handling multi-field dimension tables in Flink SQL by configuring 'value.data.structure' to 'row'. Data is stored in Redis as a delimited string and retrieved using the GET command, allowing joins on multiple fields.

```SQL
-- init data in redis --
create table sink_redis(uid VARCHAR, score double, score2 double ) with ( 'connector'='redis', 'host'='10.11.69.176','port'='6379', 'redis-mode'='single','password'='***','command'='SET', 'value.data.structure'='row')
insert into sink_redis select * from (values ('1', 10.3, 10.1))
-- 'value.data.structure'='row':value is taken from the entire row and separated by '\01' 
-- the value in redis will be: "1\x0110.3\x0110.1" --
-- init data in redis end --

-- create join table --
create table join_table with ('command'='get', 'value.data.structure'='row') like sink_redis

-- create result table --
create table result_table(uid VARCHAR, username VARCHAR, score double, score2 double) with ('connector'='print')

-- create source table --
create table source_table(uid VARCHAR, username VARCHAR, proc_time as procTime()) with ('connector'='datagen', 'fields.uid.kind'='sequence', 'fields.uid.start'='1', 'fields.uid.end'='2')

-- query --
insert
	into
	result_table
select
	s.uid,
	s.username,
	j.score,
	j.score2
from
	source_table as s
join join_table for system_time as of s.proc_time as j on
	j.uid = s.uid

result:
2> +I[2, 1e0fe885a2990edd7f13dd0b81f923713182d5c559b21eff6bda3960cba8df27c69a3c0f26466efaface8976a2e16d9f68b3, null, null]
1> +I[1, 30182e00eca2bff6e00a2d5331e8857a087792918c4379155b635a3cf42a53a1b8f3be7feb00b0c63c556641423be5537476, 10.3, 10.1]
```

--------------------------------

### Project Introduction and Improvements

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Details the enhancements made to the Flink Redis connector, including replacing Jedis with Lettuce for asynchronous operations, adding Table/SQL API support, join query caching, full-row saving, rate limiting, support for higher Flink versions, unified expiration policies, and support for Flink CDC delete operations.

```text
1. 使用Lettuce替换Jedis,同步读写改为异步读写，大幅度提升了性能 
2. 增加了Table/SQL API，增加select/维表join查询支持
3. 增加关联查询缓存(支持增量与全量)
4. 增加支持整行保存功能，用于多字段的维表关联查询
5. 增加限流功能，用于Flink SQL在线调试功能
6. 增加支持Flink高版本（包括1.12,1.13,1.14+）
7. 统一过期策略等
8. 支持flink cdc删除及其它RowKind.DELETE
9. 支持select查询

注：redis不支持两段提交无法实现刚好一次语义。
```

--------------------------------

### Flink SQL Redis Table Creation and Insertion

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Demonstrates how to create a Redis table in Flink SQL and insert data into it. It specifies connector properties like host, port, password, Redis mode, and the command to be used.

```sql
-- 创建redis表示例
create table redis_table (name varchar, age int) 
  with ('connector'='redis', 'host'='10.11.69.176', 'port'='6379','password'='test123', 
  'redis-mode'='single','command'='set');
-- 写入  
  insert into redis_table select * from (values('test', 1));
```

--------------------------------

### Redis Connector Parameters

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README.md

Lists and describes the parameters available for configuring the Redis connector in Flink SQL. This section is crucial for users to customize their Redis integration.

```APIDOC
Redis Connector Parameters:

connector: 'redis'
  - Specifies the connector type.

host: string
  - The hostname or IP address of the Redis server.

port: integer
  - The port number of the Redis server.

password: string (optional)
  - The password for authenticating with the Redis server.

redis-mode: 'single' | 'cluster' | 'sentinel'
  - The mode of the Redis deployment (single instance, cluster, or sentinel).

command: 'set' | 'get' | 'hset' | 'hget' | 'hmset' | 'hmget' | 'lpush' | 'lpop' | 'rpush' | 'rpop' | 'sadd' | 'spop' | 'zadd' | 'zrange' | 'del' | 'expire' | 'ttl'
  - The Redis command to execute for data operations. Defaults to 'set'.

scan.key: string (for GET/HGET/LRANGE/ZRANGE commands)
  - The key to scan or retrieve data from.

scan.start: integer (for LRANGE/ZRANGE commands)
  - The start index for range queries.

scan.end: integer (for LRANGE/ZRANGE commands)
  - The end index for range queries.

scan.count: integer (for LRANGE/ZRANGE commands)
  - The number of elements to retrieve in range queries.

scan.withScores: boolean (for ZRANGE command)
  - Whether to include scores in the ZRANGE results.

scan.startScore: float (for ZRANGE BYSCORE command)
  - The minimum score for range queries.

scan.endScore: float (for ZRANGE BYSCORE command)
  - The maximum score for range queries.

scan.startLex: string (for ZRANGE BYLEX command)
  - The minimum lexicographical value for range queries.

scan.endLex: string (for ZRANGE BYLEX command)
  - The maximum lexicographical value for range queries.

scan.reverse: boolean (for ZRANGE command)
  - Whether to retrieve elements in reverse order.

scan.limit: integer (for ZRANGE command)
  - The maximum number of elements to return.

scan.offset: integer (for ZRANGE command)
  - The offset for limiting the number of elements.

scan.fields: string (for HGETALL/HMGET commands)
  - Comma-separated list of fields to retrieve from a hash.

scan.pattern: string (for KEYS command)
  - A pattern to match keys.

scan.batchSize: integer
  - The batch size for operations like ZADD or HSET.

scan.timeout: integer
  - The timeout for Redis operations in milliseconds.

scan.retryTimes: integer
  - The number of times to retry failed Redis operations.

scan.retryWait: integer
  - The time to wait between retries in milliseconds.

scan.cache.ttl: integer
  - The time-to-live for the join query cache in seconds.

scan.cache.type: 'incremental' | 'full'
  - The type of join query cache (incremental or full).

scan.cache.maxSize: integer
  - The maximum size of the join query cache.

scan.cache.enabled: boolean
  - Whether the join query cache is enabled.

scan.cache.lookup.key: string
  - The key used for looking up data in the cache.

scan.cache.lookup.fields: string
  - The fields used for looking up data in the cache.

scan.cache.lookup.type: 'single' | 'multi'
  - The type of lookup for cache data.

scan.cache.lookup.return.fields: string
  - The fields to return from the cache lookup.

scan.cache.lookup.return.type: 'single' | 'multi'
  - The type of return data from the cache lookup.

scan.cache.lookup.return.all: boolean
  - Whether to return all fields from the cache lookup.

scan.cache.lookup.return.all.fields: string
  - Comma-separated list of all fields to return from the cache lookup.

scan.cache.lookup.return.all.type: 'single' | 'multi'
  - The type of return data for all fields from the cache lookup.

scan.cache.lookup.return.all.count: integer
  - The count of elements to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score: boolean
  - Whether to return scores for all fields from the cache lookup.

scan.cache.lookup.return.all.score.type: 'single' | 'multi'
  - The type of return score for all fields from the cache lookup.

scan.cache.lookup.return.all.score.count: integer
  - The count of scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.reverse: boolean
  - Whether to return scores in reverse order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.offset: integer
  - The offset for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.limit: integer
  - The limit for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.start: float
  - The start score for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.end: float
  - The end score for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start: string
  - The start lexicographical value for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end: string
  - The end lexicographical value for scores to return for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.reverse: boolean
  - Whether to return scores in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.offset: integer
  - The offset for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.limit: integer
  - The limit for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score: float
  - The start score for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score: float
  - The end score for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.type: 'single' | 'multi'
  - The type of start score for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.type: 'single' | 'multi'
  - The type of end score for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.count: integer
  - The count of start scores for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.count: integer
  - The count of end scores for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.reverse: boolean
  - Whether to return start scores in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.reverse: boolean
  - Whether to return end scores in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.offset: integer
  - The offset for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.offset: integer
  - The offset for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.limit: integer
  - The limit for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.limit: integer
  - The limit for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.start: float
  - The start score for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.start: float
  - The start score for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.end: float
  - The end score for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.end: float
  - The end score for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start: string
  - The start lexicographical value for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start: string
  - The start lexicographical value for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end: string
  - The end lexicographical value for start scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end: string
  - The end lexicographical value for end scores to return in reverse order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.reverse: boolean
  - Whether to return start scores in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.reverse: boolean
  - Whether to return end scores in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.offset: integer
  - The offset for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.offset: integer
  - The offset for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.limit: integer
  - The limit for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.limit: integer
  - The limit for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score: float
  - The start score for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score: float
  - The start score for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score: float
  - The end score for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score: float
  - The end score for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.type: 'single' | 'multi'
  - The type of start score for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.type: 'single' | 'multi'
  - The type of start score for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.type: 'single' | 'multi'
  - The type of end score for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.type: 'single' | 'multi'
  - The type of end score for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.count: integer
  - The count of start scores for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.count: integer
  - The count of start scores for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.count: integer
  - The count of end scores for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.count: integer
  - The count of end scores for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.reverse: boolean
  - Whether to return start scores in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.reverse: boolean
  - Whether to return start scores in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.reverse: boolean
  - Whether to return end scores in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.reverse: boolean
  - Whether to return end scores in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.offset: integer
  - The offset for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.offset: integer
  - The offset for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.offset: integer
  - The offset for end scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.offset: integer
  - The offset for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.limit: integer
  - The limit for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.limit: integer
  - The limit for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.limit: integer
  - The limit for end scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.limit: integer
  - The limit for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.start: float
  - The start score for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.start: float
  - The start score for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.start: float
  - The end score for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.start: float
  - The end score for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.end: float
  - The end score for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.end: float
  - The end score for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.end: float
  - The end score for end scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.end: float
  - The end score for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.lex.start: string
  - The start lexicographical value for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.lex.start: string
  - The start lexicographical value for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.lex.start: string
  - The end lexicographical value for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.lex.start: string
  - The end lexicographical value for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.lex.reverse: boolean
  - Whether to return start scores in reverse lexicographical order for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.lex.reverse: boolean
  - Whether to return start scores in reverse lexicographical order for start scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.end.score.lex.reverse: boolean
  - Whether to return end scores in reverse lexicographical order for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.end.score.lex.reverse: boolean
  - Whether to return end scores in reverse lexicographical order for end scores to return in reverse order for end scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.start.score.lex.start.score.lex.offset: integer
  - The offset for start scores to return in reverse lexicographical order for start scores to return in reverse order for start scores to return in reverse lexicographical order for scores to return in reverse lexicographical order for all fields from the cache lookup.

scan.cache.lookup.return.all.score.lex.end.score.lex.start.score.lex.offset: integer
  - The offset for start scores to
```

--------------------------------

### General Scan and Audit Parameters

Source: https://github.com/jeff-zou/flink-connector-redis/blob/main/README-en.md

Configuration parameters for scanning Redis keys, handling sorted sets, and enabling audit logging.

```APIDOC
General Scan and Audit Parameters:
  scan.key: (none) | String | redis key
  scan.addition.key: (none) | String | redis addition key, eg: map hashfield
  scan.range.start: (none) | Integer | lrange start
  scan.range.stop: (none) | Integer | lrange start
  scan.count: (none) | Integer | srandmember count
  zset.zremrangeby: (none) | String | After executing zadd, whether to execute zremrangeby, Valid values are: SCORE, LEX, RANK
  audit.log: false | Boolean | Turn on the audit log switch
```