This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bbb04afe997 [fix](metacache) Refresh stale external table lookup state
(#69501)
bbb04afe997 is described below
commit bbb04afe9974df20bd8a71eccc2f8d2b06f87963
Author: 924060929 <[email protected]>
AuthorDate: Mon Oct 26 16:31:32 2026 +0800
[fix](metacache) Refresh stale external table lookup state (#69501)
### What problem does this PR solve?
Issue Number: None
Problem Summary: External table refresh invalidated connector metadata
but retained the FE table-name snapshot and object/ID lookup. After a
remote drop or same-name recreation, listings and lookups could continue
using the old identity. Explicit REFRESH TABLE now invalidates lookup
state and replays the same decision on follower FEs. Ordinary content
mutations retain their narrower scope and do not reload table names.
An ordinary base-table handle lookup that definitively confirms absence
also repairs the matching stale lookup state. Before remote I/O, it
captures the retained database, names and table registrations and
metadata-refresh generation. Cleanup commits only while these identities
remain current; old missing responses cannot remove refreshed or
recreated state. Views are checked separately, and connector failures
are propagated rather than treated as absence. A cold names snapshot or
a session bypassing shared caches conservatively skips this repair.
The guarded commit retires names before fencing dependent object/ID
publication, preventing an in-flight numeric lookup from repopulating
the dropped table. Case-folding modes invalidate the database lookup,
connector and derived-metadata domains to clear historical remote
spellings. Physical removal and resource-release callbacks run after all
enclosing guards are released and retain exact ownership of old entries.
Numeric loads publish their ID with the object, and reloaded objects
expose an update timestamp before schema loading.
### Release note
External table refresh reloads cached membership and lookup identity.
Live lookups confirming a remote drop automatically clear matching stale
state, including historical case-equivalent metadata, while preserving
refreshed or recreated tables.
---
.../doris/connector/cache/CatalogMetaCache.java | 5 +
.../apache/doris/connector/cache/MetaCache.java | 47 ++
.../doris/connector/cache/MetaCacheCleanup.java | 79 +++
.../doris/connector/cache/ScopedMetaCache.java | 43 +-
.../connector/cache/ScopedMetaCacheRegistry.java | 52 +-
.../connector/cache/CatalogMetaCacheTest.java | 213 ++++++
.../doris/connector/es/EsConnectorRestClient.java | 12 +-
.../connector/es/EsConnectorRestClientTest.java | 29 +
.../connector/paimon/PaimonConnectorMetadata.java | 3 +-
.../PaimonConnectorMetadataReadAuthTest.java | 6 +-
.../paimon/PaimonConnectorMetadataTest.java | 9 +
.../paimon/RecordingPaimonCatalogOps.java | 4 +
.../connector/spi/ConnectorTableMetadataOps.java | 5 +-
.../org/apache/doris/catalog/RefreshManager.java | 59 +-
.../apache/doris/datasource/ExternalCatalog.java | 8 +
.../apache/doris/datasource/ExternalDatabase.java | 81 ++-
.../org/apache/doris/datasource/ExternalTable.java | 13 +
.../doris/datasource/log/ExternalObjectLog.java | 10 +
.../datasource/metacache/FeMetaCacheEntry.java | 62 +-
.../plugin/PluginDrivenExternalTable.java | 28 +-
.../apache/doris/catalog/RefreshManagerTest.java | 196 +++++-
.../org/apache/doris/catalog/RefreshTableTest.java | 17 +-
.../doris/datasource/ExternalDatabaseTest.java | 105 +++
.../datasource/metacache/FeMetaCacheEntryTest.java | 40 ++
.../plugin/PluginDrivenExternalTableTest.java | 22 +-
.../plugin/PluginDrivenMissingTableTest.java | 718 +++++++++++++++++++++
.../hive/test_hive_external_lookup_refresh.out | 35 +
.../hive/test_hive_case_sensibility.groovy | 3 +-
.../hive/test_hive_external_lookup_refresh.groovy | 96 +++
.../test_iceberg_hadoop_case_sensibility.groovy | 5 +-
.../test_iceberg_hms_case_sensibility.groovy | 3 +-
.../test_iceberg_rest_case_sensibility.groovy | 5 +-
32 files changed, 1962 insertions(+), 51 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
index d84c389b04b..45b53b48fef 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java
@@ -151,6 +151,11 @@ public final class CatalogMetaCache implements
AutoCloseable {
registry.invalidate(ScopePath.catalog());
}
+ /** Runs a short auxiliary-index update at the same publication boundary
as catalog invalidation. */
+ public void invalidateCatalogAndRun(Runnable commitAction) {
+ registry.invalidateAndRunUnderPublicationGate(ScopePath.catalog(),
commitAction);
+ }
+
public void invalidateDatabase(String database) {
registry.invalidate(ScopePath.database(database));
}
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
index 07afd00576d..f3b425e0463 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java
@@ -18,7 +18,9 @@
package org.apache.doris.connector.cache;
import java.util.Objects;
+import java.util.Optional;
import java.util.function.BiConsumer;
+import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -89,6 +91,51 @@ public final class MetaCache<K, V> {
Objects.requireNonNull(commitAction, "commitAction can not be
null"));
}
+ /** Captures an exact retained registration without loading or holding
locks across a remote read. */
+ public Optional<Observation> observe(K key, Object expectedValue) {
+ K nonNullKey = Objects.requireNonNull(key, "key");
+ return delegate.observe(nonNullKey, definition.scope(nonNullKey),
+ Objects.requireNonNull(expectedValue, "expectedValue"));
+ }
+
+ /**
+ * A short guarded action may nest observations in parent-to-child order.
It must not perform I/O,
+ * re-enter this cache, or invalidate this cache's owner. Returning false
leaves this entry unchanged.
+ * Conditional invalidation still removes the observed value if its
dependency cleanup throws.
+ */
+ public static final class Observation {
+ private final ObservationOperation operation;
+
+ Observation(ObservationOperation operation) {
+ this.operation = operation;
+ }
+
+ public boolean runIfCurrent(BooleanSupplier action) {
+ return operation.run(false, Objects.requireNonNull(action,
"action"), () -> {
+ });
+ }
+
+ public boolean invalidateIfCurrent(BooleanSupplier action) {
+ return invalidateIfCurrent(action, () -> {
+ });
+ }
+
+ /**
+ * Runs the final action after retiring this registration, before
releasing its publication guard.
+ * It follows the same parent-to-child lock order and
no-I/O/no-reentry contract as the guarded action.
+ * It also runs when authorized dependency cleanup throws, but never
when authorization returns false.
+ */
+ public boolean invalidateIfCurrent(BooleanSupplier action, Runnable
afterInvalidation) {
+ return operation.run(true, Objects.requireNonNull(action,
"action"),
+ Objects.requireNonNull(afterInvalidation,
"afterInvalidation"));
+ }
+ }
+
+ @FunctionalInterface
+ interface ObservationOperation {
+ boolean run(boolean invalidate, BooleanSupplier action, Runnable
afterInvalidation);
+ }
+
public void invalidateKey(K key) {
delegate.invalidateKey(Objects.requireNonNull(key, "key can not be
null"));
}
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java
new file mode 100644
index 00000000000..fb6f235aec3
--- /dev/null
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java
@@ -0,0 +1,79 @@
+// 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.connector.cache;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/** Defers physical removal and resource release until an enclosing guarded
cache mutation has finished. */
+public final class MetaCacheCleanup {
+ private static final ThreadLocal<Deque<Runnable>> PENDING = new
ThreadLocal<>();
+
+ private MetaCacheCleanup() {
+ }
+
+ /**
+ * Logical invalidation still completes inside the action. Detached
registrations and removal callbacks
+ * are cleaned synchronously after the outermost action returns, including
when that action fails.
+ */
+ public static void runDeferred(Runnable action) {
+ if (PENDING.get() != null) {
+ action.run();
+ return;
+ }
+ Deque<Runnable> pending = new ArrayDeque<>();
+ PENDING.set(pending);
+ Throwable failure = null;
+ try {
+ action.run();
+ } catch (RuntimeException | Error e) {
+ failure = e;
+ }
+ try {
+ Runnable cleanup;
+ while ((cleanup = pending.pollFirst()) != null) {
+ try {
+ cleanup.run();
+ } catch (RuntimeException | Error e) {
+ if (failure == null) {
+ failure = e;
+ } else if (failure != e) {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ } finally {
+ PENDING.remove();
+ }
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure != null) {
+ throw (Error) failure;
+ }
+ }
+
+ static void execute(Runnable cleanup) {
+ Deque<Runnable> pending = PENDING.get();
+ if (pending == null) {
+ cleanup.run();
+ } else {
+ pending.addLast(cleanup);
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java
index 7dfaf88eda9..b342dfa5023 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java
@@ -271,6 +271,45 @@ public final class ScopedMetaCache<K, V> implements
AutoCloseable {
}
}
+ Optional<MetaCache.Observation> observe(K key, ScopePath path, Object
expectedValue) {
+ VersionedValue<K, V> value = currentVersionedValue(key, path);
+ if (value == null || value.value != expectedValue) {
+ return Optional.empty();
+ }
+ ScopedMetaCacheRegistry.ScopeObservation scope =
registry.observe(value.scopeSnapshot);
+ return Optional.of(new MetaCache.Observation((invalidate, action,
afterInvalidation) -> deferRemovals(() -> {
+ BooleanSupplier commit = () -> scope.runIfCurrent(() -> {
+ synchronized (value.keyNode) {
+ if (closed.get() || data.getIfPresent(key) != value ||
!value.isCurrent(registry, keyNodes)) {
+ return false;
+ }
+ boolean invalidateValue = invalidate;
+ try {
+ if (!action.getAsBoolean()) {
+ invalidateValue = false;
+ return false;
+ }
+ return true;
+ } finally {
+ // Once authorized, a throwing dependency cleanup must
not retain this stale value.
+ if (invalidateValue) {
+ exactInvalidationSequence =
exactInvalidationSequence.add(BigInteger.ONE);
+ if (!activeBulkStarts.isEmpty()) {
+ exactInvalidations.put(key,
exactInvalidationSequence);
+ }
+ replaceKeyState(value.keyNode);
+ value.keyNode.loadPublicationState.set(new
Object());
+ removeCurrentVersion(key, value);
+ afterInvalidation.run();
+ }
+ }
+ }
+ });
+ return invalidate ?
bulkInvalidationGate.write(commit::getAsBoolean)
+ : bulkInvalidationGate.readBoolean(commit);
+ })));
+ }
+
public V getIfPresent(K key, ScopePath path) {
Objects.requireNonNull(key, "key can not be null");
Objects.requireNonNull(path, "path can not be null");
@@ -1124,7 +1163,9 @@ public final class ScopedMetaCache<K, V> implements
AutoCloseable {
DeferredRemoval<K, V> removal;
while ((removal = removalDeferral.removals.pollFirst()) != null) {
try {
- completeRemoval(removal.versioned, removal.cause,
removal.budgetEviction);
+ DeferredRemoval<K, V> detached = removal;
+ MetaCacheCleanup.execute(() ->
+ completeRemoval(detached.versioned,
detached.cause, detached.budgetEviction));
} catch (RuntimeException | Error e) {
if (failure == null) {
failure = e;
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java
index 918b2829a79..7f749acbab5 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java
@@ -212,6 +212,32 @@ public final class ScopedMetaCacheRegistry implements
AutoCloseable {
tryPrune(invalidated.existing);
}
+ /** Keeps a short auxiliary-index mutation ordered with value publication,
then cleans values outside the gate. */
+ void invalidateAndRunUnderPublicationGate(ScopePath path, Runnable
commitAction) {
+ Objects.requireNonNull(path, "path can not be null");
+ Runnable action = Objects.requireNonNull(commitAction, "commitAction
can not be null");
+ checkOpen();
+ AtomicReference<InvalidatedScope> invalidated = new
AtomicReference<>();
+ try {
+ publicationGate.write(() -> {
+ List<ScopeNode> existing = resolveExisting(path);
+ bumpPublicationStates(existing);
+ if (existing.size() == path.level().ordinal() + 1) {
+ ScopeNode target = existing.get(existing.size() - 1);
+ invalidated.set(new InvalidatedScope(existing,
replaceState(target)));
+ }
+ action.run();
+ return null;
+ });
+ } finally {
+ InvalidatedScope scope = invalidated.get();
+ if (scope != null) {
+ cleanDetachedState(scope.oldState);
+ tryPrune(scope.existing);
+ }
+ }
+ }
+
void invalidate(List<ScopePath> paths, Runnable afterStateReplacement) {
Objects.requireNonNull(paths, "paths can not be null");
Objects.requireNonNull(afterStateReplacement, "afterStateReplacement
can not be null");
@@ -242,6 +268,26 @@ public final class ScopedMetaCacheRegistry implements
AutoCloseable {
invalidated.forEach(scope -> tryPrune(scope.existing));
}
+ ScopeObservation observe(ScopeSnapshot snapshot) {
+ return new ScopeObservation(snapshot);
+ }
+
+ final class ScopeObservation {
+ private final ScopeSnapshot snapshot;
+ private final PublicationState publicationState;
+
+ private ScopeObservation(ScopeSnapshot snapshot) {
+ this.snapshot = snapshot;
+ this.publicationState =
snapshot.leafNode().descendantPublication.get();
+ }
+
+ boolean runIfCurrent(BooleanSupplier action) {
+ return publicationGate.readBoolean(() ->
snapshot.isCurrent(ScopedMetaCacheRegistry.this)
+ && snapshot.leafNode().descendantPublication.get() ==
publicationState
+ && action.getAsBoolean());
+ }
+ }
+
public ScopeMetrics metrics() {
ScopeMetricsAccumulator accumulator = new ScopeMetricsAccumulator();
collectMetrics(root, accumulator);
@@ -433,12 +479,16 @@ public final class ScopedMetaCacheRegistry implements
AutoCloseable {
}
private void cleanDetachedState(ScopeState state) {
+ MetaCacheCleanup.execute(() -> cleanDetachedStateNow(state));
+ }
+
+ private void cleanDetachedStateNow(ScopeState state) {
state.entries.forEach((address, ownershipToken) -> {
if (state.entries.remove(address, ownershipToken)) {
address.removeExpected(ownershipToken);
}
});
- state.children.values().forEach(child ->
cleanDetachedState(child.current.get()));
+ state.children.values().forEach(child ->
cleanDetachedStateNow(child.current.get()));
state.children.clear();
}
diff --git
a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java
b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java
index 3bb6ad66e80..f5f1beda662 100644
---
a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java
+++
b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java
@@ -25,13 +25,226 @@ import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
class CatalogMetaCacheTest {
private static final CacheSpec SPEC = CacheSpec.of(true,
CacheSpec.CACHE_NO_TTL, 100);
+ @Test
+ void deferredCleanupDetachesImmediatelyAndPreservesRecreatedValues()
throws Exception {
+ Object guard = new Object();
+ AtomicInteger removals = new AtomicInteger();
+ AtomicInteger lockedRemovals = new AtomicInteger();
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ MetaCache<String, Object> cache =
owner.create(MetaCacheDefinition.<String, Object>builder(
+ "deferred", SPEC, ignored -> ScopePath.table("db", "t"))
+ .removalListener((key, value, cause) -> {
+ if (Thread.holdsLock(guard)) {
+ lockedRemovals.incrementAndGet();
+ }
+ removals.incrementAndGet();
+ }).build());
+ Object old = new Object();
+ Object recreated = new Object();
+ cache.put("t", old);
+ MetaCacheCleanup.runDeferred(() -> {
+ synchronized (guard) {
+ MetaCacheCleanup.runDeferred(() ->
owner.invalidateTable("db", "t"));
+ Assertions.assertNull(cache.getIfPresent("t"));
+ Assertions.assertEquals(0, removals.get());
+ try {
+ // New publication can finish while physical cleanup
of the old scope is pending.
+ worker.submit(() -> cache.put("t", recreated)).get(10,
TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new AssertionError(e);
+ }
+ Assertions.assertSame(recreated, cache.getIfPresent("t"));
+ }
+ });
+ Assertions.assertEquals(1, removals.get());
+ Assertions.assertSame(recreated, cache.getIfPresent("t"));
+ MetaCacheCleanup.runDeferred(() -> {
+ synchronized (guard) {
+ Assertions.assertTrue(cache.observe("t",
recreated).get().invalidateIfCurrent(() -> true));
+ Assertions.assertNull(cache.getIfPresent("t"));
+ Assertions.assertEquals(1, removals.get());
+ }
+ });
+ Assertions.assertEquals(2, removals.get());
+ Assertions.assertEquals(0, lockedRemovals.get());
+ } finally {
+ worker.shutdownNow();
+ }
+ }
+
+ @Test
+ void deferredCleanupDrainsAfterFailureAndKeepsOriginalException() {
+ AtomicInteger cleaned = new AtomicInteger();
+ IllegalStateException original = new IllegalStateException("action");
+ IllegalArgumentException cleanupFailure = new
IllegalArgumentException("cleanup");
+ Assertions.assertSame(original,
Assertions.assertThrows(IllegalStateException.class, () ->
+ MetaCacheCleanup.runDeferred(() -> {
+ MetaCacheCleanup.execute(() -> {
+ throw cleanupFailure;
+ });
+ MetaCacheCleanup.execute(cleaned::incrementAndGet);
+ throw original;
+ })));
+ Assertions.assertEquals(1, cleaned.get());
+ Assertions.assertArrayEquals(new Throwable[] {cleanupFailure},
original.getSuppressed());
+ // The thread-local context is cleared even after failure.
+ MetaCacheCleanup.execute(cleaned::incrementAndGet);
+ Assertions.assertEquals(2, cleaned.get());
+ }
+
+ @Test
+ void
observationFinalActionSeesRetiredNamesAndAlsoRunsAfterAuthorizedFailure()
throws Exception {
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ MetaCache<String, Object> cache =
owner.create(MetaCacheDefinition.<String, Object>builder(
+ "observed-final-action", SPEC, ignored ->
ScopePath.catalog()).build());
+ Object snapshot = new Object();
+ cache.put("db", snapshot);
+ MetaCache.Observation observed = cache.observe("db",
snapshot).get();
+ AtomicInteger finalized = new AtomicInteger();
+ Runnable afterRetirement = () -> {
+ try {
+ // A cache-only reader can check visibility without
entering the held publication gate.
+ Assertions.assertNull(worker.submit(() ->
cache.getIfPresent("db")).get(10, TimeUnit.SECONDS));
+ } catch (Exception e) {
+ throw new AssertionError(e);
+ }
+ finalized.incrementAndGet();
+ };
+ Assertions.assertFalse(observed.invalidateIfCurrent(() -> false,
afterRetirement));
+ Assertions.assertEquals(0, finalized.get());
+ Assertions.assertSame(snapshot, cache.getIfPresent("db"));
+ Assertions.assertTrue(observed.invalidateIfCurrent(() -> true,
afterRetirement));
+ Assertions.assertEquals(1, finalized.get());
+ Assertions.assertFalse(observed.invalidateIfCurrent(() -> true,
afterRetirement));
+ cache.put("db", snapshot);
+ Assertions.assertThrows(IllegalStateException.class, () ->
+ cache.observe("db", snapshot).get().invalidateIfCurrent(()
-> {
+ throw new IllegalStateException("authorized cleanup
failed");
+ }, afterRetirement));
+ Assertions.assertEquals(2, finalized.get());
+ } finally {
+ worker.shutdownNow();
+ }
+ }
+
+ @Test
+ void observationRejectsReplacementScopeChangeAndSameInstanceReinsertion() {
+ CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ try {
+ MetaCache<String, Object> cache =
owner.create(MetaCacheDefinition.<String, Object>builder(
+ "observed", SPEC, ignored -> ScopePath.table("db",
"t")).build());
+ Object old = new Object();
+ cache.put("t", old);
+ MetaCache.Observation first = cache.observe("t", old).get();
+ cache.put("t", new Object());
+ Assertions.assertFalse(first.invalidateIfCurrent(() -> true));
+ cache.put("t", old);
+ Assertions.assertFalse(first.invalidateIfCurrent(() -> true));
+ MetaCache.Observation beforeRefresh = cache.observe("t",
old).get();
+ owner.invalidateTable("db", "t");
+ cache.put("t", old);
+ Assertions.assertFalse(beforeRefresh.invalidateIfCurrent(() ->
true));
+ MetaCache.Observation current = cache.observe("t", old).get();
+ Assertions.assertFalse(current.invalidateIfCurrent(() -> false));
+ Assertions.assertSame(old, cache.getIfPresent("t"));
+ Assertions.assertTrue(current.invalidateIfCurrent(() -> true));
+ Assertions.assertNull(cache.getIfPresent("t"));
+ cache.put("t", old);
+ MetaCache.Observation beforeClose = cache.observe("t", old).get();
+ owner.close();
+ Assertions.assertFalse(beforeClose.invalidateIfCurrent(() ->
true));
+ } finally {
+ owner.close();
+ }
+ }
+
+ @Test
+ void nestedObservationLeavesNamesIntactWhenObjectWasReplaced() {
+ try (CatalogMetaCache namesOwner = CatalogMetaCache.unmanaged();
+ CatalogMetaCache objectsOwner = CatalogMetaCache.unmanaged()) {
+ MetaCache<String, Object> names =
namesOwner.create(MetaCacheDefinition.<String, Object>builder(
+ "names", SPEC, ignored -> ScopePath.catalog()).build());
+ MetaCache<String, Object> objects =
objectsOwner.create(MetaCacheDefinition.<String, Object>builder(
+ "objects", SPEC, ignored -> ScopePath.catalog()).build());
+ Object snapshot = new Object();
+ Object old = new Object();
+ names.put("db", snapshot);
+ objects.put("t", old);
+ MetaCache.Observation nameObservation = names.observe("db",
snapshot).get();
+ MetaCache.Observation objectObservation = objects.observe("t",
old).get();
+ Object replacement = new Object();
+ objects.put("t", replacement);
+ AtomicInteger actions = new AtomicInteger();
+ Assertions.assertFalse(nameObservation.invalidateIfCurrent(() ->
+ objectObservation.invalidateIfCurrent(() -> {
+ actions.incrementAndGet();
+ return true;
+ })));
+ Assertions.assertSame(snapshot, names.getIfPresent("db"));
+ Assertions.assertSame(replacement, objects.getIfPresent("t"));
+ Assertions.assertEquals(0, actions.get());
+ MetaCache.Observation currentObject = objects.observe("t",
replacement).get();
+ Assertions.assertThrows(IllegalStateException.class, () ->
nameObservation.invalidateIfCurrent(() ->
+ currentObject.invalidateIfCurrent(() -> {
+ throw new IllegalStateException("dependency cleanup
failed");
+ })));
+ Assertions.assertNull(names.getIfPresent("db"));
+ Assertions.assertNull(objects.getIfPresent("t"));
+ }
+ }
+
+ @Test
+ void observedRemovalFencesPreexistingBulkAndSingleLoads() throws Exception
{
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ MetaCache<String, Object> cache =
owner.create(MetaCacheDefinition.<String, Object>builder(
+ "objects", SPEC, ignored -> ScopePath.table("db",
"t")).build());
+ Object old = new Object();
+ cache.put("t", old);
+ try (MetaCache.BulkLoad<String, Object> bulk =
cache.beginBulkLoad(ScopePath.database("db"))) {
+ MetaCache.Observation observation = cache.observe("t",
old).get();
+ Assertions.assertTrue(observation.invalidateIfCurrent(() ->
true));
+ Assertions.assertFalse(bulk.publish("t", old));
+ Assertions.assertNull(cache.getIfPresent("t"));
+ }
+ Future<Object> load = worker.submit(() -> cache.get("t", key -> {
+ started.countDown();
+ try {
+ Assertions.assertTrue(release.await(10, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ throw new AssertionError(e);
+ }
+ return old;
+ }));
+ Assertions.assertTrue(started.await(10, TimeUnit.SECONDS));
+ Object current = new Object();
+ cache.put("t", current);
+ Assertions.assertTrue(cache.observe("t",
current).get().invalidateIfCurrent(() -> true));
+ release.countDown();
+ Assertions.assertSame(old, load.get(10, TimeUnit.SECONDS));
+ Assertions.assertNull(cache.getIfPresent("t"));
+ } finally {
+ release.countDown();
+ worker.shutdownNow();
+ }
+ }
+
@Test
void unmanagedOwnershipMustBeSelectedExplicitly() throws Exception {
Assertions.assertTrue(Modifier.isPrivate(CatalogMetaCache.class.getDeclaredConstructor().getModifiers()));
diff --git
a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java
b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java
index c509f45004f..d90c821f48b 100644
---
a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java
+++
b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java
@@ -106,10 +106,16 @@ public class EsConnectorRestClient {
OkHttpClient client = httpSslEnable ? getOrCreateSslClient() :
PLAIN_CLIENT;
String node = currentNode();
try (Response response = executeResponse(client, node, path, null)) {
- return response.isSuccessful();
+ if (response.isSuccessful()) {
+ return true;
+ }
+ if (response.code() == 404) {
+ return false;
+ }
+ throw new DorisConnectorException("Failed to check index " +
indexName
+ + ": HTTP " + response.code());
} catch (IOException e) {
- LOG.warn("existIndex error", e);
- return false;
+ throw new DorisConnectorException("Failed to check index " +
indexName, e);
}
}
diff --git
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
index 16439e765bd..d1354c93744 100644
---
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
+++
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
@@ -17,10 +17,14 @@
package org.apache.doris.connector.es;
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import com.sun.net.httpserver.HttpServer;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -34,6 +38,31 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public class EsConnectorRestClientTest {
+ @Test
+ public void indexAbsenceRequiresNotFoundResponse() throws Exception {
+ AtomicInteger status = new AtomicInteger(200);
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/t/_mapping", exchange -> {
+ exchange.sendResponseHeaders(status.get(), -1);
+ exchange.close();
+ });
+ server.start();
+ EsConnectorRestClient client = new EsConnectorRestClient(
+ new String[]{"http://127.0.0.1:"; +
server.getAddress().getPort()}, null, null, false, null);
+ try {
+ Assertions.assertTrue(client.existIndex("t"));
+ status.set(404);
+ Assertions.assertFalse(client.existIndex("t"));
+ for (int failure : new int[]{401, 403, 500}) {
+ status.set(failure);
+ Assertions.assertThrows(DorisConnectorException.class, () ->
client.existIndex("t"));
+ }
+ } finally {
+ server.stop(0);
+ }
+ Assertions.assertThrows(DorisConnectorException.class, () ->
client.existIndex("t"));
+ }
+
@Test
public void testConstructorWithAuth() throws Exception {
EsConnectorRestClient client = new EsConnectorRestClient(
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index c9de0bba921..87d313e1642 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -285,8 +285,7 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
return Optional.<ConnectorTableHandle>of(handle);
});
} catch (Exception e) {
- LOG.warn("Failed to get Paimon table handle: {}.{}", dbName,
tableName, e);
- return Optional.empty();
+ throw new DorisConnectorException("Failed to get Paimon table
handle: " + dbName + "." + tableName, e);
}
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java
index b85499d9e99..ec2174d7029 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java
@@ -18,7 +18,6 @@
package org.apache.doris.connector.paimon;
import org.apache.doris.connector.spi.DorisConnectorException;
-import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.types.DataTypes;
@@ -31,7 +30,6 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
/**
* M-11 (rereview2 #6) read-path Kerberos doAs tests: every remote metadata
READ RPC must run INSIDE
@@ -156,8 +154,8 @@ public class PaimonConnectorMetadataReadAuthTest {
RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
RecordingConnectorContext ctx = new RecordingConnectorContext();
ctx.failAuth = true;
- Optional<ConnectorTableHandle> result = metadata(ops,
ctx).getTableHandle(null, "db1", "t1");
- Assertions.assertFalse(result.isPresent());
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadata(ops, ctx).getTableHandle(null, "db1", "t1"));
Assertions.assertTrue(ops.log.isEmpty(),
"auth failure must abort BEFORE the getTable seam runs");
Assertions.assertEquals(1, ctx.authCount);
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java
index 4a7c5316f9f..564b4809509 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java
@@ -181,6 +181,15 @@ public class PaimonConnectorMetadataTest {
Assertions.assertFalse(handleOpt.isPresent());
}
+ @Test
+ public void getTableHandlePropagatesLookupFailure() {
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.tableLookupFailure = new IllegalStateException("metastore
timeout");
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadataWith(ops).getTableHandle(null, "db1", "t1"));
+ Assertions.assertSame(ops.tableLookupFailure, failure.getCause());
+ }
+
@Test
public void getColumnHandlesReloadFallbackReloadsWhenTransientTableNull() {
RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
index 73147f40620..a6465faf6c6 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
@@ -69,6 +69,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
boolean throwDatabaseNotExist;
boolean throwTableNotExist;
+ RuntimeException tableLookupFailure;
// ---- B3 DDL capture fields (inputs the metadata layer passed to the
seam) ----
Schema lastCreatedSchema;
@@ -163,6 +164,9 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public Table getTable(Identifier identifier) throws
Catalog.TableNotExistException {
+ if (tableLookupFailure != null) {
+ throw tableLookupFailure;
+ }
log.add("getTable:" + identifier.getFullName());
lastGetTableId = identifier;
if (throwTableNotExist) {
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java
index 4d29727e1b8..64263b639f5 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java
@@ -62,7 +62,10 @@ import java.util.Optional;
*/
public interface ConnectorTableMetadataOps {
- /** Retrieves a table handle for the given database and table name. */
+ /**
+ * Retrieves a table handle for the given database and table name. Empty
means the table does not exist;
+ * authentication, transport and other lookup failures must propagate
rather than report absence.
+ */
@ConnectorMustImplement
default Optional<ConnectorTableHandle> getTableHandle(
ConnectorSession session, String dbName,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
index 8e5586b344f..9c1eaaf899d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java
@@ -170,8 +170,8 @@ public class RefreshManager {
return;
}
long updateTime = System.currentTimeMillis();
- refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime);
- ExternalObjectLog log = ExternalObjectLog.createForRefreshTable(
+ refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime, true);
+ ExternalObjectLog log =
ExternalObjectLog.createForExplicitRefreshTable(
catalog.getId(), db.getFullName(), table.getName(),
updateTime);
env.getEditLog().logRefreshExternalTable(log);
}
@@ -209,8 +209,12 @@ public class RefreshManager {
try {
invalidateAllConnectorCachesIfPresent(catalog);
} finally {
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(
- log.getCatalogId(), dbId, localDbName, tableId,
localTableName);
+ if (log.isInvalidateTableLookup() &&
catalog.getLowerCaseTableNames() != 0) {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(log.getCatalogId(), dbId,
localDbName);
+ } else {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(
+ log.getCatalogId(), dbId, localDbName, tableId,
localTableName);
+ }
}
LOG.info("database object cache is cold when replaying refresh
table; "
+ "invalidated caches by local names {}.{}: {}",
@@ -225,14 +229,25 @@ public class RefreshManager {
((PluginDrivenExternalCatalog)
catalog).getConnector().invalidateDb(db.get().getRemoteName());
}
} finally {
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(
- log.getCatalogId(), dbId, db.get().getFullName(),
tableId, localTableName);
+ try {
+ if (log.isInvalidateTableLookup()) {
+ db.get().invalidateTableLookup(localTableName);
+ }
+ } finally {
+ if (log.isInvalidateTableLookup() &&
catalog.getLowerCaseTableNames() != 0) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(
+ log.getCatalogId(), dbId,
db.get().getFullName());
+ } else {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(
+ log.getCatalogId(), dbId,
db.get().getFullName(), tableId, localTableName);
+ }
+ }
}
LOG.info("table object cache is cold when replaying refresh table;
invalidated caches by local name {}: {}",
localTableName, log.debugForRefreshTable());
return;
}
- refreshTableInternal(db.get(), table.get(), log.getLastUpdateTime());
+ refreshTableInternal(db.get(), table.get(), log.getLastUpdateTime(),
log.isInvalidateTableLookup());
}
private void replayRenameTable(ExternalObjectLog log, ExternalCatalog
catalog,
@@ -301,6 +316,13 @@ public class RefreshManager {
}
public void refreshTableInternal(ExternalDatabase db, ExternalTable table,
long updateTime) {
+ refreshTableInternal(db, table, updateTime, false);
+ }
+
+ private void refreshTableInternal(ExternalDatabase db, ExternalTable
table, long updateTime,
+ boolean invalidateTableLookup) {
+ boolean invalidateDatabase = invalidateTableLookup &&
table.getCatalog().getLowerCaseTableNames() != 0;
+ table.advanceMetadataRefreshGeneration();
table.unsetObjectCreated();
if (updateTime > 0) {
table.setUpdateTime(updateTime);
@@ -308,11 +330,28 @@ public class RefreshManager {
// Connector metadata is the row-count source, so invalidate it before
engine and row-count caches.
try {
if (table.getCatalog() instanceof PluginDrivenExternalCatalog) {
- ((PluginDrivenExternalCatalog)
table.getCatalog()).getConnector()
- .invalidateTable(db.getRemoteName(),
table.getRemoteName());
+ if (invalidateDatabase) {
+ // A refreshed local name may resolve to a different
remote spelling in this database.
+ ((PluginDrivenExternalCatalog)
table.getCatalog()).getConnector()
+ .invalidateDb(db.getRemoteName());
+ } else {
+ ((PluginDrivenExternalCatalog)
table.getCatalog()).getConnector()
+ .invalidateTable(db.getRemoteName(),
table.getRemoteName());
+ }
}
} finally {
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(table);
+ try {
+ if (invalidateTableLookup) {
+ db.invalidateTableLookup(table.getName());
+ }
+ } finally {
+ if (invalidateDatabase) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(
+ table.getCatalog().getId(), db.getId(),
db.getFullName());
+ } else {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(table);
+ }
+ }
}
LOG.info("refresh table {}, id {} from db {} in catalog {}, update
time: {}",
table.getName(), table.getId(), db.getFullName(),
db.getCatalog().getName(), updateTime);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
index 2ee07578704..1a8acd018f8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -1219,6 +1219,14 @@ public abstract class ExternalCatalog
this.dbIdNameIndex = new IdNameIndex("external database");
}
+ /** Cache-only identity fence for a read through a retained database
instance. */
+ public Optional<FeMetaCacheEntry.Observation>
observeDatabase(ExternalDatabase<?> database) {
+ if (!initialized || databases == null) {
+ return Optional.empty();
+ }
+ return databases.observe(database.getFullName(), database);
+ }
+
public void addDatabaseForTest(ExternalDatabase<? extends ExternalTable>
db) {
buildMetaCache();
// Test helpers only seed object/id state and keep names cache cold
unless the test fills it explicitly.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java
index 3d24d0c3b78..d59987d2e8a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java
@@ -32,6 +32,7 @@ import
org.apache.doris.common.lock.MonitoredReentrantReadWriteLock;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.Util;
import org.apache.doris.connector.cache.CacheSpec;
+import org.apache.doris.connector.cache.MetaCacheCleanup;
import org.apache.doris.datasource.infoschema.ExternalInfoSchemaDatabase;
import org.apache.doris.datasource.infoschema.ExternalMysqlDatabase;
import org.apache.doris.datasource.metacache.FeMetaCacheEntry;
@@ -318,8 +319,10 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
}
}
- // Step 4: Build and return the table instance using the resolved
names and other parameters
- return buildTableInternal(remoteTableName, localTableName, tblId,
catalog, db);
+ // A successful object reload is a metadata refresh even if the caller
never requests its schema.
+ T table = buildTableInternal(remoteTableName, localTableName, tblId,
catalog, db);
+ table.setUpdateTime(System.currentTimeMillis());
+ return table;
}
protected abstract T buildTableInternal(String remoteTableName, String
localTableName, long tblId,
@@ -347,6 +350,50 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
return Optional.ofNullable(tables.getIfPresent(tableName));
}
+ /**
+ * Captures only a shared base table with a matching retained names
snapshot. Guard order is catalog DB,
+ * table names, then table object, matching incremental name/object
publication. No remote load is performed.
+ */
+ public Optional<Runnable> observeTableLookup(ExternalTable table, Runnable
invalidateMetadata) {
+ if (!initialized ||
extCatalog.shouldBypassSchemaCache(SessionContext.current())) {
+ return Optional.empty();
+ }
+ NameCacheValue names = getTableNamesValue(false);
+ if (names == null ||
!table.getRemoteName().equals(names.remoteNameOfLocalName(table.getName()))) {
+ return Optional.empty();
+ }
+ long metadataRefreshGeneration = table.getMetadataRefreshGeneration();
+ return extCatalog.observeDatabase(this).flatMap(database ->
+ tableNames.observe("", names).flatMap(membership ->
+ tables.observe(table.getName(), table).map(object ->
() -> MetaCacheCleanup.runDeferred(() -> {
+ // Catalog reset/connector replacement holds this
monitor too. Only local invalidation
+ // runs here; remote reads happened before
entering this guarded commit.
+ synchronized (extCatalog) {
+ if (!initialized || !extCatalog.isInitialized()
+ ||
table.getMetadataRefreshGeneration() != metadataRefreshGeneration
+ ||
Env.getCurrentEnv().getCatalogMgr().getCatalog(extCatalog.getId())
+ != extCatalog) {
+ return;
+ }
+ database.runIfCurrent(() ->
membership.invalidateIfCurrent(
+ () -> object.runIfCurrent(() -> {
+ invalidateMetadata.run();
+ return true;
+ }),
+ () -> {
+ // Retire names before fencing
object loads: a numeric reader may have
+ // captured its name already and
begun loading through the old snapshot.
+ if (isStoredTableNamesLowerCase()
|| isTableNamesCaseInsensitive()) {
+
tables.invalidateAllAndRun(tableIdNameIndex::clear);
+ } else {
+
tables.invalidateKeyAndRun(table.getName(),
+ () ->
tableIdNameIndex.removeName(table.getName()));
+ }
+ }));
+ }
+ }))));
+ }
+
/**
* Resolve the retained local table name for replay without loading table
metadata.
*
@@ -544,6 +591,10 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
}
long expectedTableId = Util.genIdByName(extCatalog.getName(), name,
finalName);
tableIdNameIndex.checkCanPut(expectedTableId, finalName);
+ return getTableAndPublishId(finalName);
+ }
+
+ private T getTableAndPublishId(String finalName) {
return tables.getAndRunIfCurrent(
finalName,
(localTableName, table) ->
!localTableName.equals(tableIdNameIndex.getName(table.getId())),
@@ -743,14 +794,34 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
}
}
+ /** Refreshes table membership and lookup identity without loading remote
metadata. */
+ public void invalidateTableLookup(String tableName) {
+ if (!isInitialized()) {
+ return;
+ }
+ // Folding or comparing local names without case can hide another
retained remote spelling.
+ // Fence every object/ID load in this database when refreshing that
lookup identity.
+ if (isStoredTableNamesLowerCase() || isTableNamesCaseInsensitive()) {
+ invalidateAllTableCache();
+ return;
+ }
+ String localTableName = resolveTableNameForInvalidation(tableName);
+ try {
+ invalidateTableCache(localTableName);
+ } finally {
+ resetMetaCacheNames();
+ }
+ }
+
private void invalidateAllTableCache() {
if (tableNames != null) {
tableNames.invalidateAll();
}
if (tables != null) {
- tables.invalidateAll();
+ tables.invalidateAllAndRun(tableIdNameIndex::clear);
+ } else {
+ tableIdNameIndex.clear();
}
- tableIdNameIndex.clear();
}
@Override
@@ -760,7 +831,7 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
if (tableName == null) {
return null;
}
- return tables.get(tableName);
+ return getTableAndPublishId(tableName);
}
public long getLastUpdateTime() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java
index 4c874fe3721..3e81cb70aec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java
@@ -92,6 +92,8 @@ public class ExternalTable implements TableIf, Writable,
GsonPostProcessable {
// record the table update time, like insert/alter/delete
protected volatile long updateTime = 0;
+ // Fences missing-table responses that started before a metadata-only
refresh of this object.
+ private transient volatile long metadataRefreshGeneration;
protected long dbId;
protected boolean objectCreated;
@@ -338,6 +340,17 @@ public class ExternalTable implements TableIf, Writable,
GsonPostProcessable {
this.updateTime = updateTime;
}
+ public long getMetadataRefreshGeneration() {
+ return metadataRefreshGeneration;
+ }
+
+ /** Ordered with guarded missing-table cleanup, which holds the same
catalog monitor. */
+ public void advanceMetadataRefreshGeneration() {
+ synchronized (catalog) {
+ metadataRefreshGeneration++;
+ }
+ }
+
@Override
public long getLastCheckTime() {
return 0;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java
index 30464457e60..308b3cd5184 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java
@@ -61,6 +61,9 @@ public class ExternalObjectLog implements Writable {
@SerializedName(value = "newPartitionNames")
private List<String> newPartitionNames;
+ @SerializedName(value = "invalidateTableLookup")
+ private boolean invalidateTableLookup;
+
@SerializedName(value = "lastUpdateTime")
private long lastUpdateTime;
@@ -85,6 +88,13 @@ public class ExternalObjectLog implements Writable {
return externalObjectLog;
}
+ public static ExternalObjectLog createForExplicitRefreshTable(long
catalogId, String dbName, String tblName,
+ long updateTime) {
+ ExternalObjectLog log = createForRefreshTable(catalogId, dbName,
tblName, updateTime);
+ log.setInvalidateTableLookup(true);
+ return log;
+ }
+
public static ExternalObjectLog createForRefreshPartitions(long catalogId,
String dbName, String tblName,
List<String> modifiedPartNames, List<String> newPartNames, long
updateTime) {
ExternalObjectLog externalObjectLog = new ExternalObjectLog();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java
index 1cb7f0dcf9c..7de440c881b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java
@@ -34,12 +34,14 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
+import java.util.function.BooleanSupplier;
import java.util.function.Function;
import java.util.function.Predicate;
import javax.annotation.Nullable;
@@ -239,7 +241,8 @@ public class FeMetaCacheEntry<K, V> {
&& (!effectiveEnabled || data.getIfPresent(key) ==
value);
if (current && required.test(key, value) && isCurrent(stripe,
token)) {
try {
- action.accept(key, value);
+ // A warm-value index repair must share the
publication barrier with bulk invalidation.
+ data.compareAndSet(key, value, value, () ->
action.accept(key, value));
} catch (RuntimeException | Error throwable) {
data.invalidateKey(key);
throw throwable;
@@ -254,6 +257,55 @@ public class FeMetaCacheEntry<K, V> {
}
}
+ public Optional<Observation> observe(K key, Object expectedValue) {
+ StripeState<K> stripe = stripeState(key);
+ synchronized (stripe) {
+ return data.observe(key, expectedValue)
+ .map(value -> new Observation(value, stripe, () ->
bumpAction(stripe, key)));
+ }
+ }
+
+ /** Orders conditional removal with both cache publication and auxiliary
identity actions. */
+ public static final class Observation {
+ private final MetaCache.Observation value;
+ private final Object stripe;
+ private final Runnable invalidateActions;
+
+ private Observation(MetaCache.Observation value, Object stripe,
Runnable invalidateActions) {
+ this.value = value;
+ this.stripe = stripe;
+ this.invalidateActions = invalidateActions;
+ }
+
+ public boolean runIfCurrent(BooleanSupplier action) {
+ synchronized (stripe) {
+ return value.runIfCurrent(action);
+ }
+ }
+
+ public boolean invalidateIfCurrent(BooleanSupplier action) {
+ return invalidateIfCurrent(action, () -> {
+ });
+ }
+
+ /** Retires the observed value before fencing dependent publication,
under the same stripe and guard. */
+ public boolean invalidateIfCurrent(BooleanSupplier action, Runnable
afterInvalidation) {
+ synchronized (stripe) {
+ return value.invalidateIfCurrent(() -> {
+ boolean invalidate = true;
+ try {
+ invalidate = action.getAsBoolean();
+ return invalidate;
+ } finally {
+ if (invalidate) {
+ invalidateActions.run();
+ }
+ }
+ }, afterInvalidation);
+ }
+ }
+ }
+
public V getIfPresent(K key) {
return data.getIfPresent(key);
}
@@ -360,6 +412,12 @@ public class FeMetaCacheEntry<K, V> {
}
public void invalidateAll() {
+ invalidateAllAndRun(() -> {
+ });
+ }
+
+ public void invalidateAllAndRun(Runnable commitAction) {
+ Runnable action = Objects.requireNonNull(commitAction, "commitAction
can not be null");
for (int i = 0; i < stripeCount; i++) {
StripeState<K> stripe = stripeStates.get(i);
if (stripe != null) {
@@ -370,7 +428,7 @@ public class FeMetaCacheEntry<K, V> {
}
}
}
- owner.invalidateCatalog();
+ owner.invalidateCatalogAndRun(action);
}
public void forEach(BiConsumer<K, V> consumer) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java
index 6cbea40af75..93d78e42e67 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java
@@ -119,8 +119,34 @@ public class PluginDrivenExternalTable extends
ExternalTable {
*/
public Optional<ConnectorTableHandle> resolveConnectorTableHandle(
ConnectorSession session, ConnectorMetadata metadata) {
+ PluginDrivenExternalCatalog owner = (PluginDrivenExternalCatalog)
catalog;
+ Connector connector = owner.getConnector();
+ ExternalDatabase<?> database = getDb();
+ boolean invalidateDatabase = owner.getLowerCaseTableNames() != 0;
+ Optional<Runnable> observation = database.observeTableLookup(this, ()
-> {
+ try {
+ if (invalidateDatabase) {
+ connector.invalidateDb(db.getRemoteName());
+ } else {
+ connector.invalidateTable(db.getRemoteName(),
getRemoteName());
+ }
+ } finally {
+ if (invalidateDatabase) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(
+ owner.getId(), database.getId(),
database.getFullName());
+ } else {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(this);
+ }
+ }
+ });
String dbName = db != null ? db.getRemoteName() : "";
- return metadata.getTableHandle(session, dbName, getRemoteName());
+ Optional<ConnectorTableHandle> handle =
metadata.getTableHandle(session, dbName, getRemoteName());
+ // A view has no table handle. Check live view existence before
treating an empty handle as a
+ // dropped table; the cached isView flag can outlive a remote view
deletion.
+ if (!handle.isPresent() && !metadata.viewExists(session, dbName,
getRemoteName())) {
+ observation.ifPresent(Runnable::run);
+ }
+ return handle;
}
/**
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java
index 44059931620..dedd1b42b82 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java
@@ -40,12 +40,15 @@ import org.apache.doris.datasource.test.TestExternalCatalog;
import org.apache.doris.datasource.test.TestExternalDatabase;
import org.apache.doris.datasource.test.TestExternalTable;
import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.gson.GsonUtils;
import com.google.common.util.concurrent.MoreExecutors;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.MockedConstruction;
@@ -53,6 +56,7 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -116,6 +120,7 @@ public class RefreshManagerTest {
mockedEnv.close();
mockedEnv = null;
}
+ MutableCatalogProvider.reset();
}
@Test
@@ -167,6 +172,145 @@ public class RefreshManagerTest {
CATALOG_ID, DATABASE_ID, DATABASE_NAME, TABLE_ID, TABLE_NAME);
}
+ @Test
+ public void testExplicitRefreshReloadsTableNamesAndObject() throws
Exception {
+ useMutableCatalog();
+ TestExternalTable oldTable = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotNull(oldTable);
+
+ MutableCatalogProvider.putTable("new_column");
+ new RefreshManager().handleRefreshTable(CATALOG_NAME, DATABASE_NAME,
TABLE_NAME, false);
+
+ Assertions.assertNull(database.getCachedTableNamesForTest());
+ Assertions.assertNull(database.getCachedTableForTest(TABLE_NAME));
+ TestExternalTable refreshed = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotSame(oldTable, refreshed);
+ ArgumentCaptor<ExternalObjectLog> log =
ArgumentCaptor.forClass(ExternalObjectLog.class);
+ Mockito.verify(editLog).logRefreshExternalTable(log.capture());
+ Assertions.assertTrue(log.getValue().isInvalidateTableLookup());
+ }
+
+ @Test
+ public void
testExplicitRefreshReplayRetainsReplacementUpdateTimeBeforeSchemaLoad() {
+ useMutableCatalog();
+ TestExternalTable oldTable = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotNull(oldTable);
+ long refreshTime = System.currentTimeMillis();
+
+ new
RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, refreshTime));
+
+ TestExternalTable replacement = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotSame(oldTable, replacement);
+ Assertions.assertTrue(replacement.getUpdateTime() >= refreshTime);
+ }
+
+ @Test
+ public void testContentRefreshRetainsTableNamesAndObject() {
+ useMutableCatalog();
+ TestExternalTable table = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotNull(table);
+
+ new RefreshManager().refreshTableAfterExternalMutation(table);
+
+ Assertions.assertNotNull(database.getCachedTableNamesForTest());
+ Assertions.assertSame(table,
database.getCachedTableForTest(TABLE_NAME));
+ ArgumentCaptor<ExternalObjectLog> log =
ArgumentCaptor.forClass(ExternalObjectLog.class);
+ Mockito.verify(editLog).logRefreshExternalTable(log.capture());
+ Assertions.assertFalse(log.getValue().isInvalidateTableLookup());
+ }
+
+ @Test
+ public void testExplicitRefreshReplayClearsNamesWhenTableObjectIsCold() {
+ useMutableCatalog();
+ TestExternalTable table = database.getTableNullable(TABLE_NAME);
+ Assertions.assertNotNull(table);
+ database.evictTableObjectForTest(TABLE_NAME);
+ MutableCatalogProvider.removeTable();
+
+ new
RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L));
+
+ Assertions.assertNull(database.getCachedTableNamesForTest());
+ Assertions.assertTrue(database.getTableNamesWithLock().isEmpty());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void
testExplicitRefreshReplayInvalidatesDatabaseInCaseInsensitiveMode(int mode) {
+ useMutableCatalog(mode);
+ Assertions.assertNotNull(database.getTableNullable(TABLE_NAME));
+ ExternalObjectLog log =
ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L);
+
+ new RefreshManager().replayRefreshTable(
+ GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(log),
ExternalObjectLog.class));
+
+ Assertions.assertNull(database.getCachedTableNamesForTest());
+ Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID,
DATABASE_NAME);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void
testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMode(int mode)
{
+ useMutableCatalog(mode);
+ catalog.setInitializedForTest(false);
+
+ new
RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L));
+
+ Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID,
DATABASE_NAME);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void
testExplicitRefreshReplayInvalidatesConnectorDatabaseInCaseInsensitiveMode(int
mode) {
+ PluginCatalogFixture fixture = usePluginCatalog(mode);
+ ExternalTable oldTable = new ExternalTable(TABLE_ID, TABLE_NAME,
TABLE_NAME,
+ fixture.catalog, fixture.database,
TableIf.TableType.PLUGIN_EXTERNAL_TABLE);
+ String otherSpelling = TABLE_NAME.toUpperCase(java.util.Locale.ROOT);
+ ExternalTable otherTable = new ExternalTable(
+ Util.genIdByName(CATALOG_NAME, DATABASE_NAME, otherSpelling),
otherSpelling, otherSpelling,
+ fixture.catalog, fixture.database,
TableIf.TableType.PLUGIN_EXTERNAL_TABLE);
+ fixture.database.addTableForTest(oldTable);
+ fixture.database.addTableForTest(otherTable);
+
+ new
RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L));
+
+ Mockito.verify(fixture.connector).invalidateDb(DATABASE_NAME);
+ Mockito.verify(fixture.connector,
Mockito.never()).invalidateTable(DATABASE_NAME, TABLE_NAME);
+ Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID,
DATABASE_NAME);
+
Assertions.assertFalse(fixture.database.getTableForReplay(otherSpelling).isPresent());
+ }
+
+ @Test
+ public void
testExplicitRefreshInLowerCaseModeClearsHistoricalRemoteSpellings() {
+ PluginCatalogFixture fixture = usePluginCatalog(1);
+ ExternalTable table = new ExternalTable(TABLE_ID, TABLE_NAME, "FOO",
+ fixture.catalog, fixture.database,
TableIf.TableType.PLUGIN_EXTERNAL_TABLE);
+ fixture.database.addTableForTest(table);
+ Map<String, String> connectorEntries = new HashMap<>();
+ connectorEntries.put("Foo", "old incarnation");
+ connectorEntries.put("FOO", "current incarnation");
+ Mockito.doAnswer(inv -> {
+ connectorEntries.clear();
+ return null;
+ }).when(fixture.connector).invalidateDb(DATABASE_NAME);
+ Mockito.doAnswer(inv -> {
+ connectorEntries.remove(inv.getArgument(1));
+ return null;
+ }).when(fixture.connector).invalidateTable(Mockito.eq(DATABASE_NAME),
Mockito.anyString());
+
+ new
RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable(
+ CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L));
+
+ Assertions.assertTrue(connectorEntries.isEmpty());
+ Mockito.verify(fixture.connector).invalidateDb(DATABASE_NAME);
+ Mockito.verify(fixture.connector,
Mockito.never()).invalidateTable(Mockito.anyString(), Mockito.anyString());
+ Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID,
DATABASE_NAME);
+ }
+
@Test
public void
testRefreshTableRemovesRowCountLoadedDuringConnectorInvalidation() throws
Exception {
PluginCatalogFixture fixture = usePluginCatalog();
@@ -366,8 +510,14 @@ public class RefreshManagerTest {
}
private PluginCatalogFixture usePluginCatalog() {
+ return usePluginCatalog(0);
+ }
+
+ private PluginCatalogFixture usePluginCatalog(int lowerCaseTableNames) {
Connector connector = Mockito.mock(Connector.class);
- Map<String, String> properties = Collections.singletonMap("type",
"test");
+ Map<String, String> properties = new HashMap<>();
+ properties.put("type", "test");
+ properties.put(ExternalCatalog.LOWER_CASE_TABLE_NAMES,
Integer.toString(lowerCaseTableNames));
PluginDrivenExternalCatalog pluginCatalog = new
PluginDrivenExternalCatalog(
CATALOG_ID, CATALOG_NAME, null, properties, "", connector);
pluginCatalog.setInitializedForTest(true);
@@ -405,6 +555,45 @@ public class RefreshManagerTest {
Assertions.assertEquals(NEW_TABLE_NAME,
constraintManager.newTableName.getTbl());
}
+ private void useMutableCatalog() {
+ useMutableCatalog(0);
+ }
+
+ private void useMutableCatalog(int lowerCaseTableNames) {
+ MutableCatalogProvider.putTable("old_column");
+ Map<String, String> properties = new HashMap<>();
+ properties.put("catalog_provider.class",
MutableCatalogProvider.class.getName());
+ properties.put(ExternalCatalog.LOWER_CASE_TABLE_NAMES,
Integer.toString(lowerCaseTableNames));
+ catalog = new TestExternalCatalog(CATALOG_ID, CATALOG_NAME, "",
properties, "");
+ catalog.setInitializedForTest(true);
+ database = new CountingDatabase(catalog, DATABASE_ID, DATABASE_NAME,
DATABASE_NAME);
+ database.setInitializedForTest(true);
+ catalog.addDatabaseForTest(database);
+ testingCatalogMgr.setCatalog(catalog);
+ }
+
+ public static class MutableCatalogProvider implements
TestExternalCatalog.TestCatalogProvider {
+ private static final Map<String, Map<String, List<Column>>> METADATA =
new HashMap<>();
+
+ static void putTable(String columnName) {
+ METADATA.computeIfAbsent(DATABASE_NAME, ignored -> new HashMap<>())
+ .put(TABLE_NAME, Collections.singletonList(new
Column(columnName, PrimitiveType.INT)));
+ }
+
+ static void removeTable() {
+ METADATA.get(DATABASE_NAME).remove(TABLE_NAME);
+ }
+
+ static void reset() {
+ METADATA.clear();
+ }
+
+ @Override
+ public Map<String, Map<String, List<Column>>> getMetadata() {
+ return METADATA;
+ }
+ }
+
public static class EmptyCatalogProvider implements
TestExternalCatalog.TestCatalogProvider {
@Override
public Map<String, Map<String, List<Column>>> getMetadata() {
@@ -471,6 +660,11 @@ public class RefreshManagerTest {
public CatalogIf<? extends DatabaseIf<? extends TableIf>>
getCatalog(long id) {
return catalog != null && id == catalog.getId() ? catalog : null;
}
+
+ @Override
+ public CatalogIf<? extends DatabaseIf<? extends TableIf>>
getCatalog(String name) {
+ return catalog != null && name.equals(catalog.getName()) ? catalog
: null;
+ }
}
private static class TestingEnv extends Env {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java
index 36edaacdf01..40252053a5d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java
@@ -104,13 +104,16 @@ public class RefreshTableTest extends TestWithFeService {
Assertions.assertFalse(table.isObjectCreated());
long l3 = table.getUpdateTime();
Assertions.assertTrue(l3 >= l2);
- table.getFullSchema();
- // only table.getFullSchema() can change table.lastUpdateTime
- long l4 = table.getUpdateTime();
- Assertions.assertTrue(l4 > l3);
- // updateTime is equal to schema update time as default
- long l5 = table.getUpdateTime();
- Assertions.assertTrue(l5 >= l4);
+ TestExternalTable refreshed = (TestExternalTable)
test1.getDbNullable("db1").getTable("tbl11").get();
+ Assertions.assertNotSame(table, refreshed);
+ // information_schema reads UPDATE_TIME without forcing schema loading.
+ Assertions.assertTrue(refreshed.getUpdateTime() >= l3);
+ Assertions.assertSame(refreshed,
test1.getDbNullable("db1").getTableNullable(refreshed.getId()));
+ refreshed.getFullSchema();
+ // Schema loading updates the replacement table, not the invalidated
table object.
+ long l4 = refreshed.getUpdateTime();
+ Assertions.assertTrue(l4 >= l3);
+ Assertions.assertEquals(l3, table.getUpdateTime());
// external info schema db
ExternalInfoSchemaDatabase infoDb = (ExternalInfoSchemaDatabase)
test1.getDbNullable(InfoSchemaDb.DATABASE_NAME);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java
index d2d9a91e772..8c0d1b91a69 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java
@@ -55,6 +55,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
+import java.util.function.BiPredicate;
public class ExternalDatabaseTest extends TestWithFeService {
@Override
@@ -63,6 +65,53 @@ public class ExternalDatabaseTest extends TestWithFeService {
createDefaultCtx().setThreadLocalInfo();
}
+ @Test
+ public void testExplicitRefreshInvalidatesNamesObjectAndId() {
+ NameMissTableCatalogProvider.reset();
+ try {
+ NameMissTableCatalogProvider.putTable("db1", "Foo");
+ NameMissTableCatalog catalog = new NameMissTableCatalog(0);
+ InspectableDatabase db = new InspectableDatabase(catalog, 200L,
"db1", "db1");
+ db.setInitializedForTest(true);
+ TestExternalTable oldTable = db.getTableNullable("Foo");
+ Assertions.assertNotNull(oldTable);
+
+ NameMissTableCatalogProvider.MOCKED_META.get("db1").remove("Foo");
+ db.invalidateTableLookup("Foo");
+
+ Assertions.assertNull(db.getCachedTableForTest("Foo"));
+
Assertions.assertNull(db.getCachedTableNameByIdForTest(oldTable.getId()));
+ Assertions.assertTrue(db.getTableNamesWithLock().isEmpty());
+ } finally {
+ NameMissTableCatalogProvider.reset();
+ }
+ }
+
+ @Test
+ public void testExplicitRefreshFencesCaseEquivalentOldObject() {
+ NameMissTableCatalogProvider.reset();
+ try {
+ NameMissTableCatalogProvider.putTable("db_ci", "Dest");
+ CaseInsensitiveCatalog catalog = new CaseInsensitiveCatalog();
+ InspectableDatabase db = new InspectableDatabase(catalog, 210L,
"db_ci", "db_ci");
+ db.setInitializedForTest(true);
+ TestExternalTable oldTable = db.getTableNullable("Dest");
+ Assertions.assertNotNull(oldTable);
+
+
NameMissTableCatalogProvider.MOCKED_META.get("db_ci").remove("Dest");
+ NameMissTableCatalogProvider.putTable("db_ci", "DEST");
+ db.resetMetaCacheNames();
+ Assertions.assertTrue(db.getTableNamesWithLock().contains("DEST"));
+ db.invalidateTableLookup("DEST");
+
+ Assertions.assertNull(db.getCachedTableForTest("Dest"));
+
Assertions.assertNull(db.getCachedTableNameByIdForTest(oldTable.getId()));
+ Assertions.assertNotSame(oldTable, db.getTableNullable("DEST"));
+ } finally {
+ NameMissTableCatalogProvider.reset();
+ }
+ }
+
@Test
public void testResetMetaCacheNamesKeepsObjectAndIdCache() {
InspectableCatalog catalog = new InspectableCatalog();
@@ -262,6 +311,62 @@ public class ExternalDatabaseTest extends
TestWithFeService {
}
}
+ @Test
+ public void testNumericLookupAfterRefreshPublishesReplacementId() throws
Exception {
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ ExecutorService queryExecutor = Executors.newSingleThreadExecutor();
+ CountDownLatch nameCaptured = new CountDownLatch(1);
+ CountDownLatch releaseLookup = new CountDownLatch(1);
+ try {
+ InspectableCatalog catalog = new InspectableCatalog();
+ InspectableDatabase db = new InspectableDatabase(catalog, 245L,
"db1", "db1");
+ db.setInitializedForTest(true);
+ TestExternalTable oldTable = db.getTableNullable("tbl_base");
+ Assertions.assertNotNull(oldTable);
+
+ FeMetaCacheEntry<String, TestExternalTable> objectEntry =
+ new FeMetaCacheEntry<String, TestExternalTable>(
+ "numeric_lookup_refresh_race",
+ ignored -> new TestExternalTable(oldTable.getId(),
oldTable.getName(),
+ oldTable.getRemoteName(), catalog, db),
+ CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, 10L),
+ refreshExecutor,
+ false) {
+ @Override
+ public TestExternalTable get(String key) {
+ nameCaptured.countDown();
+ awaitLatch(releaseLookup);
+ return super.get(key);
+ }
+
+ @Override
+ public TestExternalTable getAndRunIfCurrent(String key,
+ BiPredicate<String, TestExternalTable>
actionRequired,
+ BiConsumer<String, TestExternalTable>
currentValueAction) {
+ nameCaptured.countDown();
+ awaitLatch(releaseLookup);
+ return super.getAndRunIfCurrent(key,
actionRequired, currentValueAction);
+ }
+ };
+ setTablesEntryForTest(db, objectEntry);
+
+ Future<TestExternalTable> lookup = queryExecutor.submit(() ->
db.getTableNullable(oldTable.getId()));
+ Assertions.assertTrue(nameCaptured.await(3L, TimeUnit.SECONDS));
+ db.invalidateTableLookup(oldTable.getName());
+ releaseLookup.countDown();
+
+ TestExternalTable replacement = lookup.get(3L, TimeUnit.SECONDS);
+ Assertions.assertNotNull(replacement);
+ Assertions.assertNotSame(oldTable, replacement);
+ Assertions.assertSame(replacement,
objectEntry.getIfPresent(oldTable.getName()));
+ Assertions.assertEquals(oldTable.getName(),
db.getCachedTableNameByIdForTest(oldTable.getId()));
+ } finally {
+ releaseLookup.countDown();
+ queryExecutor.shutdownNow();
+ refreshExecutor.shutdownNow();
+ }
+ }
+
@Test
public void testHotNamedLookupSkipsLockWhenIdAlreadyMapped() throws
Exception {
ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java
index ba4673d8065..826cd9ff942 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java
@@ -25,7 +25,9 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -287,6 +289,44 @@ public class FeMetaCacheEntryTest {
}
}
+ @Test
+ public void testBulkInvalidationOrdersNewObjectAndIdPublication() throws
Exception {
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ ExecutorService workers = Executors.newFixedThreadPool(2);
+ CountDownLatch indexClearStarted = new CountDownLatch(1);
+ CountDownLatch finishIndexClear = new CountDownLatch(1);
+ CountDownLatch newLookupStarted = new CountDownLatch(1);
+ Map<String, Integer> ids = new ConcurrentHashMap<>();
+ try {
+ FeMetaCacheEntry<String, Integer> entry = new FeMetaCacheEntry<>(
+ "objects", String::length, ENABLED, refreshExecutor,
false);
+ entry.getAndRunIfCurrent("old", (key, value) -> ids.put(key,
value));
+
+ Future<?> invalidate = workers.submit(() ->
entry.invalidateAllAndRun(() -> {
+ indexClearStarted.countDown();
+ await(finishIndexClear);
+ ids.clear();
+ }));
+ await(indexClearStarted);
+ Future<Integer> newLookup = workers.submit(() -> {
+ newLookupStarted.countDown();
+ return entry.getAndRunIfCurrent("new", (key, value) ->
ids.put(key, value));
+ });
+ await(newLookupStarted);
+ finishIndexClear.countDown();
+
+ invalidate.get(3L, TimeUnit.SECONDS);
+ Assertions.assertEquals(Integer.valueOf(3), newLookup.get(3L,
TimeUnit.SECONDS));
+ Assertions.assertEquals(Integer.valueOf(3),
entry.getIfPresent("new"));
+ Assertions.assertEquals(Integer.valueOf(3), ids.get("new"));
+ Assertions.assertFalse(ids.containsKey("old"));
+ } finally {
+ finishIndexClear.countDown();
+ workers.shutdownNow();
+ refreshExecutor.shutdownNow();
+ }
+ }
+
@Test
public void testColdValueRemainsHiddenUntilAuxiliaryIndexPublication()
throws Exception {
ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java
index f21ed9f8006..64deaeaab12 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java
@@ -101,6 +101,16 @@ public class PluginDrivenExternalTableTest {
return s;
}
+ private static PluginDrivenExternalTable tableWithCachelessDatabase() {
+ PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(database.getRemoteName()).thenReturn("");
+ Mockito.when(database.observeTableLookup(Mockito.eq(table),
Mockito.any()))
+ .thenReturn(Optional.empty());
+ Deencapsulation.setField(table, "db", database);
+ return table;
+ }
+
// ==================== ยง4.4 W3: per-handle write-admission capability
probes ====================
/**
@@ -134,7 +144,7 @@ public class PluginDrivenExternalTableTest {
ConnectorSession session = noneScopedSession();
Mockito.when(catalog.buildConnectorSession()).thenReturn(session);
Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session);
- PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS);
+ PluginDrivenExternalTable table = tableWithCachelessDatabase();
Deencapsulation.setField(table, "catalog", catalog);
return table;
}
@@ -207,7 +217,7 @@ public class PluginDrivenExternalTableTest {
ConnectorSession session = noneScopedSession();
Mockito.when(catalog.buildConnectorSession()).thenReturn(session);
Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session);
- PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS);
+ PluginDrivenExternalTable table = tableWithCachelessDatabase();
Deencapsulation.setField(table, "catalog", catalog);
return table;
}
@@ -249,7 +259,7 @@ public class PluginDrivenExternalTableTest {
ConnectorSession session = noneScopedSession();
Mockito.when(catalog.buildConnectorSession()).thenReturn(session);
Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session);
- PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS);
+ PluginDrivenExternalTable table = tableWithCachelessDatabase();
Deencapsulation.setField(table, "catalog", catalog);
return table;
}
@@ -317,8 +327,7 @@ public class PluginDrivenExternalTableTest {
Mockito.when(catalog.buildConnectorSession()).thenReturn(session);
Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session);
- PluginDrivenExternalTable table =
- Mockito.mock(PluginDrivenExternalTable.class,
Mockito.CALLS_REAL_METHODS);
+ PluginDrivenExternalTable table = tableWithCachelessDatabase();
Deencapsulation.setField(table, "catalog", catalog);
SchemaCacheValue scv = Mockito.mock(SchemaCacheValue.class);
Mockito.when(scv.getSchema()).thenReturn(BASE_SCHEMA);
@@ -450,8 +459,7 @@ public class PluginDrivenExternalTableTest {
PluginDrivenExternalCatalog catalog =
Mockito.mock(PluginDrivenExternalCatalog.class);
Mockito.when(catalog.getConnector()).thenReturn(connector);
Mockito.when(catalog.buildConnectorSession()).thenReturn(session);
- PluginDrivenExternalTable table = Mockito.mock(
- PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS);
+ PluginDrivenExternalTable table = tableWithCachelessDatabase();
Deencapsulation.setField(table, "catalog", catalog);
Mockito.doReturn(99L).when(table).getId();
ConnectContext ctx = new ConnectContext();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java
new file mode 100644
index 00000000000..6062697ee67
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java
@@ -0,0 +1,718 @@
+// 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.datasource.plugin;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.RefreshManager;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.Util;
+import org.apache.doris.connector.spi.Connector;
+import org.apache.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.ConnectorMetadata;
+import org.apache.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.ConnectorTableSchema;
+import org.apache.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.datasource.CacheException;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.SchemaCacheKey;
+import org.apache.doris.datasource.SchemaCacheValue;
+import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.datasource.metacache.ExternalMetaCache;
+import org.apache.doris.datasource.metacache.FeMetaCacheEntry;
+import org.apache.doris.datasource.metacache.IdNameIndex;
+import org.apache.doris.datasource.metacache.NameCacheValue;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+
+/** Tests missing-table repair through real FE lookup/schema caches and a
controllable connector. */
+public class PluginDrivenMissingTableTest extends TestWithFeService {
+ private static final AtomicLong IDS = new AtomicLong(910000);
+ private final Map<String, String> remoteTables = new ConcurrentHashMap<>();
+ private final Set<String> remoteViews = ConcurrentHashMap.newKeySet();
+ private Connector connector;
+ private ConnectorMetadata metadata;
+ private TestCatalog catalog;
+ private TestDatabase database;
+ private final Map<String, String> remoteCache = new ConcurrentHashMap<>();
+ private ExecutorService worker;
+ private CountDownLatch missingStarted;
+ private CountDownLatch releaseMissing;
+ private AtomicBoolean blockMissing;
+ private boolean failLookup;
+ private boolean failInvalidation;
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ FeConstants.runningUnitTest = true;
+ createDefaultCtx().setThreadLocalInfo();
+ }
+
+ @BeforeEach
+ void setup() {
+ failLookup = false;
+ failInvalidation = false;
+ remoteTables.put("a", "old_column");
+ remoteTables.put("b", "sibling_column");
+ missingStarted = new CountDownLatch(1);
+ releaseMissing = new CountDownLatch(1);
+ blockMissing = new AtomicBoolean();
+ worker = Executors.newFixedThreadPool(2);
+ connector = Mockito.mock(Connector.class);
+ metadata = Mockito.mock(ConnectorMetadata.class);
+
Mockito.when(connector.getMetadata(Mockito.any())).thenReturn(metadata);
+ Mockito.doAnswer(call -> {
+ if (failInvalidation) {
+ throw new IllegalStateException("connector cleanup failed");
+ }
+ remoteCache.remove(call.getArgument(1));
+ return null;
+ }).when(connector).invalidateTable(Mockito.anyString(),
Mockito.anyString());
+ Mockito.doAnswer(call -> {
+ remoteCache.clear();
+ return null;
+ }).when(connector).invalidateDb(Mockito.anyString());
+ Mockito.when(metadata.getTableHandle(Mockito.any(),
Mockito.anyString(), Mockito.anyString()))
+ .thenAnswer(call -> {
+ if (failLookup) {
+ throw new IllegalStateException("metastore timeout");
+ }
+ String name = call.getArgument(2);
+ if (remoteViews.contains(name)) {
+ return Optional.empty();
+ }
+ String column = remoteTables.get(name);
+ if (column == null) {
+ if (blockMissing.compareAndSet(true, false)) {
+ missingStarted.countDown();
+ Assertions.assertTrue(releaseMissing.await(10,
TimeUnit.SECONDS));
+ }
+ return Optional.empty();
+ }
+ return Optional.of(new Handle(name, column));
+ });
+ Mockito.when(metadata.viewExists(Mockito.any(), Mockito.anyString(),
Mockito.anyString()))
+ .thenAnswer(call -> remoteViews.contains(call.getArgument(2)));
+ Mockito.when(metadata.getTableSchema(Mockito.any(),
Mockito.any())).thenAnswer(call -> {
+ Handle handle = call.getArgument(1);
+ return new ConnectorTableSchema(handle.name,
Collections.singletonList(new ConnectorColumn(
+ handle.column, ConnectorType.of("INT", -1, -1), "", true,
null, true)),
+ null, Collections.emptyMap());
+ });
+ Mockito.when(metadata.fromRemoteColumnName(Mockito.any(),
Mockito.anyString(),
+ Mockito.anyString(), Mockito.anyString())).thenAnswer(call ->
call.getArgument(3));
+ Mockito.when(metadata.fromRemoteTableName(Mockito.any(),
Mockito.anyString(), Mockito.anyString()))
+ .thenAnswer(call -> call.getArgument(2));
+ Mockito.when(metadata.fromRemoteDatabaseName(Mockito.any(),
Mockito.anyString()))
+ .thenAnswer(call -> call.getArgument(1));
+ catalog = new TestCatalog(IDS.incrementAndGet());
+ CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr();
+ manager.getIdToCatalog().put(catalog.getId(), catalog);
+ Map<String, CatalogIf<?>> byName = Deencapsulation.getField(manager,
"nameToCatalog");
+ byName.put(catalog.getName(), catalog);
+ installDatabase();
+ }
+
+ @AfterEach
+ void cleanup() throws Exception {
+ releaseMissing.countDown();
+ worker.shutdownNow();
+ Assertions.assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS));
+
Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalog(catalog.getId());
+ CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr();
+ manager.getIdToCatalog().remove(catalog.getId());
+ Map<String, CatalogIf<?>> byName = Deencapsulation.getField(manager,
"nameToCatalog");
+ byName.remove(catalog.getName());
+ remoteTables.clear();
+ remoteViews.clear();
+ remoteCache.clear();
+ }
+
+ private void installDatabase() {
+ catalog.setInitializedForTest(true);
+ database = new TestDatabase();
+ database.setInitializedForTest(true);
+ catalog.addDatabaseForTest(database);
+ }
+
+ private org.apache.doris.connector.cache.MetaCache<SchemaCacheKey,
SchemaCacheValue> schemas() {
+ ExternalMetaCache cache =
Deencapsulation.invoke(Env.getCurrentEnv().getExtMetaCacheMgr(), "engine",
"default");
+ return cache.entry(catalog.getId(), "schema", SchemaCacheKey.class,
SchemaCacheValue.class);
+ }
+
+ private PluginDrivenExternalTable warmTarget() {
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ PluginDrivenExternalTable table = database.getTableNullable("a");
+ table.getFullSchema();
+ database.getTableNullable("b").getFullSchema();
+ remoteCache.put("a", "old");
+ remoteCache.put("b", "sibling");
+ // Keep membership/object entries hot while forcing the next schema
read to consult the source.
+ schemas().invalidateKey(new
SchemaCacheKey(table.getOrBuildNameMapping()));
+ return table;
+ }
+
+ @Test
+ void missingTableRepairsNamesObjectsIdsAndConnectorMetadata() {
+ PluginDrivenExternalTable old = warmTarget();
+ PluginDrivenExternalTable sibling = database.getTableNullable("b");
+ List<Column> siblingSchema = sibling.getFullSchema();
+ remoteTables.remove("a");
+ Assertions.assertThrows(CacheException.class, old::getFullSchema);
+ Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock());
+ Assertions.assertNull(database.getTableNullable(old.getId()));
+ Assertions.assertNull(remoteCache.get("a"));
+ Assertions.assertEquals("sibling", remoteCache.get("b"));
+ Assertions.assertSame(sibling, database.getTableNullable("b"));
+ Assertions.assertSame(siblingSchema, sibling.getFullSchema());
+ }
+
+ @Test
+ void missingHandleRepairsNamesBeforeSchemaLoading() {
+ PluginDrivenExternalTable table = warmTarget();
+ remoteTables.remove("a");
+ ConnectorSession session = catalog.buildCrossStatementSession();
+ try {
+ Assertions.assertFalse(table.resolveConnectorTableHandle(session,
metadata).isPresent());
+ } finally {
+ session.getStatementScope().closeAll();
+ }
+ Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock());
+ Assertions.assertNull(database.getTableNullable(table.getId()));
+ Assertions.assertNull(remoteCache.get("a"));
+ }
+
+ @Test
+ void existingViewWithoutTableHandleKeepsLookupCaches() {
+ remoteViews.add("v");
+ PluginDrivenExternalTable view = database.getTableNullable("v");
+ Assertions.assertNotNull(view);
+ remoteCache.put("v", "view metadata");
+ ConnectorSession session = catalog.buildCrossStatementSession();
+ try {
+ Assertions.assertFalse(view.resolveConnectorTableHandle(session,
metadata).isPresent());
+ Assertions.assertFalse(view.resolveConnectorTableHandle(session,
metadata).isPresent());
+ } finally {
+ session.getStatementScope().closeAll();
+ }
+ Assertions.assertSame(view, database.getTableNullable("v"));
+ Assertions.assertSame(view, database.getTableNullable(view.getId()));
+ Assertions.assertTrue(database.getTableNamesWithLock().contains("v"));
+ Assertions.assertEquals("view metadata", remoteCache.get("v"));
+
+ remoteViews.remove("v");
+ readMissingHandle(view);
+ Assertions.assertNull(database.getTableNullable(view.getId()));
+ Assertions.assertFalse(database.getTableNamesWithLock().contains("v"));
+ }
+
+ @Test
+ void delayedMissingDoesNotInvalidateRecreatedTable() throws Exception {
+ verifyDelayedMissing(false);
+ }
+
+ @Test
+ void delayedMissingDoesNotInvalidateExactKeyReplacement() throws Exception
{
+ verifyDelayedMissing(true);
+ }
+
+ @Test
+ void delayedMissingDoesNotInvalidateMetadataOnlyRefresh() throws Exception
{
+ PluginDrivenExternalTable table = warmTarget();
+ remoteTables.remove("a");
+ blockMissing.set(true);
+ Future<?> missing = worker.submit(() -> readMissingHandle(table));
+ Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS));
+
+ remoteTables.put("a", "new_column");
+ new RefreshManager().refreshTableInternal(database, table, 123L);
+ List<Column> refreshedSchema = table.getFullSchema();
+ remoteCache.put("a", "refreshed");
+ int nameLoads = catalog.nameLoads.get();
+
+ releaseMissing.countDown();
+ missing.get(10, TimeUnit.SECONDS);
+ Assertions.assertSame(table, database.getTableNullable("a"));
+ Assertions.assertSame(table, database.getTableNullable(table.getId()));
+ Assertions.assertSame(refreshedSchema, table.getFullSchema());
+ Assertions.assertEquals("new_column",
refreshedSchema.get(0).getName());
+ Assertions.assertEquals("refreshed", remoteCache.get("a"));
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ Assertions.assertEquals(nameLoads, catalog.nameLoads.get());
+ }
+
+ private void verifyDelayedMissing(boolean replaceWithoutScopeInvalidation)
throws Exception {
+ PluginDrivenExternalTable old = warmTarget();
+ remoteTables.remove("a");
+ blockMissing.set(true);
+ Future<CacheException> result = worker.submit(() ->
Assertions.assertThrows(CacheException.class,
+ old::getFullSchema));
+ Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS));
+ remoteTables.put("a", "new_column");
+ PluginDrivenExternalTable replacement;
+ if (replaceWithoutScopeInvalidation) {
+ replacement = database.buildTableInternal("a", "a", old.getId(),
catalog, database);
+ database.addTableForTest(replacement);
+ schemas().put(new
SchemaCacheKey(replacement.getOrBuildNameMapping()),
replacement.initSchema().get());
+ } else {
+ new RefreshManager().handleRefreshTable(catalog.getName(), "db",
"a", false);
+ replacement = database.getTableNullable("a");
+ }
+ List<Column> newSchema = replacement.getFullSchema();
+ remoteCache.put("a", "new");
+ database.getTableNamesWithLock();
+ int nameLoads = catalog.nameLoads.get();
+ releaseMissing.countDown();
+ result.get(10, TimeUnit.SECONDS);
+ Assertions.assertSame(replacement, database.getTableNullable("a"));
+ Assertions.assertSame(replacement,
database.getTableNullable(replacement.getId()));
+ Assertions.assertSame(newSchema, replacement.getFullSchema());
+ Assertions.assertTrue(replacement.isObjectCreated());
+ Assertions.assertEquals("new_column", newSchema.get(0).getName());
+ Assertions.assertEquals("new", remoteCache.get("a"));
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ Assertions.assertEquals(nameLoads, catalog.nameLoads.get());
+ }
+
+ @Test
+ void delayedMissingDoesNotInvalidateReplacementDatabase() throws Exception
{
+ PluginDrivenExternalTable old = warmTarget();
+ remoteTables.remove("a");
+ blockMissing.set(true);
+ Future<?> result = worker.submit(() -> readMissingHandle(old));
+ Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS));
+ remoteTables.put("a", "replacement_column");
+ database = new TestDatabase();
+ database.setInitializedForTest(true);
+ catalog.addDatabaseForTest(database);
+ PluginDrivenExternalTable replacement = database.getTableNullable("a");
+ List<Column> schema = replacement.getFullSchema();
+ remoteCache.put("a", "replacement");
+ database.getTableNamesWithLock();
+ int loads = catalog.nameLoads.get();
+ releaseMissing.countDown();
+ result.get(10, TimeUnit.SECONDS);
+ Assertions.assertSame(replacement, database.getTableNullable("a"));
+ Assertions.assertSame(schema, replacement.getFullSchema());
+ Assertions.assertEquals("replacement", remoteCache.get("a"));
+ Assertions.assertEquals(loads, catalog.nameLoads.get());
+ }
+
+ @Test
+ void delayedMissingDoesNotInvalidateReplacementCatalog() throws Exception {
+ PluginDrivenExternalTable old = warmTarget();
+ remoteTables.remove("a");
+ blockMissing.set(true);
+ Future<?> result = worker.submit(() -> readMissingHandle(old));
+ Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS));
+ long catalogId = catalog.getId();
+ catalog = new TestCatalog(catalogId);
+ CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr();
+ manager.getIdToCatalog().put(catalogId, catalog);
+ Map<String, CatalogIf<?>> byName = Deencapsulation.getField(manager,
"nameToCatalog");
+ byName.put(catalog.getName(), catalog);
+ remoteTables.put("a", "replacement_column");
+ installDatabase();
+ PluginDrivenExternalTable replacement = database.getTableNullable("a");
+ List<Column> schema = replacement.getFullSchema();
+ remoteCache.put("a", "replacement");
+ database.getTableNamesWithLock();
+ int loads = catalog.nameLoads.get();
+ releaseMissing.countDown();
+ result.get(10, TimeUnit.SECONDS);
+ Assertions.assertSame(replacement, database.getTableNullable("a"));
+ Assertions.assertSame(schema, replacement.getFullSchema());
+ Assertions.assertEquals("replacement", remoteCache.get("a"));
+ Assertions.assertEquals(loads, catalog.nameLoads.get());
+ }
+
+ @Test
+ void failedConnectorCleanupStillRemovesObservedLookup() {
+ PluginDrivenExternalTable old = warmTarget();
+ remoteTables.remove("a");
+ failInvalidation = true;
+ Assertions.assertThrows(IllegalStateException.class, () ->
readMissingHandle(old));
+ Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock());
+ Assertions.assertNull(database.getTableNullable(old.getId()));
+ }
+
+ private void readMissingHandle(PluginDrivenExternalTable table) {
+ ConnectorSession session = catalog.buildCrossStatementSession();
+ try {
+ Assertions.assertFalse(table.resolveConnectorTableHandle(session,
metadata).isPresent());
+ } finally {
+ session.getStatementScope().closeAll();
+ }
+ }
+
+ @Test
+ void changedCanonicalSpellingFencesBothExistingAndNewOldObjectReads()
throws Exception {
+ catalog.lowerCaseMode = 2;
+ try {
+ remoteTables.remove("a");
+ remoteTables.put("Foo", "old_column");
+ PluginDrivenExternalTable old = database.getTableNullable("foo");
+ old.getFullSchema();
+ schemas().invalidateKey(new
SchemaCacheKey(old.getOrBuildNameMapping()));
+ remoteTables.remove("Foo");
+ blockMissing.set(true);
+ Future<CacheException> missing = worker.submit(() ->
Assertions.assertThrows(CacheException.class,
+ old::getFullSchema));
+ Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS));
+ remoteTables.put("FOO", "new_column");
+ FeMetaCacheEntry<String, NameCacheValue> names =
Deencapsulation.getField(database, "tableNames");
+ names.invalidateKey("");
+ PluginDrivenExternalTable replacement =
database.getTableNullable("foo");
+ Assertions.assertEquals("FOO", replacement.getName());
+ List<Column> schema = replacement.getFullSchema();
+ remoteCache.put("FOO", "new");
+ int loads = catalog.nameLoads.get();
+ releaseMissing.countDown();
+ missing.get(10, TimeUnit.SECONDS);
+ Assertions.assertSame(replacement,
database.getTableNullable("foo"));
+ Assertions.assertSame(schema, replacement.getFullSchema());
+ Assertions.assertEquals("new", remoteCache.get("FOO"));
+ Assertions.assertEquals(loads, catalog.nameLoads.get());
+ // A fresh request holding the old object must not combine it with
the new names snapshot.
+ ConnectorSession session = catalog.buildCrossStatementSession();
+ try {
+
Assertions.assertFalse(old.resolveConnectorTableHandle(session,
metadata).isPresent());
+ } finally {
+ session.getStatementScope().closeAll();
+ }
+ Assertions.assertSame(replacement,
database.getTableNullable("foo"));
+ Assertions.assertSame(schema, replacement.getFullSchema());
+ Assertions.assertEquals("new", remoteCache.get("FOO"));
+ } finally {
+ catalog.lowerCaseMode = 0;
+ }
+ }
+
+ @Test
+ void lookupFailureDoesNotRepairNamesOrEvictObjects() {
+ PluginDrivenExternalTable table = warmTarget();
+ remoteTables.remove("a");
+ failLookup = true;
+ Assertions.assertThrows(IllegalStateException.class,
table::getFullSchema);
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ Assertions.assertSame(table, database.getTableNullable(table.getId()));
+ Assertions.assertEquals("old", remoteCache.get("a"));
+ Assertions.assertEquals(1, catalog.nameLoads.get());
+ }
+
+ @Test
+ void numericLookupCapturedBeforeRefreshRepublishesItsIdentity() throws
Exception {
+ PluginDrivenExternalTable old = warmTarget();
+ IdNameIndex original = Deencapsulation.getField(database,
"tableIdNameIndex");
+ IdNameIndex index = Mockito.spy(original);
+ Deencapsulation.setField(database, "tableIdNameIndex", index);
+ CountDownLatch captured = new CountDownLatch(1);
+ CountDownLatch proceed = new CountDownLatch(1);
+ AtomicBoolean pause = new AtomicBoolean(true);
+ Mockito.doAnswer(call -> {
+ String name = original.getName(old.getId());
+ if (pause.compareAndSet(true, false)) {
+ captured.countDown();
+ Assertions.assertTrue(proceed.await(10, TimeUnit.SECONDS));
+ }
+ return name;
+ }).when(index).getName(old.getId());
+ Future<PluginDrivenExternalTable> lookup = worker.submit(() ->
database.getTableNullable(old.getId()));
+ try {
+ Assertions.assertTrue(captured.await(10, TimeUnit.SECONDS));
+ remoteTables.put("a", "new_column");
+ new RefreshManager().handleRefreshTable(catalog.getName(), "db",
"a", false);
+ Assertions.assertNull(original.getName(old.getId()));
+ proceed.countDown();
+ PluginDrivenExternalTable replacement = lookup.get(10,
TimeUnit.SECONDS);
+ Assertions.assertNotSame(old, replacement);
+ Assertions.assertEquals("a", index.getName(old.getId()));
+ Assertions.assertSame(replacement,
database.getTableNullable(old.getId()));
+ } finally {
+ proceed.countDown();
+ }
+ }
+
+ @Test
+ void foldedNamesRefreshClearsHistoricalRemoteSpellings() throws Exception {
+ catalog.lowerCaseMode = 1;
+ remoteTables.remove("a");
+ remoteTables.put("Foo", "first");
+ Assertions.assertNotNull(database.getTableNullable("foo"));
+ remoteCache.put("Foo", "first");
+ remoteTables.remove("Foo");
+ remoteTables.put("FOO", "second");
+ FeMetaCacheEntry<String, NameCacheValue> names =
Deencapsulation.getField(database, "tableNames");
+ names.invalidateKey("");
+ database.evictTableObjectForTest("foo");
+ PluginDrivenExternalTable second = database.getTableNullable("foo");
+ Assertions.assertEquals("FOO", second.getRemoteName());
+ remoteCache.put("FOO", "second");
+ remoteTables.remove("FOO");
+ remoteTables.put("Foo", "third");
+ new RefreshManager().handleRefreshTable(catalog.getName(), "db",
"foo", false);
+ Assertions.assertTrue(remoteCache.isEmpty());
+ PluginDrivenExternalTable replacement =
database.getTableNullable("foo");
+ Assertions.assertEquals("Foo", replacement.getRemoteName());
+ Assertions.assertEquals("third",
replacement.getFullSchema().get(0).getName());
+ }
+
+ @Test
+ void automaticRepairClearsHistoricalCaseInsensitiveSpellings() {
+ verifyAutomaticRepairClearsHistoricalSpellings(2);
+ }
+
+ @Test
+ void automaticRepairClearsHistoricalFoldedSpellings() {
+ verifyAutomaticRepairClearsHistoricalSpellings(1);
+ }
+
+ private void verifyAutomaticRepairClearsHistoricalSpellings(int mode) {
+ catalog.lowerCaseMode = mode;
+ remoteTables.remove("a");
+ remoteTables.put("Foo", "first_column");
+ PluginDrivenExternalTable first = database.getTableNullable("foo");
+ first.getFullSchema();
+ remoteCache.put("Foo", "first");
+ remoteTables.remove("Foo");
+ remoteTables.put("FOO", "second_column");
+ FeMetaCacheEntry<String, NameCacheValue> names =
Deencapsulation.getField(database, "tableNames");
+ names.invalidateKey("");
+ if (mode == 1) {
+ database.evictTableObjectForTest("foo");
+ }
+ PluginDrivenExternalTable second = database.getTableNullable("foo");
+ second.getFullSchema();
+ remoteCache.put("FOO", "second");
+ remoteTables.remove("FOO");
+ remoteTables.put("Foo", "third_column");
+ readMissingHandle(second);
+ Assertions.assertTrue(remoteCache.isEmpty());
+ PluginDrivenExternalTable third = database.getTableNullable("foo");
+ Assertions.assertNotSame(first, third);
+ Assertions.assertEquals("Foo", third.getRemoteName());
+ Assertions.assertEquals("third_column",
third.getFullSchema().get(0).getName());
+ }
+
+ @Test
+ void numericLookupCannotRefillBetweenObjectAndNamesRetirement() throws
Exception {
+ PluginDrivenExternalTable old = warmTarget();
+ IdNameIndex original = Deencapsulation.getField(database,
"tableIdNameIndex");
+ IdNameIndex index = Mockito.spy(original);
+ Deencapsulation.setField(database, "tableIdNameIndex", index);
+ CountDownLatch captured = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ CountDownLatch progress = new CountDownLatch(1);
+ CountDownLatch finishBuild = new CountDownLatch(1);
+ AtomicBoolean capture = new AtomicBoolean(true);
+ database.numericBuildProgress = progress;
+ database.finishNumericBuild = finishBuild;
+ Mockito.doAnswer(call -> {
+ String name = original.getName(old.getId());
+ if (capture.compareAndSet(true, false)) {
+ captured.countDown();
+ Assertions.assertTrue(resume.await(10, TimeUnit.SECONDS));
+ }
+ return name;
+ }).when(index).getName(old.getId());
+ FeMetaCacheEntry<String, PluginDrivenExternalTable> objects =
+ Mockito.spy(Deencapsulation.<FeMetaCacheEntry<String,
PluginDrivenExternalTable>>getField(
+ database, "tables"));
+ Deencapsulation.setField(database, "tables", objects);
+ Mockito.doAnswer(call -> {
+ Optional<FeMetaCacheEntry.Observation> observed =
(Optional<FeMetaCacheEntry.Observation>) call.callRealMethod();
+ FeMetaCacheEntry.Observation observation =
Mockito.spy(observed.get());
+ org.mockito.stubbing.Answer<Boolean> afterObjectGuard = invocation
-> {
+ boolean current = (boolean) invocation.callRealMethod();
+ if (current) {
+ // Pause the missing repair after its object guard returns
but before names retire.
+ resume.countDown();
+ Assertions.assertTrue(progress.await(10,
TimeUnit.SECONDS));
+ }
+ return current;
+ };
+
Mockito.doAnswer(afterObjectGuard).when(observation).runIfCurrent(Mockito.any(BooleanSupplier.class));
+
Mockito.doAnswer(afterObjectGuard).when(observation).invalidateIfCurrent(Mockito.any(BooleanSupplier.class));
+ return Optional.of(observation);
+ }).when(objects).observe("a", old);
+ Future<PluginDrivenExternalTable> numeric = worker.submit(() -> {
+ PluginDrivenExternalTable result =
database.getTableNullable(old.getId());
+ progress.countDown();
+ return result;
+ });
+ try {
+ Assertions.assertTrue(captured.await(10, TimeUnit.SECONDS));
+ remoteTables.remove("a");
+ readMissingHandle(old);
+ finishBuild.countDown();
+ numeric.get(10, TimeUnit.SECONDS);
+ Assertions.assertNull(objects.getIfPresent("a"));
+ Assertions.assertNull(original.getName(old.getId()));
+ Assertions.assertEquals(Set.of("b"),
database.getTableNamesWithLock());
+ } finally {
+ resume.countDown();
+ finishBuild.countDown();
+ }
+ }
+
+ @Test
+ void ordinaryWritesDoNotReloadNames() {
+ verifyOrdinaryWrite(0);
+ }
+
+ @Test
+ void foldedNameOrdinaryWritesDoNotReloadNames() {
+ verifyOrdinaryWrite(1);
+ }
+
+ @Test
+ void caseInsensitiveOrdinaryWritesDoNotReloadNames() {
+ verifyOrdinaryWrite(2);
+ }
+
+ private void verifyOrdinaryWrite(int mode) {
+ catalog.lowerCaseMode = mode;
+ PluginDrivenExternalTable table = warmTarget();
+ int loads = catalog.nameLoads.get();
+ new RefreshManager().refreshTableInternal(database, table, 123L);
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ Assertions.assertSame(table, database.getTableNullable("a"));
+ Assertions.assertEquals(loads, catalog.nameLoads.get());
+ }
+
+ @Test
+ void sessionBypassDoesNotInvalidateSharedNames() {
+ PluginDrivenExternalTable table = warmTarget();
+ remoteTables.remove("a");
+ catalog.bypassSchema = true;
+ readMissingHandle(table);
+ Assertions.assertEquals(Set.of("a", "b"),
database.getTableNamesWithLock());
+ Assertions.assertSame(table, database.getTableNullable(table.getId()));
+ Assertions.assertEquals("old", remoteCache.get("a"));
+ }
+
+ private static class Handle implements ConnectorTableHandle {
+ private final String name;
+ private final String column;
+
+ Handle(String name, String column) {
+ this.name = name;
+ this.column = column;
+ }
+ }
+
+ private class TestCatalog extends PluginDrivenExternalCatalog {
+ private final AtomicInteger nameLoads = new AtomicInteger();
+ private boolean bypassSchema;
+ private int lowerCaseMode;
+
+ TestCatalog(long id) {
+ super(id, "missing_test_" + id, "", new HashMap<>(), "",
connector);
+ }
+
+ @Override
+ protected void initLocalObjectsImpl() {
+ // The fixture injects the connector directly instead of loading a
production plugin.
+ }
+
+ @Override
+ public int getLowerCaseTableNames() {
+ return lowerCaseMode;
+ }
+
+ @Override
+ public Connector getConnector() {
+ return connector;
+ }
+
+ @Override
+ protected List<String> listDatabaseNames() {
+ return Collections.singletonList("db");
+ }
+
+ @Override
+ protected List<String> listTableNamesFromRemote(SessionContext
context, String dbName) {
+ nameLoads.incrementAndGet();
+ List<String> names = new ArrayList<>(remoteTables.keySet());
+ names.addAll(remoteViews);
+ return names;
+ }
+
+ @Override
+ public boolean shouldBypassSchemaCache(SessionContext context) {
+ return bypassSchema;
+ }
+ }
+
+ private class TestDatabase extends
ExternalDatabase<PluginDrivenExternalTable> {
+ private CountDownLatch numericBuildProgress;
+ private CountDownLatch finishNumericBuild;
+
+ TestDatabase() {
+ super(catalog, Util.genIdByName(catalog.getName(), "db"), "db",
"db");
+ }
+
+ @Override
+ protected PluginDrivenExternalTable buildTableInternal(String
remoteName, String localName, long id,
+ ExternalCatalog owner, ExternalDatabase db) {
+ if (numericBuildProgress != null) {
+ numericBuildProgress.countDown();
+ try {
+ Assertions.assertTrue(finishNumericBuild.await(10,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ throw new AssertionError(e);
+ }
+ }
+ return new PluginDrivenExternalTable(id, localName, remoteName,
owner, db) {
+ @Override
+ protected synchronized void makeSureInitialized() {
+ // The fixture registers this database explicitly;
connector schema loading remains real.
+ objectCreated = true;
+ }
+ };
+ }
+ }
+}
diff --git
a/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out
b/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out
new file mode 100644
index 00000000000..7730655d851
--- /dev/null
+++
b/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out
@@ -0,0 +1,35 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !names_before --
+owner_sibling
+owner_target
+
+-- !schema_before --
+before_col int Yes true \N
+
+-- !sibling_before --
+sibling_col int Yes true \N
+
+-- !names_after_drop --
+owner_sibling
+
+-- !sibling_after --
+sibling_col int Yes true \N
+
+-- !names_after_database_refresh --
+owner_sibling
+owner_target
+
+-- !schema_after_database_refresh --
+after_col text Yes true \N
+
+-- !schema_after_recreation --
+recreated_col bigint Yes true \N
+
+-- !observed_names_before --
+owner_observed
+
+-- !observed_schema_before --
+observed_col int Yes true \N
+
+-- !observed_names_after --
+
diff --git
a/regression-test/suites/external_table_p0/hive/test_hive_case_sensibility.groovy
b/regression-test/suites/external_table_p0/hive/test_hive_case_sensibility.groovy
index a8f7eb2f7da..2f604a4c804 100644
---
a/regression-test/suites/external_table_p0/hive/test_hive_case_sensibility.groovy
+++
b/regression-test/suites/external_table_p0/hive/test_hive_case_sensibility.groovy
@@ -138,7 +138,8 @@ suite("test_hive_case_sensibility", "p0,external") {
qt_sql11 """show tables from case_db2 like "%case_tbl14%""""
// empty
qt_sql12 """show tables from case_db2 like "%case_tbl21%""""
- order_qt_sql13 """select * from information_schema.tables
where TABLE_SCHEMA="case_db1";"""
+ // UPDATE_TIME is populated when an external table object is
loaded; it is not a case-mapping result.
+ order_qt_sql13 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where TABLE_SCHEMA="case_db1";"""
order_qt_sql14 """select * from information_schema.columns
where TABLE_SCHEMA="case_db1";"""
// 4. insert
diff --git
a/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy
b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy
new file mode 100644
index 00000000000..6885f32d91d
--- /dev/null
+++
b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy
@@ -0,0 +1,96 @@
+// 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.
+
+suite("test_hive_external_lookup_refresh", "p0,external") {
+ String enabled = context.config.otherConfigs.get("enableHiveTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Hive test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String hmsPort = context.config.otherConfigs.get("hive3HmsPort")
+ String hdfsPort = context.config.otherConfigs.get("hive3HdfsPort")
+ for (String catalog : ["hive_lookup_source", "hive_lookup_target"]) {
+ sql "DROP CATALOG IF EXISTS ${catalog}"
+ sql """CREATE CATALOG ${catalog} PROPERTIES (
+ 'type' = 'hms',
+ 'hadoop.username' = 'hadoop',
+ 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}',
+ 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}'
+ )"""
+ }
+
+ // Two independent catalogs model a change made through another engine's
cache domain.
+ sql "DROP DATABASE IF EXISTS hive_lookup_source.fe_lookup_refresh_db FORCE"
+ sql "CREATE DATABASE hive_lookup_source.fe_lookup_refresh_db"
+ sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target
(before_col INT)"
+ sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_sibling
(sibling_col INT)"
+ order_qt_names_before "SHOW TABLES FROM
hive_lookup_target.fe_lookup_refresh_db"
+ qt_schema_before "DESC
hive_lookup_target.fe_lookup_refresh_db.owner_target"
+ qt_sibling_before "DESC
hive_lookup_target.fe_lookup_refresh_db.owner_sibling"
+ // Validate the stable projection used by case-mapping suites without
requiring BE data execution.
+ sql """EXPLAIN SELECT * REPLACE(CAST(NULL AS DATETIME) AS UPDATE_TIME)
+ FROM hive_lookup_target.information_schema.tables WHERE TABLE_SCHEMA =
'fe_lookup_refresh_db'"""
+
+ sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target"
+ sql "REFRESH TABLE hive_lookup_target.fe_lookup_refresh_db.owner_target"
+ // Listing first proves refresh itself invalidated membership, without a
failing SELECT repairing it.
+ order_qt_names_after_drop "SHOW TABLES FROM
hive_lookup_target.fe_lookup_refresh_db"
+ test {
+ sql "SELECT * FROM
hive_lookup_target.fe_lookup_refresh_db.owner_target"
+ exception "does not exist"
+ }
+ test {
+ sql "EXPLAIN SELECT * FROM
hive_lookup_target.fe_lookup_refresh_db.owner_target"
+ exception "does not exist"
+ }
+ qt_sibling_after "DESC
hive_lookup_target.fe_lookup_refresh_db.owner_sibling"
+
+ sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target
(after_col STRING)"
+ sql "REFRESH DATABASE hive_lookup_target.fe_lookup_refresh_db"
+ order_qt_names_after_database_refresh "SHOW TABLES FROM
hive_lookup_target.fe_lookup_refresh_db"
+ qt_schema_after_database_refresh "DESC
hive_lookup_target.fe_lookup_refresh_db.owner_target"
+
+ sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target"
+ sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target
(recreated_col BIGINT)"
+ sql "REFRESH TABLE hive_lookup_target.fe_lookup_refresh_db.owner_target"
+ qt_schema_after_recreation "DESC
hive_lookup_target.fe_lookup_refresh_db.owner_target"
+
+ sql "DROP CATALOG IF EXISTS hive_lookup_observed"
+ sql """CREATE CATALOG hive_lookup_observed PROPERTIES (
+ 'type' = 'hms',
+ 'hadoop.username' = 'hadoop',
+ 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}',
+ 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}',
+ 'schema.cache.ttl-second' = '0'
+ )"""
+ sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_observed
(observed_col INT)"
+ // Schema/HMS table reads are live, while the FE table-name and object
caches stay enabled.
+ order_qt_observed_names_before """SHOW TABLES FROM
hive_lookup_observed.fe_lookup_refresh_db
+ LIKE 'owner_observed'"""
+ qt_observed_schema_before "DESC
hive_lookup_observed.fe_lookup_refresh_db.owner_observed"
+ sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_observed"
+ test {
+ sql "SELECT * FROM
hive_lookup_observed.fe_lookup_refresh_db.owner_observed"
+ // MVCC observes absence before a later schema lookup reports the
now-missing cached object.
+ exception "failed to load schema cache value"
+ }
+ order_qt_observed_names_after """SHOW TABLES FROM
hive_lookup_observed.fe_lookup_refresh_db
+ LIKE 'owner_observed'"""
+
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_hadoop_case_sensibility.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_hadoop_case_sensibility.groovy
index aa2882625ea..50b68064426 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_hadoop_case_sensibility.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_hadoop_case_sensibility.groovy
@@ -146,9 +146,10 @@ suite("test_iceberg_hadoop_case_sensibility",
"p0,external") {
// for case 1,2, like is case insensible, and case_tbl22 is not
created, so will show CASE_TBL22
qt_sql12 """show tables from iceberg_hadoop_case_db2 like
"%case_tbl22%""""
- order_qt_sql13 """select * from information_schema.tables where
TABLE_SCHEMA="iceberg_hadoop_case_db2";"""
+ // UPDATE_TIME is populated when an external table object is
loaded; it is not a case-mapping result.
+ order_qt_sql13 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where
TABLE_SCHEMA="iceberg_hadoop_case_db2";"""
order_qt_sql14 """select * from information_schema.columns where
TABLE_SCHEMA="iceberg_hadoop_case_db2";"""
- order_qt_sql131 """select * from information_schema.tables where
TABLE_SCHEMA="ICEBERG_HADOOP_CASE_DB2";"""
+ order_qt_sql131 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where
TABLE_SCHEMA="ICEBERG_HADOOP_CASE_DB2";"""
order_qt_sql141 """select * from information_schema.columns where
TABLE_SCHEMA="ICEBERG_HADOOP_CASE_DB2";"""
// 4. insert
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_hms_case_sensibility.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_hms_case_sensibility.groovy
index 87be56fded9..3d3f11f8fa8 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_hms_case_sensibility.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_hms_case_sensibility.groovy
@@ -142,7 +142,8 @@ suite("test_iceberg_hms_case_sensibility", "p0,external") {
qt_sql11 """show tables from iceberg_hms_case_db2 like
"%case_tbl14%"""" // empty
qt_sql12 """show tables from iceberg_hms_case_db2 like
"%case_tbl21%""""
- order_qt_sql13 """select * from information_schema.tables where
TABLE_SCHEMA="iceberg_hms_case_db1";"""
+ // UPDATE_TIME is populated when an external table object is
loaded; it is not a case-mapping result.
+ order_qt_sql13 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where
TABLE_SCHEMA="iceberg_hms_case_db1";"""
order_qt_sql14 """select * from information_schema.columns where
TABLE_SCHEMA="iceberg_hms_case_db1";"""
// 4. insert
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_case_sensibility.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_case_sensibility.groovy
index ec77c7af0ec..75adb6e9513 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_case_sensibility.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_case_sensibility.groovy
@@ -147,9 +147,10 @@ suite("test_iceberg_rest_case_sensibility", "p0,external")
{
// for case 1,2, like is case insensible, and case_tbl22 is not
created, so will show CASE_TBL22
qt_sql12 """show tables from iceberg_rest_case_db2 like
"%case_tbl22%""""
- order_qt_sql13 """select * from information_schema.tables where
TABLE_SCHEMA="iceberg_rest_case_db2";"""
+ // UPDATE_TIME is populated when an external table object is
loaded; it is not a case-mapping result.
+ order_qt_sql13 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where
TABLE_SCHEMA="iceberg_rest_case_db2";"""
order_qt_sql14 """select * from information_schema.columns where
TABLE_SCHEMA="iceberg_rest_case_db2";"""
- order_qt_sql131 """select * from information_schema.tables where
TABLE_SCHEMA="ICEBERG_REST_CASE_DB2";"""
+ order_qt_sql131 """select * replace(cast(null as datetime) as
UPDATE_TIME) from information_schema.tables where
TABLE_SCHEMA="ICEBERG_REST_CASE_DB2";"""
order_qt_sql141 """select * from information_schema.columns where
TABLE_SCHEMA="ICEBERG_REST_CASE_DB2";"""
// 4. insert
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
