Gabriel39 commented on code in PR #70216:
URL: https://github.com/apache/doris/pull/70216#discussion_r4489351666
##########
be/src/runtime/exec_env.cpp:
##########
@@ -59,6 +61,17 @@ void
ExecEnv::set_storage_engine(std::unique_ptr<BaseStorageEngine>&& engine) {
void ExecEnv::set_write_cooldown_meta_executors() {
_write_cooldown_meta_executors =
std::make_unique<WriteCooldownMetaExecutors>();
}
+
+void ExecEnv::init_fluss_union_tail_cache() {
Review Comment:
Fixed in f1371632dfc. Both lifecycle definitions now live outside BE_TEST.
Before the fix, compiling the actual exec_env.cpp and exec_env_init.cpp with
BE_TEST undefined succeeded, but a relocatable link requiring both lifecycle
symbols failed. The same check after the fix resolves both unconditional calls.
A focused ASAN executable relinked with the newly compiled runtime object also
passes 39 Fluss union-reader tests and one production CachePolicy
unregistration probe. The focused executable retains the previously disclosed
local Rust duplicate-symbol linker workaround; normal full BE linking remains
for CI.
##########
fe/be-java-extensions/fluss-scanner/src/main/java/org/apache/doris/fluss/FlussTableInfoCache.java:
##########
@@ -0,0 +1,125 @@
+// 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.doris.fluss;
+
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+/**
+ * Table infos kept for a short time from one scan range to the next, for
every range this BE reads.
+ *
+ * <p>{@code Connection#getTable} refreshes the table's metadata and asks for
its info, and a fluss server
+ * answers both from ZooKeeper. A point lookup is one range, so under
concurrent lookups those two requests
+ * per query queued on the tablet server in front of the lookups themselves. A
range now builds its table
+ * from an info another range loaded within the TTL, and leaves the bucket
metadata to the fluss reader,
+ * which fetches what its connection is missing and refreshes what a server
rejects.
+ *
+ * <p>A dropped, recreated or altered table is seen once its entry expires; a
range that needs a column its
+ * entry lacks reloads it at once ({@link #reload}). Keyed by the whole client
configuration, like
+ * {@link FlussConnectionPool}, so catalogs on different clusters never share
an entry; the key can hold
+ * credentials and is never logged.
+ */
+final class FlussTableInfoCache {
+
+ /** FE sends this Doris cache option outside the {@code fluss.client.*}
client configuration. */
+ static final String TTL_PROPERTY = "fluss.metadata.cache.ttl.sec";
+ static final long DEFAULT_TTL_SEC = 10;
+
+ static final FlussTableInfoCache INSTANCE = new
FlussTableInfoCache(System::nanoTime);
+
+ private final LongSupplier nanoTime;
+ private final ConcurrentHashMap<Key, Entry> entries = new
ConcurrentHashMap<>();
+
+ FlussTableInfoCache(LongSupplier nanoTime) {
+ this.nanoTime = nanoTime;
+ }
+
+ /** The info loaded for this table within {@code ttlNanos}, or a new one
from {@code load}. */
+ TableInfo get(Map<String, String> config, TablePath tablePath, long
ttlNanos, Supplier<TableInfo> load) {
+ if (ttlNanos <= 0) {
+ return load.get();
+ }
+ Entry entry = entries.get(new Key(config, tablePath));
+ if (entry != null && nanoTime.getAsLong() - entry.expiresAtNanos < 0) {
Review Comment:
Fixed in f1371632dfc. Entries now record when metadata finished loading, and
hits check their age against the requesting TTL as well as the retained entry's
expiry. A 600-second entry therefore reloads once it is too old for a
one-second request; cache hits do not reset its age. Added virtual-clock tests
cover the older warm entry, fresh reuse and the exact shorter-TTL boundary,
plus TTL starting after the load completes. An embedded scanner test warms a
600-second entry and then requests a one-second TTL. The shorter-TTL
reproductions fail before the fix and pass afterward. All 107 scanner tests and
FE Checkstyle pass.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
