leaves12138 commented on code in PR #8857: URL: https://github.com/apache/paimon/pull/8857#discussion_r3655591048
########## paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/AsyncLsmCompactor.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.lookup.sort.db; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** Coordinates asynchronous compaction, failure propagation, backpressure, and shutdown. */ +class AsyncLsmCompactor { + + private final LsmLevels levels; + private final LsmCompactor compactor; + private final ExecutorService executor; + private final int levelZeroFileCountTrigger; + private final LsmCompactor.FileSupplier fileSupplier; + private final LsmCompactor.FileDeleter fileDeleter; + private final ThreadLocal<List<File>> deferredCompactionDeletes; + + @Nullable private Future<?> compactionFuture; + + AsyncLsmCompactor( + LsmLevels levels, + CompactorFactory compactorFactory, + ExecutorService executor, + int levelZeroFileCountTrigger, + LsmCompactor.FileSupplier fileSupplier, + LsmCompactor.FileDeleter fileDeleter) { + this.levels = levels; + this.executor = executor; + this.levelZeroFileCountTrigger = levelZeroFileCountTrigger; + this.fileSupplier = fileSupplier; + this.fileDeleter = fileDeleter; + this.deferredCompactionDeletes = new ThreadLocal<>(); + this.compactor = compactorFactory.create(this::deferOrDeleteCompactedFile); + } + + void scheduleIfNeeded() throws IOException { + checkFailure(); + if (!needsCompaction() || compactionFuture != null) { + return; + } + + try { + compactionFuture = + executor.submit( + () -> { + while (needsCompaction()) { Review Comment: Could we reject non-positive `level0FileNumCompactTrigger` values before starting asynchronous compaction? The public builder currently accepts `0` or a negative value. With `0`, `flush()` schedules the worker and then always enters backpressure (`L0 file count >= 0`), while the worker loops forever because `needsCompaction()` remains true even after L0 becomes empty. A minimal repro prints `beforeFlush` and still does not return after 5 seconds. This is a regression from the previous single `maybeCompact` call; please validate the trigger as strictly positive and add a regression test. ########## paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java: ########## @@ -549,7 +607,8 @@ private SstFileMetadata findFileForKey(List<SstFileMetadata> sortedFiles, Memory private SstFileMetadata writeMemTableToSst(TreeMap<MemorySlice, byte[]> data) throws IOException { File sstFile = newSstFile(); - SortLookupStoreWriter writer = storeFactory.createWriter(sstFile, null); + SortLookupStoreWriter writer = + storeFactory.createWriter(sstFile, bloomFilterBuilderFactory.apply(data.size())); Review Comment: This is not an unchanged pre-existing limitation for this production path: `LocalKvDb` did not write Bloom filters before this PR, while this PR enables them by default for `ClusteringKeyIndex`, which uses `RowCompactedSerializer` comparator equality. Therefore the PR turns comparator-equal NaN encodings into false-negative lookups after flush. I think this should remain blocking unless the Bloom filter is disabled for such comparators or the hash input is made comparator-consistent. -- 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]
