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 d84c389b04b186..45b53b48fef542 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 void invalidateCatalog() { 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 07afd00576d4bf..e30a24f72f9487 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,10 @@ package org.apache.doris.connector.cache; import java.util.Objects; +import java.util.Optional; import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; @@ -89,6 +92,34 @@ public boolean compareAndSet(K key, V expectedValue, V updatedValue, Runnable co Objects.requireNonNull(commitAction, "commitAction can not be null")); } + /** Captures an exact retained registration without loading or holding locks across a remote read. */ + public Optional 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 BiFunction operation; + + Observation(BiFunction operation) { + this.operation = operation; + } + + public boolean runIfCurrent(BooleanSupplier action) { + return operation.apply(false, Objects.requireNonNull(action, "action")); + } + + public boolean invalidateIfCurrent(BooleanSupplier action) { + return operation.apply(true, Objects.requireNonNull(action, "action")); + } + } + 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 00000000000000..fb6f235aec3d97 --- /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> 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 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 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 b58ad13bd9dfa5..960822e9c558f4 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 @@ -270,6 +270,44 @@ public V getWithPublicationAction(K key, ScopePath path, Function loader, } } + Optional observe(K key, ScopePath path, Object expectedValue) { + VersionedValue 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) -> 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); + } + } + } + }); + 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"); @@ -1118,7 +1156,9 @@ private void drainDeferredRemovals(RemovalDeferral removalDeferral) { DeferredRemoval removal; while ((removal = removalDeferral.removals.pollFirst()) != null) { try { - completeRemoval(removal.versioned, removal.cause, removal.budgetEviction); + DeferredRemoval 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 918b2829a7927b..7f749acbab512b 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 @@ void invalidate(ScopePath path, Runnable afterStateReplacement) { 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 invalidated = new AtomicReference<>(); + try { + publicationGate.write(() -> { + List 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 paths, Runnable afterStateReplacement) { Objects.requireNonNull(paths, "paths can not be null"); Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); @@ -242,6 +268,26 @@ void invalidate(List paths, Runnable afterStateReplacement) { 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 @@ private void bumpPublicationStates(List nodes) { } 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 3bb6ad66e80d68..9aebbc312da086 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,190 @@ 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 cache = owner.create(MetaCacheDefinition.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 observationRejectsReplacementScopeChangeAndSameInstanceReinsertion() { + CatalogMetaCache owner = CatalogMetaCache.unmanaged(); + try { + MetaCache cache = owner.create(MetaCacheDefinition.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 names = namesOwner.create(MetaCacheDefinition.builder( + "names", SPEC, ignored -> ScopePath.catalog()).build()); + MetaCache objects = objectsOwner.create(MetaCacheDefinition.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 cache = owner.create(MetaCacheDefinition.builder( + "objects", SPEC, ignored -> ScopePath.table("db", "t")).build()); + Object old = new Object(); + cache.put("t", old); + try (MetaCache.BulkLoad 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 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 c509f45004fce2..d90c821f48b6c0 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 boolean existIndex(String indexName) { 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 16439e765bd05b..d1354c93744c2f 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 @@ */ 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 245da688dd84f6..b7b365e81ebc4b 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 Optional getTableHandle( return Optional.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 b85499d9e99fc6..ec2174d7029246 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.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 void getTableHandleRunsSeamInsideAuthenticator() { RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); RecordingConnectorContext ctx = new RecordingConnectorContext(); ctx.failAuth = true; - Optional 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 4a7c5316f9f903..564b4809509e05 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 void getTableHandleEmptyWhenTableMissing() { 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 73147f40620a0d..a6465faf6c62ca 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 @@ public List listTables(String databaseName) throws Catalog.DatabaseNotEx @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 4d29727e1b84d9..64263b639f59a4 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 @@ */ 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 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 8e5586b344f79c..9c1eaaf899dacc 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 void handleRefreshTable(String catalogName, String dbName, String tableNa 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 void replayRefreshTable(ExternalObjectLog log) { 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 void replayRefreshTable(ExternalObjectLog log) { ((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 void refreshExternalTableFromEvent(String catalogName, String localDbName } 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 void refreshTableInternal(ExternalDatabase db, ExternalTable table, long // 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 3934483b97bf48..d828add101d375 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 @@ -1190,6 +1190,14 @@ public void gsonPostProcess() throws IOException { this.dbIdNameIndex = new IdNameIndex("external database"); } + /** Cache-only identity fence for a read through a retained database instance. */ + public Optional observeDatabase(ExternalDatabase database) { + if (!initialized || databases == null) { + return Optional.empty(); + } + return databases.observe(database.getFullName(), database); + } + public void addDatabaseForTest(ExternalDatabase 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 3d24d0c3b78ce9..8cf5501a960f3e 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.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 T buildTableForInit(String remoteTableName, String localTableName, long t } } - // 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,44 @@ public Optional getTableForReplay(long tableId) { 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 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.invalidateIfCurrent(() -> { + try { + invalidateMetadata.run(); + } finally { + tableIdNameIndex.removeName(table.getName()); + } + return true; + }))); + } + })))); + } + /** * Resolve the retained local table name for replay without loading table metadata. * @@ -544,6 +585,10 @@ public T getTableNullable(String tableName) { } 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 +788,34 @@ protected void invalidateTableCache(String localTableName) { } } + /** 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 +825,7 @@ public T getTableNullable(long tableId) { 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 4c874fe372179d..3e81cb70aec630 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 void setUpdateTime(long updateTime) { 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 30464457e60a80..308b3cd5184100 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 newPartitionNames; + @SerializedName(value = "invalidateTableLookup") + private boolean invalidateTableLookup; + @SerializedName(value = "lastUpdateTime") private long lastUpdateTime; @@ -85,6 +88,13 @@ public static ExternalObjectLog createForRefreshTable(long catalogId, String dbN 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 modifiedPartNames, List 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 1cb7f0dcf9c444..13a4a428de732b 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.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 V getAndRunIfCurrent(K key, BiPredicate actionRequired, && (!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,49 @@ public V getAndRunIfCurrent(K key, BiPredicate actionRequired, } } + public Optional observe(K key, Object expectedValue) { + StripeState 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) { + synchronized (stripe) { + return value.invalidateIfCurrent(() -> { + boolean invalidate = true; + try { + invalidate = action.getAsBoolean(); + return invalidate; + } finally { + if (invalidate) { + invalidateActions.run(); + } + } + }); + } + } + } + public V getIfPresent(K key) { return data.getIfPresent(key); } @@ -360,6 +406,12 @@ public void invalidateKeyAndRun(K key, Runnable afterInvalidation) { } 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 stripe = stripeStates.get(i); if (stripe != null) { @@ -370,7 +422,7 @@ public void invalidateAll() { } } } - owner.invalidateCatalog(); + owner.invalidateCatalogAndRun(action); } public void forEach(BiConsumer 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 6cbea40af7505b..a461d17c240237 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,24 @@ public PluginDrivenExternalTable(long id, String name, String remoteName, */ public Optional resolveConnectorTableHandle( ConnectorSession session, ConnectorMetadata metadata) { + PluginDrivenExternalCatalog owner = (PluginDrivenExternalCatalog) catalog; + Connector connector = owner.getConnector(); + ExternalDatabase database = getDb(); + Optional observation = database.observeTableLookup(this, () -> { + try { + connector.invalidateTable(db.getRemoteName(), getRemoteName()); + } finally { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(this); + } + }); String dbName = db != null ? db.getRemoteName() : ""; - return metadata.getTableHandle(session, dbName, getRemoteName()); + Optional 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 44059931620e82..dedd1b42b82720 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.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.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 void tearDown() { mockedEnv.close(); mockedEnv = null; } + MutableCatalogProvider.reset(); } @Test @@ -167,6 +172,145 @@ public void testReplayNameBasedRefreshInvalidatesColdTableByLogName() { 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 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 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 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 @@ private Connector usePluginCatalogWithDisabledDatabaseObjectCache() { } private PluginCatalogFixture usePluginCatalog() { + return usePluginCatalog(0); + } + + private PluginCatalogFixture usePluginCatalog(int lowerCaseTableNames) { Connector connector = Mockito.mock(Connector.class); - Map properties = Collections.singletonMap("type", "test"); + Map 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 @@ private void assertColdRenameMigrated() { Assertions.assertEquals(NEW_TABLE_NAME, constraintManager.newTableName.getTbl()); } + private void useMutableCatalog() { + useMutableCatalog(0); + } + + private void useMutableCatalog(int lowerCaseTableNames) { + MutableCatalogProvider.putTable("old_column"); + Map 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>> 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>> getMetadata() { + return METADATA; + } + } + public static class EmptyCatalogProvider implements TestExternalCatalog.TestCatalogProvider { @Override public Map>> getMetadata() { @@ -471,6 +660,11 @@ void setCatalog(CatalogIf> catalog) { public CatalogIf> getCatalog(long id) { return catalog != null && id == catalog.getId() ? catalog : null; } + + @Override + public CatalogIf> 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 36edaacdf0197c..40252053a5d5a8 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 void testRefreshTable() throws Exception { 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 d2d9a91e77209c..8c0d1b91a69cfa 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.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 @@ protected void runBeforeAll() throws Exception { 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 @@ protected void beforeCurrentValueActionForTest(String key, TestExternalTable val } } + @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 objectEntry = + new FeMetaCacheEntry( + "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 actionRequired, + BiConsumer currentValueAction) { + nameCaptured.countDown(); + awaitLatch(releaseLookup); + return super.getAndRunIfCurrent(key, actionRequired, currentValueAction); + } + }; + setTablesEntryForTest(db, objectEntry); + + Future 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 22bbc80ad788f0..bcadcd3f7149f8 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 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; @@ -271,6 +273,44 @@ protected void beforeCurrentValueActionForTest(String key, Integer value) { } } + @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 ids = new ConcurrentHashMap<>(); + try { + FeMetaCacheEntry 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 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 f21ed9f8006890..64deaeaab129f3 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 @@ private static ConnectorSession noneScopedSession() { 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 @@ private static PluginDrivenExternalTable capabilityTable(boolean handlePresent, 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 @@ private static PluginDrivenExternalTable writeTargetTable(ConnectorTableHandle r 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 @@ private static PluginDrivenExternalTable syntheticPredicateTable(ConnectorTableH 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 @@ private static PluginDrivenExternalTable pluginTable(List synth 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 void resolveWriteColumnsRunsAndRestoresPluginContextClassLoader() { 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 00000000000000..d62b8f21101058 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java @@ -0,0 +1,608 @@ +// 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; + +/** 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 remoteTables = new ConcurrentHashMap<>(); + private final Set remoteViews = ConcurrentHashMap.newKeySet(); + private Connector connector; + private ConnectorMetadata metadata; + private TestCatalog catalog; + private TestDatabase database; + private final Map 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> 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> 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 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 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 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 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 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 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> byName = Deencapsulation.getField(manager, "nameToCatalog"); + byName.put(catalog.getName(), catalog); + remoteTables.put("a", "replacement_column"); + installDatabase(); + PluginDrivenExternalTable replacement = database.getTableNullable("a"); + List 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 missing = worker.submit(() -> Assertions.assertThrows(CacheException.class, + old::getFullSchema)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + remoteTables.put("FOO", "new_column"); + FeMetaCacheEntry names = Deencapsulation.getField(database, "tableNames"); + names.invalidateKey(""); + PluginDrivenExternalTable replacement = database.getTableNullable("foo"); + Assertions.assertEquals("FOO", replacement.getName()); + List 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 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 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 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 listDatabaseNames() { + return Collections.singletonList("db"); + } + + @Override + protected List listTableNamesFromRemote(SessionContext context, String dbName) { + nameLoads.incrementAndGet(); + List names = new ArrayList<>(remoteTables.keySet()); + names.addAll(remoteViews); + return names; + } + + @Override + public boolean shouldBypassSchemaCache(SessionContext context) { + return bypassSchema; + } + } + + private class TestDatabase extends ExternalDatabase { + TestDatabase() { + super(catalog, Util.genIdByName(catalog.getName(), "db"), "db", "db"); + } + + @Override + protected PluginDrivenExternalTable buildTableInternal(String remoteName, String localName, long id, + ExternalCatalog owner, ExternalDatabase db) { + 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 00000000000000..7730655d8510a0 --- /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_external_lookup_refresh.groovy b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy new file mode 100644 index 00000000000000..3d3ce7135e3902 --- /dev/null +++ b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy @@ -0,0 +1,93 @@ +// 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" + + 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'""" + +}