This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new ce08516a59 [#9970] feat(server): Gravitino REST supports the
hierarchical schema (#11175)
ce08516a59 is described below
commit ce08516a596d9d2dde24d8ec7b5fb62056bb244c
Author: roryqi <[email protected]>
AuthorDate: Fri May 22 15:50:07 2026 +0800
[#9970] feat(server): Gravitino REST supports the hierarchical schema
(#11175)
### What changes were proposed in this pull request?
Expose hierarchical schema support through the **Gravitino
server REST API**, building on the already-merged core hierarchical
schema (#11036, #11074) and Iceberg REST support (#11118).
- **`SchemaOperations`**: add a `parentSchema` query parameter to list
schemas; scope create-schema authorization at the `SCHEMA` level. A new
**`CreateSchemaAuthorizationExecutor`** injects the parent schema for
nested names (`A:B:C`) so the authorization inheritance chain (`A:B → A
→ CATALOG`) is evaluated. Wired through `AuthorizeExecutorFactory` /
`AuthorizationRequest` (new `CREATE_SCHEMA` type).
- **`IcebergCatalogCapability`**: accept logical nested schema names
using the configured external separator (e.g. `A:B:C`), rejecting empty
segments; `IcebergCatalog` passes the configured separator.
- **`IcebergCatalogOperations.listSchemas`**: list schemas under a
parent namespace and convert multi-level Iceberg namespaces back to
logical names.
### Why are the changes needed?
This is the remaining "Gravitino REST" portion of the
hierarchical-schema feature; the core and Iceberg-REST parts were
already split out and merged.
Fix: #9970
### Does this PR introduce _any_ user-facing change?
- `GET .../schemas` accepts an optional `parentSchema` query parameter.
- Creating a schema with a separator-delimited name (default `:`)
creates a nested schema; the create-schema privilege can now be granted
at SCHEMA scope and inherits to descendants.
### How was this patch tested?
- New unit tests: `TestIcebergCatalogCapability`,
`TestIcebergCatalogOperations` (listSchemas), `TestSchemaOperations`
(parentSchema listing).
- New `HierarchicalSchemaAuthorizationIT` covering end-to-end nested schema
creation, authorization inheritance, and ownership on an Iceberg
catalog.
- `./gradlew test -PskipITs` for the affected modules passes; spotless
clean.
---
.../java/org/apache/gravitino/SupportsSchemas.java | 22 ++
.../catalog/lakehouse/iceberg/IcebergCatalog.java | 3 +-
.../iceberg/IcebergCatalogCapability.java | 55 ++++
.../iceberg/IcebergCatalogOperations.java | 28 +-
.../iceberg/TestIcebergCatalogCapability.java | 106 +++++++
.../iceberg/TestIcebergCatalogOperations.java | 55 +++-
clients/client-java/build.gradle.kts | 5 +
.../apache/gravitino/client/BaseSchemaCatalog.java | 42 ++-
.../gravitino/client/TestRelationalCatalog.java | 37 +++
.../HierarchicalSchemaAuthorizationIT.java | 330 +++++++++++++++++++++
.../gravitino/api/supports_schemas.py | 18 +-
.../gravitino/client/base_schema_catalog.py | 21 +-
.../tests/unittests/test_base_schema_catalog.py | 100 +++++++
.../catalog/SchemaOperationDispatcher.java | 41 +++
.../gravitino/hook/SchemaHookDispatcher.java | 108 ++++++-
.../gravitino/utils/HierarchicalSchemaUtil.java | 17 +-
.../catalog/TestSchemaOperationDispatcher.java | 55 ++++
.../gravitino/hook/TestSchemaHookDispatcher.java | 133 ++++++++-
.../gravitino/integration/test/util/BaseIT.java | 9 +-
.../annotations/AuthorizationRequest.java | 3 +-
.../authorization/AuthorizeExecutorFactory.java | 2 +
.../CreateSchemaAuthorizationExecutor.java | 105 +++++++
.../server/web/rest/SchemaOperations.java | 55 +++-
.../server/web/rest/TestSchemaOperations.java | 40 +++
24 files changed, 1334 insertions(+), 56 deletions(-)
diff --git a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
index 96fe10da07..42284ed24d 100644
--- a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
+++ b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
@@ -46,6 +46,28 @@ public interface SupportsSchemas {
*/
String[] listSchemas() throws NoSuchCatalogException;
+ /**
+ * List the schemas directly under the given parent schema.
+ *
+ * <p>This is only meaningful for catalogs that support hierarchical
(multi-level) schemas, such
+ * as an Iceberg catalog accessed through the Gravitino REST server with a
configured schema
+ * separator. For example, when the schemas {@code a}, {@code a:b} and
{@code a:b:c} exist, this
+ * method invoked with parent {@code a:b} returns {@code [a:b:c]}. For a
flat catalog, or a parent
+ * schema that has no children, an empty array is returned.
+ *
+ * @param parentSchema The parent (possibly hierarchical) schema name whose
direct children are
+ * listed, e.g. {@code "a"} or {@code "a:b"}. Must not be null or blank.
+ * @return An array of schema names directly under the given parent schema.
+ * @throws IllegalArgumentException If {@code parentSchema} is null or blank.
+ * @throws NoSuchCatalogException If the catalog does not exist.
+ * @throws NoSuchSchemaException If the parent schema does not exist.
+ */
+ default String[] listSchemas(String parentSchema)
+ throws NoSuchCatalogException, NoSuchSchemaException {
+ throw new UnsupportedOperationException(
+ "Listing schemas under a parent schema is not supported by this
catalog");
+ }
+
/**
* Check if a schema exists.
*
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java
index 9dc1d53414..653831464d 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java
@@ -31,6 +31,7 @@ import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.JdbcCredential;
import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
/** Implementation of an Apache Iceberg catalog in Apache Gravitino. */
public class IcebergCatalog extends BaseCatalog<IcebergCatalog> {
@@ -71,7 +72,7 @@ public class IcebergCatalog extends
BaseCatalog<IcebergCatalog> {
@Override
public Capability newCapability() {
- return new IcebergCatalogCapability();
+ return new
IcebergCatalogCapability(HierarchicalSchemaUtil.schemaSeparator());
}
@Override
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogCapability.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogCapability.java
index 90a0906629..2bac6ed4db 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogCapability.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogCapability.java
@@ -20,12 +20,67 @@ package org.apache.gravitino.catalog.lakehouse.iceberg;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
public class IcebergCatalogCapability implements Capability {
+
+ private final String schemaSeparator;
+
+ /**
+ * Creates a capability with the given external schema separator.
+ *
+ * @param schemaSeparator the external separator used in logical schema
names (e.g. {@code ":"})
+ */
+ public IcebergCatalogCapability(String schemaSeparator) {
+ this.schemaSeparator = schemaSeparator;
+ }
+
@Override
public CapabilityResult columnDefaultValue() {
// Iceberg column default value is WIP, see
// https://github.com/apache/iceberg/pull/4525
return CapabilityResult.unsupported("Iceberg does not support column
default value.");
}
+
+ /**
+ * Validates the schema name specification for Iceberg.
+ *
+ * <p>For {@link Scope#SCHEMA}, Iceberg accepts:
+ *
+ * <ul>
+ * <li>Regular flat schema names matching the default name pattern.
+ * <li>Logical hierarchical schema names using the configured external
separator (e.g. {@code
+ * "A:B:C"} with separator {@code ":"}). Each segment must be
non-empty and individually
+ * satisfy the default SCHEMA name rules (reserved words and name
pattern).
+ * </ul>
+ *
+ * <p>For all other scopes, the default validation rules apply.
+ */
+ @Override
+ public CapabilityResult specificationOnName(Scope scope, String name) {
+ if (scope == Scope.SCHEMA && name.contains(schemaSeparator)) {
+ return validateSegments(name, schemaSeparator);
+ }
+ return Capability.super.specificationOnName(scope, name);
+ }
+
+ private CapabilityResult validateSegments(String name, String separator) {
+ String[] segments = HierarchicalSchemaUtil.splitSchemaName(name,
separator);
+ for (String segment : segments) {
+ if (segment.isEmpty()) {
+ return CapabilityResult.unsupported(
+ String.format(
+ "The SCHEMA name '%s' contains an empty segment after
splitting by '%s'.",
+ name, separator));
+ }
+ CapabilityResult segmentResult =
Capability.super.specificationOnName(Scope.SCHEMA, segment);
+ if (!segmentResult.supported()) {
+ return CapabilityResult.unsupported(
+ String.format(
+ "The SCHEMA name '%s' contains an illegal segment '%s': %s",
+ name, segment, segmentResult.unsupportedMessage()));
+ }
+ }
+ return CapabilityResult.SUPPORTED;
+ }
}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
index a0e985ea49..0e7f8d48bd 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
@@ -57,6 +57,7 @@ import
org.apache.gravitino.iceberg.common.authentication.SupportsKerberos;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper.IcebergTableChange;
import
org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy;
+import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.Representation;
@@ -72,6 +73,7 @@ import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -159,13 +161,31 @@ public class IcebergCatalogOperations
@Override
public NameIdentifier[] listSchemas(Namespace namespace) throws
NoSuchCatalogException {
try {
+ String separator = HierarchicalSchemaUtil.schemaSeparator();
+ org.apache.iceberg.catalog.Namespace icebergParent;
+ if (namespace.length() == 2) {
+ icebergParent = org.apache.iceberg.catalog.Namespace.empty();
+ } else {
+ String parentPath =
+ String.join(separator, Arrays.copyOfRange(namespace.levels(), 2,
namespace.length()));
+ icebergParent =
+
IcebergIdentifierUtils.getIcebergNamespaceFromSchemaName(parentPath, separator);
+ }
+
List<org.apache.iceberg.catalog.Namespace> namespaces =
- icebergCatalogWrapper
- .listNamespace(IcebergCatalogWrapperHelper.getIcebergNamespace())
- .namespaces();
+ icebergCatalogWrapper.listNamespace(icebergParent).namespaces();
+ Namespace catalogNamespace = Namespace.of(namespace.level(0),
namespace.level(1));
return namespaces.stream()
- .map(icebergNamespace -> NameIdentifier.of(namespace,
icebergNamespace.toString()))
+ .map(
+ icebergNamespace -> {
+ // Convert the multi-level Iceberg namespace back to a logical
schema name using the
+ // configured separator so upper layers receive consistent
logical names.
+ String logicalName =
+ IcebergIdentifierUtils.icebergNamespaceToSchemaName(
+ icebergNamespace, separator);
+ return NameIdentifier.of(catalogNamespace, logicalName);
+ })
.toArray(NameIdentifier[]::new);
} catch (NoSuchNamespaceException e) {
throw new NoSuchSchemaException(
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogCapability.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogCapability.java
new file mode 100644
index 0000000000..0172f2fb2a
--- /dev/null
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogCapability.java
@@ -0,0 +1,106 @@
+/*
+ * 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.catalog.lakehouse.iceberg;
+
+import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.connector.capability.CapabilityResult;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergCatalogCapability {
+
+ @Test
+ public void testSchemaNameSupportsConfiguredSeparator() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.SCHEMA, "team:sales");
+ Assertions.assertTrue(result.supported());
+ }
+
+ @Test
+ public void testSchemaNameSupportsDeepNesting() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
+ capability.specificationOnName(Capability.Scope.SCHEMA,
"team:sales:reports");
+ Assertions.assertTrue(result.supported());
+ }
+
+ @Test
+ public void testSchemaNameRejectsEmptySegments() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
+ capability.specificationOnName(Capability.Scope.SCHEMA, "team::sales");
+ Assertions.assertFalse(result.supported());
+ }
+
+ @Test
+ public void testSchemaNameRejectsLeadingSeparator() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.SCHEMA, ":sales");
+ Assertions.assertFalse(result.supported());
+ }
+
+ @Test
+ public void testSchemaNameRejectsTrailingSeparator() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.SCHEMA, "sales:");
+ Assertions.assertFalse(result.supported());
+ }
+
+ @Test
+ public void testFlatSchemaNameAllowed() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.SCHEMA, "flat");
+ Assertions.assertTrue(result.supported());
+ }
+
+ @Test
+ public void testNonSchemaScopeStillUsesDefaultRules() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.TABLE, "table:name");
+ Assertions.assertFalse(result.supported());
+ }
+
+ @Test
+ public void testDefaultConstructorUsesColonSeparator() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability(":");
+
+ CapabilityResult result =
capability.specificationOnName(Capability.Scope.SCHEMA, "team:sales");
+ Assertions.assertTrue(result.supported());
+ }
+
+ @Test
+ public void testCustomSeparatorSlash() {
+ IcebergCatalogCapability capability = new IcebergCatalogCapability("/");
+
+ CapabilityResult validResult =
+ capability.specificationOnName(Capability.Scope.SCHEMA, "team/sales");
+ Assertions.assertTrue(validResult.supported());
+
+ CapabilityResult invalidResult =
+ capability.specificationOnName(Capability.Scope.SCHEMA, "team//sales");
+ Assertions.assertFalse(invalidResult.supported());
+ }
+}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
index 06e95e4bd8..dfbd129a23 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
@@ -18,14 +18,26 @@
*/
package org.apache.gravitino.catalog.lakehouse.iceberg;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import com.google.common.collect.ImmutableMap;
+import java.util.Arrays;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
+import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import org.apache.iceberg.rest.responses.ListNamespacesResponse;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class TestIcebergCatalogOperations {
+
+ private static final String METALAKE = "metalake";
+ private static final String CATALOG = "catalog";
+
@Test
public void testTestConnection() {
IcebergCatalogOperations catalogOperations = new
IcebergCatalogOperations();
@@ -34,7 +46,7 @@ public class TestIcebergCatalogOperations {
GravitinoRuntimeException.class,
() ->
catalogOperations.testConnection(
- NameIdentifier.of("metalake", "catalog"),
+ NameIdentifier.of(METALAKE, CATALOG),
Catalog.Type.RELATIONAL,
"iceberg",
"comment",
@@ -42,4 +54,45 @@ public class TestIcebergCatalogOperations {
Assertions.assertTrue(
exception.getMessage().contains("Failed to run listNamespace on
Iceberg catalog"));
}
+
+ @Test
+ public void testListSchemasConvertsMultiLevelNamespacesToLogicalNames() {
+ IcebergCatalogWrapper mockWrapper = mock(IcebergCatalogWrapper.class);
+ IcebergCatalogOperations catalogOperations = new
IcebergCatalogOperations();
+ catalogOperations.icebergCatalogWrapper = mockWrapper;
+
+ org.apache.iceberg.catalog.Namespace flatNs =
org.apache.iceberg.catalog.Namespace.of("mydb");
+ org.apache.iceberg.catalog.Namespace hierarchicalNs =
+ org.apache.iceberg.catalog.Namespace.of("A", "B", "C");
+ ListNamespacesResponse mockResponse =
+ ListNamespacesResponse.builder().addAll(Arrays.asList(flatNs,
hierarchicalNs)).build();
+ when(mockWrapper.listNamespace(any())).thenReturn(mockResponse);
+
+ NameIdentifier[] result =
catalogOperations.listSchemas(Namespace.of(METALAKE, CATALOG));
+
+ Assertions.assertEquals(2, result.length);
+ // Flat namespace stays as-is.
+ Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"mydb".equals(id.name())));
+ // Multi-level Iceberg namespace is joined with the configured separator.
+ Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"A:B:C".equals(id.name())));
+ }
+
+ @Test
+ public void testListSchemasFlatOnlyReturnsUnchangedNames() {
+ IcebergCatalogWrapper mockWrapper = mock(IcebergCatalogWrapper.class);
+ IcebergCatalogOperations catalogOperations = new
IcebergCatalogOperations();
+ catalogOperations.icebergCatalogWrapper = mockWrapper;
+
+ org.apache.iceberg.catalog.Namespace ns1 =
org.apache.iceberg.catalog.Namespace.of("db1");
+ org.apache.iceberg.catalog.Namespace ns2 =
org.apache.iceberg.catalog.Namespace.of("db2");
+ ListNamespacesResponse mockResponse =
+ ListNamespacesResponse.builder().addAll(Arrays.asList(ns1,
ns2)).build();
+ when(mockWrapper.listNamespace(any())).thenReturn(mockResponse);
+
+ NameIdentifier[] result =
catalogOperations.listSchemas(Namespace.of(METALAKE, CATALOG));
+
+ Assertions.assertEquals(2, result.length);
+ Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db1".equals(id.name())));
+ Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db2".equals(id.name())));
+ }
}
diff --git a/clients/client-java/build.gradle.kts
b/clients/client-java/build.gradle.kts
index 11929c8fec..b3850d71fe 100644
--- a/clients/client-java/build.gradle.kts
+++ b/clients/client-java/build.gradle.kts
@@ -59,6 +59,7 @@ dependencies {
testImplementation(libs.mockito.core)
testImplementation(libs.mockserver.netty)
testImplementation(libs.mockserver.client.java)
+ testImplementation(libs.h2db)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
@@ -85,6 +86,10 @@ tasks.test {
dependsOn(":catalogs:catalog-model:jar",
":catalogs:catalog-model:runtimeJars")
dependsOn(":catalogs:catalog-hive:jar",
":catalogs:catalog-hive:runtimeJars")
dependsOn(":catalogs:catalog-kafka:jar",
":catalogs:catalog-kafka:runtimeJars")
+ dependsOn(
+ ":catalogs:catalog-lakehouse-iceberg:jar",
+ ":catalogs:catalog-lakehouse-iceberg:runtimeJars"
+ )
}
}
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
index 45ae377ca2..0553abfc4b 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
@@ -18,11 +18,13 @@
*/
package org.apache.gravitino.client;
+import com.google.common.base.Preconditions;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.MetadataObjects;
@@ -143,16 +145,25 @@ abstract class BaseSchemaCatalog extends CatalogDTO
*/
@Override
public String[] listSchemas() throws NoSuchCatalogException {
+ return doListSchemas(Collections.emptyMap());
+ }
- EntityListResponse resp =
- restClient.get(
- formatSchemaRequestPath(schemaNamespace()),
- EntityListResponse.class,
- Collections.emptyMap(),
- ErrorHandlers.schemaErrorHandler());
- resp.validate();
-
- return
Arrays.stream(resp.identifiers()).map(NameIdentifier::name).toArray(String[]::new);
+ /**
+ * List the schemas directly under the given parent schema.
+ *
+ * @param parentSchema The parent (possibly hierarchical) schema name whose
direct children are
+ * listed, e.g. {@code "a"} or {@code "a:b"}. Must not be null or blank.
+ * @return A list of schema names directly under the given parent schema.
+ * @throws IllegalArgumentException if {@code parentSchema} is null or blank.
+ * @throws NoSuchCatalogException if the catalog with specified namespace
does not exist.
+ * @throws NoSuchSchemaException if the parent schema does not exist.
+ */
+ @Override
+ public String[] listSchemas(String parentSchema)
+ throws NoSuchCatalogException, NoSuchSchemaException {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(parentSchema), "parentSchema must not be null
or blank");
+ return doListSchemas(Collections.singletonMap("parentSchema",
parentSchema));
}
/**
@@ -371,4 +382,17 @@ abstract class BaseSchemaCatalog extends CatalogDTO
public boolean dropFunction(NameIdentifier ident) {
return functionOperations.dropFunction(ident);
}
+
+ private String[] doListSchemas(Map<String, String> queryParams) {
+ EntityListResponse resp =
+ restClient.get(
+ formatSchemaRequestPath(schemaNamespace()),
+ queryParams,
+ EntityListResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.schemaErrorHandler());
+ resp.validate();
+
+ return
Arrays.stream(resp.identifiers()).map(NameIdentifier::name).toArray(String[]::new);
+ }
}
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestRelationalCatalog.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestRelationalCatalog.java
index ff0a29c032..b10e9f981e 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestRelationalCatalog.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestRelationalCatalog.java
@@ -190,6 +190,43 @@ public class TestRelationalCatalog extends TestBase {
Assertions.assertTrue(ex2.getMessage().contains("unparsed error"));
}
+ @Test
+ public void testListSchemasUnderParent() throws JsonProcessingException {
+ Namespace schemaNs = Namespace.of(metalakeName, catalogName);
+ String schemaPath =
withSlash(RelationalCatalog.formatSchemaRequestPath(schemaNs));
+
+ // A non-blank parentSchema must be forwarded as the "parentSchema" query
parameter.
+ NameIdentifier child = NameIdentifier.of(schemaNs, "a:b:c");
+ EntityListResponse resp = new EntityListResponse(new NameIdentifier[]
{child});
+ buildMockResource(
+ Method.GET, schemaPath, ImmutableMap.of("parentSchema", "a:b"), null,
resp, SC_OK);
+ String[] schemas = catalog.asSchemas().listSchemas("a:b");
+ Assertions.assertEquals(1, schemas.length);
+ Assertions.assertEquals("a:b:c", schemas[0]);
+
+ // A null or blank parentSchema is rejected before any request is sent.
+ SupportsSchemas schemas2 = catalog.asSchemas();
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
schemas2.listSchemas(null));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
schemas2.listSchemas(""));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
schemas2.listSchemas(" "));
+
+ // Test throw NoSuchSchemaException when the parent schema does not exist.
+ ErrorResponse errorResp =
+ ErrorResponse.notFound(NoSuchSchemaException.class.getSimpleName(),
"schema not found");
+ buildMockResource(
+ Method.GET,
+ schemaPath,
+ ImmutableMap.of("parentSchema", "missing"),
+ null,
+ errorResp,
+ SC_NOT_FOUND);
+ SupportsSchemas supportsSchemas = catalog.asSchemas();
+ Throwable ex =
+ Assertions.assertThrows(
+ NoSuchSchemaException.class, () ->
supportsSchemas.listSchemas("missing"));
+ Assertions.assertTrue(ex.getMessage().contains("schema not found"));
+ }
+
@Test
public void testCreateSchema() throws JsonProcessingException {
String schemaName = "schema1";
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/HierarchicalSchemaAuthorizationIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/HierarchicalSchemaAuthorizationIT.java
new file mode 100644
index 0000000000..e4c1142a39
--- /dev/null
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/HierarchicalSchemaAuthorizationIT.java
@@ -0,0 +1,330 @@
+/*
+ * 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.client.integration.test.authorization;
+
+import static org.junit.Assert.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.Privilege;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObject;
+import org.apache.gravitino.authorization.SecurableObjects;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.exceptions.ForbiddenException;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+
+/**
+ * Integration tests for hierarchical schema authorization.
+ *
+ * <p>Tests cover:
+ *
+ * <ol>
+ * <li>Admin can create hierarchical schemas; auto-creates parent chain.
+ * <li>Normal user cannot create hierarchical schemas without grants.
+ * <li>Granting {@code create_schema} on a parent schema allows creating
direct children.
+ * <li>Granting {@code create_schema} on an ancestor schema inherits down to
all descendants.
+ * <li>List schemas returns only top-level schemas by default.
+ * <li>{@code use_schema} on a hierarchical schema allows loading and
listing it.
+ * <li>Drop hierarchical schema requires ownership or catalog ownership.
+ * </ol>
+ */
+@Tag("gravitino-docker-test")
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class HierarchicalSchemaAuthorizationIT extends
BaseRestApiAuthorizationIT {
+
+ private static final String CATALOG = "hierarchical_catalog";
+ private static final String ROLE = "hierarchical_role";
+
+ /** Top-level schemas created by admin at setup. */
+ private static final String ROOT_A = "A";
+
+ private static final String SCHEMA_AB = "A:B";
+ private static final String SCHEMA_ABC = "A:B:C";
+ private static final String SCHEMA_ABD = "A:B:D";
+
+ @BeforeAll
+ @Override
+ public void startIntegrationTest() throws Exception {
+ // Enable authorization and configure the hierarchical schema separator.
+ customConfigs.put(Configs.SCHEMA_SEPARATOR.getKey(), ":");
+ super.startIntegrationTest();
+
+ // Create an Iceberg catalog because ':' hierarchical schema names are
only supported there.
+ // The Iceberg JDBC catalog is backed by an in-memory H2 database purely
for test convenience:
+ // it is lightweight and needs no extra Docker container, so the IT can
exercise the
+ // hierarchical-namespace behavior without a heavyweight metastore. H2 is
NOT a supported
+ // production Iceberg backend.
+ Map<String, String> catalogProperties = new HashMap<>();
+ catalogProperties.put("catalog-backend", "jdbc");
+ catalogProperties.put("warehouse",
"/tmp/gravitino-it-hierarchical-schema");
+ catalogProperties.put(
+ "uri",
"jdbc:h2:mem:gravitino-it-hierarchical-schema;DB_CLOSE_DELAY=-1;MODE=MYSQL");
+ catalogProperties.put("jdbc-driver", "org.h2.Driver");
+ catalogProperties.put("jdbc-initialize", "true");
+ client
+ .loadMetalake(METALAKE)
+ .createCatalog(
+ CATALOG, Catalog.Type.RELATIONAL, "lakehouse-iceberg", "comment",
catalogProperties);
+
+ // Grant the normal user a role with USE_CATALOG so it can interact with
the catalog.
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+ List<SecurableObject> securableObjects = new ArrayList<>();
+ List<Privilege> privileges = new ArrayList<>();
+ privileges.add(Privileges.UseCatalog.allow());
+ securableObjects.add(SecurableObjects.ofCatalog(CATALOG, privileges));
+ metalake.createRole(ROLE, new HashMap<>(), securableObjects);
+ metalake.grantRolesToUser(ImmutableList.of(ROLE), NORMAL_USER);
+ }
+
+ /**
+ * Admin creates a hierarchical schema {@code A:B:C}. The server
automatically ensures the parent
+ * chain {@code A} and {@code A:B} exists. Verify all three schemas are
accessible by the admin.
+ */
+ @Test
+ @Order(1)
+ public void testAdminCreatesHierarchicalSchemaAutoCreatesParentChain() {
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ // Creating "A:B:C" should auto-create "A" and "A:B".
+ catalog.asSchemas().createSchema(SCHEMA_ABC, "hierarchical schema", new
HashMap<>());
+
+ // Default listing only returns top-level schemas.
+ String[] schemas = catalog.asSchemas().listSchemas();
+ List<String> schemaList = Arrays.asList(schemas);
+
+ assertTrue(schemaList.contains(ROOT_A), "Parent 'A' should be
auto-created");
+ assertFalse(
+ schemaList.contains(SCHEMA_AB),
+ "Default listSchemas() should not include hierarchical schema A:B");
+ assertFalse(
+ schemaList.contains(SCHEMA_ABC),
+ "Default listSchemas() should not include hierarchical schema A:B:C");
+
+ // Verify hierarchical schemas exist via direct load.
+ assertEquals(SCHEMA_AB, catalog.asSchemas().loadSchema(SCHEMA_AB).name());
+ assertEquals(SCHEMA_ABC,
catalog.asSchemas().loadSchema(SCHEMA_ABC).name());
+ }
+
+ /**
+ * Normal user without {@code create_schema} on the parent schema cannot
create a hierarchical
+ * schema.
+ */
+ @Test
+ @Order(2)
+ public void testNormalUserCannotCreateHierarchicalSchemaWithoutGrant() {
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ // Creating "A:B:D" requires create_schema on parent "A:B" (or any
ancestor / catalog).
+ assertThrows(
+ ForbiddenException.class,
+ () -> catalogByNormalUser.asSchemas().createSchema(SCHEMA_ABD, "test",
new HashMap<>()));
+ }
+
+ /**
+ * Granting {@code create_schema} on the parent schema {@code A:B} allows
the normal user to
+ * create a direct child {@code A:B:D}.
+ */
+ @Test
+ @Order(3)
+ public void testCreateSchemaWithGrantOnParentSchema() {
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+
+ // Grant create_schema on parent "A:B".
+ metalake.grantPrivilegesToRole(
+ ROLE,
+ MetadataObjects.of(CATALOG, SCHEMA_AB, MetadataObject.Type.SCHEMA),
+ ImmutableList.of(Privileges.UseSchema.allow(),
Privileges.CreateSchema.allow()));
+
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ // Normal user can now create a child of "A:B".
+ catalogByNormalUser.asSchemas().createSchema(SCHEMA_ABD, "child of A:B",
new HashMap<>());
+
+ // Verify "A:B:D" was created.
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ assertEquals(SCHEMA_ABD,
catalog.asSchemas().loadSchema(SCHEMA_ABD).name());
+ }
+
+ /**
+ * Privilege inheritance: granting {@code create_schema} on ancestor {@code
A} allows creating
+ * schemas anywhere in the {@code A:*} subtree via the inheritance chain.
+ */
+ @Test
+ @Order(4)
+ public void testCreateSchemaInheritedFromAncestorGrant() {
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+
+ // Grant create_schema on ancestor "A". Due to inheritance, this covers
"A:B:*" as well.
+ metalake.grantPrivilegesToRole(
+ ROLE,
+ MetadataObjects.of(CATALOG, ROOT_A, MetadataObject.Type.SCHEMA),
+ ImmutableList.of(Privileges.UseSchema.allow(),
Privileges.CreateSchema.allow()));
+
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ // Normal user can now create "A:B:E" via inheritance from "A".
+ catalogByNormalUser.asSchemas().createSchema("A:B:E", "inherited from A",
new HashMap<>());
+
+ // Verify "A:B:E" was created.
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ assertEquals("A:B:E", catalog.asSchemas().loadSchema("A:B:E").name());
+ }
+
+ /**
+ * By default {@code listSchemas()} returns only top-level schemas (those
without the separator).
+ * Hierarchical schemas are accessible through the parent-aware filter.
+ */
+ @Test
+ @Order(5)
+ public void testListSchemasReturnsTopLevelByDefault() {
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ String[] schemas = catalog.asSchemas().listSchemas();
+
+ // Top-level schemas should include "A" (and any others created by prior
tests).
+ List<String> schemaList = Arrays.asList(schemas);
+ assertTrue(schemaList.contains(ROOT_A));
+
+ // Hierarchical schemas should NOT appear in the default listing.
+ schemaList.forEach(
+ name ->
+ assertFalse(
+ name.contains(":"),
+ "Default listSchemas() should return only top-level schemas,
got: " + name));
+ }
+
+ /**
+ * Granting {@code use_schema} on a hierarchical schema allows the normal
user to load it, even
+ * without grants on intermediate parent schemas.
+ */
+ @Test
+ @Order(6)
+ public void testUseSchemaPrivilegeOnHierarchicalSchema() {
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+
+ // Grant use_schema specifically on "A:B:C".
+ metalake.grantPrivilegesToRole(
+ ROLE,
+ MetadataObjects.of(CATALOG, SCHEMA_ABC, MetadataObject.Type.SCHEMA),
+ ImmutableList.of(Privileges.UseSchema.allow()));
+
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ // Normal user can now load "A:B:C".
+ Schema loaded = catalogByNormalUser.asSchemas().loadSchema(SCHEMA_ABC);
+ assertEquals(SCHEMA_ABC, loaded.name());
+ }
+
+ /**
+ * Dropping a hierarchical schema requires ownership or catalog ownership.
Normal user without
+ * ownership cannot drop; after becoming owner, they can.
+ */
+ @Test
+ @Order(7)
+ public void testDropHierarchicalSchemaRequiresOwnership() {
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ // Normal user cannot drop "A:B:C" without ownership.
+ assertThrows(
+ ForbiddenException.class,
+ () -> catalogByNormalUser.asSchemas().dropSchema(SCHEMA_ABC, false));
+
+ // Set normal user as owner of "A:B:C".
+ metalake.setOwner(
+ MetadataObjects.of(CATALOG, SCHEMA_ABC, MetadataObject.Type.SCHEMA),
+ NORMAL_USER,
+ Owner.Type.USER);
+
+ // Normal user can now drop it.
+ catalogByNormalUser.asSchemas().dropSchema(SCHEMA_ABC, false);
+
+ // Verify "A:B:C" is gone; parent "A:B" should still exist.
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ assertThrows(NoSuchSchemaException.class, () ->
catalog.asSchemas().loadSchema(SCHEMA_ABC));
+ assertEquals(ROOT_A, catalog.asSchemas().loadSchema(ROOT_A).name());
+ }
+
+ /**
+ * Tests that {@code create_schema} cannot be bound to a SCHEMA-level
securable object before the
+ * canBindTo change (this is now allowed). Grants {@code
CreateSchema.allow()} on a schema object
+ * and verifies no exception is thrown.
+ */
+ @Test
+ @Order(8)
+ public void testCreateSchemaPrivilegeCanBindToSchema() {
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+
+ // This should NOT throw — CreateSchema.canBindTo(SCHEMA) is now true.
+ metalake.grantPrivilegesToRole(
+ ROLE,
+ MetadataObjects.of(CATALOG, ROOT_A, MetadataObject.Type.SCHEMA),
+ ImmutableList.of(Privileges.CreateSchema.allow()));
+
+ // Verify the grant was applied by checking normal user can create under
"A".
+ Catalog catalogByNormalUser =
normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ catalogByNormalUser.asSchemas().createSchema("A:F", "via schema-level
grant", new HashMap<>());
+
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+ assertEquals("A:F", catalog.asSchemas().loadSchema("A:F").name());
+ }
+
+ /**
+ * Listing with an explicit {@code parentSchema} returns the direct children
of that parent,
+ * exposing the hierarchical schemas that the default top-level listing
hides. Uses a
+ * self-contained {@code P:Q} subtree so the assertions do not depend on
schemas created or
+ * dropped by earlier ordered tests.
+ */
+ @Test
+ @Order(9)
+ public void testListSchemasUnderParentReturnsChildren() {
+ Catalog catalog = client.loadMetalake(METALAKE).loadCatalog(CATALOG);
+
+ catalog.asSchemas().createSchema("P:Q:R", "child R", new HashMap<>());
+ catalog.asSchemas().createSchema("P:Q:S", "child S", new HashMap<>());
+
+ // Children are not visible in the default top-level listing.
+ List<String> topLevel = Arrays.asList(catalog.asSchemas().listSchemas());
+ assertTrue(topLevel.contains("P"), "Auto-created parent 'P' should be a
top-level schema");
+ assertFalse(topLevel.contains("P:Q"), "Default listSchemas() should not
include 'P:Q'");
+ assertFalse(topLevel.contains("P:Q:R"), "Default listSchemas() should not
include 'P:Q:R'");
+
+ // Listing under "P:Q" returns its direct children as logical hierarchical
names.
+ List<String> children =
Arrays.asList(catalog.asSchemas().listSchemas("P:Q"));
+ assertTrue(children.contains("P:Q:R"), "listSchemas(\"P:Q\") should
include P:Q:R");
+ assertTrue(children.contains("P:Q:S"), "listSchemas(\"P:Q\") should
include P:Q:S");
+ }
+}
diff --git a/clients/client-python/gravitino/api/supports_schemas.py
b/clients/client-python/gravitino/api/supports_schemas.py
index 36f0ac6a98..7a61400b6f 100644
--- a/clients/client-python/gravitino/api/supports_schemas.py
+++ b/clients/client-python/gravitino/api/supports_schemas.py
@@ -16,7 +16,7 @@
# under the License.
from abc import ABC, abstractmethod
-from typing import Dict, List
+from typing import Dict, List, Optional
from gravitino.api.schema import Schema
from gravitino.api.schema_change import SchemaChange
@@ -30,15 +30,29 @@ class SupportsSchemas(ABC):
"""
@abstractmethod
- def list_schemas(self) -> List[str]:
+ def list_schemas(self, parent_schema: Optional[str] = None) -> List[str]:
"""List schemas under the entity.
If an entity such as a table, view exists, its parent schemas must
also exist and must be
returned by this discovery method. For example, if table a.b.t exists,
this method invoked as
listSchemas(a) must return [b] in the result array
+ If ``parent_schema`` is provided, list the schemas directly under the
given parent schema.
+ This is only meaningful for catalogs that support hierarchical
(multi-level) schemas, such
+ as an Iceberg catalog accessed through the Gravitino REST server with
a configured schema
+ separator. For example, when the schemas ``a``, ``a:b`` and ``a:b:c``
exist, this method
+ invoked with parent ``a:b`` returns ``[a:b:c]``. For a flat catalog,
or a parent schema that
+ has no children, an empty list is returned.
+
+ Args:
+ parent_schema: The parent (possibly hierarchical) schema name
whose direct children are
+ listed, e.g. ``"a"`` or ``"a:b"``. When ``None``, all schemas
under the catalog are
+ listed. Must not be blank when provided.
+
Raises:
+ IllegalArgumentException: If ``parent_schema`` is provided but
blank.
NoSuchCatalogException: If the catalog does not exist.
+ NoSuchSchemaException: If ``parent_schema`` is provided but does
not exist.
Returns:
A list of schema names under the namespace.
diff --git a/clients/client-python/gravitino/client/base_schema_catalog.py
b/clients/client-python/gravitino/client/base_schema_catalog.py
index 89872f893a..a2cbae094b 100644
--- a/clients/client-python/gravitino/client/base_schema_catalog.py
+++ b/clients/client-python/gravitino/client/base_schema_catalog.py
@@ -119,17 +119,32 @@ class BaseSchemaCatalog(
def as_function_catalog(self):
return self
- def list_schemas(self) -> List[str]:
- """List all the schemas under the given catalog namespace.
+ def list_schemas(self, parent_schema: Optional[str] = None) -> List[str]:
+ """List the schemas under the given catalog namespace, or directly
under the given parent
+ schema when ``parent_schema`` is provided.
+
+ Args:
+ parent_schema: The parent (possibly hierarchical) schema name
whose direct children are
+ listed, e.g. ``"a"`` or ``"a:b"``. When ``None``, all schemas
under the catalog are
+ listed. Must not be blank when provided.
Raises:
+ IllegalArgumentException if ``parent_schema`` is provided but
blank.
NoSuchCatalogException if the catalog with specified namespace
does not exist.
+ NoSuchSchemaException if ``parent_schema`` is provided but does
not exist.
Returns:
- A list of schema names under the given catalog namespace.
+ A list of schema names under the given catalog namespace or
parent schema.
"""
+ params = None
+ if parent_schema is not None:
+ if not parent_schema.strip():
+ raise IllegalArgumentException("parentSchema must not be null
or blank")
+ params = {"parentSchema": parent_schema}
+
resp = self.rest_client.get(
BaseSchemaCatalog.format_schema_request_path(self._schema_namespace()),
+ params=params,
error_handler=SCHEMA_ERROR_HANDLER,
)
entity_list_response = EntityListResponse.from_json(
diff --git a/clients/client-python/tests/unittests/test_base_schema_catalog.py
b/clients/client-python/tests/unittests/test_base_schema_catalog.py
new file mode 100644
index 0000000000..907bb4ab57
--- /dev/null
+++ b/clients/client-python/tests/unittests/test_base_schema_catalog.py
@@ -0,0 +1,100 @@
+# 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.
+
+import unittest
+from unittest.mock import patch, Mock
+
+from gravitino.client.relational_catalog import RelationalCatalog
+from gravitino.dto.audit_dto import AuditDTO
+from gravitino.dto.responses.entity_list_response import EntityListResponse
+from gravitino.exceptions.base import (
+ IllegalArgumentException,
+ NoSuchSchemaException,
+)
+from gravitino.name_identifier import NameIdentifier
+from gravitino.namespace import Namespace
+from gravitino.utils import HTTPClient, Response
+
+
+class TestBaseSchemaCatalog(unittest.TestCase):
+ metalake_name = "test_metalake"
+ catalog_name = "test_catalog"
+ catalog_namespace = Namespace.of(metalake_name)
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.rest_client = HTTPClient("http://localhost:8090";)
+ cls.catalog = RelationalCatalog(
+ catalog_namespace=cls.catalog_namespace,
+ name=cls.catalog_name,
+ catalog_type=RelationalCatalog.Type.RELATIONAL,
+ provider="test_provider",
+ audit=AuditDTO("anonymous"),
+ rest_client=cls.rest_client,
+ )
+
+ def _get_mock_http_resp(self, json_str: str, return_code: int = 200):
+ mock_http_resp = Mock()
+ mock_http_resp.getcode.return_value = return_code
+ mock_http_resp.read.return_value = json_str.encode("utf-8")
+ mock_http_resp.info.return_value = None
+ mock_http_resp.url = None
+ return Response(mock_http_resp)
+
+ def test_list_schemas(self):
+ schema_a = NameIdentifier.of(self.metalake_name, self.catalog_name,
"a")
+ schema_b = NameIdentifier.of(self.metalake_name, self.catalog_name,
"b")
+
+ resp_body = EntityListResponse(_code=0, _idents=[schema_a, schema_b])
+ mock_resp = self._get_mock_http_resp(resp_body.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.get",
return_value=mock_resp
+ ) as mock_get:
+ schemas = self.catalog.as_schemas().list_schemas()
+ self.assertEqual(["a", "b"], schemas)
+ # No parentSchema query param when listing all schemas.
+ self.assertIsNone(mock_get.call_args.kwargs["params"])
+
+ def test_list_schemas_under_parent(self):
+ schema_abc = NameIdentifier.of(self.metalake_name, self.catalog_name,
"a:b:c")
+
+ resp_body = EntityListResponse(_code=0, _idents=[schema_abc])
+ mock_resp = self._get_mock_http_resp(resp_body.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.get",
return_value=mock_resp
+ ) as mock_get:
+ schemas = self.catalog.as_schemas().list_schemas("a:b")
+ self.assertEqual(["a:b:c"], schemas)
+ # The parent schema is passed through as a query param.
+ self.assertEqual(
+ {"parentSchema": "a:b"}, mock_get.call_args.kwargs["params"]
+ )
+
+ def test_list_schemas_with_blank_parent(self):
+ for blank in ["", " "]:
+ with self.assertRaises(IllegalArgumentException):
+ self.catalog.as_schemas().list_schemas(blank)
+
+ def test_list_schemas_under_missing_parent(self):
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.get",
+ side_effect=NoSuchSchemaException("schema not found"),
+ ):
+ with self.assertRaises(NoSuchSchemaException):
+ self.catalog.as_schemas().list_schemas("a:b")
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
index b61c99e0d7..3d282683d5 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
@@ -23,6 +23,7 @@ import static
org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePro
import static
org.apache.gravitino.utils.NameIdentifierUtil.getCatalogIdentifier;
import java.time.Instant;
+import java.util.List;
import java.util.Map;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.EntityStore;
@@ -43,6 +44,8 @@ import org.apache.gravitino.lock.TreeLockUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.PrincipalUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -339,10 +342,48 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
} catch (Exception e) {
throw new RuntimeException(e);
}
+
+ cleanupOrphanedAncestors(catalogIdent, ident);
return droppedFromCatalog;
});
}
+ /**
+ * Reconciles auto-created ancestor entities after a hierarchical schema
leaf is dropped. This
+ * mirrors {@code IcebergNamespaceHookDispatcher.dropNamespace}: walk the
ancestors
+ * innermost-to-outermost and delete the orphaned Gravitino entity for each
ancestor that no
+ * longer exists in the catalog, stopping at the first ancestor that still
exists (its existence
+ * implies all of its outer ancestors exist). For a flat schema name this is
a no-op.
+ *
+ * @param catalogIdent the identifier of the catalog the schema belongs to
+ * @param ident the identifier of the dropped (leaf) schema
+ */
+ private void cleanupOrphanedAncestors(NameIdentifier catalogIdent,
NameIdentifier ident) {
+ String separator = HierarchicalSchemaUtil.schemaSeparator();
+ List<String> ancestorNames =
HierarchicalSchemaUtil.getAncestorNames(ident.name(), separator);
+ String metalake = ident.namespace().level(0);
+ String catalog = ident.namespace().level(1);
+ for (int i = ancestorNames.size() - 1; i >= 0; i--) {
+ NameIdentifier ancestorIdent =
+ NameIdentifierUtil.ofSchema(metalake, catalog, ancestorNames.get(i));
+ boolean ancestorExistsInCatalog =
+ doWithCatalog(
+ catalogIdent,
+ c -> c.doWithSchemaOps(s -> s.schemaExists(ancestorIdent)),
+ RuntimeException.class);
+ if (ancestorExistsInCatalog) {
+ break;
+ }
+ try {
+ store.delete(ancestorIdent, SCHEMA, true);
+ } catch (NoSuchEntityException e) {
+ LOG.warn("The orphaned ancestor schema does not exist in the store:
{}", ancestorIdent, e);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
private void importSchema(NameIdentifier identifier) {
EntityCombinedSchema schema = internalLoadSchema(identifier);
if (schema.imported()) {
diff --git
a/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java
b/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java
index 6114f7d2fb..dae07ffbc6 100644
--- a/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java
+++ b/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java
@@ -18,10 +18,13 @@
*/
package org.apache.gravitino.hook;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.gravitino.Entity;
import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
@@ -36,6 +39,9 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
+import org.apache.gravitino.lock.LockType;
+import org.apache.gravitino.lock.TreeLockUtils;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.PrincipalUtils;
@@ -59,24 +65,94 @@ public class SchemaHookDispatcher implements
SchemaDispatcher {
@Override
public Schema createSchema(NameIdentifier ident, String comment, Map<String,
String> properties)
throws NoSuchCatalogException, SchemaAlreadyExistsException {
- Schema schema = dispatcher.createSchema(ident, comment, properties);
+ // The inner NormalizeDispatcher case-folds the schema name based on
catalog capabilities, so
+ // the entity is stored under the normalized identifier. Normalize here
too so ownership is
+ // attached to the identifiers the manager sees and ancestor probing
matches stored names.
+ NameIdentifier normalizedIdent =
+ CapabilityHelpers.applyCapabilities(
+ ident, Capability.Scope.SCHEMA,
GravitinoEnv.getInstance().catalogManager());
- // Set the creator as the owner of the schema.
- OwnerDispatcher ownerManager =
GravitinoEnv.getInstance().ownerDispatcher();
- if (ownerManager != null) {
- // The inner NormalizeDispatcher case-folds the schema name based on
catalog capabilities,
- // so the entity is stored under the normalized identifier. Apply the
same normalization
- // here so the owner is attached to the same identifier the manager sees.
- NameIdentifier normalizedIdent =
- CapabilityHelpers.applyCapabilities(
- ident, Capability.Scope.SCHEMA,
GravitinoEnv.getInstance().catalogManager());
- ownerManager.setOwner(
- normalizedIdent.namespace().level(0),
- NameIdentifierUtil.toMetadataObject(normalizedIdent,
Entity.EntityType.SCHEMA),
- PrincipalUtils.getCurrentUserName(),
- Owner.Type.USER);
+ // Serialize probe -> create -> owner-assignment on the catalog so
concurrent hierarchical
+ // creates cannot both claim a shared, newly-created ancestor (which would
let the later create
+ // overwrite the first creator's ownership). We lock the catalog node --
the same node the inner
+ // SchemaOperationDispatcher.createSchema write-locks -- so the
acquisition is reentrant; a
+ // deeper (branch-scoped) lock would hold the catalog node in READ mode
and deadlock against
+ // that inner WRITE acquisition.
+ NameIdentifier catalogIdent =
+ NameIdentifierUtil.ofCatalog(
+ normalizedIdent.namespace().level(0),
normalizedIdent.namespace().level(1));
+ return TreeLockUtils.doWithTreeLock(
+ catalogIdent,
+ LockType.WRITE,
+ () -> {
+ // For a hierarchical schema name (e.g. "A:B:C") the store
auto-creates a row for each
+ // missing ancestor ("A", "A:B"). Probe BEFORE the create which
ancestors are new, so
+ // ownership is assigned only to schemas this request actually
creates and a pre-existing
+ // ancestor's owner is never overwritten.
+ List<NameIdentifier> newAncestors =
findMissingAncestors(normalizedIdent);
+
+ Schema schema = dispatcher.createSchema(ident, comment, properties);
+
+ // Set the creator as the owner of the new schema and of any
ancestors it created. This
+ // mirrors IcebergNamespaceHookDispatcher.createNamespace so
ownership-based
+ // authorization -- which treats ownership of an ancestor schema as
ownership of the
+ // whole subtree -- behaves the same on the Gravitino and Iceberg
REST surfaces.
+ OwnerDispatcher ownerManager =
GravitinoEnv.getInstance().ownerDispatcher();
+ if (ownerManager != null) {
+ List<MetadataObject> ownedObjects = new
ArrayList<>(newAncestors.size() + 1);
+ for (NameIdentifier ancestor : newAncestors) {
+ ownedObjects.add(
+ NameIdentifierUtil.toMetadataObject(ancestor,
Entity.EntityType.SCHEMA));
+ }
+ ownedObjects.add(
+ NameIdentifierUtil.toMetadataObject(normalizedIdent,
Entity.EntityType.SCHEMA));
+ // All objects are SCHEMA-typed, so the batch path (single object
type) is valid.
+ ownerManager.setOwners(
+ normalizedIdent.namespace().level(0),
+ ownedObjects,
+ PrincipalUtils.getCurrentUserName(),
+ Owner.Type.USER);
+ }
+ return schema;
+ });
+ }
+
+ /**
+ * Returns the identifiers of the (already-normalized) ancestor schemas of
{@code normalizedIdent}
+ * that do not yet exist, ordered outermost-to-innermost. Returns an empty
list for a flat (non
+ * hierarchical) schema name.
+ *
+ * <p>This issues one {@code schemaExists} probe per missing ancestor while
the caller holds the
+ * catalog WRITE lock, so it blocks other schema operations on the catalog
for the duration. The
+ * cost is bounded by the schema nesting depth, which is expected to be
small (typically 2-3
+ * levels); the innermost-to-outermost short-circuit below means a request
that nests under an
+ * existing parent issues only a single probe. If much deeper nesting
becomes common this should
+ * be revisited (e.g. a single prefix query instead of per-level probes).
+ */
+ private List<NameIdentifier> findMissingAncestors(NameIdentifier
normalizedIdent) {
+ String separator = HierarchicalSchemaUtil.schemaSeparator();
+ String schemaName = normalizedIdent.name();
+ List<NameIdentifier> missing = new ArrayList<>();
+ if (!schemaName.contains(separator)) {
+ return missing;
+ }
+ String metalake = normalizedIdent.namespace().level(0);
+ String catalog = normalizedIdent.namespace().level(1);
+ List<String> ancestorNames =
HierarchicalSchemaUtil.getAncestorNames(schemaName, separator);
+ // Walk innermost-to-outermost: in the hierarchical schema model the
existence of an inner
+ // ancestor implies all of its outer ancestors exist, so we can stop
probing at the first
+ // ancestor that already exists.
+ for (int i = ancestorNames.size() - 1; i >= 0; i--) {
+ NameIdentifier ancestorIdent =
+ NameIdentifierUtil.ofSchema(metalake, catalog, ancestorNames.get(i));
+ if (dispatcher.schemaExists(ancestorIdent)) {
+ break;
+ }
+ missing.add(ancestorIdent);
}
- return schema;
+ // Reverse to outermost-to-innermost, the order ownership assignment
consumes.
+ Collections.reverse(missing);
+ return missing;
}
@Override
diff --git
a/core/src/main/java/org/apache/gravitino/utils/HierarchicalSchemaUtil.java
b/core/src/main/java/org/apache/gravitino/utils/HierarchicalSchemaUtil.java
index 4d6a3fae91..85bc82462d 100644
--- a/core/src/main/java/org/apache/gravitino/utils/HierarchicalSchemaUtil.java
+++ b/core/src/main/java/org/apache/gravitino/utils/HierarchicalSchemaUtil.java
@@ -111,6 +111,21 @@ public final class HierarchicalSchemaUtil {
return StringUtils.isNotBlank(name) && name.contains(separator);
}
+ /**
+ * Splits a (possibly hierarchical) schema name into its segments using the
external separator.
+ * Trailing empty segments are preserved (split limit {@code -1}) so callers
can detect malformed
+ * names that contain empty segments (e.g. {@code "A::B"} or {@code "A:"}).
+ *
+ * <p>Example: {@code "A:B:C"} with separator {@code ":"} → {@code ["A",
"B", "C"]}
+ *
+ * @param name the schema name to split
+ * @param separator the external separator
+ * @return the segments of the schema name
+ */
+ public static String[] splitSchemaName(String name, String separator) {
+ return name.split(Pattern.quote(separator), -1);
+ }
+
/**
* Returns all ancestor schema names of the given schema name, ordered from
outermost to innermost
* (but excluding the name itself). Returns an empty list for top-level
(non-HierarchicalSchema)
@@ -125,7 +140,7 @@ public final class HierarchicalSchemaUtil {
public static List<String> getAncestorNames(String schemaName, String
separator) {
Preconditions.checkArgument(StringUtils.isNotBlank(schemaName),
"schemaName must not be blank");
Preconditions.checkArgument(StringUtils.isNotBlank(separator), "separator
must not be blank");
- String[] parts = schemaName.split(Pattern.quote(separator), -1);
+ String[] parts = splitSchemaName(schemaName, separator);
List<String> ancestors = new ArrayList<>();
for (int i = 1; i < parts.length; i++) {
ancestors.add(String.join(separator, Arrays.copyOf(parts, i)));
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
index f57a8c4038..084eb6c105 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
@@ -295,4 +295,59 @@ public class TestSchemaOperationDispatcher extends
TestOperationDispatcher {
Assertions.assertThrows(
RuntimeException.class, () -> dispatcher.dropSchema(schemaIdent,
false));
}
+
+ @Test
+ public void testDropHierarchicalSchemaCleansUpOrphanedAncestors() throws
IOException {
+ // Clear any spy stubs leaked from other tests sharing the static
entityStore.
+ reset(entityStore);
+ // Only the leaf "orphanA:orphanB:orphanC" is created in the catalog and
the store. Names are
+ // unique to this test because the catalog connector keeps its schemas in
shared static state.
+ NameIdentifier leaf = NameIdentifier.of(metalake, catalog,
"orphanA:orphanB:orphanC");
+ dispatcher.createSchema(leaf, "comment", ImmutableMap.of("k1", "v1", "k2",
"v2"));
+
+ // Simulate the ancestor entities the relational store auto-creates for a
hierarchical name.
+ NameIdentifier ancestorAb = NameIdentifier.of(metalake, catalog,
"orphanA:orphanB");
+ NameIdentifier ancestorA = NameIdentifier.of(metalake, catalog, "orphanA");
+ putSchemaEntity(ancestorAb);
+ putSchemaEntity(ancestorA);
+
+ // The catalog only knows the leaf, so dropping it leaves the ancestors
orphaned in the store;
+ // since neither ancestor exists in the catalog, both entities must be
cleaned up.
+ Assertions.assertTrue(dispatcher.dropSchema(leaf, false));
+ Assertions.assertFalse(entityStore.exists(ancestorAb, SCHEMA));
+ Assertions.assertFalse(entityStore.exists(ancestorA, SCHEMA));
+ }
+
+ @Test
+ public void testDropHierarchicalSchemaKeepsAncestorsThatStillExist() throws
IOException {
+ // Clear any spy stubs leaked from other tests sharing the static
entityStore.
+ reset(entityStore);
+ // Both the parent "keepA:keepB" and the leaf "keepA:keepB:keepC" exist in
the catalog and the
+ // store. Names are unique to this test to avoid the connector's shared
static schema state.
+ NameIdentifier parentAb = NameIdentifier.of(metalake, catalog,
"keepA:keepB");
+ NameIdentifier leaf = NameIdentifier.of(metalake, catalog,
"keepA:keepB:keepC");
+ dispatcher.createSchema(parentAb, "comment", ImmutableMap.of("k1", "v1",
"k2", "v2"));
+ dispatcher.createSchema(leaf, "comment", ImmutableMap.of("k1", "v1", "k2",
"v2"));
+
+ // Simulate the top-level ancestor entity.
+ NameIdentifier ancestorA = NameIdentifier.of(metalake, catalog, "keepA");
+ putSchemaEntity(ancestorA);
+
+ // "keepA:keepB" still exists in the catalog, so cleanup stops there and
keeps it and "keepA".
+ Assertions.assertTrue(dispatcher.dropSchema(leaf, false));
+ Assertions.assertTrue(entityStore.exists(parentAb, SCHEMA));
+ Assertions.assertTrue(entityStore.exists(ancestorA, SCHEMA));
+ }
+
+ private void putSchemaEntity(NameIdentifier ident) throws IOException {
+ SchemaEntity entity =
+ SchemaEntity.builder()
+ .withId(idGenerator.nextId())
+ .withName(ident.name())
+ .withNamespace(ident.namespace())
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("tester").withCreateTime(Instant.now()).build())
+ .build();
+ entityStore.put(entity, true);
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java
b/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java
index ce14f9f4d0..3e05c3c83c 100644
--- a/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java
+++ b/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java
@@ -18,30 +18,45 @@
*/
package org.apache.gravitino.hook;
+import static org.apache.gravitino.Configs.TREE_LOCK_CLEAN_INTERVAL;
+import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY;
+import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import com.google.common.collect.ImmutableList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Schema;
+import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.Owner;
import org.apache.gravitino.authorization.OwnerDispatcher;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.catalog.SchemaDispatcher;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
+import org.apache.gravitino.lock.LockManager;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
public class TestSchemaHookDispatcher {
@@ -54,6 +69,7 @@ public class TestSchemaHookDispatcher {
// state into the GravitinoEnv singleton across tests.
private OwnerDispatcher savedOwnerDispatcher;
private CatalogManager savedCatalogManager;
+ private LockManager savedLockManager;
@BeforeEach
public void setUp() throws Exception {
@@ -70,8 +86,13 @@ public class TestSchemaHookDispatcher {
// initialized. Read the field directly via reflection to capture the
current value safely.
savedCatalogManager =
(CatalogManager) FieldUtils.readField(GravitinoEnv.getInstance(),
"catalogManager", true);
+ savedLockManager =
+ (LockManager) FieldUtils.readField(GravitinoEnv.getInstance(),
"lockManager", true);
FieldUtils.writeField(GravitinoEnv.getInstance(), "ownerDispatcher",
mockOwnerDispatcher, true);
FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
mockCatalogManager, true);
+ // createSchema now acquires a catalog-level tree lock, so wire up a real
LockManager.
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "lockManager", new
LockManager(newLockConfig()), true);
hookDispatcher = new SchemaHookDispatcher(mockDispatcher);
}
@@ -80,6 +101,15 @@ public class TestSchemaHookDispatcher {
FieldUtils.writeField(
GravitinoEnv.getInstance(), "ownerDispatcher", savedOwnerDispatcher,
true);
FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
savedCatalogManager, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
savedLockManager, true);
+ }
+
+ private static Config newLockConfig() {
+ Config config = mock(Config.class);
+ doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
+ doReturn(1000L).when(config).get(TREE_LOCK_MIN_NODE_IN_MEMORY);
+ doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
+ return config;
}
@Test
@@ -90,7 +120,7 @@ public class TestSchemaHookDispatcher {
doThrow(new RuntimeException("Set owner failed"))
.when(mockOwnerDispatcher)
- .setOwner(any(), any(), any(), any());
+ .setOwners(any(), anyList(), any(), any());
RuntimeException thrown =
Assertions.assertThrows(
@@ -103,7 +133,7 @@ public class TestSchemaHookDispatcher {
@Test
public void testCreateSchemaSetsOwnerWithNormalizedIdentifier() throws
Exception {
// Use a case-insensitive capability so the schema name is normalized to
lower case before
- // setOwner is called, mirroring what NormalizeDispatcher would do for the
manager.
+ // setOwners is called, mirroring what NormalizeDispatcher would do for
the manager.
when(mockCatalogWrapper.capabilities()).thenReturn(new
CaseInsensitiveCapability());
NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog",
"MY_SCHEMA");
@@ -112,28 +142,115 @@ public class TestSchemaHookDispatcher {
hookDispatcher.createSchema(ident, "comment", Collections.emptyMap());
- ArgumentCaptor<MetadataObject> captor =
ArgumentCaptor.forClass(MetadataObject.class);
- verify(mockOwnerDispatcher)
- .setOwner(eq("test_metalake"), captor.capture(), any(),
eq(Owner.Type.USER));
+ List<MetadataObject> owned = captureOwnedObjects();
+ Assertions.assertEquals(1, owned.size(), "A flat schema only assigns
ownership to the leaf");
Assertions.assertEquals(
"my_schema",
- captor.getValue().name(),
- "Schema name passed to setOwner must be lowercased by
Capability.Scope.SCHEMA"
+ owned.get(0).name(),
+ "Schema name passed to setOwners must be lowercased by
Capability.Scope.SCHEMA"
+ " normalization");
// Schema's namespace is [metalake, catalog];
NameIdentifierUtil.toMetadataObject uses
// level(1) as parent. Catalog is not subject to per-scope name
normalization here, so
// parent is just the catalog name -- there is no schema component to
normalize.
Assertions.assertEquals(
"test_catalog",
- captor.getValue().parent(),
+ owned.get(0).parent(),
"Schema parent must be the catalog name (level(1) of the namespace);
SCHEMA's namespace"
+ " has no schema component to normalize");
}
+ @Test
+ public void testCreateHierarchicalSchemaOwnsNewAncestors() throws Exception {
+ // A capability that permits hierarchical (":"-separated) schema names so
the hierarchical name
+ // is not rejected during normalization.
+ when(mockCatalogWrapper.capabilities()).thenReturn(new
HierarchicalCapability());
+
+ NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog",
"A:B:C");
+ Schema mockSchema = mock(Schema.class);
+ when(mockDispatcher.createSchema(any(), any(),
any())).thenReturn(mockSchema);
+ // No ancestor exists yet, so creating "A:B:C" auto-creates "A" and "A:B".
+ when(mockDispatcher.schemaExists(any())).thenReturn(false);
+
+ hookDispatcher.createSchema(ident, "comment", Collections.emptyMap());
+
+ List<String> ownedNames =
+
captureOwnedObjects().stream().map(MetadataObject::name).collect(Collectors.toList());
+ Assertions.assertEquals(
+ Arrays.asList("A", "A:B", "A:B:C"),
+ ownedNames,
+ "Creator must own every newly-created ancestor plus the leaf,
outermost-to-innermost");
+ }
+
+ @Test
+ public void testCreateHierarchicalSchemaKeepsExistingAncestorOwner() throws
Exception {
+ when(mockCatalogWrapper.capabilities()).thenReturn(new
HierarchicalCapability());
+
+ NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog",
"A:B:C");
+ Schema mockSchema = mock(Schema.class);
+ when(mockDispatcher.createSchema(any(), any(),
any())).thenReturn(mockSchema);
+ // "A" already exists (and has its own owner); only "A:B" and the leaf are
newly created.
+ NameIdentifier existingA = NameIdentifier.of("test_metalake",
"test_catalog", "A");
+ when(mockDispatcher.schemaExists(any())).thenReturn(false);
+ when(mockDispatcher.schemaExists(eq(existingA))).thenReturn(true);
+
+ hookDispatcher.createSchema(ident, "comment", Collections.emptyMap());
+
+ List<String> ownedNames =
+
captureOwnedObjects().stream().map(MetadataObject::name).collect(Collectors.toList());
+ Assertions.assertEquals(
+ Arrays.asList("A:B", "A:B:C"),
+ ownedNames,
+ "Pre-existing ancestor 'A' must keep its owner; only newly-created
schemas are claimed");
+ }
+
+ @Test
+ public void testDropSchemaRemovesPrivileges() {
+ NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog",
"A:B:C");
+ when(mockDispatcher.dropSchema(eq(ident), eq(false))).thenReturn(true);
+
+ try (MockedStatic<AuthorizationUtils> authz =
Mockito.mockStatic(AuthorizationUtils.class)) {
+ authz
+ .when(
+ () ->
+ AuthorizationUtils.getMetadataObjectLocation(
+ any(NameIdentifier.class), any(Entity.EntityType.class)))
+ .thenReturn(ImmutableList.of("/test"));
+
+ boolean dropped = hookDispatcher.dropSchema(ident, false);
+
+ Assertions.assertTrue(dropped, "Drop result must be propagated from the
inner dispatcher");
+ verify(mockDispatcher).dropSchema(eq(ident), eq(false));
+ // Privileges for the dropped schema must be removed.
+ authz.verify(
+ () ->
+ AuthorizationUtils.authorizationPluginRemovePrivileges(
+ eq(ident), eq(Entity.EntityType.SCHEMA),
eq(ImmutableList.of("/test"))));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<MetadataObject> captureOwnedObjects() {
+ ArgumentCaptor<List<MetadataObject>> captor =
ArgumentCaptor.forClass(List.class);
+ verify(mockOwnerDispatcher)
+ .setOwners(eq("test_metalake"), captor.capture(), any(),
eq(Owner.Type.USER));
+ return captor.getValue();
+ }
+
private static class CaseInsensitiveCapability implements Capability {
@Override
public CapabilityResult caseSensitiveOnName(Scope scope) {
return CapabilityResult.unsupported("case-insensitive");
}
}
+
+ /** Accepts hierarchical SCHEMA names so normalization does not reject
":"-separated names. */
+ private static class HierarchicalCapability implements Capability {
+ @Override
+ public CapabilityResult specificationOnName(Scope scope, String name) {
+ if (scope == Scope.SCHEMA) {
+ return CapabilityResult.SUPPORTED;
+ }
+ return Capability.super.specificationOnName(scope, name);
+ }
+ }
}
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
index b003b6e5f8..5f328fb023 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
@@ -119,11 +119,15 @@ public class BaseIT {
public static final String DOWNLOAD_CLICKHOUSE_JDBC_DRIVER_URL =
"https://repo1.maven.org/maven2/com/clickhouse/clickhouse-jdbc/0.7.1/clickhouse-jdbc-0.7.1-all.jar";;
+ public static final String DOWNLOAD_SQLITE_JDBC_DRIVER_URL =
+
"https://repo1.maven.org/maven2/org/xerial/sqlite-jdbc/3.42.0.0/sqlite-jdbc-3.42.0.0.jar";;
+
public static final Map<String, Pattern>
SUPPORTED_CLEAN_CONFLICTS_DRIVER_TYPES =
ImmutableMap.of(
"mysql", Pattern.compile("mysql-connector-java-([\\d.]+)\\.jar"),
"postgresql", Pattern.compile("postgresql-([\\d.]+)\\.jar"),
- "clickhouse",
Pattern.compile("clickhouse-jdbc-([\\d.]+)(-all)?\\.jar"));
+ "clickhouse",
Pattern.compile("clickhouse-jdbc-([\\d.]+)(-all)?\\.jar"),
+ "sqlite", Pattern.compile("sqlite-jdbc-([\\d.]+)\\.jar"));
private TestDatabaseName META_DATA;
private MySQLContainer MYSQL_CONTAINER;
@@ -201,7 +205,8 @@ public class BaseIT {
String[] driverUrls = {
DOWNLOAD_MYSQL_JDBC_DRIVER_URL,
DOWNLOAD_POSTGRESQL_JDBC_DRIVER_URL,
- DOWNLOAD_CLICKHOUSE_JDBC_DRIVER_URL
+ DOWNLOAD_CLICKHOUSE_JDBC_DRIVER_URL,
+ DOWNLOAD_SQLITE_JDBC_DRIVER_URL
};
String[] dirs = getJdbcDriverDownloadDirs();
downloadJdbcDrivers(driverUrls, dirs);
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
index 93105922eb..3d84fa84f2 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
@@ -33,6 +33,7 @@ public @interface AuthorizationRequest {
ASSOCIATE_TAG,
ASSOCIATE_POLICY,
RUN_JOB,
- LOAD_TABLE
+ LOAD_TABLE,
+ CREATE_SCHEMA
}
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
index 7b13d63c2f..5badd704d4 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
@@ -54,6 +54,8 @@ public class AuthorizeExecutorFactory {
entityType,
secondaryExpression,
secondaryExpressionCondition);
+ case CREATE_SCHEMA -> new CreateSchemaAuthorizationExecutor(
+ parameters, args, expression, metadataContext, pathParams,
entityType);
};
}
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CreateSchemaAuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CreateSchemaAuthorizationExecutor.java
new file mode 100644
index 0000000000..31799bb2e7
--- /dev/null
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CreateSchemaAuthorizationExecutor.java
@@ -0,0 +1,105 @@
+/*
+ * 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.server.web.filter.authorization;
+
+import java.lang.reflect.Parameter;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.dto.requests.SchemaCreateRequest;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationRequest;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+
+/**
+ * Authorization executor for {@code createSchema} operations.
+ *
+ * <p>For hierarchical schema names (e.g. {@code A:B:C}), injects the parent
schema ({@code A:B})
+ * into the metadata context so that the standard expression can evaluate
{@code CREATE_SCHEMA}
+ * against the already-existing parent. {@link
+ * org.apache.gravitino.server.authorization.jcasbin.JcasbinAuthorizer} then
walks the inheritance
+ * chain ({@code A:B → A → CATALOG}) automatically.
+ *
+ * <p>For top-level schemas no SCHEMA is injected; the expression falls back
to CATALOG-level
+ * checks.
+ */
+public class CreateSchemaAuthorizationExecutor extends
CommonAuthorizerExecutor {
+
+ /**
+ * Creates an authorization executor for a {@code createSchema} request and,
for hierarchical
+ * schema names, injects the parent schema into the metadata context so the
standard expression
+ * can evaluate {@code CREATE_SCHEMA} against the already-existing parent.
+ *
+ * @param parameters the parameters of the intercepted method
+ * @param args the arguments passed to the intercepted method
+ * @param expression the authorization expression to evaluate
+ * @param metadataContext the mutable metadata context bound to the
authorization expression
+ * @param pathParams the path parameters of the request
+ * @param entityType the optional entity type of the request
+ */
+ public CreateSchemaAuthorizationExecutor(
+ Parameter[] parameters,
+ Object[] args,
+ String expression,
+ Map<Entity.EntityType, NameIdentifier> metadataContext,
+ Map<String, Object> pathParams,
+ Optional<String> entityType) {
+ super(expression, metadataContext, pathParams, entityType);
+ injectParentSchema(parameters, args);
+ }
+
+ private void injectParentSchema(Parameter[] parameters, Object[] args) {
+ SchemaCreateRequest request = extractRequest(parameters, args);
+ // Skip injection when the name is missing/blank; let
SchemaCreateRequest.validate() surface the
+ // intended 400 instead of throwing during authorization interception
(returned as a 500).
+ if (request == null || StringUtils.isBlank(request.getName())) {
+ return;
+ }
+
+ NameIdentifier catalogIdent =
metadataContext.get(Entity.EntityType.CATALOG);
+ if (catalogIdent == null) {
+ return;
+ }
+ String metalake = catalogIdent.namespace().level(0);
+ String catalog = catalogIdent.name();
+
+ String separator = HierarchicalSchemaUtil.schemaSeparator();
+ String schemaName = request.getName();
+ String[] levels = HierarchicalSchemaUtil.splitSchemaName(schemaName,
separator);
+
+ if (levels.length > 1) {
+ String parentPath = String.join(separator, Arrays.copyOf(levels,
levels.length - 1));
+ metadataContext.put(
+ Entity.EntityType.SCHEMA, NameIdentifierUtil.ofSchema(metalake,
catalog, parentPath));
+ }
+ }
+
+ private SchemaCreateRequest extractRequest(Parameter[] parameters, Object[]
args) {
+ for (int i = 0; i < parameters.length; i++) {
+ AuthorizationRequest annotation =
parameters[i].getAnnotation(AuthorizationRequest.class);
+ if (annotation != null
+ && annotation.type() ==
AuthorizationRequest.RequestType.CREATE_SCHEMA) {
+ return (SchemaCreateRequest) args[i];
+ }
+ }
+ return null;
+ }
+}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java
b/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java
index bbacfdf9f5..5b7f52ad89 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java
@@ -35,6 +35,7 @@ import javax.ws.rs.QueryParam;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Entity;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
@@ -53,8 +54,10 @@ import org.apache.gravitino.metrics.MetricNames;
import org.apache.gravitino.server.authorization.MetadataAuthzHelper;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationMetadata;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationRequest;
import
org.apache.gravitino.server.authorization.expression.AuthorizationExpressionConstants;
import org.apache.gravitino.server.web.Utils;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.slf4j.Logger;
@@ -86,14 +89,24 @@ public class SchemaOperations {
public Response listSchemas(
@PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
String metalake,
- @PathParam("catalog") @AuthorizationMetadata(type =
Entity.EntityType.CATALOG)
- String catalog) {
- LOG.info("Received list schema request for catalog: {}.{}", metalake,
catalog);
+ @PathParam("catalog") @AuthorizationMetadata(type =
Entity.EntityType.CATALOG) String catalog,
+ @DefaultValue("") @QueryParam("parentSchema") String parentSchema) {
+ LOG.info(
+ "Received list schema request for catalog: {}.{}, parentSchema: {}",
+ metalake,
+ catalog,
+ parentSchema);
try {
return Utils.doAs(
httpRequest,
() -> {
- Namespace schemaNS = NamespaceUtil.ofSchema(metalake, catalog);
+ Namespace schemaNS;
+ if (StringUtils.isBlank(parentSchema)) {
+ schemaNS = NamespaceUtil.ofSchema(metalake, catalog);
+ } else {
+ validateParentSchema(parentSchema);
+ schemaNS = Namespace.of(metalake, catalog, parentSchema);
+ }
NameIdentifier[] idents = dispatcher.listSchemas(schemaNS);
idents =
MetadataAuthzHelper.filterByExpression(
@@ -102,7 +115,12 @@ public class SchemaOperations {
Entity.EntityType.SCHEMA,
idents);
Response response = Utils.ok(new EntityListResponse(idents));
- LOG.info("List {} schemas in catalog {}.{}", idents.length,
metalake, catalog);
+ LOG.info(
+ "List {} schemas in catalog {}.{} (parentSchema='{}')",
+ idents.length,
+ metalake,
+ catalog,
+ parentSchema);
return response;
});
} catch (Exception e) {
@@ -115,13 +133,14 @@ public class SchemaOperations {
@Timed(name = "create-schema." + MetricNames.HTTP_PROCESS_DURATION, absolute
= true)
@ResponseMetered(name = "create-schema", absolute = true)
@AuthorizationExpression(
- expression = "ANY(OWNER, METALAKE, CATALOG) || ANY_USE_CATALOG &&
ANY_CREATE_SCHEMA",
- accessMetadataType = MetadataObject.Type.CATALOG)
+ expression = "ANY(OWNER, METALAKE, CATALOG, SCHEMA) || ANY_USE_CATALOG
&& ANY_CREATE_SCHEMA",
+ accessMetadataType = MetadataObject.Type.SCHEMA)
public Response createSchema(
@PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
String metalake,
@PathParam("catalog") @AuthorizationMetadata(type =
Entity.EntityType.CATALOG) String catalog,
- SchemaCreateRequest request) {
+ @AuthorizationRequest(type =
AuthorizationRequest.RequestType.CREATE_SCHEMA)
+ SchemaCreateRequest request) {
LOG.info("Received create schema request: {}.{}.{}", metalake, catalog,
request.getName());
try {
return Utils.doAs(
@@ -244,4 +263,24 @@ public class SchemaOperations {
return ExceptionHandlers.handleSchemaException(OperationType.DROP,
schema, catalog, e);
}
}
+
+ /**
+ * Validates the {@code parentSchema} query parameter. The value is a
logical (possibly
+ * hierarchical) schema name, so it must not contain empty segments (e.g.
{@code "A::B"} or {@code
+ * "A:"}) before it is passed to {@link Namespace#of}.
+ *
+ * @param parentSchema the non-blank {@code parentSchema} query parameter
+ * @throws IllegalArgumentException if the value contains an empty segment
+ */
+ private static void validateParentSchema(String parentSchema) {
+ String separator = HierarchicalSchemaUtil.schemaSeparator();
+ for (String segment : HierarchicalSchemaUtil.splitSchemaName(parentSchema,
separator)) {
+ if (segment.isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The parentSchema '%s' contains an empty segment after
splitting by '%s'.",
+ parentSchema, separator));
+ }
+ }
+ }
}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java
index 02ce325f3b..8981e5e29d 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.server.web.rest;
import static org.apache.gravitino.Configs.CACHE_ENABLED;
import static org.apache.gravitino.Configs.ENABLE_AUTHORIZATION;
+import static org.apache.gravitino.Configs.SCHEMA_SEPARATOR;
import static org.apache.gravitino.Configs.TREE_LOCK_CLEAN_INTERVAL;
import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY;
import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY;
@@ -27,6 +28,8 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableList;
@@ -43,6 +46,7 @@ import org.apache.gravitino.Audit;
import org.apache.gravitino.Config;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.catalog.SchemaDispatcher;
import org.apache.gravitino.catalog.SchemaOperationDispatcher;
@@ -94,6 +98,7 @@ public class TestSchemaOperations extends BaseOperationsTest {
Mockito.doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
Mockito.doReturn(false).when(config).get(CACHE_ENABLED);
Mockito.doReturn(false).when(config).get(ENABLE_AUTHORIZATION);
+ Mockito.doReturn(":").when(config).get(SCHEMA_SEPARATOR);
FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
}
@@ -177,6 +182,41 @@ public class TestSchemaOperations extends
BaseOperationsTest {
Assertions.assertEquals(RuntimeException.class.getSimpleName(),
errorResp2.getType());
}
+ @Test
+ public void testListSchemasWithParentSchemaPassToDispatcher() {
+ NameIdentifier ident = NameIdentifier.of(metalake, catalog, "A:sales");
+ when(dispatcher.listSchemas(any())).thenReturn(new NameIdentifier[]
{ident});
+
+ Response resp =
+ target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas")
+ .queryParam("parentSchema", "A")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+ EntityListResponse listResp = resp.readEntity(EntityListResponse.class);
+ Assertions.assertEquals(0, listResp.getCode());
+ Assertions.assertEquals(1, listResp.identifiers().length);
+ Assertions.assertEquals("A:sales", listResp.identifiers()[0].name());
+
+ verify(dispatcher).listSchemas(eq(Namespace.of(metalake, catalog, "A")));
+ }
+
+ @Test
+ public void testListSchemasWithMalformedParentSchemaReturnsBadRequest() {
+ Response resp =
+ target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas")
+ .queryParam("parentSchema", "A::B")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
resp.getStatus());
+ // The malformed parentSchema must be rejected before reaching the
dispatcher.
+ verify(dispatcher, never()).listSchemas(any());
+ }
+
@Test
public void testCreateSchema() {
SchemaCreateRequest req =
