github-actions[bot] commented on code in PR #70216:
URL: https://github.com/apache/doris/pull/70216#discussion_r4489012191
##########
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:
[P1] Define the tail-cache lifecycle methods in production builds
Both new definitions sit inside the existing `#ifdef BE_TEST` block (lines
50–87), while `ExecEnv::init_mem_env()` and `ExecEnv::destroy()` call them
unconditionally from `exec_env_init.cpp`. A normal non-`BE_TEST` backend
therefore has undefined references to both lifecycle methods at link time. The
focused `BE_TEST` executable includes these definitions and cannot expose this
failure. Move both definitions outside the test-only guard.
##########
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:
[P2] Apply the requesting TTL when accepting a cached table info
If a catalog with TTL 600 seconds warms this process-wide entry, a catalog
with the same client configuration and TTL 1 second still accepts it at age two
seconds: the key excludes the Doris TTL, and this check uses only the first
caller's expiry. `ALTER CATALOG` changing TTL from 600 to 1 has the same
behavior, so scanner metadata can remain cached for the old duration despite
the new setting. Store the load time and compare its age with the requesting
TTL, or isolate entries by TTL, and cover a warm entry followed by a shorter
positive TTL. This is separate from the fixed FE wire-key mismatch.
--
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]
