jerryshao commented on code in PR #13771:
URL: https://github.com/apache/gravitino/pull/13771#discussion_r4258293930
##########
catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/integration/test/CatalogKafkaIT.java:
##########
@@ -192,6 +193,69 @@ public void testCatalog() throws ExecutionException,
InterruptedException {
Assertions.assertFalse(adminClient.listTopics().names().get().isEmpty());
}
+ @Test
+ public void testCatalogReregistrationLoadsExistingTopic()
+ throws ExecutionException, InterruptedException {
+ String catalogName =
GravitinoITUtils.genRandomName("test_catalog_reregistration");
+ String topicName =
GravitinoITUtils.genRandomName("test_topic_reregistration");
+ NameIdentifier topicIdent = NameIdentifier.of(DEFAULT_SCHEMA_NAME,
topicName);
+ Map<String, String> properties = ImmutableMap.of(BOOTSTRAP_SERVERS,
kafkaBootstrapServers);
+
+ adminClient.createTopics(ImmutableList.of(new NewTopic(topicName, 1,
(short) 1))).all().get();
+
+ Catalog originalCatalog = createCatalog(catalogName, "original catalog",
properties);
+ Assertions.assertEquals(
+ topicName,
originalCatalog.asTopicCatalog().loadTopic(topicIdent).name());
+
+ Assertions.assertTrue(metalake.dropCatalog(catalogName, true));
+ Assertions.assertTrue(topicExistsInKafka(topicName));
+
+ Catalog reregisteredCatalog = createCatalog(catalogName, "re-registered
catalog", properties);
+ Topic restoredTopic =
+ Assertions.assertDoesNotThrow(
+ () -> reregisteredCatalog.asTopicCatalog().loadTopic(topicIdent));
+ Assertions.assertEquals(topicName, restoredTopic.name());
+ checkTopicReadWrite(topicName);
+
+
Assertions.assertTrue(reregisteredCatalog.asTopicCatalog().dropTopic(topicIdent));
+ awaitTopicDeletedInKafka(topicName);
+ Assertions.assertTrue(metalake.dropCatalog(catalogName, true));
+ }
+
+ @Test
+ public void testActiveTopicRegistrationCannotBeRebound()
+ throws ExecutionException, InterruptedException {
+ String originalCatalogName =
+ GravitinoITUtils.genRandomName("test_active_topic_original_catalog");
+ String conflictingCatalogName =
+
GravitinoITUtils.genRandomName("test_active_topic_conflicting_catalog");
+ String topicName =
GravitinoITUtils.genRandomName("test_active_topic_registration");
+ NameIdentifier topicIdent = NameIdentifier.of(DEFAULT_SCHEMA_NAME,
topicName);
+ Map<String, String> properties = ImmutableMap.of(BOOTSTRAP_SERVERS,
kafkaBootstrapServers);
+
+ adminClient.createTopics(ImmutableList.of(new NewTopic(topicName, 1,
(short) 1))).all().get();
+
+ Catalog originalCatalog = createCatalog(originalCatalogName, "original
catalog", properties);
+ Assertions.assertEquals(
+ topicName,
originalCatalog.asTopicCatalog().loadTopic(topicIdent).name());
+
+ Catalog conflictingCatalog =
+ createCatalog(conflictingCatalogName, "conflicting catalog",
properties);
+ RESTException exception =
+ Assertions.assertThrows(
+ RESTException.class, () ->
conflictingCatalog.asTopicCatalog().loadTopic(topicIdent));
Review Comment:
[Important] This assertion is correct, and that is the problem: the topic
refusal reaches Java clients as an untyped `RESTException`, unlike the
identical refusal for schemas, tables and views.
`TopicOperationDispatcher.java:119` throws `UnsupportedOperationException`,
`TopicExceptionHandler` has no branch for it (`ExceptionHandlers.java:670-693`)
so `BaseExceptionHandler` maps it to `UNSUPPORTED_OPERATION_CODE` 1006
(`ExceptionHandlers.java:1247-1249`) - the same code the schema/table/view
paths produce. On the client, `TableErrorHandler`
(`ErrorHandlers.java:423-424`), `ViewErrorHandler` (:478-479) and
`SchemaErrorHandler` (:531-532) all rethrow 1006 as
`UnsupportedOperationException`, which is exactly what
`CatalogHive2IT.java:1518-1521` and `:1547-1549` assert. `TopicErrorHandler`
(:779-824) has no 1006 case, so it falls through to `RestErrorHandler.accept`,
which throws `RESTException("Unable to process: %s", ...)` (:1436).
So the one entity type where the operator cannot clean up the external
identifier is also the one where the error arrives as a generic transport
failure: callers can only string-match the message, and code that already
handles this PR's refusal for tables will miss it for topics.
Add `case ErrorConstants.UNSUPPORTED_OPERATION_CODE: throw new
UnsupportedOperationException(errorMessage);` to `TopicErrorHandler` and assert
`UnsupportedOperationException` here, matching `CatalogHive2IT`.
Verified by: reading `TopicExceptionHandler`, `BaseExceptionHandler.handle`,
all four client error handlers and `RestErrorHandler.accept` on this branch;
the `contains(...)` check does hold either way because `formatErrorMessage`
keeps the server message.
##########
catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogCapability.java:
##########
@@ -23,10 +23,30 @@
public class PaimonCatalogCapability implements Capability {
+ private final boolean hiveBackend;
+
+ /** Creates a Paimon catalog capability with the default case-sensitive name
behavior. */
+ public PaimonCatalogCapability() {
+ this(false);
+ }
+
+ PaimonCatalogCapability(boolean hiveBackend) {
+ this.hiveBackend = hiveBackend;
+ }
+
@Override
public CapabilityResult columnDefaultValue() {
// See https://github.com/apache/paimon/pull/1425/files
return CapabilityResult.unsupported(
"Paimon set column default value through table properties instead of
column info.");
}
+
+ @Override
+ public CapabilityResult caseSensitiveOnName(Scope scope) {
+ if (hiveBackend && (scope == Scope.SCHEMA || scope == Scope.TABLE || scope
== Scope.VIEW)) {
Review Comment:
[Question] Does the upgrade case @mchades raised for Iceberg also apply
here, and should the description mention Paimon at all?
The thread on `IcebergCatalogCapability` closed with "a pre-upgrade `Orders`
registration stays stored as `Orders` while every request now normalizes to
`orders`" being accepted as a known limitation. This block makes the same
change for `lakehouse-paimon` with `catalog-backend=hive`, where
`internalCreateTable`/`internalCreateView` likewise stored `ident.name()`
verbatim before this PR. On a case-sensitive metadata store (PostgreSQL), any
such row becomes unreachable, and the subsequent load goes down the new
create-only import path.
What I cannot settle from this repository is whether such rows can exist: it
depends on whether Paimon's Hive catalog accepted a mixed-case name at create
time (its `allow-upper-case` behaviour), which is Paimon-side code not vendored
here. If it did, this inherits exactly the limitation that was accepted for
Iceberg - fine, but then it belongs in the same note. If it never did, the fold
is purely a fix and there is nothing to record.
Either way, the "Does this PR introduce any user-facing change?" section
names only Hive-backed Iceberg; it does not mention that `lakehouse-paimon` on
a Hive backend now folds schema, table and view names too, which is a behaviour
change for anyone scripting mixed-case Paimon identifiers.
Verified by: `git diff origin/main...HEAD` on `PaimonCatalog.java:74-76` and
this file; `PaimonConstants.java:24`; the resolved/open threads on
`IcebergCatalogCapability` L65; and the PR description on this head.
##########
catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java:
##########
@@ -71,7 +71,8 @@ public ViewCatalog asViewCatalog() {
@Override
public Capability newCapability() {
- return new PaimonCatalogCapability();
+ String backend =
entity().getProperties().get(PaimonConstants.CATALOG_BACKEND);
+ return new
PaimonCatalogCapability(PaimonCatalogBackend.HIVE.name().equalsIgnoreCase(backend));
Review Comment:
[Nit] `entity().getProperties()` is dereferenced without a null check, in a
path that runs eagerly for every catalog load.
`CatalogEntity.properties` is declared `@Nullable` (`CatalogEntity.java:70`,
`Field.optional("properties", ...)` at :56), and `POConverters.fromCatalogPO`
passes whatever
`JsonUtils.anyFieldMapper().readValue(catalogPO.getProperties(), Map.class)`
returns (:279-280), which is null for a stored `null`.
`CatalogManager.initCatalogWrapper` calls `catalog.capability()` eagerly inside
the isolated class loader (`CatalogManager.java:1827`), so a null map would
fail the catalog at load, not at first use.
I could not construct a reachable path: `validatePropertyForCreate` runs
first (:1823) and both `catalog-backend` and `warehouse` are required, so a
Paimon catalog cannot be created with no properties. `IcebergCatalog.java:89`
has the identical shape, so this is consistency rather than a live bug - but
`entity().toCatalogInfo().properties()` already maps null to an empty map
(`CatalogEntity.java:151`) if you want it closed off in both places.
Verified by: reading `CatalogEntity`, `POConverters.fromCatalogPO`,
`CatalogManager.initCatalogWrapper` and `BaseCatalog.capability()` (:374-379)
on this branch; `TestGravitinoPaimonTable.java:129-136` does build a
`CatalogEntity` with no properties, so the null state is constructible in tests.
##########
catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogCapability.java:
##########
@@ -62,6 +62,7 @@ public CapabilityResult caseSensitiveOnName(Scope scope) {
switch (scope) {
case SCHEMA:
case TABLE:
+ case VIEW:
Review Comment:
[Question] With Hive, Iceberg-on-Hive and now Paimon-on-Hive folded,
`lakehouse-hudi` with `catalog-backend=hms` is the one remaining
Hive-metastore-backed provider that stays case-sensitive. Intentional?
`HudiCatalogPropertiesMetadata.java:34-51` makes `catalog-backend` a
required enum over `BackendType`, and `CatalogUtils.java:44-47` resolves `HMS`
to `HudiHMSBackend`, i.e. a real Hive metastore. `HudiCatalog` declares no
`newCapability()` override, so it keeps `Capability`'s default case-sensitive
rules for SCHEMA/TABLE, while HMS resolves `Orders` and `orders` to the same
object. The consequence is the duplicate-identity case this PR exists to
prevent, one level down: two Gravitino registrations, with two entity IDs and
two sets of tag/privilege relations, for one HMS table.
I did not find a path where this produces the new refusal - the Hudi catalog
is read-only, so Gravitino never writes `gravitino.identifier` into those
tables - which is why this is a question rather than a blocking finding. But on
a shared HMS, which is the premise of #13770, a table created through a
Gravitino Hive catalog does carry the identifier and can be read through a Hudi
catalog on the same metastore.
If it is out of scope, saying so on this thread (or in the description) is
enough.
Verified by: reading `HudiCatalogPropertiesMetadata`, `utils/CatalogUtils`,
`HudiCatalog` and grepping every `implements Capability` in `catalogs/` on this
branch - the only `caseSensitiveOnName` overrides are Hive, Iceberg, Paimon and
Glue (Glue folds SCHEMA/TABLE and exposes no views).
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestBackendIT.java:
##########
@@ -33,6 +35,17 @@
@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
public abstract class SparkIcebergCatalogRestBackendIT extends
SparkIcebergCatalogIT {
+ /**
+ * Tests replacing columns without assuming the case-sensitivity of the
catalog behind the REST
+ * service. Gravitino cannot infer that backend capability through the REST
protocol.
+ */
+ @Override
+ @Test
+ @EnabledIf("supportsReplaceColumns")
+ protected void testAlterTableReplaceColumns() {
+ runAlterTableReplaceColumnsTest(false);
Review Comment:
[Nit] This override is the right way to scope the test, but it is now the
only place the REST-backed limitation is written down.
On `main` this suite runs the shared REPLACE COLUMNS test with the uppercase
`%S` spelling and passes. On this branch it would fail, because
`IcebergCatalog.newCapability()` folds only for `catalog-backend=hive` while
`rest` routinely fronts a Hive metastore - the case @mchades and the previous
pass both raised and you declined as a known limitation. The override encodes
that decision in a test javadoc; the PR description's user-facing-change
section still says only that "Hive-backed Iceberg schema and table identifiers
follow the backend's case-insensitive behavior", which reads as if the gap does
not exist.
One sentence in the description (and ideally in the Iceberg catalog docs)
saying that an Iceberg catalog configured with `catalog-backend=rest` keeps
exact-case identifiers and will refuse a mixed-case load that previously
resolved would make the limitation discoverable to the operators who hit it.
Verified by: `git show
origin/main:spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java`
against `SparkCommonIT.java:618-658` on this head,
`IcebergCatalog.java:88-93`, and the PR description on `594b2d2`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
