Akash3121 commented on code in PR #10813:
URL: https://github.com/apache/paimon/pull/10813#discussion_r4384734196
##########
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();
+ long snapshotId = latest == null ? -1L : latest.id();
Review Comment:
[P1] Keep deletion vectors aligned with the streamed file update
Loading deletion vectors from the latest table snapshot can make them newer
than the `beforeFiles / dataFiles` being applied. If snapshot N adds a deletion
vector for an old data file and snapshot N+1 rewrites that file and removes the
vector, a lagging query operator processing N reads N+1's now-empty vector
state while its lookup levels still contain the old file. The deleted row is
then returned until the N+1 file update arrives. This is observable in
`QueryExecutorOperator`, which applies updates individually while
`KvQueryServer` serves concurrent requests. Please propagate the update's
snapshot/index state with the file event, or otherwise update files and
deletion vectors from the same snapshot.
--
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]
