luoyuxia commented on code in PR #3442:
URL: https://github.com/apache/fluss/pull/3442#discussion_r3445484874
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java:
##########
@@ -889,6 +894,36 @@ private int[] getKeyRowProjection() {
return projection;
}
+ private boolean lookupCoversAllData(
+ List<ResolvedExpression> filters, Map<Integer, LogicalType>
primaryKeyTypes) {
+ if (!isDataLakeEnabled || !isPartitioned()) {
+ return true;
+ }
+ return PushdownUtils.partitionExists(
+ tablePath, flussConfig, resolveLookupPartition(filters,
primaryKeyTypes));
+ }
+
+ private PartitionSpec resolveLookupPartition(
Review Comment:
Not sure whether it can be simplify to the following:
```
private boolean lookupCoversAllData(GenericRowData lookupRow) {
if (!isDataLakeEnabled || !isPartitioned()) {
return true;
}
// lookupRow holds all primary key fields (in pk-projection order);
partition keys are a
// subset of the primary key, so the partition can be fully resolved
from it.
RowType flussRowType =
FlinkConversions.toFlussRowType(tableOutputType);
PartitionGetter partitionGetter =
new PartitionGetter(
flussRowType.project(primaryKeyIndexes),
flussRowType.project(partitionKeyIndexes).getFieldNames());
PartitionSpec partitionSpec =
partitionGetter
.getResolvedPartitionSpec(new
FlinkAsFlussRow(lookupRow))
.toPartitionSpec();
return PushdownUtils.partitionExists(tablePath, flussConfig,
partitionSpec);
}
```
--
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]
