luoyuxia commented on code in PR #3864:
URL: https://github.com/apache/fluss/pull/3864#discussion_r3719148979
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java:
##########
@@ -311,8 +396,253 @@ private LookupResultForBucket lookupInternal(
}
}
+ private CachedLakeTableLookuper acquireCachedLookuper(
+ LookupContext context,
+ TableConfig tableConfig,
+ Configuration clusterConf,
+ long currentLakeConfigVersion,
+ long cacheSizeBytes) {
+ CachedLakeTableLookuper cachedLookuper =
+ tryAcquireCachedLookuper(
+ context,
+ tableConfig,
+ clusterConf,
+ currentLakeConfigVersion,
+ cacheSizeBytes);
+ if (cachedLookuper != null) {
+ return cachedLookuper;
+ }
+
+ int maxEvictions = lakeTableLookupers.asMap().size();
+ for (int evictions = 0; evictions < maxEvictions; evictions++) {
+ // Evict only after compute releases the target table's cache
lock. Updating a
+ // different table mapping from inside compute can deadlock with a
concurrent
+ // replacement performing the inverse update.
+ if (!evictLeastRecentlyUsed(context.tableId)) {
+ break;
+ }
+ cachedLookuper =
+ tryAcquireCachedLookuper(
+ context,
+ tableConfig,
+ clusterConf,
+ currentLakeConfigVersion,
+ cacheSizeBytes);
+ if (cachedLookuper != null) {
+ return cachedLookuper;
+ }
+ }
+ throw capacityThrottledException(context, cacheSizeBytes);
+ }
+
+ /**
+ * Makes one atomic attempt to acquire a matching cached lookuper without
evicting other tables.
+ *
+ * @return the acquired lookuper, or {@code null} if its capacity cannot
be reserved
+ */
+ private @Nullable CachedLakeTableLookuper tryAcquireCachedLookuper(
+ LookupContext context,
+ TableConfig tableConfig,
+ Configuration clusterConf,
+ long currentLakeConfigVersion,
+ long cacheSizeBytes) {
+ CachedLakeTableLookuper cachedLookuper =
+ lakeTableLookupers
+ .asMap()
+ .compute(
+ context.tableId,
+ (ignored, currentLookuper) -> {
+ CachedLakeTableLookuper selectedLookuper =
currentLookuper;
+ // Create the lookuper lazily, and
recreate it after schema,
+ // lake configuration, or effective cache
size changes so it
+ // reloads lake table/query state and uses
the current settings.
+ if (!matchesLookupConfiguration(
+ selectedLookuper,
+ context,
+ currentLakeConfigVersion,
+ cacheSizeBytes)) {
+ selectedLookuper =
+ tryCreateCachedLookuper(
+ context,
+ tableConfig,
+ clusterConf,
+
currentLakeConfigVersion,
+ cacheSizeBytes,
+ currentLookuper);
+ if (selectedLookuper == null) {
+ // Preserve the current mapping
and leave compute before
+ // attempting to evict another
table.
+ return currentLookuper;
+ }
+ }
+ // Pin the lookuper before leaving the
atomic cache update.
+ // Eviction or invalidation can then defer
closing it until this
+ // lookup releases it.
+ selectedLookuper.acquire(ticker.read());
+ return selectedLookuper;
+ });
+ return matchesLookupConfiguration(
+ cachedLookuper, context, currentLakeConfigVersion,
cacheSizeBytes)
+ ? cachedLookuper
+ : null;
+ }
+
+ private static boolean matchesLookupConfiguration(
+ @Nullable CachedLakeTableLookuper cachedLookuper,
+ LookupContext context,
+ long currentLakeConfigVersion,
+ long cacheSizeBytes) {
+ return cachedLookuper != null
+ && cachedLookuper.schemaId == context.schemaId
+ && cachedLookuper.lakeConfigVersion == currentLakeConfigVersion
+ && cachedLookuper.cacheSizeBytes == cacheSizeBytes;
+ }
+
+ /**
+ * Creates a lookuper after atomically reserving its configured cache
capacity.
+ *
+ * @return the new cached lookuper, or {@code null} if its capacity cannot
be reserved
+ */
+ private @Nullable CachedLakeTableLookuper tryCreateCachedLookuper(
+ LookupContext context,
+ TableConfig tableConfig,
+ Configuration clusterConf,
+ long currentLakeConfigVersion,
+ long cacheSizeBytes,
+ @Nullable CachedLakeTableLookuper currentLookuper) {
+ File tableLookupDir =
+ FlussPaths.historicalLookupTableDir(
+ getOrPreparePaimonLookupTempDir(clusterConf),
+ context.tablePath,
+ context.tableId);
+ if (currentLookuper == null) {
+ // A cache miss must obtain capacity before creating any local
lookup resources.
+ Reservation reservation =
budgetManager.tryReserve(context.tableId, cacheSizeBytes);
+ if (reservation == null) {
+ return null;
+ }
+ try {
+ LakeTableLookuper lookuper =
+ createLakeTableLookuper(
+ context.tablePath,
+ tableLookupDir.getAbsolutePath(),
+ tableConfig,
+ clusterConf);
+ return new CachedLakeTableLookuper(
+ context.tableId,
+ context.tablePath,
+ context.schemaId,
+ currentLakeConfigVersion,
+ cacheSizeBytes,
+ tableLookupDir,
+ reservation,
+ lookuper);
+ } catch (Throwable throwable) {
+ budgetManager.release(reservation);
+ throw throwable;
+ }
+ }
+
+ // Build the replacement first so a creation failure leaves the
current lookuper and its
+ // reservation unchanged in the cache.
+ LakeTableLookuper lookuper =
+ createLakeTableLookuper(
+ context.tablePath,
+ tableLookupDir.getAbsolutePath(),
+ tableConfig,
+ clusterConf);
+ // Replace the reservation atomically: the old and replacement cache
sizes never count
+ // against the global budget at the same time.
+ Reservation reservation =
+ budgetManager.tryReplace(currentLookuper.reservation,
cacheSizeBytes);
+ if (reservation == null) {
+ // The candidate was never published, while the current lookuper
remains usable.
+ closeLookuper(lookuper, tableLookupDir);
+ return null;
+ }
Review Comment:
The create-before-replace ordering is intentional. `tryReplace()` mutates
the budget state rather than performing a read-only capacity check. Reserving
first would retire the current reservation before the replacement lookuper
exists, and a later creation failure would require a dedicated rollback
operation that remains correct across concurrent limit changes and removal
callbacks.
The current Paimon lookuper is lazy: construction only stores configuration.
The catalog, IOManager, local directories, and lookup files are initialized on
the first `lookup()`, after `tryReplace()` succeeds and the candidate is
published. Therefore, failed replacement admission does not materialize
unreserved local cache data. We accept the small amount of temporary
wrapper/plugin construction so creation failures leave the existing lookuper
and reservation unchanged.
--
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]
