platinumhamburg commented on code in PR #3630:
URL: https://github.com/apache/fluss/pull/3630#discussion_r3600514655
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java:
##########
@@ -115,19 +125,89 @@ public CompletableFuture<LookupResult> lookup(InternalRow
lookupKey) {
tableInfo.getTablePath(),
metadataUpdater);
} catch (PartitionNotExistException e) {
- return CompletableFuture.completedFuture(new
LookupResult(Collections.emptyList()));
+ return mayFallbackToHistoricalLookup(
+ bucketingFunction.bucketing(bkBytes, numBuckets),
pkBytes, lookupKey);
}
}
int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
TableBucket tableBucket = new TableBucket(tableInfo.getTableId(),
partitionId, bucketId);
+ return lookupBucket(tableBucket, pkBytes, insertIfNotExists, null,
lookupKey);
+ }
+
+ /** Falls back to historical lookup if the lookup key belongs to a
historical partition. */
+ private CompletableFuture<LookupResult> mayFallbackToHistoricalLookup(
+ int bucketId, byte[] keyBytes, InternalRow lookupKey) {
+ String originalPartitionName = partitionGetter.getPartition(lookupKey);
+ if (!isHistoricalLookupCandidatePartition(
+ tableInfo, originalPartitionName, Instant.now())) {
Review Comment:
The Coordinator removes expired partitions using the CoordinatorServer
clock, but this code calculates the retention boundary again using the client
clock.
For an hourly partitioned table with retention set to 1, the Coordinator may
be just past 10:00 and remove partition `2024010108`, while a client is still
just before 10:00. The client refreshes metadata and sees that the partition is
gone, but this check still treats it as retained and returns an empty result. A
few seconds later, the same lookup starts using the historical path without any
data or metadata change.
Can this decision use the Coordinator's time instead of `Instant.now()` on
each client? A test with the Coordinator and client on opposite sides of an
hour boundary would cover this case.
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java:
##########
@@ -115,19 +125,89 @@ public CompletableFuture<LookupResult> lookup(InternalRow
lookupKey) {
tableInfo.getTablePath(),
metadataUpdater);
} catch (PartitionNotExistException e) {
- return CompletableFuture.completedFuture(new
LookupResult(Collections.emptyList()));
+ return mayFallbackToHistoricalLookup(
+ bucketingFunction.bucketing(bkBytes, numBuckets),
pkBytes, lookupKey);
}
}
int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
TableBucket tableBucket = new TableBucket(tableInfo.getTableId(),
partitionId, bucketId);
+ return lookupBucket(tableBucket, pkBytes, insertIfNotExists, null,
lookupKey);
+ }
+
+ /** Falls back to historical lookup if the lookup key belongs to a
historical partition. */
+ private CompletableFuture<LookupResult> mayFallbackToHistoricalLookup(
+ int bucketId, byte[] keyBytes, InternalRow lookupKey) {
+ String originalPartitionName = partitionGetter.getPartition(lookupKey);
+ if (!isHistoricalLookupCandidatePartition(
Review Comment:
`tableInfo` is fixed when the `Lookuper` is created, but
`table.auto-partition.num-retention` can be changed while the lookuper is
running.
Suppose a client creates a lookuper with retention set to 7. On January 10,
retention is changed to 1, so the Coordinator removes partition `20240105`. The
existing lookuper still checks `20240105` with retention 7 and returns an empty
result, while a client using the updated table info sends the same lookup
through the historical path.
Can we use the current table configuration here, or let the Coordinator make
this decision? The current IT changes retention before creating the lookuper,
so it does not cover this case.
--
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]
