[hotfix-#1179][lookup] When lookup-type is 'NONE', each data will tri…

…gger the logic of querying data.
DTStack · Aug 24, 2022 · cb9c8b6 · cb9c8b6
1 parent 0278622
commit cb9c8b6
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 31 deletions.
diff --git a/...main/java/com/dtstack/chunjun/connector/cassandra/source/CassandraDynamicTableSource.java b/...main/java/com/dtstack/chunjun/connector/cassandra/source/CassandraDynamicTableSource.java
@@ -142,18 +142,18 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  // 通过该参数得到类型转换器，将数据库中的字段转成对应的类型
  final RowType rowType = (RowType) tableSchema.toRowDataType().getLogicalType();
 
- if (cassandraLookupConf.getCache().equalsIgnoreCase(CacheType.LRU.toString())) {
- return ParallelAsyncTableFunctionProvider.of(
- new CassandraLruTableFunction(
+ if (cassandraLookupConf.getCache().equalsIgnoreCase(CacheType.ALL.toString())) {
+ return ParallelTableFunctionProvider.of(
+ new CassandraAllTableFunction(
  cassandraLookupConf,
  new CassandraRowConverter(
  rowType, Arrays.asList(tableSchema.getFieldNames())),
  tableSchema.getFieldNames(),
  keyNames),
  cassandraLookupConf.getParallelism());
  }
- return ParallelTableFunctionProvider.of(
- new CassandraAllTableFunction(
+ return ParallelAsyncTableFunctionProvider.of(
+ new CassandraLruTableFunction(
  cassandraLookupConf,
  new CassandraRowConverter(
  rowType, Arrays.asList(tableSchema.getFieldNames())),

diff --git a/.../src/main/java/com/dtstack/chunjun/connector/hbase/table/BaseHBaseDynamicTableSource.java b/.../src/main/java/com/dtstack/chunjun/connector/hbase/table/BaseHBaseDynamicTableSource.java
@@ -102,12 +102,12 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  }
  fillKerberosConf();
  hbaseSchema.setTableName(hBaseConf.getTable());
- if (lookupConf.getCache().equalsIgnoreCase(CacheType.LRU.toString())) {
- return ParallelAsyncTableFunctionProvider.of(
- getAbstractLruTableFunction(), lookupConf.getParallelism());
+ if (lookupConf.getCache().equalsIgnoreCase(CacheType.ALL.toString())) {
+ return ParallelTableFunctionProvider.of(
+ getAbstractAllTableFunction(), lookupConf.getParallelism());
  }
- return ParallelTableFunctionProvider.of(
- getAbstractAllTableFunction(), lookupConf.getParallelism());
+ return ParallelAsyncTableFunctionProvider.of(
+ getAbstractLruTableFunction(), lookupConf.getParallelism());
  }
 
  @Override

diff --git a/...-base/src/main/java/com/dtstack/chunjun/connector/jdbc/source/JdbcDynamicTableSource.java b/...-base/src/main/java/com/dtstack/chunjun/connector/jdbc/source/JdbcDynamicTableSource.java
@@ -86,9 +86,9 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  // 通过该参数得到类型转换器，将数据库中的字段转成对应的类型
  final RowType rowType = (RowType) physicalSchema.toRowDataType().getLogicalType();
 
- if (lookupConf.getCache().equalsIgnoreCase(CacheType.LRU.toString())) {
- return ParallelAsyncTableFunctionProvider.of(
- new JdbcLruTableFunction(
+ if (lookupConf.getCache().equalsIgnoreCase(CacheType.ALL.toString())) {
+ return ParallelTableFunctionProvider.of(
+ new JdbcAllTableFunction(
  jdbcConf,
  jdbcDialect,
  lookupConf,
@@ -97,8 +97,8 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  rowType),
  lookupConf.getParallelism());
  }
- return ParallelTableFunctionProvider.of(
- new JdbcAllTableFunction(
+ return ParallelAsyncTableFunctionProvider.of(
+ new JdbcLruTableFunction(
  jdbcConf,
  jdbcDialect,
  lookupConf,

diff --git a/...r-kudu/src/main/java/com/dtstack/chunjun/connector/kudu/table/KuduDynamicTableSource.java b/...r-kudu/src/main/java/com/dtstack/chunjun/connector/kudu/table/KuduDynamicTableSource.java
@@ -143,18 +143,18 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  // 通过该参数得到类型转换器，将数据库中的字段转成对应的类型
  final RowType rowType = (RowType) tableSchema.toRowDataType().getLogicalType();
 
- if (kuduLookupConf.getCache().equalsIgnoreCase(CacheType.LRU.toString())) {
- return ParallelAsyncTableFunctionProvider.of(
- new KuduLruTableFunction(
+ if (kuduLookupConf.getCache().equalsIgnoreCase(CacheType.ALL.toString())) {
+ return ParallelTableFunctionProvider.of(
+ new KuduAllTableFunction(
  kuduLookupConf,
  new KuduRowConverter(
  rowType, Arrays.asList(tableSchema.getFieldNames())),
  tableSchema.getFieldNames(),
  keyNames),
  kuduLookupConf.getParallelism());
  }
- return ParallelTableFunctionProvider.of(
- new KuduAllTableFunction(
+ return ParallelAsyncTableFunctionProvider.of(
+ new KuduLruTableFunction(
  kuduLookupConf,
  new KuduRowConverter(rowType, Arrays.asList(tableSchema.getFieldNames())),
  tableSchema.getFieldNames(),

diff --git a/...dis/src/main/java/com/dtstack/chunjun/connector/redis/source/RedisDynamicTableSource.java b/...dis/src/main/java/com/dtstack/chunjun/connector/redis/source/RedisDynamicTableSource.java
@@ -72,19 +72,18 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
  }
  final RowType rowType = (RowType) physicalSchema.toRowDataType().getLogicalType();
 
- if (lookupConf.getCache().equalsIgnoreCase(CacheType.LRU.toString())) {
- return ParallelAsyncTableFunctionProvider.of(
- new RedisLruTableFunction(
- redisConf, lookupConf, new RedisRowConverter(rowType)),
+ if (lookupConf.getCache().equalsIgnoreCase(CacheType.ALL.toString())) {
+ return ParallelTableFunctionProvider.of(
+ new RedisAllTableFunction(
+ redisConf,
+ lookupConf,
+ physicalSchema.getFieldNames(),
+ keyNames,
+ new RedisRowConverter(rowType)),
  lookupConf.getParallelism());
  }
- return ParallelTableFunctionProvider.of(
- new RedisAllTableFunction(
- redisConf,
- lookupConf,
- physicalSchema.getFieldNames(),
- keyNames,
- new RedisRowConverter(rowType)),
+ return ParallelAsyncTableFunctionProvider.of(
+ new RedisLruTableFunction(redisConf, lookupConf, new RedisRowConverter(rowType)),
  lookupConf.getParallelism());
  }