wwj6591812 commented on code in PR #4383:
URL: https://github.com/apache/paimon/pull/4383#discussion_r1816622010
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java:
##########
@@ -244,11 +240,17 @@ public LookupRuntimeProvider
getLookupRuntimeProvider(LookupContext context) {
boolean enableAsync = options.get(LOOKUP_ASYNC);
int asyncThreadNumber = options.get(LOOKUP_ASYNC_THREAD_NUMBER);
return LookupRuntimeProviderFactory.create(
- new FileStoreLookupFunction(table, projection, joinKey,
predicate),
+ getFileStoreLookupFunction(
+ context, timeTravelDisabledTable(table), projection,
joinKey),
enableAsync,
asyncThreadNumber);
}
+ protected FileStoreLookupFunction getFileStoreLookupFunction(
+ LookupContext context, Table table, int[] projection, int[]
joinKey) {
Review Comment:
Delete parameter context?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -172,11 +200,21 @@ protected Integer
inferSourceParallelism(StreamExecutionEnvironment env) {
protected void scanSplitsForInference() {
if (splitStatistics == null) {
if (table instanceof DataTable) {
- List<PartitionEntry> partitionEntries =
- table.newReadBuilder()
- .withFilter(predicate)
- .newScan()
- .listPartitionEntries();
+ List<PartitionEntry> partitionEntries;
+ try {
+ partitionEntries =
+ table.newReadBuilder()
+ .withFilter(predicate)
+ .newScan()
+ .listPartitionEntries();
+ } catch (Exception ignore) {
+ partitionEntries =
Review Comment:
1、What exception will be catch?
2、Add a log with ignore?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -172,11 +200,21 @@ protected Integer
inferSourceParallelism(StreamExecutionEnvironment env) {
protected void scanSplitsForInference() {
if (splitStatistics == null) {
if (table instanceof DataTable) {
- List<PartitionEntry> partitionEntries =
- table.newReadBuilder()
- .withFilter(predicate)
- .newScan()
- .listPartitionEntries();
+ List<PartitionEntry> partitionEntries;
+ try {
+ partitionEntries =
+ table.newReadBuilder()
+ .withFilter(predicate)
+ .newScan()
+ .listPartitionEntries();
+ } catch (Exception ignore) {
+ partitionEntries =
Review Comment:
What exception will be catch?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java:
##########
@@ -244,11 +240,17 @@ public LookupRuntimeProvider
getLookupRuntimeProvider(LookupContext context) {
boolean enableAsync = options.get(LOOKUP_ASYNC);
int asyncThreadNumber = options.get(LOOKUP_ASYNC_THREAD_NUMBER);
return LookupRuntimeProviderFactory.create(
- new FileStoreLookupFunction(table, projection, joinKey,
predicate),
+ getFileStoreLookupFunction(
+ context, timeTravelDisabledTable(table), projection,
joinKey),
enableAsync,
asyncThreadNumber);
}
+ protected FileStoreLookupFunction getFileStoreLookupFunction(
Review Comment:
Why this method is protected?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -60,6 +63,16 @@ public abstract class FlinkTableSource
private static final Logger LOG =
LoggerFactory.getLogger(FlinkTableSource.class);
+ private static final List<String> TIME_TRAVEL_OPTIONS =
Review Comment:
Mode this to TimeTravelUtil?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
