JingsongLi commented on code in PR #10813:
URL: https://github.com/apache/paimon/pull/10813#discussion_r4385469128
##########
paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java:
##########
@@ -137,68 +180,103 @@ public void refreshFiles(
.computeIfAbsent(bucket, k -> new BucketLookupState());
state.lock.writeLock().lock();
try {
+ // Load before touching the levels so a failure leaves the
previous view intact.
+ // Compaction can mark a row deleted without rewriting its data
file. Streaming
+ // plans leave deletion files empty (SnapshotReaderImpl fills them
for batch plans
+ // only), and the updated file may not appear in this split at
all, so the bucket
+ // index is reloaded from the latest snapshot. That snapshot can
be newer than the
+ // splits just applied; a newer vector hides a row only until the
next refresh
+ // installs its data file. Unchanged index files are not read
again.
+ Map<String, DeletionVector> deletionVectors =
+ loadDeletionVectors(partition, bucket, state);
if (state.lookupLevels == null) {
// Initial phase: ignore beforeFiles as they represent
deletions from previous state
state.lookupLevels = createLookupLevels(partition, bucket,
dataFiles);
} else {
state.lookupLevels.getLevels().update(beforeFiles, dataFiles);
}
+ state.deletionVectors = deletionVectors;
} finally {
state.lock.writeLock().unlock();
}
}
- private LookupLevels<KeyValue> createLookupLevels(
+ private Map<String, DeletionVector> loadDeletionVectors(
+ BinaryRow partition, int bucket, BucketLookupState state) {
+ if (!deletionVectorsEnabled) {
+ return Collections.emptyMap();
+ }
+ List<IndexFileMeta> fileMetas = new ArrayList<>();
+ for (IndexManifestEntry entry : currentDeletionVectorEntries()) {
+ if (entry.bucket() == bucket &&
entry.partition().equals(partition)) {
+ fileMetas.add(entry.indexFile());
+ }
+ }
+ if (fileMetas.equals(state.deletionVectorFiles)) {
+ return state.deletionVectors;
+ }
+ if (fileMetas.isEmpty()) {
+ state.deletionVectorFiles = fileMetas;
+ return Collections.emptyMap();
+ }
+ // Record the file list only after the read succeeds. A failure leaves
the previous
+ // list and vectors in place, so the next refresh retries instead of
trusting them.
+ Map<String, DeletionVector> loaded =
+ indexFileHandler.readAllDeletionVectors(partition, bucket,
fileMetas);
+ state.deletionVectorFiles = fileMetas;
+ return loaded;
+ }
+
+ private List<IndexManifestEntry> currentDeletionVectorEntries() {
+ Snapshot latest = snapshotManager.latestSnapshot();
Review Comment:
[P1] Pair deletion vectors with the snapshot of the supplied files
`latestSnapshot()` can be newer than the files being installed. I reproduced
this using real writes/compaction: at S1, compact 20 keys including `(1, 10)`
and retain its planned data files; write `(1, 99)` and partially compact S2;
then call `refreshFiles` with the held S1 files. This code reads S2's vector,
masks key 1's position in S1, and lookup returns **null** (key 2 remains
visible). The same test against the parent implementation returns **10**. Key 1
exists in both S1 and S2, so this is neither a valid stale value nor the
current value.
This interleaving is reachable when QueryFileMonitor has planned/enqueued an
S1 event and S2 commits before QueryExecutorOperator processes it; the query
server continues serving lookups between events. A miss can silently drop a
lookup-join match. Please carry the event/split snapshot ID (or matching DV
metadata) through `refreshFiles` and load the corresponding deletion vectors,
so file levels and visibility are installed from the same snapshot. Add a
regression with a held file plan and an intervening update/compaction.
--
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]
