This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 0c47a5eca7 [Cherry-pick to branch-1.3] [#14640] fix(core): preserve
native comments and isolate metadata operations (#14641) (#14724)
0c47a5eca7 is described below
commit 0c47a5eca708dd1640d6a25b37e748663c778e46
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 19:44:00 2026 +0800
[Cherry-pick to branch-1.3] [#14640] fix(core): preserve native comments
and isolate metadata operations (#14641) (#14724)
### What changes were proposed in this pull request?
Backport 61d5c847538adc240dac008162815401e9568fff (#14641) to
`branch-1.3`, with cherry-pick conflicts resolved against the 1.3 APIs
and storage schema.
Preserve complete native column comments in responses while storing a
bounded, NUL-free projection. Complete catalog initialization,
configuration checks, and any enabled backend verification before
publishing their entities, coordinate local store/cache publication, and
narrow catalog creation and table DROP locks. Use provider existing-name
rules for column references and wait for asynchronous JDBC staging-table
cleanup in marked Trino SQL assertions.
### Why are the changes needed?
Addresses #14640 on the release branch: native comments must not break
metadata import, and slow source DDL must not unnecessarily block
sibling metadata operations. Preserve the branch's existing false-DROP
registration protection, local change-log bookkeeping, and case folding
before default provider name validation.
### Does this PR introduce _any_ user-facing change?
On **branch-1.3**, stored top-level column comments are bounded to **256
UTF-16 units**, matching its existing VARCHAR(256) schema. CREATE, ADD
COLUMN, and comment updates reject NUL or oversized top-level column
comments before source DDL; accepted comments are not truncated.
Nested-field ALTER comments retain provider behavior. Full external
source comments remain available in responses. PostgreSQL existing
top-level column references accept native names while new top-level
names retain creation validation. No schema migration or new
configuration is required.
The lock changes cover catalog creation and table DROP. Table PURGE,
view DROP, and cold table import retain their current behavior. This
does not add SQL cancellation/timeouts or cross-node source/store
atomicity. Orphaned-schema cleanup remains best-effort and rechecks
schema existence before removing registrations.
### How was this patch tested?
- 183 targeted tests: 159 core tests covering the changed classes and
related normalization, table-existence, partition, and hook callers; 12
PostgreSQL capability tests; 12 Trino SQL-harness tests.
- Eight regression cases fail when the affected production classes are
temporarily restored to `branch-1.3`. Two additional
normalize-dispatcher cases fail before the case-folding compatibility
fix, covering existing column and AFTER references; all pass with the
final fixes.
- H2 2.2.224 VARCHAR(256) probe accepts 256 ASCII characters, 254 ASCII
characters plus an emoji, and 128 emoji; rejects 255 ASCII characters
plus an emoji and 256 emoji with SQLState 22001.
- Scoped Spotless checks and `git diff --check` pass.
- The Trino harness unit tests ran with a temporary Gradle init script
that removes this module's package exclusion under `-PskipITs`, without
provisioning its Docker environment. Full Docker Trino SQL integration
and the full core suite were not run locally. No entity-store SQL or
migration scripts were changed.
---------
Co-authored-by: Qi Yu <[email protected]>
Co-authored-by: yuqi <[email protected]>
---
.../apache/gravitino/catalog/jdbc/JdbcCatalog.java | 12 +-
.../catalog/jdbc/JdbcCatalogOperations.java | 9 +-
.../gravitino/catalog/jdbc/config/JdbcConfig.java | 4 +-
.../catalog/jdbc/utils/JdbcConnectionTimeouts.java | 7 +-
.../jdbc/TestJdbcCatalogOperationsProbe.java | 4 +-
.../postgresql/PostgreSqlCatalogCapability.java | 2 +-
.../TestPostgreSqlCatalogCapability.java | 7 +-
.../gravitino/catalog/CapabilityHelpers.java | 22 +-
.../apache/gravitino/catalog/CatalogManager.java | 125 ++++++----
.../gravitino/catalog/ManagedTableOperations.java | 9 +-
.../catalog/TableOperationDispatcher.java | 118 ++++++----
.../gravitino/connector/capability/Capability.java | 8 +-
.../java/org/apache/gravitino/lock/TreeLock.java | 12 +-
.../org/apache/gravitino/meta/ColumnEntity.java | 46 +++-
.../java/org/apache/gravitino/TestCatalog.java | 2 +-
.../gravitino/catalog/TestCapabilityHelpers.java | 58 +++++
.../gravitino/catalog/TestCatalogManager.java | 225 ++++++++++++++++--
.../catalog/TestManagedTableOperations.java | 59 +++++
.../catalog/TestTableNormalizeDispatcher.java | 55 +++++
.../catalog/TestTableOperationDispatcher.java | 255 +++++++++++++++++++++
.../apache/gravitino/meta/TestColumnEntity.java | 15 ++
docs/jdbc-catalog-connection-validation.md | 13 +-
docs/manage-relational-metadata-using-gravitino.md | 32 +++
.../integration/test/TestTrinoQueryIT.java | 194 ++++++++++++++++
.../connector/integration/test/TrinoQueryIT.java | 127 +++++++---
.../testsets/jdbc-postgresql/00006_datatype.sql | 5 +-
26 files changed, 1235 insertions(+), 190 deletions(-)
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
index b4c8593397..1bf5d6c1ed 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
@@ -136,21 +136,21 @@ public abstract class JdbcCatalog extends
BaseCatalog<JdbcCatalog> {
/**
* Verifies the backend is reachable at create time with a dedicated,
time-bounded one-off probe
* connection (see {@link JdbcCatalogOperations#probeConnection}). On
failure the probe raises a
- * {@link ConnectionFailedException}, which propagates so {@code
CatalogManager} rolls back the
- * just-persisted catalog entity.
+ * {@link ConnectionFailedException}, which propagates so {@code
CatalogManager} rejects creation
+ * before persisting the catalog entity.
*
* <p>The probe does not borrow from the lazy pool: it opens its own
connection bounded by both
* the connect and the read/login timeout, so a backend that accepts the
socket but never speaks
- * the protocol fails fast instead of blocking the metalake write lock. The
probe connection is
- * opened, validated and closed within the call, so there is no pool or
connection leak.
+ * the protocol fails fast instead of blocking the target catalog write
lock. The probe connection
+ * is opened, validated and closed within the call, so there is no pool or
connection leak.
*/
@Override
public void validateOnCreate() {
try {
((JdbcCatalogOperations)
ops()).probeConnection(entity().nameIdentifier());
} catch (RuntimeException e) {
- // Includes ConnectionFailedException; rethrow unchanged so
CatalogManager rolls back and the
- // connection-failed taxonomy the caller relies on is preserved.
+ // Includes ConnectionFailedException; rethrow unchanged so
CatalogManager rejects creation
+ // before publication and preserves the connection-failed taxonomy the
caller relies on.
throw e;
} catch (Exception e) {
// Defensive: the probe only throws runtime exceptions, so wrap any
unexpected checked
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
index c5f067cc3c..2059d4f482 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
@@ -267,9 +267,10 @@ public class JdbcCatalogOperations implements
CatalogOperations, SupportsSchemas
* pool, whose connections carry only a connect timeout so normal operations
stay unbounded), this
* probe applies both the connect timeout and the socket-read (and, for
PostgreSQL, login) timeout
* from {@link JdbcConnectionTimeouts#probeConnectionProperties}. That
matters because the probe
- * runs under the metalake write lock during catalog creation: a backend
that accepts the TCP
- * connection but never speaks the database protocol would otherwise block
forever and starve the
- * whole metalake. The connection is opened, validated and closed within
this call.
+ * runs under the target catalog write lock during catalog creation: a
backend that accepts the
+ * TCP connection but never speaks the database protocol would otherwise
block forever, delaying
+ * operations that need this catalog lock or an ancestor write lock. The
connection is opened,
+ * validated and closed within this call.
*
* @param catalogIdent the identifier of the catalog being verified
* @throws ConnectionFailedException if the backend cannot be reached within
the bounded timeout
@@ -310,7 +311,7 @@ public class JdbcCatalogOperations implements
CatalogOperations, SupportsSchemas
properties.setProperty("password", password);
}
// Bound both the connect and the read/login timeout for this probe so an
accept-but-silent
- // backend fails fast instead of hanging under the metalake write lock.
+ // backend fails fast instead of hanging under the target catalog write
lock.
JdbcConnectionTimeouts.probeConnectionProperties(jdbcConfig, jdbcUrl)
.forEach(properties::setProperty);
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
index 6d2388ec77..2115a380f8 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
@@ -113,8 +113,8 @@ public class JdbcConfig extends Config {
* normal operations get only the <em>connect</em> timeout (a read timeout
there would abort
* legitimate long-running operations), while the one-off create-time
verification probe gets both
* the <em>connect</em> and the <em>read/login</em> timeout, because that
probe runs under the
- * metalake write lock and a backend that accepts the socket but never
speaks the protocol would
- * otherwise block the connect-only bound forever. An operator value set
directly on the
+ * target catalog write lock and a backend that accepts the socket but never
speaks the protocol
+ * would otherwise block the connect-only bound forever. An operator value
set directly on the
* connection or in the {@code jdbc-url} is never overridden.
*
* <p>The value is bounded by {@link Integer#MAX_VALUE} milliseconds to fit
the drivers' integer
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
index a20d98164d..490291ed4a 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
@@ -41,11 +41,11 @@ import org.slf4j.LoggerFactory;
* query).
* <li>{@link #probeConnectionProperties} โ used for the one-off
<b>create-time verification
* probe</b>. It bounds the connect timeout <em>and</em> the socket-read
(and, for PostgreSQL,
- * login) timeout, because that probe runs under the metalake write
lock: a backend that
+ * login) timeout, because that probe runs under the target catalog
write lock: a backend that
* accepts the TCP connection but never speaks the database protocol (a
load balancer with a
* dead backend, a wrong port, a proxy that accepts and never forwards)
would otherwise make
* the connect-only bound wait on the driver's default read timeout,
which is "wait forever",
- * and hold the lock for the whole metalake.
+ * and hold the target catalog lock.
* </ul>
*
* <p>DBCP has no portable timeout; the real bound lives in the driver, and
both the parameter name
@@ -139,7 +139,8 @@ public final class JdbcConnectionTimeouts {
* {@link JdbcConfig#getConnectTimeoutMs()} value, or an empty map when the
driver family is
* unknown or the operator has already set the properties. Bounding the read
timeout here (but not
* on the pool, see {@link #connectionProperties}) ensures a backend that
accepts the socket but
- * never speaks the protocol fails the probe instead of blocking the
metalake write lock forever.
+ * never speaks the protocol fails the probe instead of blocking the target
catalog write lock
+ * forever.
*
* @param config the catalog JDBC configuration; must not be null
* @param jdbcUrl the JDBC URL whose query parameters are inspected for
operator-set timeouts; may
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
index d39bcbb386..5276ef35bd 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
@@ -55,8 +55,8 @@ public class TestJdbcCatalogOperationsProbe {
/**
* A probe against a server that accepts the TCP connection but never
replies must fail within the
- * configured read timeout, not hang. This is the
metalake-write-lock-starvation case: before the
- * read timeout was injected on the probe, the driver's handshake read
waited forever.
+ * configured read timeout, not hang and starve catalog operations. Before
the read timeout was
+ * injected on the probe, the driver's handshake read waited forever.
*/
@Test
public void testProbeAgainstAcceptButSilentServerIsBounded() throws
Exception {
diff --git
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/PostgreSqlCatalogCapability.java
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/PostgreSqlCatalogCapability.java
index 07f5c1e144..218642d51a 100644
---
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/PostgreSqlCatalogCapability.java
+++
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/PostgreSqlCatalogCapability.java
@@ -46,7 +46,7 @@ public class PostgreSqlCatalogCapability implements
Capability {
/** {@inheritDoc} */
@Override
public CapabilityResult specificationOnExistingName(Scope scope, String
name) {
- if (scope != Scope.TABLE && scope != Scope.SCHEMA) {
+ if (scope != Scope.TABLE && scope != Scope.SCHEMA && scope !=
Scope.COLUMN) {
return specificationOnName(scope, name);
}
if (MetadataObjects.METADATA_OBJECT_RESERVED_NAME.equals(name)) {
diff --git
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/TestPostgreSqlCatalogCapability.java
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/TestPostgreSqlCatalogCapability.java
index 6f33006344..6f77cb9e10 100644
---
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/TestPostgreSqlCatalogCapability.java
+++
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/TestPostgreSqlCatalogCapability.java
@@ -28,15 +28,18 @@ public class TestPostgreSqlCatalogCapability {
private final PostgreSqlCatalogCapability capability = new
PostgreSqlCatalogCapability();
@Test
- void testNativeDeletionNamesRemainDistinctFromCreationNames() {
+ void testNativeNamesRemainDistinctFromCreationNames() {
for (Capability.Scope scope :
- new Capability.Scope[] {Capability.Scope.TABLE,
Capability.Scope.SCHEMA}) {
+ new Capability.Scope[] {
+ Capability.Scope.TABLE, Capability.Scope.SCHEMA,
Capability.Scope.COLUMN
+ }) {
Assertions.assertTrue(
capability.specificationOnExistingName(scope,
"native\"name").supported());
Assertions.assertFalse(capability.specificationOnName(scope,
"native\"name").supported());
Assertions.assertFalse(
capability.specificationOnExistingName(scope,
"bad\0name").supported());
Assertions.assertFalse(capability.specificationOnExistingName(scope,
"").supported());
+ Assertions.assertFalse(capability.specificationOnExistingName(scope,
"*").supported());
Assertions.assertFalse(
capability.specificationOnExistingName(scope,
"่กจ".repeat(22)).supported());
}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CapabilityHelpers.java
b/core/src/main/java/org/apache/gravitino/catalog/CapabilityHelpers.java
index 39c1facc49..e66dc69cdd 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CapabilityHelpers.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CapabilityHelpers.java
@@ -27,6 +27,7 @@ import java.util.Arrays;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.connector.capability.CapabilityResult;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.file.FilesetChange;
import org.apache.gravitino.model.ModelChange;
@@ -399,15 +400,24 @@ public class CapabilityHelpers {
return TableChange.rename(newName, newSchemaName);
}
+ private static void applyExistingColumnName(String name, Capability
capabilities) {
+ CapabilityResult result =
+ capabilities.specificationOnExistingName(Capability.Scope.COLUMN,
name);
+ Preconditions.checkArgument(
+ result.supported(), result.unsupportedMessage() + " Illegal name: " +
name);
+ }
+
private static TableChange applyCapabilities(
TableChange.ColumnChange change, Capability capabilities) {
- String[] fieldName = applyCaseSensitiveOnColumnName(change.fieldName(),
capabilities);
- applyNameSpecification(Capability.Scope.COLUMN, fieldName[0],
capabilities);
-
if (change instanceof TableChange.AddColumn) {
return applyCapabilities((TableChange.AddColumn) change, capabilities);
+ }
+
+ // Preserve branch-1.3 validation after case folding for case-insensitive
providers.
+ String[] fieldName = applyCaseSensitiveOnColumnName(change.fieldName(),
capabilities);
+ applyExistingColumnName(fieldName[0], capabilities);
- } else if (change instanceof TableChange.UpdateColumnNullability) {
+ if (change instanceof TableChange.UpdateColumnNullability) {
return applyCapabilities((TableChange.UpdateColumnNullability) change,
capabilities);
} else if (change instanceof TableChange.UpdateColumnDefaultValue) {
@@ -435,7 +445,7 @@ public class CapabilityHelpers {
String afterFieldName =
applyCaseSensitiveOnName(
Capability.Scope.COLUMN, afterPosition.getColumn(),
capabilities);
- applyNameSpecification(Capability.Scope.COLUMN, afterFieldName,
capabilities);
+ applyExistingColumnName(afterFieldName, capabilities);
return TableChange.updateColumnPosition(
fieldName, TableChange.ColumnPosition.after(afterFieldName));
}
@@ -503,7 +513,7 @@ public class CapabilityHelpers {
private static TableChange applyCapabilities(
TableChange.RenameColumn renameColumn, Capability capabilities) {
- String[] fieldName = applyCapabilities(renameColumn.fieldName(),
capabilities);
+ String[] fieldName =
applyCaseSensitiveOnColumnName(renameColumn.fieldName(), capabilities);
String newName = renameColumn.getNewName();
if (fieldName.length == 1) {
newName = applyCapabilitiesOnName(Capability.Scope.COLUMN, newName,
capabilities);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 967759479d..8eb9fc1737 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -56,6 +56,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -826,65 +827,70 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
.build();
return TreeLockUtils.doWithTreeLock(
- metalakeIdent,
+ ident,
LockType.WRITE,
() -> {
- checkMetalake(metalakeIdent, store);
+ // Keep close() from retiring resources between initialization and
publication. The
+ // lock order remains tree lock, lifecycle lock, then the per-key
cache operation.
lifecycleLock.readLock().lock();
- boolean needClean = false;
try {
checkOpen();
- needClean = true;
- store.put(e, false /* overwrite */);
- try (CatalogLease lease =
- createAndCacheCatalogLease(ident, () ->
createCatalogWrapper(e, mergedConfig))) {
- Catalog catalog = toCatalogInfo(lease.wrapper());
- needClean = false;
- return catalog;
- }
-
- } catch (EntityAlreadyExistsException e1) {
- needClean = false;
- LOG.warn("Catalog {} already exists", ident, e1);
- throw new CatalogAlreadyExistsException("Catalog %s already
exists", ident);
-
- } catch (IllegalArgumentException | NoSuchMetalakeException e2) {
- throw e2;
-
- } catch (Exception e3) {
- catalogCache.invalidate(ident);
- LOG.error("Failed to create catalog {}", ident, e3);
- if (e3 instanceof RuntimeException) {
- throw (RuntimeException) e3;
- }
- throw new RuntimeException(e3);
-
- } finally {
+ checkMetalake(metalakeIdent, store);
+ AtomicBoolean entityStored = new AtomicBoolean(false);
try {
- if (needClean) {
- // since we put the catalog entity into the store but failed
to create the catalog
- // instance,
- // we need to clean up the entity stored.
- try {
- if (store.delete(ident, EntityType.CATALOG, true)) {
- // This cleanup deletion writes a DROP record to the
entity change log. Mark it
- // as
- // a local mutation so the change-log poller consumes that
record's token
- // instead
- // of one meant for a later mutation of the same
identifier. Without this, the
- // poller could treat a subsequent local change as remote
and spuriously
- // invalidate (and asynchronously close) a cached catalog
wrapper that is still
- // in
- // use, causing a NullPointerException.
- markLocalMutation(ident);
- }
- } catch (IOException e4) {
- LOG.error("Failed to clean up catalog {}", ident, e4);
+ // Reject known duplicates before contacting the backend;
store.put still arbitrates
+ // concurrent creates from other server instances.
+ if (store.exists(ident, EntityType.CATALOG)) {
+ throw new CatalogAlreadyExistsException("Catalog %s already
exists", ident);
+ }
+ CatalogWrapper wrapper = createCatalogWrapper(e, mergedConfig);
+ boolean published = false;
+ try {
+ Preconditions.checkState(
+ wrapper.tryAcquire(), "A new catalog wrapper cannot be
retired");
+ try (CatalogLease lease = new CatalogLease(wrapper)) {
+ // Snapshot construction can also fail; finish it before
making the row visible.
+ Catalog info = toCatalogInfo(lease.wrapper());
+ // Initializing outside compute keeps slow backend
verification out of cache
+ // locks. Serialize the short store/cache publication with
acquireCatalogLease
+ // so a listCatalogsInfo reader cannot build a second
wrapper in between them.
+ catalogCache
+ .asMap()
+ .compute(
+ ident,
+ (key, previous) -> {
+ try {
+ store.put(e, false /* overwrite */);
+ } catch (NoSuchEntityException missing) {
+ throw new NoSuchMetalakeException(
+ missing, "Metalake %s does not exist",
metalakeIdent);
+ } catch (EntityAlreadyExistsException duplicate) {
+ throw new CatalogAlreadyExistsException(
+ duplicate, "Catalog %s already exists",
ident);
+ } catch (IOException ioe) {
+ throw new RuntimeException("Failed to store
catalog " + ident, ioe);
+ }
+ entityStored.set(true);
+ return wrapper;
+ });
+ published = true;
+ return info;
+ }
+ } finally {
+ if (!published) {
+ wrapper.retire();
}
}
- } finally {
- lifecycleLock.readLock().unlock();
+ } catch (RuntimeException re) {
+ rollbackFailedCatalogCreate(ident, entityStored.get());
+ throw re;
+ } catch (Exception ex) {
+ rollbackFailedCatalogCreate(ident, entityStored.get());
+ LOG.error("Failed to create catalog {}", ident, ex);
+ throw new RuntimeException(ex);
}
+ } finally {
+ lifecycleLock.readLock().unlock();
}
});
}
@@ -2081,4 +2087,23 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
throw new RuntimeException(ioe);
}
}
+
+ /**
+ * Removes the catalog entity left behind by a failed publication when
{@code store.put}
+ * succeeded.
+ */
+ private void rollbackFailedCatalogCreate(NameIdentifier ident, boolean
entityStored) {
+ if (!entityStored) {
+ return;
+ }
+ catalogCache.invalidate(ident);
+ try {
+ if (store.delete(ident, EntityType.CATALOG, true)) {
+ // The cleanup DROP must consume its own token, not a later local
mutation's token.
+ markLocalMutation(ident);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to clean up catalog {}", ident, e);
+ }
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
b/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
index 6b48716b0f..358e0c03ac 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
@@ -306,6 +306,7 @@ public abstract class ManagedTableOperations implements
TableCatalog {
for (TableChange.ColumnChange change : columnChanges) {
if (change instanceof TableChange.AddColumn addColumn) {
String columnName = DOT.join(addColumn.getFieldName());
+ ColumnEntity.validateCommentForWrite(columnName,
addColumn.getComment());
boolean exists = newColumns.stream().anyMatch(col ->
col.name().equals(columnName));
if (exists) {
throw new IllegalArgumentException(String.format("Column %s already
exists", columnName));
@@ -443,6 +444,8 @@ public abstract class ManagedTableOperations implements
TableCatalog {
Optional<Integer> newPosition,
Optional<Boolean> newNullable,
Optional<Boolean> newAutoIncrement) {
+ newComment.ifPresent(
+ comment -> ColumnEntity.validateCommentForWrite(oldColumn.name(),
comment));
return ColumnEntity.builder()
.withId(oldColumn.id())
.withName(newName.orElse(oldColumn.name()))
@@ -498,7 +501,11 @@ public abstract class ManagedTableOperations implements
TableCatalog {
return columns == null
? Collections.emptyList()
: IntStream.range(0, columns.length)
- .mapToObj(i -> ColumnEntity.toColumnEntity(columns[i], i,
idGenerator.nextId(), audit))
+ .mapToObj(
+ i -> {
+ ColumnEntity.validateCommentForWrite(columns[i].name(),
columns[i].comment());
+ return ColumnEntity.toColumnEntity(columns[i], i,
idGenerator.nextId(), audit);
+ })
.collect(Collectors.toList());
}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
index 786550ff1f..c358a2efdd 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -410,41 +410,45 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
@Override
public boolean dropTable(NameIdentifier ident) {
NameIdentifier schemaIdentifier = getSchemaIdentifier(ident);
- return TreeLockUtils.doWithTreeLock(
- schemaIdentifier,
- LockType.WRITE,
- () -> {
- NameIdentifier catalogIdent = getCatalogIdentifier(ident);
- boolean droppedFromCatalog =
- doWithCatalog(
- catalogIdent,
- c -> c.doWithTableOps(t -> t.dropTable(ident)),
- RuntimeException.class);
-
- boolean isManagedTable = isManagedEntity(catalogIdent,
Capability.Scope.TABLE);
- if (isManagedTable) {
- return droppedFromCatalog;
- }
+ Pair<Boolean, Boolean> result =
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.WRITE,
+ () -> {
+ NameIdentifier catalogIdent = getCatalogIdentifier(ident);
+ boolean droppedFromCatalog =
+ doWithCatalog(
+ catalogIdent,
+ c -> c.doWithTableOps(t -> t.dropTable(ident)),
+ RuntimeException.class);
+
+ boolean isManagedTable = isManagedEntity(catalogIdent,
Capability.Scope.TABLE);
+ if (isManagedTable) {
+ return Pair.of(droppedFromCatalog, isManagedTable);
+ }
- // A false result is ambiguous: the external table may have been
renamed or dropped out of
- // band, or the catalog user may be unable to see it. Preserve the
registration, because
- // deleting it would lose Gravitino-only metadata such as tags and
privileges, including
- // DENY privileges, of a table that may still exist. A true
out-of-band drop can therefore
- // leave a stale registration that requires separate cleanup.
- if (droppedFromCatalog) {
- try {
- store.delete(ident, TABLE);
- } catch (NoSuchEntityException e) {
- LOG.warn("The table to be dropped does not exist in the store:
{}", ident, e);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- // Run unconditionally: an out-of-band drop may have left orphaned
schema entities. The
- // cleanup is best-effort and stops as soon as a schema still exists.
- OrphanedSchemaCleanup.cleanUp(this, catalogIdent, schemaIdentifier);
- return droppedFromCatalog;
- });
+ // A false result is ambiguous: the external table may have been
renamed or dropped
+ // out of band, or the catalog user may be unable to see it.
Keep the registration
+ // to preserve Gravitino-only metadata such as tags and
privileges,
+ // including DENY privileges, of a table that may still exist. A
true out-of-band drop
+ // can therefore leave a stale registration that requires
separate cleanup.
+ if (droppedFromCatalog) {
+ try {
+ store.delete(ident, TABLE);
+ } catch (NoSuchEntityException e) {
+ LOG.warn("The table to be dropped does not exist in the
store: {}", ident, e);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return Pair.of(droppedFromCatalog, isManagedTable);
+ });
+ if (!result.getRight()) {
+ // Run even after a false drop result: out-of-band drops can leave
orphaned schemas.
+ // Release the table lock before cleanup to avoid a lock upgrade.
+ cleanUpOrphanedSchema(getCatalogIdentifier(ident), schemaIdentifier);
+ }
+ return result.getLeft();
}
/**
@@ -795,7 +799,7 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
return Objects.equal(left.name(), right.name())
&& columnPosition == right.position()
&& Objects.equal(left.dataType(), right.dataType())
- && Objects.equal(left.comment(), right.comment())
+ && Objects.equal(ColumnEntity.commentForStorage(left.comment()),
right.comment())
&& left.nullable() == right.nullable()
&& left.autoIncrement() == right.autoIncrement()
&& Objects.equal(left.defaultValue(), right.defaultValue());
@@ -818,10 +822,10 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
String.format(
"type[catalog=%s, store=%s]", catalogColumn.dataType(),
entityColumn.dataType()));
}
- if (!Objects.equal(catalogColumn.comment(), entityColumn.comment())) {
+ String storedComment =
ColumnEntity.commentForStorage(catalogColumn.comment());
+ if (!Objects.equal(storedComment, entityColumn.comment())) {
differences.add(
- String.format(
- "comment[catalog=%s, store=%s]", catalogColumn.comment(),
entityColumn.comment()));
+ String.format("comment[catalog=%s, store=%s]", storedComment,
entityColumn.comment()));
}
if (catalogColumn.nullable() != entityColumn.nullable()) {
differences.add(
@@ -1026,7 +1030,7 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
.withName(column.name())
.withPosition(columnPair.getLeft())
.withDataType(column.dataType())
- .withComment(column.comment())
+ .withComment(ColumnEntity.commentForStorage(column.comment()))
.withNullable(column.nullable())
.withAutoIncrement(column.autoIncrement())
.withDefaultValue(column.defaultValue())
@@ -1150,10 +1154,39 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
combinedTable.tableFromGravitino().id()));
}
+ private void cleanUpOrphanedSchema(NameIdentifier catalogIdent,
NameIdentifier schemaIdent) {
+ try {
+ // A queued schema writer also blocks new readers. Avoid queuing one
behind another table's
+ // source DDL when the schema still exists and there is nothing to clean
up.
+ boolean schemaExists =
+ TreeLockUtils.doWithTreeLock(
+ schemaIdent,
+ LockType.READ,
+ () ->
+ doWithCatalog(
+ catalogIdent,
+ c -> c.doWithSchemaOps(s -> s.schemaExists(schemaIdent)),
+ RuntimeException.class));
+ if (schemaExists) {
+ return;
+ }
+ TreeLockUtils.doWithTreeLock(
+ schemaIdent,
+ LockType.WRITE,
+ () -> {
+ // Recheck under the write lock before deleting any registration.
+ OrphanedSchemaCleanup.cleanUp(this, catalogIdent, schemaIdent);
+ return null;
+ });
+ } catch (Exception e) {
+ LOG.warn("Failed to clean up orphaned schema entities starting from {}",
schemaIdent, e);
+ }
+ }
+
private static void validateColumns(Column[] columns) {
for (Column column : columns) {
ColumnEntity.NAME.validate(column.name(), COLUMN);
- ColumnEntity.COMMENT.validate(column.comment(), COLUMN);
+ ColumnEntity.validateCommentForWrite(column.name(), column.comment());
}
}
@@ -1162,7 +1195,7 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
if (addColumn.getFieldName().length == 1) {
ColumnEntity.NAME.validate(addColumn.getFieldName()[0], COLUMN);
- ColumnEntity.COMMENT.validate(addColumn.getComment(), COLUMN);
+ ColumnEntity.validateCommentForWrite(addColumn.getFieldName()[0],
addColumn.getComment());
}
} else if (change instanceof TableChange.RenameColumn) {
TableChange.RenameColumn renameColumn = (TableChange.RenameColumn)
change;
@@ -1172,7 +1205,8 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
} else if (change instanceof TableChange.UpdateColumnComment) {
TableChange.UpdateColumnComment updateComment =
(TableChange.UpdateColumnComment) change;
if (updateComment.getFieldName().length == 1) {
- ColumnEntity.COMMENT.validate(updateComment.getNewComment(), COLUMN);
+ ColumnEntity.validateCommentForWrite(
+ updateComment.getFieldName()[0], updateComment.getNewComment());
}
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/connector/capability/Capability.java
b/core/src/main/java/org/apache/gravitino/connector/capability/Capability.java
index c48e1766ed..164e6dc461 100644
---
a/core/src/main/java/org/apache/gravitino/connector/capability/Capability.java
+++
b/core/src/main/java/org/apache/gravitino/connector/capability/Capability.java
@@ -85,13 +85,13 @@ public interface Capability {
}
/**
- * Checks a table or schema name on an existing-table deletion path.
+ * Checks an existing table or schema name for deletion, or an existing
column reference.
*
* <p>The default preserves the create-time restrictions. A catalog may
accept additional native
- * names only when its deletion operations safely quote them. This does not
relax creation or
- * rename validation.
+ * names only when the corresponding operations safely quote them. This does
not relax creation or
+ * rename-target validation.
*
- * @param scope the table or schema scope
+ * @param scope the table, schema, or column scope
* @param name the existing name to check
* @return whether the existing name is supported
*/
diff --git a/core/src/main/java/org/apache/gravitino/lock/TreeLock.java
b/core/src/main/java/org/apache/gravitino/lock/TreeLock.java
index a9c2d7013a..3186979307 100644
--- a/core/src/main/java/org/apache/gravitino/lock/TreeLock.java
+++ b/core/src/main/java/org/apache/gravitino/lock/TreeLock.java
@@ -39,22 +39,22 @@ import org.slf4j.LoggerFactory;
* /metalake readLock
* /metalake/catalog readLock
* /metalake/catalog/db1 readLock
- * /metalake/catalog/db/table1 readLock
+ * /metalake/catalog/db1/table1 readLock
* </pre>
*
- * If we need to alter a table `metalake.catalog.db1.table1` (without changing
the name of it), the
- * lock manager will lock the following:
+ * If we need to drop a table `metalake.catalog.db1.table1`, the lock manager
will lock the
+ * following:
*
* <pre>
* / readLock
* /metalake readLock
* /metalake/catalog readLock
* /metalake/catalog/db1 readLock
- * /metalake/catalog/db/table1 writeLock
+ * /metalake/catalog/db1/table1 writeLock
* </pre>
*
- * When we need to rename a table or drop a table
`metalake.catalog.db1.table1`, the lock manager
- * will lock the following:
+ * When we need to rename a table `metalake.catalog.db1.table1`, the lock
manager will lock the
+ * following:
*
* <pre>
* / readLock
diff --git a/core/src/main/java/org/apache/gravitino/meta/ColumnEntity.java
b/core/src/main/java/org/apache/gravitino/meta/ColumnEntity.java
index 68a1618ed6..cf70c121ba 100644
--- a/core/src/main/java/org/apache/gravitino/meta/ColumnEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/ColumnEntity.java
@@ -20,9 +20,11 @@
package org.apache.gravitino.meta;
import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import java.util.Collections;
import java.util.Map;
+import javax.annotation.Nullable;
import lombok.ToString;
import org.apache.gravitino.Audit;
import org.apache.gravitino.Auditable;
@@ -162,13 +164,55 @@ public class ColumnEntity implements Entity, Auditable {
return new Builder();
}
+ /**
+ * Validates a user-supplied column comment before writing source or managed
metadata. H2 counts
+ * UTF-16 units in VARCHAR limits, so accepted comments must fit that
portable bound without the
+ * lossy projection used for external imports.
+ *
+ * @param columnName the column name used in validation errors
+ * @param comment the requested comment, which may be null
+ * @throws IllegalArgumentException if the comment contains NUL or exceeds
the storage limit
+ */
+ public static void validateCommentForWrite(String columnName, @Nullable
String comment) {
+ Preconditions.checkArgument(
+ comment == null || comment.indexOf('\0') < 0,
+ "Column comment for %s must not contain NUL characters",
+ columnName);
+ COMMENT.validate(comment, EntityType.COLUMN);
+ Preconditions.checkArgument(
+ comment == null || comment.length() <=
EntityFieldLimits.MAX_COMMENT_LENGTH,
+ "Column comment for %s must not exceed %s UTF-16 units",
+ columnName,
+ EntityFieldLimits.MAX_COMMENT_LENGTH);
+ }
+
+ /**
+ * Projects a source column comment into the portable entity-store
representation. The full source
+ * comment remains available from the connector table.
+ *
+ * @param comment the source comment
+ * @return a NUL-free comment bounded by the storage capacity
+ */
+ @Nullable
+ public static String commentForStorage(@Nullable String comment) {
+ if (comment == null) {
+ return null;
+ }
+ String safe = comment.replace("\0", "");
+ int end = Math.min(safe.length(), EntityFieldLimits.MAX_COMMENT_LENGTH);
+ if (end < safe.length() && end > 0 &&
Character.isHighSurrogate(safe.charAt(end - 1))) {
+ end--;
+ }
+ return safe.substring(0, end);
+ }
+
public static ColumnEntity toColumnEntity(
Column column, int position, long uid, AuditInfo audit) {
return builder()
.withId(uid)
.withName(column.name())
.withPosition(position)
- .withComment(column.comment())
+ .withComment(commentForStorage(column.comment()))
.withDataType(column.dataType())
.withNullable(column.nullable())
.withAutoIncrement(column.autoIncrement())
diff --git a/core/src/test/java/org/apache/gravitino/TestCatalog.java
b/core/src/test/java/org/apache/gravitino/TestCatalog.java
index b306f91157..e05c6286a0 100644
--- a/core/src/test/java/org/apache/gravitino/TestCatalog.java
+++ b/core/src/test/java/org/apache/gravitino/TestCatalog.java
@@ -86,7 +86,7 @@ public class TestCatalog extends BaseCatalog<TestCatalog> {
* Mirrors the JDBC override: runs a real {@code testConnection} probe at
create time instead of
* only eagerly initializing {@code ops()}. {@code ops()} is still invoked
(so a failing {@code
* initialize()} still surfaces here), and the probe additionally throws
when the catalog was
- * created with {@code need-fail=true}, exercising the create-time
verification/rollback path.
+ * created with {@code need-fail=true}, exercising the create-time
verification failure path.
*/
@Override
public void validateOnCreate() {
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCapabilityHelpers.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCapabilityHelpers.java
index b1b4ca1618..86207c2716 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCapabilityHelpers.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCapabilityHelpers.java
@@ -21,13 +21,71 @@ package org.apache.gravitino.catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.connector.capability.CapabilityResult;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
public class TestCapabilityHelpers {
+ @Test
+ void testNewColumnDoesNotUseExistingNameRules() {
+ Capability capability =
+ new Capability() {
+ @Override
+ public CapabilityResult specificationOnExistingName(Scope scope,
String name) {
+ throw new AssertionError("New columns must use creation rules");
+ }
+ };
+ TableChange[] changes =
+ CapabilityHelpers.applyCapabilities(
+ capability,
+ TableChange.addColumn(new String[] {"new_column"},
Types.IntegerType.get()));
+ Assertions.assertEquals(1, changes.length);
+ }
+
+ /** Checks existing references and new names use their respective provider
rules. */
+ @Test
+ public void testExistingColumnReferencesDoNotUseCreationPattern() {
+ Capability strict =
+ new Capability() {
+ @Override
+ public CapabilityResult specificationOnName(Scope scope, String
name) {
+ return name.matches("[a-zA-Z_][a-zA-Z0-9_]*")
+ ? CapabilityResult.SUPPORTED
+ : CapabilityResult.unsupported("invalid new name");
+ }
+
+ @Override
+ public CapabilityResult specificationOnExistingName(Scope scope,
String name) {
+ return CapabilityResult.SUPPORTED;
+ }
+ };
+ TableChange[] changes =
+ CapabilityHelpers.applyCapabilities(
+ strict,
+ TableChange.updateColumnComment(new String[] {"native col"},
"comment"),
+ TableChange.renameColumn(new String[] {"native col"},
"valid_name"),
+ TableChange.deleteColumn(new String[] {"native col"}, false),
+ TableChange.updateColumnPosition(
+ new String[] {"native col"},
TableChange.ColumnPosition.after("other native col")));
+ Assertions.assertEquals(4, changes.length);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ CapabilityHelpers.applyCapabilities(
+ strict, TableChange.addColumn(new String[] {"new col"},
Types.IntegerType.get())));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ CapabilityHelpers.applyCapabilities(
+ strict, TableChange.renameColumn(new String[] {"native col"},
"new col")));
+ }
+
@Test
void testGetCapabilityPropagatesNoSuchCatalogException() {
NameIdentifier tableIdent =
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 0d2287d985..3ec95226e0 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -48,6 +48,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -111,6 +112,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
@@ -173,6 +175,189 @@ public class TestCatalogManager {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCatalogCreateIsInvisibleUntilValidationCompletes(boolean
failValidation)
+ throws Exception {
+ NameIdentifier ident = NameIdentifier.of(metalake, "publication_test");
+ CountDownLatch validationStarted = new CountDownLatch(1);
+ CountDownLatch releaseValidation = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ RuntimeException failure = new IllegalArgumentException("backend
validation failed");
+ try (CatalogManager manager =
+ Mockito.spy(new CatalogManager(config, entityStore, new
RandomIdGenerator()))) {
+ Mockito.doAnswer(
+ invocation -> {
+ validationStarted.countDown();
+ Assertions.assertTrue(releaseValidation.await(10,
TimeUnit.SECONDS));
+ if (failValidation) {
+ throw failure;
+ }
+ return invocation.callRealMethod();
+ })
+ .when(manager)
+ .createCatalogWrapper(any(CatalogEntity.class), Mockito.anyMap());
+ Future<Catalog> creation =
+ executor.submit(
+ () ->
+ manager.createCatalog(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
catalogProps()));
+ try {
+ Assertions.assertTrue(validationStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertEquals(
+ 0,
+ executor
+ .submit(() -> manager.listCatalogs(Namespace.of(metalake)))
+ .get(3, TimeUnit.SECONDS)
+ .length);
+ Assertions.assertEquals(
+ 0,
+ executor
+ .submit(() -> manager.listCatalogsInfo(Namespace.of(metalake)))
+ .get(3, TimeUnit.SECONDS)
+ .length);
+ Assertions.assertNull(manager.getCatalogCache().getIfPresent(ident));
+ releaseValidation.countDown();
+ if (failValidation) {
+ Exception error =
+ Assertions.assertThrows(Exception.class, () -> creation.get(5,
TimeUnit.SECONDS));
+ Assertions.assertSame(failure, error.getCause());
+ Assertions.assertFalse(entityStore.exists(ident,
EntityType.CATALOG));
+ } else {
+ Assertions.assertEquals(ident.name(), creation.get(5,
TimeUnit.SECONDS).name());
+ Assertions.assertEquals(1,
manager.listCatalogsInfo(Namespace.of(metalake)).length);
+ Mockito.verify(manager, Mockito.never())
+ .createCatalogWrapper(any(CatalogEntity.class), eq(null));
+ }
+ } finally {
+ releaseValidation.countDown();
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+ }
+
+ @Test
+ void testCatalogInsertAndCachePublicationDoNotPermitDuplicateLoad() throws
Exception {
+ InMemoryEntityStore store = Mockito.spy(new InMemoryEntityStore());
+ store.initialize(config);
+ store.put(metalakeEntity, true);
+ NameIdentifier ident = NameIdentifier.of(metalake, "publication_gap");
+ CountDownLatch inserted = new CountDownLatch(1);
+ CountDownLatch releaseInsert = new CountDownLatch(1);
+ CountDownLatch loadStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Mockito.doAnswer(
+ invocation -> {
+ Object result = invocation.callRealMethod();
+ inserted.countDown();
+ Assertions.assertTrue(releaseInsert.await(10, TimeUnit.SECONDS));
+ return result;
+ })
+ .when(store)
+ .put(any(CatalogEntity.class), eq(false));
+ try (CatalogManager manager =
+ Mockito.spy(new CatalogManager(config, store, new
RandomIdGenerator()))) {
+ Mockito.doAnswer(
+ invocation -> {
+ loadStarted.countDown();
+ return invocation.callRealMethod();
+ })
+ .when(manager)
+ .acquireCatalogLease(ident);
+ Future<Catalog> creation =
+ executor.submit(
+ () ->
+ manager.createCatalog(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
catalogProps()));
+ try {
+ Assertions.assertTrue(inserted.await(5, TimeUnit.SECONDS));
+ Future<Catalog[]> listing =
+ executor.submit(() ->
manager.listCatalogsInfo(Namespace.of(metalake)));
+ Assertions.assertTrue(loadStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertThrows(
+ TimeoutException.class, () -> listing.get(200,
TimeUnit.MILLISECONDS));
+ releaseInsert.countDown();
+ Assertions.assertEquals(ident.name(), creation.get(5,
TimeUnit.SECONDS).name());
+ Assertions.assertEquals(ident.name(), listing.get(5,
TimeUnit.SECONDS)[0].name());
+ Mockito.verify(manager, Mockito.times(1))
+ .createCatalogWrapper(any(CatalogEntity.class), Mockito.anyMap());
+ Mockito.verify(manager, Mockito.never())
+ .createCatalogWrapper(any(CatalogEntity.class), eq(null));
+ } finally {
+ releaseInsert.countDown();
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ void testFailedCatalogInsertClosesInitializedWrapper() throws Exception {
+ InMemoryEntityStore store = Mockito.spy(new InMemoryEntityStore());
+ store.initialize(config);
+ store.put(metalakeEntity, true);
+ IOException failure = new IOException("catalog insert failed");
+ Mockito.doThrow(failure).when(store).put(any(CatalogEntity.class),
eq(false));
+ AtomicReference<CatalogManager.CatalogWrapper> initialized = new
AtomicReference<>();
+ try (CatalogManager manager =
+ Mockito.spy(new CatalogManager(config, store, new
RandomIdGenerator()))) {
+ Mockito.doAnswer(
+ invocation -> {
+ CatalogManager.CatalogWrapper wrapper =
+ (CatalogManager.CatalogWrapper)
invocation.callRealMethod();
+ initialized.set(wrapper);
+ return wrapper;
+ })
+ .when(manager)
+ .createCatalogWrapper(any(CatalogEntity.class), Mockito.anyMap());
+ RuntimeException error =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ manager.createCatalog(
+ NameIdentifier.of(metalake, "failed_insert"),
+ Catalog.Type.RELATIONAL,
+ provider,
+ "comment",
+ catalogProps()));
+ Assertions.assertSame(failure, error.getCause());
+ Assertions.assertNotNull(initialized.get());
+ Assertions.assertTrue(initialized.get().isRetired());
+ Assertions.assertEquals(0, initialized.get().activeOperations());
+ Assertions.assertNull(initialized.get().catalog());
+ Mockito.verify(store, Mockito.never()).delete(any(),
eq(EntityType.CATALOG), eq(true));
+ } finally {
+ store.close();
+ }
+ }
+
+ /** Checks a catalog creation does not take a sibling catalog lock. */
+ @Test
+ public void testCatalogCreationDoesNotLockSiblingCatalogs() throws Exception
{
+ try (TreeLockTestSupport.HeldLock sibling =
+ TreeLockTestSupport.HeldLock.acquire(
+ NameIdentifier.of(metalake, "existing_sibling"), LockType.READ)) {
+ TreeLockTestSupport.assertRunsConcurrentlyWith(
+ sibling,
+ () ->
+ catalogManager.createCatalog(
+ NameIdentifier.of(metalake, "independent_catalog_create"),
+ Catalog.Type.RELATIONAL,
+ provider,
+ "comment",
+ ImmutableMap.of(
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3")));
+ }
+ }
+
@Test
void testJdbcDriverDefaultAfterProviderConfigMerge(@TempDir Path
catalogPackage)
throws IOException {
@@ -615,7 +800,7 @@ public class TestCatalogManager {
catalogManager.createCatalog(
failIdent, Catalog.Type.RELATIONAL, provider, "comment",
failProps));
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(failIdent));
- // Store rollback: the just-persisted entity is deleted.
+ // Failed backend validation must not leave a catalog entity.
Assertions.assertThrows(
NoSuchEntityException.class,
() -> entityStore.get(failIdent, EntityType.CATALOG,
CatalogEntity.class));
@@ -1095,7 +1280,7 @@ public class TestCatalogManager {
}
@Test
- void testFailedCreateCatalogCleanupMarksLocalMutation() throws Exception {
+ void testFailedCatalogValidationDoesNotSuppressRemoteInvalidation() throws
Exception {
ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore();
store.initialize(config);
store.put(metalakeEntity, true);
@@ -1103,8 +1288,7 @@ public class TestCatalogManager {
CatalogManager manager = new CatalogManager(config, store, new
RandomIdGenerator());
NameIdentifier ident = NameIdentifier.of("metalake",
"failed_create_cleanup");
- // A creation that fails validation (key1 is required but missing) stores
the entity and then
- // rolls it back via store.delete(), which writes a DROP record to the
entity change log.
+ // Validation failure must leave no catalog row. A later valid create can
reuse the name.
Map<String, String> invalidProps =
ImmutableMap.of(
"provider", "test", PROPERTY_KEY2, "value2", PROPERTY_KEY5_PREFIX
+ "1", "v");
@@ -1129,25 +1313,13 @@ public class TestCatalogManager {
Assertions.assertNotNull(manager.loadCatalogAndWrap(ident));
Assertions.assertNotNull(manager.getCatalogCache().getIfPresent(ident));
- // Simulate a subsequent local mutation (e.g. disableCatalog) that writes
an ALTER record.
- manager.markLocalMutation(ident);
-
- // The poller delivers both records in one batch: the DROP from the
failed-create cleanup and
- // the ALTER from the local mutation. The cleanup DROP must carry its own
local-mutation token
- // (the fix); otherwise it consumes the ALTER's token, the ALTER is
treated as remote, and the
- // in-use cached wrapper is spuriously invalidated and asynchronously
closed.
+ // Validation failed before persistence, so no cleanup DROP or local token
should exist.
+ // A remote ALTER must therefore invalidate the newly created catalog.
store
.listener
.get()
.onEntityChange(
List.of(
- new EntityChangeRecord(
- 1L,
- "metalake",
- "CATALOG",
- "metalake.failed_create_cleanup",
- OperateType.DROP,
- 0L),
new EntityChangeRecord(
2L,
"metalake",
@@ -1156,10 +1328,9 @@ public class TestCatalogManager {
OperateType.ALTER,
0L)));
- Assertions.assertNotNull(
+ Assertions.assertNull(
manager.getCatalogCache().getIfPresent(ident),
- "Cache should NOT be invalidated: the failed-create cleanup DROP must
be tracked as a "
- + "local mutation so it does not steal the token meant for the
later ALTER");
+ "Failed validation must not leave a local token that suppresses a
remote ALTER");
manager.close();
}
@@ -1943,4 +2114,16 @@ public class TestCatalogManager {
Assertions.assertFalse(testProps.containsKey(ID_KEY),
"`gravitino.identifier` is missing");
}
+
+ private static Map<String, String> catalogProps() {
+ return ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
b/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
index c660a40b67..8fb23c1c2a 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
@@ -121,6 +121,65 @@ public class TestManagedTableOperations {
Assertions.assertEquals("schema1", schema.name());
}
+ @Test
+ void testUnicodeCommentValidationAndRoundTrip() {
+ NameIdentifier ident =
+ NameIdentifierUtil.ofTable(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"unicode_comments");
+ for (String comment : new String[] {"a".repeat(255) + "๐",
"๐".repeat(256), "bad\0comment"}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperations.createTable(
+ ident,
+ new Column[] {Column.of("c", Types.StringType.get(),
comment)},
+ "comment",
+ StringIdentifier.newPropertiesWithId(
+ StringIdentifier.fromId(idGenerator.nextId()), Map.of()),
+ new Transform[0],
+ Distributions.NONE,
+ new SortOrder[0],
+ Indexes.EMPTY_INDEXES));
+ }
+ String comment = "๐".repeat(128);
+ Table created =
+ tableOperations.createTable(
+ ident,
+ new Column[] {Column.of("c", Types.StringType.get(), comment)},
+ "comment",
+ StringIdentifier.newPropertiesWithId(
+ StringIdentifier.fromId(idGenerator.nextId()), Map.of()),
+ new Transform[0],
+ Distributions.NONE,
+ new SortOrder[0],
+ Indexes.EMPTY_INDEXES);
+ Assertions.assertEquals(comment, created.columns()[0].comment());
+ Assertions.assertEquals(comment,
tableOperations.loadTable(ident).columns()[0].comment());
+ String mixed = "a".repeat(254) + "๐";
+ Table altered =
+ tableOperations.alterTable(
+ ident,
+ TableChange.updateColumnComment(new String[] {"c"}, mixed),
+ TableChange.addColumn(new String[] {"added"},
Types.StringType.get(), comment));
+ Assertions.assertEquals(mixed, altered.columns()[0].comment());
+ Assertions.assertEquals(comment, altered.columns()[1].comment());
+ for (String invalid : new String[] {"a".repeat(255) + "๐",
"๐".repeat(256), "bad\0comment"}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperations.alterTable(
+ ident, TableChange.updateColumnComment(new String[] {"c"},
invalid)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperations.alterTable(
+ ident,
+ TableChange.addColumn(
+ new String[] {"rejected"}, Types.StringType.get(),
invalid)));
+ Assertions.assertEquals(mixed,
tableOperations.loadTable(ident).columns()[0].comment());
+ Assertions.assertEquals(2,
tableOperations.loadTable(ident).columns().length);
+ }
+ }
+
@Test
public void testCreateAndListTables() {
// Create a table
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestTableNormalizeDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestTableNormalizeDispatcher.java
index 7f1b8d90a2..fef9f9d03d 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableNormalizeDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableNormalizeDispatcher.java
@@ -54,6 +54,8 @@ import org.apache.gravitino.utils.ThrowableFunction;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
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.Mockito;
@@ -82,6 +84,8 @@ public class TestTableNormalizeDispatcher extends
TestOperationDispatcher {
.thenReturn(CapabilityResult.unsupported("case insensitive"));
Mockito.when(capability.specificationOnName(Mockito.any(),
Mockito.anyString()))
.thenReturn(CapabilityResult.SUPPORTED);
+ Mockito.when(capability.specificationOnExistingName(Mockito.any(),
Mockito.anyString()))
+ .thenCallRealMethod();
Mockito.when(catalog.capability()).thenReturn(capability);
CatalogTestUtils.mockDoWithCatalog(manager, catalog);
CatalogManager.CatalogWrapper wrapper =
Mockito.mock(CatalogManager.CatalogWrapper.class);
@@ -125,6 +129,57 @@ public class TestTableNormalizeDispatcher extends
TestOperationDispatcher {
Mockito.verify(capability,
Mockito.times(3)).caseSensitiveOnName(Capability.Scope.COLUMN);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testExistingColumnReferencesAreCaseFoldedBeforeValidation(boolean
uppercaseReference)
+ throws Exception {
+ TableDispatcher delegate = Mockito.mock(TableDispatcher.class);
+ CatalogManager manager = Mockito.mock(CatalogManager.class);
+ BaseCatalog<?> catalog = Mockito.mock(BaseCatalog.class);
+ Capability capability =
+ new Capability() {
+ @Override
+ public CapabilityResult caseSensitiveOnName(Scope scope) {
+ return scope == Scope.COLUMN
+ ? CapabilityResult.unsupported("case insensitive columns")
+ : CapabilityResult.SUPPORTED;
+ }
+
+ @Override
+ public CapabilityResult specificationOnName(Scope scope, String
name) {
+ return scope != Scope.COLUMN || name.matches("[a-z_][a-z0-9_]*")
+ ? CapabilityResult.SUPPORTED
+ : CapabilityResult.unsupported("lowercase column names
required");
+ }
+ };
+ Mockito.when(catalog.capability()).thenReturn(capability);
+ CatalogTestUtils.mockDoWithCatalog(manager, catalog);
+ TableNormalizeDispatcher dispatcher = new
TableNormalizeDispatcher(delegate, manager);
+ NameIdentifier ident = NameIdentifier.of("metalake", "catalog", "schema",
"table");
+ String[] column = {uppercaseReference ? "COL" : "col"};
+ String previous = uppercaseReference ? "previous" : "PREVIOUS";
+ dispatcher.alterTable(
+ ident,
+ TableChange.updateColumnComment(column, "comment"),
+ TableChange.renameColumn(column, "RENAMED"),
+ TableChange.deleteColumn(column, false),
+ TableChange.updateColumnPosition(column,
TableChange.ColumnPosition.after(previous)));
+
+ ArgumentCaptor<TableChange[]> captor =
ArgumentCaptor.forClass(TableChange[].class);
+ Mockito.verify(delegate).alterTable(Mockito.eq(ident), captor.capture());
+ TableChange[] changes = captor.getValue();
+ Assertions.assertEquals(4, changes.length);
+ for (TableChange change : changes) {
+ Assertions.assertArrayEquals(
+ new String[] {"col"}, ((TableChange.ColumnChange)
change).fieldName());
+ }
+ Assertions.assertEquals("renamed", ((TableChange.RenameColumn)
changes[1]).getNewName());
+ Assertions.assertEquals(
+ "previous",
+ ((TableChange.After) ((TableChange.UpdateColumnPosition)
changes[3]).getPosition())
+ .getColumn());
+ }
+
@Test
void testDeletionCanUseProviderValidatedNativeNamesWithoutRelaxingCreation()
throws Exception {
TableDispatcher delegate = Mockito.mock(TableDispatcher.class);
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
index fec1538e2d..0ba1ffe118 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -33,6 +33,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
@@ -59,6 +60,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.reflect.MethodUtils;
import org.apache.gravitino.Config;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
@@ -121,6 +123,259 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
GravitinoEnv.getInstance(), "internalSchemaDispatcher",
schemaOperationDispatcher, true);
}
+ /** Checks blocked source DDL permits warm sibling metadata reads. */
+ @Test
+ public void testBlockedSourceDropDoesNotBlockSiblingMetadataReads() throws
Exception {
+ Namespace ns = Namespace.of(metalake, catalog, "drop_source_lock_scope");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(ns.levels()), "comment", Map.of("k1", "v1", "k2",
"v2"));
+ NameIdentifier target = NameIdentifier.of(ns, "target");
+ NameIdentifier sibling = NameIdentifier.of(ns, "sibling");
+ createTable(target);
+ createTable(sibling);
+ NameIdentifier completedDrop = NameIdentifier.of(ns, "completed_drop");
+ createTable(completedDrop);
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ TestCatalog base =
+ catalogManager.doWithCatalog(NameIdentifier.of(metalake, catalog), c
-> (TestCatalog) c);
+ TestCatalogOperations original = (TestCatalogOperations) base.ops();
+ TestCatalogOperations source = spy(original);
+ FieldUtils.writeField(base, "ops", source, true);
+ doAnswer(
+ invocation -> {
+ started.countDown();
+ Assertions.assertTrue(release.await(5, TimeUnit.SECONDS));
+ return invocation.callRealMethod();
+ })
+ .when(source)
+ .dropTable(target);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Boolean> drop = executor.submit(() ->
tableOperationDispatcher.dropTable(target));
+ try {
+ Assertions.assertTrue(started.await(5, TimeUnit.SECONDS));
+ // Cleanup for a completed drop must not queue a schema writer behind
the blocked drop.
+ Assertions.assertTrue(
+ executor
+ .submit(() -> tableOperationDispatcher.dropTable(completedDrop))
+ .get(3, TimeUnit.SECONDS));
+ executor
+ .submit(
+ () -> {
+ Assertions.assertEquals(2,
tableOperationDispatcher.listTables(ns).length);
+ Assertions.assertEquals(
+ "sibling",
tableOperationDispatcher.loadTable(sibling).name());
+ })
+ .get(3, TimeUnit.SECONDS);
+ Assertions.assertFalse(drop.isDone());
+ release.countDown();
+ Assertions.assertTrue(drop.get(5, TimeUnit.SECONDS));
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ FieldUtils.writeField(base, "ops", original, true);
+ }
+ }
+
+ @Test
+ void testSchemaCleanupProbeFailureAndRecreatedSchemaPreserveRegistration()
throws Exception {
+ for (boolean failProbe : new boolean[] {true, false}) {
+ Namespace ns = Namespace.of(metalake, catalog, "cleanup_probe_" +
failProbe);
+ NameIdentifier schemaIdent = NameIdentifier.of(ns.levels());
+ schemaOperationDispatcher.createSchema(
+ schemaIdent, "comment", Map.of("k1", "v1", "k2", "v2"));
+ NameIdentifier tableIdent = NameIdentifier.of(ns, "t");
+ createTable(tableIdent);
+ TestCatalog base =
+ catalogManager.doWithCatalog(NameIdentifier.of(metalake, catalog), c
-> (TestCatalog) c);
+ TestCatalogOperations original = (TestCatalogOperations) base.ops();
+ TestCatalogOperations source = spy(original);
+ FieldUtils.writeField(base, "ops", source, true);
+ try {
+ if (failProbe) {
+ doThrow(new IllegalStateException("schema probe failed"))
+ .when(source)
+ .schemaExists(schemaIdent);
+ } else {
+ // The schema reappears between the read probe and the write-locked
recheck.
+ doReturn(false, true).when(source).schemaExists(schemaIdent);
+ }
+ clearInvocations(entityStore);
+ Assertions.assertTrue(tableOperationDispatcher.dropTable(tableIdent));
+ verify(source, times(failProbe ? 1 : 2)).schemaExists(schemaIdent);
+ verify(entityStore, never()).delete(eq(schemaIdent), eq(SCHEMA),
anyBoolean());
+ } finally {
+ FieldUtils.writeField(base, "ops", original, true);
+ }
+ }
+ }
+
+ /** Checks external comments preserve complete responses and column
identities. */
+ @Test
+ public void
testExternalCommentsAreProjectedForStorageAndPreservedInResponses() throws
Exception {
+ Namespace ns = Namespace.of(metalake, catalog,
"external_comment_projection");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(ns.levels()), "comment", Map.of("k1", "v1", "k2",
"v2"));
+ NameIdentifier ident = NameIdentifier.of(ns, "native_table");
+ String sourceComment = "a\0" + "๐".repeat(3000) + "tail";
+ catalogManager.doWithCatalog(
+ NameIdentifier.of(metalake, catalog),
+ c -> {
+ TestCatalogOperations ops = (TestCatalogOperations) c.ops();
+ ops.createTable(
+ ident,
+ new Column[] {Column.of("c", Types.StringType.get(),
sourceComment)},
+ "comment",
+ Map.of("k1", "v1", "k2", "v2"),
+ new Transform[0]);
+ return null;
+ });
+ Table imported = tableOperationDispatcher.loadTable(ident);
+ Assertions.assertEquals(sourceComment, imported.columns()[0].comment());
+ TableEntity stored = entityStore.get(ident, TABLE, TableEntity.class);
+ String projection = ColumnEntity.commentForStorage(sourceComment);
+ Assertions.assertEquals(projection, stored.columns().get(0).comment());
+ Assertions.assertFalse(projection.contains("\0"));
+ Assertions.assertTrue(projection.length() <=
EntityFieldLimits.MAX_COMMENT_LENGTH);
+ long columnId = stored.columns().get(0).id();
+ // A second native comment change exercises the existing-column update
path.
+ String updated = "b\0" + "x".repeat(5000);
+ catalogManager.doWithCatalog(
+ NameIdentifier.of(metalake, catalog),
+ c -> {
+ ((TestCatalogOperations) c.ops())
+ .alterTable(ident, TableChange.updateColumnComment(new String[]
{"c"}, updated));
+ return null;
+ });
+ Assertions.assertEquals(
+ updated,
tableOperationDispatcher.loadTable(ident).columns()[0].comment());
+ TableEntity reloaded = entityStore.get(ident, TABLE, TableEntity.class);
+ Assertions.assertEquals(columnId, reloaded.columns().get(0).id());
+ Assertions.assertEquals(
+ ColumnEntity.commentForStorage(updated),
reloaded.columns().get(0).comment());
+ clearInvocations(entityStore);
+ Assertions.assertEquals(
+ updated,
tableOperationDispatcher.loadTable(ident).columns()[0].comment());
+ String changedSuffix = updated + "different suffix";
+ catalogManager.doWithCatalog(
+ NameIdentifier.of(metalake, catalog),
+ c -> {
+ ((TestCatalogOperations) c.ops())
+ .alterTable(
+ ident, TableChange.updateColumnComment(new String[] {"c"},
changedSuffix));
+ return null;
+ });
+ Assertions.assertEquals(
+ changedSuffix,
tableOperationDispatcher.loadTable(ident).columns()[0].comment());
+ verify(entityStore, never()).update(eq(ident), eq(TableEntity.class),
eq(TABLE), any());
+ }
+
+ @Test
+ void testColumnDifferenceUsesBoundedCommentProjection() throws Exception {
+ Column source = Column.of("c", Types.StringType.get(), "a".repeat(5000));
+ ColumnEntity stored = ColumnEntity.toColumnEntity(source, 0, 1L,
AuditInfo.EMPTY);
+ String same =
+ (String)
+ MethodUtils.invokeMethod(
+ tableOperationDispatcher, true, "columnDifferenceContent",
source, 0, stored);
+ Assertions.assertEquals("", same);
+ Column changed = Column.of("c", Types.StringType.get(), "b\0" +
"b".repeat(5000));
+ String difference =
+ (String)
+ MethodUtils.invokeMethod(
+ tableOperationDispatcher, true, "columnDifferenceContent",
changed, 0, stored);
+ Assertions.assertFalse(difference.contains("\0"));
+ Assertions.assertFalse(difference.contains("b".repeat(4097)));
+ Assertions.assertTrue(difference.contains("b".repeat(256)));
+ }
+
+ @Test
+ void testUnicodeCommentsExceedingStorageFailBeforeSourceDdl() throws
Exception {
+ Namespace ns = Namespace.of(metalake, catalog, "unicode_comment_request");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(ns.levels()), "comment", Map.of("k1", "v1", "k2",
"v2"));
+ NameIdentifier ident = NameIdentifier.of(ns, "t");
+ TestCatalog base =
+ catalogManager.doWithCatalog(NameIdentifier.of(metalake, catalog), c
-> (TestCatalog) c);
+ TestCatalogOperations original = (TestCatalogOperations) base.ops();
+ TestCatalogOperations source = spy(original);
+ FieldUtils.writeField(base, "ops", source, true);
+ try {
+ for (String comment : new String[] {"a".repeat(255) + "๐",
"๐".repeat(256)}) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.createTable(
+ ident,
+ new Column[] {Column.of("c", Types.StringType.get(),
comment)},
+ "comment",
+ Map.of("k1", "v1", "k2", "v2"),
+ new Transform[0]));
+ Assertions.assertTrue(error.getMessage().contains("256 UTF-16"));
+ }
+ verify(source, never()).createTable(any(), any(), any(), any(),
any(Transform[].class));
+ createTable(ident);
+ clearInvocations(source);
+ for (String comment : new String[] {"a".repeat(255) + "๐",
"๐".repeat(256)}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.alterTable(
+ ident,
+ TableChange.addColumn(
+ new String[] {"added"}, Types.StringType.get(),
comment)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.alterTable(
+ ident, TableChange.updateColumnComment(new String[]
{"col1"}, comment)));
+ }
+ verify(source, never()).alterTable(any(), any(TableChange[].class));
+ } finally {
+ FieldUtils.writeField(base, "ops", original, true);
+ }
+ }
+
+ /** Checks invalid request comments cannot partially apply source DDL. */
+ @Test
+ public void testNulColumnCommentsFailBeforeSourceDdl() throws Exception {
+ Namespace ns = Namespace.of(metalake, catalog, "nul_comment_request");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(ns.levels()), "comment", Map.of("k1", "v1", "k2",
"v2"));
+ NameIdentifier ident = NameIdentifier.of(ns, "t");
+ IllegalArgumentException createError =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.createTable(
+ ident,
+ new Column[] {Column.of("c", Types.StringType.get(),
"bad\0comment")},
+ "comment",
+ Map.of("k1", "v1", "k2", "v2"),
+ new Transform[0]));
+ Assertions.assertTrue(createError.getMessage().contains("Column comment
for c"));
+ Assertions.assertFalse(tableOperationDispatcher.tableExists(ident));
+ createTable(ident);
+ IllegalArgumentException addError =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.alterTable(
+ ident,
+ TableChange.addColumn(
+ new String[] {"added"}, Types.StringType.get(),
"bad\0comment")));
+ Assertions.assertTrue(addError.getMessage().contains("Column comment for
added"));
+ IllegalArgumentException alterError =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableOperationDispatcher.alterTable(
+ ident, TableChange.updateColumnComment(new String[]
{"col1"}, "bad\0comment")));
+ Assertions.assertTrue(alterError.getMessage().contains("Column comment for
col1"));
+ Assertions.assertEquals(1,
tableOperationDispatcher.loadTable(ident).columns().length);
+ }
+
@Test
public void testCreateAndListTables() throws IOException {
Namespace tableNs = Namespace.of(metalake, catalog, "schema41");
diff --git a/core/src/test/java/org/apache/gravitino/meta/TestColumnEntity.java
b/core/src/test/java/org/apache/gravitino/meta/TestColumnEntity.java
index 519bcfca36..94f578e2ce 100644
--- a/core/src/test/java/org/apache/gravitino/meta/TestColumnEntity.java
+++ b/core/src/test/java/org/apache/gravitino/meta/TestColumnEntity.java
@@ -30,6 +30,21 @@ import org.junit.jupiter.api.Test;
public class TestColumnEntity {
+ @Test
+ void testCommentStorageProjectionBoundaries() {
+ Assertions.assertNull(ColumnEntity.commentForStorage(null));
+ Assertions.assertEquals("", ColumnEntity.commentForStorage(""));
+ Assertions.assertEquals("ab", ColumnEntity.commentForStorage("a\0b"));
+ String limit = "a".repeat(256);
+ Assertions.assertEquals(limit, ColumnEntity.commentForStorage(limit));
+ Assertions.assertEquals(limit, ColumnEntity.commentForStorage(limit +
"trailing"));
+ // A supplementary character must either fit completely or be excluded
completely.
+ String prefix = "a".repeat(255);
+ Assertions.assertEquals(prefix, ColumnEntity.commentForStorage(prefix +
"๐"));
+ String exactFit = "a".repeat(254) + "๐";
+ Assertions.assertEquals(exactFit, ColumnEntity.commentForStorage(exactFit
+ "trailing"));
+ }
+
@Test
public void testColumnEntityFields() {
ColumnEntity columnEntity =
diff --git a/docs/jdbc-catalog-connection-validation.md
b/docs/jdbc-catalog-connection-validation.md
index e8e7a7d9d1..457408c826 100644
--- a/docs/jdbc-catalog-connection-validation.md
+++ b/docs/jdbc-catalog-connection-validation.md
@@ -61,9 +61,9 @@ the driver supports `Connection.isValid()`.
A relational JDBC catalog verifies that its backend is reachable when the
catalog is created,
rather than deferring the first connection to a later operation. The
verification opens a dedicated,
time-bounded one-off probe connection (separate from the pool used by normal
operations), confirms
-it, and closes it. If it fails, the create is rejected and rolled back (the
catalog entity is not
-persisted), and the caller receives a connection error that names the backend
`host:port` and the
-underlying driver message.
+it, and closes it. If it fails, creation is rejected before the catalog entity
is persisted, and the
+caller receives a connection error that names the backend `host:port` and the
underlying driver
+message.
To register a catalog whose backend is temporarily unavailable, set the
catalog property
`verify-on-create` to `false`, which skips the create-time check:
@@ -103,10 +103,11 @@ It is applied differently to the two kinds of connection:
timeout is deliberately not imposed there, because it would also abort
legitimate long-running
operations on a healthy backend (a large `ALTER TABLE`, a slow metadata
query).
- **The create-time verification probe** gets both the **connect** and the
**read** (and, for
- PostgreSQL, **login**) timeout. The probe runs under the metalake write
lock, so it must be fully
- bounded: a backend that accepts the TCP connection but never speaks the
database protocol (a load
+ PostgreSQL, **login**) timeout. The probe runs under the target catalog
write lock, so it must be
+ fully bounded: a backend that accepts the TCP connection but never speaks
the database protocol (a load
balancer with a dead backend, a wrong port, a proxy that accepts and never
forwards) would
- otherwise block a connect-only bound forever and stall the whole metalake.
+ otherwise block a connect-only bound forever, delaying operations that need
this catalog lock
+ or an ancestor write lock. Sibling catalogs can still operate while
verification runs.
A timeout value you set directly on the connection or in the `jdbc-url` query
string is never
overridden; drivers whose family is not recognized are left unbounded by
Gravitino. The pooled
diff --git a/docs/manage-relational-metadata-using-gravitino.md
b/docs/manage-relational-metadata-using-gravitino.md
index 0876445b1a..667a05ecef 100644
--- a/docs/manage-relational-metadata-using-gravitino.md
+++ b/docs/manage-relational-metadata-using-gravitino.md
@@ -280,3 +280,35 @@ identifiers =
catalog.as_table_catalog().list_tables(Namespace.of("public"))
</TabItem>
</Tabs>
+
+### External column comments and existing names
+
+Loading an external table preserves its complete source column comments in the
returned
+metadata. The entity store keeps a bounded, NUL-free projection of each column
comment,
+without changing the source comment, column identity, or column governance
metadata.
+On branch-1.3, the projection is limited to 256 UTF-16 units to match the
existing
+column-comment schema in every supported store.
+CREATE and ALTER requests reject NUL comments and comments exceeding 256
UTF-16 units on
+top-level columns before source DDL, including managed-table creation, ADD
COLUMN and comment
+updates. Nested-field ALTER comments retain provider behavior.
+Supplementary Unicode characters count as two units, so 128 emoji fit while
256 emoji do not.
+Accepted request comments are stored without truncation. The UTF-16 bound is
needed for H2's
+VARCHAR storage limit even though MySQL and PostgreSQL count code points. No
store migration
+is required.
+
+Existing top-level column references use the provider's existing-name
specification; new
+top-level columns and top-level rename targets still use creation rules.
PostgreSQL accepts quoted
+native column names while retaining its UTF-8 name-length, NUL, and wildcard
restrictions. Other providers
+retain their default validation until they explicitly support native-name
operations.
+
+Catalog creation takes the target catalog lock, permitting sibling catalogs to
operate
+while source verification runs. Catalog wrapper initialization, configuration
checks, and any
+enabled create-time backend verification complete before the entity is
persisted; local cache
+publication is coordinated with catalog loading. Failed verification does not
expose a transient
+catalog through listing. Table DROP holds the target table lock for source DDL
and registration
+removal. After releasing that table lock, best-effort cleanup checks schema
+existence under a read lock and only takes a write lock if the schema is
missing, rechecking
+before deleting orphaned registrations. This does not add SQL statement
timeouts or
+cancel blocked source DDL. Cold table import retains its existing locking
behavior. Table PURGE
+and view DROP also retain their schema write lock across source DDL and can
still block sibling
+operations; this change narrows only table DROP.
diff --git
a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TestTrinoQueryIT.java
b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TestTrinoQueryIT.java
new file mode 100644
index 0000000000..359ef7bfa9
--- /dev/null
+++
b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TestTrinoQueryIT.java
@@ -0,0 +1,194 @@
+/*
+ * 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.gravitino.trino.connector.integration.test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
+
+class TestTrinoQueryIT {
+ private static final String TABLES = "\"tb01\"\n\"tb02\"";
+ private static final String STAGING_TABLES = TABLES +
"\n\"tmp_trino_staging\"";
+ private static final Duration TIMEOUT = Duration.ofSeconds(2);
+
+ @Test
+ void testGenerationWaitsForCleanup(@TempDir Path directory) throws Exception
{
+ Files.writeString(directory.resolve("rollback.sql"), "<RETRY_UNTIL_MATCH>
SHOW TABLES;\n");
+ Files.writeString(directory.resolve("rollback.txt"), TABLES + "\n\n");
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES,
TABLES);
+ new TrinoQueryIT().executeSqlFileWithGenOutput(directory.toString(),
"rollback.sql", runner);
+ Assertions.assertEquals(TABLES + "\n",
Files.readString(directory.resolve("rollback.txt")));
+ Mockito.verify(runner, Mockito.times(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testGenerationCanReuseFinalExpectedResult(@TempDir Path directory)
throws Exception {
+ Files.writeString(directory.resolve("rollback.sql"), "<RETRY_UNTIL_MATCH>
SHOW TABLES;\n");
+ for (int trailingNewlines = 0; trailingNewlines <= 2; trailingNewlines++) {
+ Files.writeString(directory.resolve("rollback.txt"), TABLES +
"\n".repeat(trailingNewlines));
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES"))
+ .thenReturn(STAGING_TABLES, TABLES, STAGING_TABLES, TABLES);
+ TrinoQueryIT test = new TrinoQueryIT();
+ for (int generation = 0; generation < 2; generation++) {
+ test.executeSqlFileWithGenOutput(directory.toString(), "rollback.sql",
runner);
+ Assertions.assertEquals(TABLES + "\n",
Files.readString(directory.resolve("rollback.txt")));
+ }
+ Mockito.verify(runner, Mockito.times(4)).runQuery("SHOW TABLES");
+ }
+ }
+
+ @Test
+ void testGenerationRequiresExistingRetryExpectation(@TempDir Path directory)
throws Exception {
+ Files.writeString(
+ directory.resolve("rollback.sql"),
+ "CREATE TABLE t (i INT);\n<RETRY_UNTIL_MATCH> SHOW TABLES;\n");
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery(Mockito.anyString())).thenReturn("ok");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new TrinoQueryIT()
+ .executeSqlFileWithGenOutput(directory.toString(),
"rollback.sql", runner));
+ Mockito.verifyNoInteractions(runner);
+ Assertions.assertFalse(Files.exists(directory.resolve("rollback.txt")));
+ }
+
+ @Test
+ void testGenerationTimeoutPreservesGoldenFile(@TempDir Path directory)
throws Exception {
+ Files.writeString(directory.resolve("rollback.sql"), "<RETRY_UNTIL_MATCH>
SHOW TABLES;\n");
+ String previous = TABLES + "\n\n";
+ Files.writeString(directory.resolve("rollback.txt"), previous);
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES);
+ IllegalStateException error =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ new TrinoQueryIT()
+ .executeSqlFileWithGenOutput(
+ directory.toString(), "rollback.sql", runner,
Duration.ofMillis(300)));
+ Assertions.assertTrue(error.getMessage().contains("golden file
unchanged"));
+ Assertions.assertEquals(previous,
Files.readString(directory.resolve("rollback.txt")));
+ Mockito.verify(runner, Mockito.atLeast(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testGenerationKeepsStatementExpectationsAligned(@TempDir Path
directory) throws Exception {
+ Files.writeString(
+ directory.resolve("rollback.sql"), "SHOW SCHEMAS;\n<RETRY_UNTIL_MATCH>
SHOW TABLES;\n");
+ Files.writeString(directory.resolve("rollback.txt"), "\"old_schema\"\n\n"
+ TABLES + "\n\n");
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW SCHEMAS")).thenReturn("\"new_schema\"");
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES,
TABLES);
+ new TrinoQueryIT().executeSqlFileWithGenOutput(directory.toString(),
"rollback.sql", runner);
+ Assertions.assertEquals(
+ "\"new_schema\"\n\n" + TABLES + "\n",
Files.readString(directory.resolve("rollback.txt")));
+ Mockito.verify(runner).runQuery("SHOW SCHEMAS");
+ Mockito.verify(runner, Mockito.times(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testUnmarkedGenerationNeedsNoExistingGolden(@TempDir Path directory)
throws Exception {
+ Files.writeString(directory.resolve("plain.sql"), "SHOW TABLES;\n");
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(TABLES);
+ new TrinoQueryIT().executeSqlFileWithGenOutput(directory.toString(),
"plain.sql", runner);
+ Assertions.assertEquals(TABLES + "\n",
Files.readString(directory.resolve("plain.txt")));
+ Mockito.verify(runner).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testGenerationRejectsMarkedWritesBeforeExecutingSql(@TempDir Path
directory)
+ throws Exception {
+ Files.writeString(
+ directory.resolve("writes.sql"),
+ "SHOW TABLES;\n<RETRY_UNTIL_MATCH> INSERT INTO t VALUES (1);\n");
+ String previous = TABLES + "\n\n1\n\n";
+ Files.writeString(directory.resolve("writes.txt"), previous);
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new TrinoQueryIT()
+ .executeSqlFileWithGenOutput(directory.toString(),
"writes.sql", runner));
+ Mockito.verifyNoInteractions(runner);
+ Assertions.assertEquals(previous,
Files.readString(directory.resolve("writes.txt")));
+ }
+
+ @Test
+ void testSqlFileWaitsForCleanup(@TempDir Path directory) throws Exception {
+ Files.writeString(
+ directory.resolve("rollback.sql"),
+ "-- Wait for transaction rollback.\n<RETRY_UNTIL_MATCH> SHOW
TABLES;\n");
+ Files.writeString(directory.resolve("rollback.txt"), TABLES + "\n\n");
+ TrinoQueryIT.getTesterCount(directory.toString(), "pg", "rollback");
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES,
TABLES);
+ new TrinoQueryIT()
+ .executeSqlFileWithCheckResult(directory.toString(), "rollback.sql",
runner, "pg");
+ Mockito.verify(runner, Mockito.times(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testWaitsForStagingTableCleanup() {
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES,
TABLES);
+ Assertions.assertEquals(
+ TABLES,
+ TrinoQueryIT.executeQuery(runner, "<RETRY_UNTIL_MATCH> SHOW TABLES",
TABLES, TIMEOUT));
+ Mockito.verify(runner, Mockito.times(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testPersistentLeakRemainsAMismatch() {
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES);
+ String result =
+ TrinoQueryIT.executeQuery(runner, "<RETRY_UNTIL_MATCH> SHOW TABLES",
TABLES, TIMEOUT);
+ Assertions.assertEquals(STAGING_TABLES, result);
+ Assertions.assertFalse(TrinoQueryIT.match(TABLES, result));
+ Mockito.verify(runner, Mockito.atLeast(2)).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testUnmarkedQueryRunsOnce() {
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Mockito.when(runner.runQuery("SHOW TABLES")).thenReturn(STAGING_TABLES,
TABLES);
+ Assertions.assertEquals(
+ STAGING_TABLES, TrinoQueryIT.executeQuery(runner, "SHOW TABLES",
TABLES, TIMEOUT));
+ Mockito.verify(runner).runQuery("SHOW TABLES");
+ }
+
+ @Test
+ void testRejectsRetryingWrites() {
+ TrinoQueryRunner runner = Mockito.mock(TrinoQueryRunner.class);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ TrinoQueryIT.executeQuery(
+ runner, "<RETRY_UNTIL_MATCH> INSERT INTO t VALUES (1)", "",
TIMEOUT));
+ Mockito.verifyNoInteractions(runner);
+ }
+}
diff --git
a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryIT.java
b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryIT.java
index 606c78db74..29de4b0ccc 100644
---
a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryIT.java
+++
b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryIT.java
@@ -19,13 +19,16 @@
package org.apache.gravitino.trino.connector.integration.test;
import java.io.File;
-import java.io.FileOutputStream;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
@@ -37,11 +40,13 @@ 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;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.gravitino.integration.test.util.ITUtils;
import org.apache.logging.log4j.util.Strings;
import org.awaitility.Awaitility;
+import org.awaitility.core.ConditionTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,6 +62,9 @@ public class TrinoQueryIT extends TrinoQueryITBase {
private static final Map<String, TestStatus> allTestStatus = new TreeMap<>();
private static final int testParallelism = 2;
+ private static final String RETRY_FLAG = "<RETRY_UNTIL_MATCH>";
+ private static final Pattern RESULT_PATTERN =
+ Pattern.compile("((\".*?\")\\n{2,})|((\\S.*?)\\n{2,})", Pattern.DOTALL);
private static final Pattern PARAM_PATTERN =
Pattern.compile("\\$\\{(\\??[\\w_]+)\\}");
private static final Pattern QUERY_FAILED_PATTERN = Pattern.compile("^Query
\\w+ failed:");
@@ -229,26 +237,16 @@ public class TrinoQueryIT extends TrinoQueryITBase {
Matcher sqlMatcher =
Pattern.compile("([<\\w].*?);", Pattern.DOTALL |
Pattern.UNIX_LINES).matcher(sqls);
- Matcher resultMatcher =
- Pattern.compile("((\".*?\")\\n{2,})|((\\S.*?)\\n{2,})", Pattern.DOTALL
| Pattern.UNIX_LINES)
- .matcher(testResults);
+ Matcher resultMatcher = RESULT_PATTERN.matcher(testResults);
TestStatus testStatus = allTestStatus.get(TestStatus.testName(path,
catalog));
testStatus.startTime = System.currentTimeMillis();
while (sqlMatcher.find()) {
String sql = sqlMatcher.group(1);
sql = resolveParameters(sql);
- String expectResult = "";
- if (resultMatcher.find()) {
- if (resultMatcher.group(2) != null) {
- expectResult = resultMatcher.group(2).trim();
- } else {
- expectResult = resultMatcher.group(4).trim();
- }
- }
+ String expectResult = nextExpectedResult(resultMatcher);
- String result = queryRunner.runQuery(sql).trim();
- result = result.replaceAll("WARNING:.*?\\n", "");
+ String result = executeQuery(queryRunner, sql, expectResult,
Duration.ofSeconds(30));
boolean match = match(expectResult, result);
if (match) {
@@ -276,6 +274,31 @@ public class TrinoQueryIT extends TrinoQueryITBase {
LOG.info("Test progress {}/{}", passCount.get(), totalCount.get());
}
+ // Only explicitly marked read queries wait for asynchronous connector
cleanup. Keep the last
+ // result on timeout so the caller reports the expected/actual mismatch
instead of hiding a leak.
+ static String executeQuery(
+ TrinoQueryRunner runner, String sql, String expected, Duration timeout) {
+ if (!sql.startsWith(RETRY_FLAG)) {
+ return runner.runQuery(sql).trim().replaceAll("WARNING:.*?\\n", "");
+ }
+ String query = retryQuery(sql);
+ AtomicReference<String> result = new AtomicReference<>("");
+ try {
+ Awaitility.await()
+ .atMost(timeout)
+ .pollDelay(Duration.ZERO)
+ .pollInterval(Duration.ofMillis(100))
+ .until(
+ () -> {
+
result.set(runner.runQuery(query).trim().replaceAll("WARNING:.*?\\n", ""));
+ return match(expected, result.get());
+ });
+ } catch (ConditionTimeoutException e) {
+ // The existing result check will fail with the final query output.
+ }
+ return result.get();
+ }
+
/**
* * This method is used to match the result of the query. There are three
cases: 1. The expected
* result is equal to the actual result. 2. The expected result is a query
failed result, and the
@@ -475,29 +498,58 @@ public class TrinoQueryIT extends TrinoQueryITBase {
void executeSqlFileWithGenOutput(
String testSetDirName, String filename, TrinoQueryRunner queryRunner)
throws Exception {
- String path = ITUtils.joinPath(testSetDirName, filename);
- String sqls = TrinoQueryITBase.readFileToString(path);
- String resultFileName = path.replace(".sql", ".txt");
- FileOutputStream outputStream = new FileOutputStream(resultFileName);
+ executeSqlFileWithGenOutput(testSetDirName, filename, queryRunner,
Duration.ofSeconds(30));
+ }
+ void executeSqlFileWithGenOutput(
+ String testSetDirName, String filename, TrinoQueryRunner queryRunner,
Duration timeout)
+ throws Exception {
+ String path = ITUtils.joinPath(testSetDirName, filename);
+ String sqls = removeSqlComments(TrinoQueryITBase.readFileToString(path));
+ Path resultFile = Path.of(path.replace(".sql", ".txt"));
+ // Read the trusted baseline before opening the output. A transient
staging table must never
+ // become its own expected result, nor may a failed regeneration damage
the old golden file.
+ String previousResults = Files.exists(resultFile) ?
Files.readString(resultFile) : "";
+ // Match the check path's EOF normalization, including files written by
this generator.
+ Matcher resultMatcher = RESULT_PATTERN.matcher(previousResults + "\n\n");
Matcher sqlMatcher =
- Pattern.compile("(\\w.*?);", Pattern.DOTALL |
Pattern.UNIX_LINES).matcher(sqls);
-
- boolean firstLine = true;
+ Pattern.compile("([<\\w].*?);", Pattern.DOTALL |
Pattern.UNIX_LINES).matcher(sqls);
+ List<String> queries = new ArrayList<>();
+ List<String> expectations = new ArrayList<>();
while (sqlMatcher.find()) {
- if (!firstLine) {
- outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
+ String sql = resolveParameters(sqlMatcher.group(1));
+ String expected = nextExpectedResult(resultMatcher);
+ if (sql.startsWith(RETRY_FLAG)) {
+ retryQuery(sql);
+ if (expected.isEmpty()) {
+ throw new IllegalArgumentException(
+ "RETRY_UNTIL_MATCH requires an existing expected result in " +
resultFile);
+ }
}
- firstLine = false;
+ queries.add(sql);
+ expectations.add(expected);
+ }
- String sql = resolveParameters(sqlMatcher.group(1));
- String result = queryRunner.runQuery(sql).trim();
+ StringBuilder output = new StringBuilder();
+ for (int i = 0; i < queries.size(); i++) {
+ String sql = queries.get(i);
+ String result;
+ if (sql.startsWith(RETRY_FLAG)) {
+ result = executeQuery(queryRunner, sql, expectations.get(i), timeout);
+ if (!match(expectations.get(i), result)) {
+ throw new IllegalStateException(
+ "Timed out waiting for " + sql + "; golden file unchanged: " +
resultFile);
+ }
+ } else {
+ result = queryRunner.runQuery(sql).trim();
+ }
LOG.info("Execute sql:\n{}\nResult:\n{}", sql, result);
- outputStream.write(result.getBytes(StandardCharsets.UTF_8));
- outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
+ if (i > 0) {
+ output.append('\n');
+ }
+ output.append(result).append('\n');
}
-
- outputStream.close();
+ Files.writeString(resultFile, output, StandardCharsets.UTF_8);
}
static int getTesterCount(String testSetDirName, String catalog, String
testFilterPrefix)
@@ -563,4 +615,19 @@ public class TrinoQueryIT extends TrinoQueryITBase {
return catalog + ":" + simpleTesterName(filename);
}
}
+
+ private static String nextExpectedResult(Matcher matcher) {
+ if (!matcher.find()) {
+ return "";
+ }
+ return (matcher.group(2) != null ? matcher.group(2) :
matcher.group(4)).trim();
+ }
+
+ private static String retryQuery(String sql) {
+ String query = sql.substring(RETRY_FLAG.length()).trim();
+ if (!query.toUpperCase(Locale.ROOT).startsWith("SHOW ")) {
+ throw new IllegalArgumentException("RETRY_UNTIL_MATCH only supports SHOW
queries");
+ }
+ return query;
+ }
}
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
index 29b4edeae7..d3c1d0da0f 100644
---
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
@@ -71,8 +71,9 @@ INSERT INTO tb02 (f1, f2, f3, f4, f5, f6, f7, f9, f10, f11,
f12, f13, f14, f15,
VALUES ('Sample text 1', 'same9', x'65', 123.456, 7.89, 12.34, false, 100,
1000, 1000, 1992382342, DATE '2024-01-01',
NULL, TIMESTAMP '2024-01-01 08:00:00', TIMESTAMP '2024-01-01 08:00:00
UTC');
--- Failed inserts must roll back their JDBC staging tables.
-SHOW TABLES;
+-- Failed inserts must roll back their JDBC staging tables. Trino aborts
transactions
+-- asynchronously, so wait for cleanup while still failing if staging tables
remain.
+<RETRY_UNTIL_MATCH> SHOW TABLES;
CREATE TABLE tb03 (int_array ARRAY(INTEGER));
