roryqi commented on code in PR #13771:
URL: https://github.com/apache/gravitino/pull/13771#discussion_r4257161405
##########
catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java:
##########
@@ -86,7 +86,10 @@ static boolean shouldValidateWarehouseProperty(String
backend, String warehouse)
@Override
public Capability newCapability() {
- return new
IcebergCatalogCapability(HierarchicalSchemaUtil.schemaSeparator());
+ String backend =
entity().getProperties().get(IcebergConstants.CATALOG_BACKEND);
+ return new IcebergCatalogCapability(
+ HierarchicalSchemaUtil.schemaSeparator(),
+ IcebergCatalogBackend.HIVE.name().equalsIgnoreCase(backend));
Review Comment:
Thanks. Resolving this through a REST-backed catalog would require
discovering and propagating the downstream backend name semantics, which the
current capability boundary cannot do safely. I scoped the mixed-case assertion
to the direct Hive backend via supportsCaseInsensitiveTableNameAliases();
REST-backed catalogs keep the exact-case path. I suggest treating REST -> Hive
case folding as a known limitation of this patch.
##########
catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogCapability.java:
##########
@@ -48,6 +60,15 @@ public CapabilityResult supportsHierarchicalSchema() {
return CapabilityResult.SUPPORTED;
}
+ @Override
+ public CapabilityResult caseSensitiveOnName(Scope scope) {
+ if (hiveBackend && (scope == Scope.SCHEMA || scope == Scope.TABLE)) {
+ return CapabilityResult.unsupported(
+ "The Iceberg Hive backend is case insensitive for schemas and
tables.");
+ }
+ return Capability.super.caseSensitiveOnName(scope);
+ }
Review Comment:
Good catch. I added Scope.VIEW to the Hive-backend case-insensitive
capability and added normalization assertions for both Hive and non-Hive
backends in e8396991ba.
##########
core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java:
##########
@@ -166,14 +169,65 @@ public void testCreateAndLoadTopic() throws IOException {
AuditInfo.builder().withCreator("gravitino").withCreateTime(Instant.now()).build())
.build();
doReturn(unmatchedEntity).when(entityStore).get(any(),
eq(Entity.EntityType.TOPIC), any());
- Topic loadedTopic4 = topicOperationDispatcher.loadTopic(topicIdent1);
- // Succeed to import the topic entity
+ UnsupportedOperationException exception =
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> topicOperationDispatcher.loadTopic(topicIdent1));
+ Assertions.assertTrue(exception.getMessage().contains("immutable Kafka
topic UUID"));
+ verify(entityStore).put(any(TopicEntity.class), eq(false));
+ verify(entityStore).restoreSoftDeletedEntity(any(TopicEntity.class));
reset(entityStore);
TopicEntity topicEntity =
entityStore.get(topicIdent1, Entity.EntityType.TOPIC,
TopicEntity.class);
Assertions.assertEquals("test", topicEntity.auditInfo().creator());
- // Audit info is gotten from the catalog, not from the entity store
- Assertions.assertEquals("test", loadedTopic4.auditInfo().creator());
+ }
+
+ @Test
+ public void testConcurrentImportTopicReusesExistingEntity() throws
IOException {
+ Namespace topicNs = Namespace.of(metalake, catalog,
"schema_topic_concurrent_import");
+ NameIdentifier schemaIdent = NameIdentifier.of(topicNs.levels());
+ schemaOperationDispatcher.createSchema(
+ schemaIdent, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"));
+ NameIdentifier topicIdent = NameIdentifier.of(topicNs,
"topic_concurrent_import");
+ createTopicForLockTest(topicIdent);
+ TopicEntity importedTopicEntity = entityStore.get(topicIdent, TOPIC,
TopicEntity.class);
+
+ reset(entityStore);
+ doThrow(new NoSuchEntityException("mock error"))
+ .doThrow(new NoSuchEntityException("mock error"))
+ .doReturn(importedTopicEntity)
+ .when(entityStore)
+ .get(any(), eq(TOPIC), any());
+ doThrow(new EntityAlreadyExistsException("mock conflict"))
+ .when(entityStore)
+ .put(any(TopicEntity.class), eq(false));
+
+ Topic loadedTopic =
+ Assertions.assertDoesNotThrow(() ->
topicOperationDispatcher.loadTopic(topicIdent));
+ Assertions.assertEquals(topicIdent.name(), loadedTopic.name());
+ verify(entityStore).restoreSoftDeletedEntity(any(TopicEntity.class));
Review Comment:
Added an end-to-end Kafka regression test in e8396991ba. It creates and
imports a real Kafka topic, drops the catalog with cascade, recreates the
catalog against the same broker, and verifies loadTopic plus read/write
succeeds through the dispatcher/backend restoration path.
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java:
##########
@@ -630,6 +630,7 @@ protected void testAlterTableReplaceColumns() {
tableInfo = getTableInfo(tableName);
checkTableColumns(tableName, updateColumns, tableInfo);
+ sql(String.format("CREATE TABLE IF NOT EXISTS %S (id int)", tableName));
Review Comment:
Fixed in e8396991ba. The CREATE TABLE IF NOT EXISTS statement now uses the
complete post-REPLACE COLUMNS schema (id, name2, and age) before inserting the
three-column row.
--
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]
