yuqi1129 commented on code in PR #6211:
URL: https://github.com/apache/gravitino/pull/6211#discussion_r1914346589
##########
core/src/main/java/org/apache/gravitino/utils/NamespaceUtil.java:
##########
@@ -107,6 +108,17 @@ public static Namespace ofFileset(String metalake, String
catalog, String schema
return Namespace.of(metalake, catalog, schema);
}
+ /**
+ * Create a namespace for fileset from a schema name identifier.
+ *
+ * @param ident The schema name identifier.
+ * @return A namespace for fileset.
+ */
+ public static Namespace toFileset(NameIdentifier ident) {
Review Comment:
We could not get the actual meaning from the method name, what about
`toFilesetNamespace` or something similar?
.
Parameter name `ident` can be replace with `schemaIdent` to improve the
readiabitily.
##########
core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java:
##########
@@ -465,30 +471,72 @@ public static List<String> getMetadataObjectLocation(
}
break;
case SCHEMA:
- {
- Catalog catalogObj =
- GravitinoEnv.getInstance()
- .catalogDispatcher()
- .loadCatalog(
- NameIdentifier.of(ident.namespace().level(0),
ident.namespace().level(1)));
- LOG.info("Catalog provider is %s", catalogObj.provider());
- if (catalogObj.provider().equals("hive")) {
- Schema schema =
GravitinoEnv.getInstance().schemaDispatcher().loadSchema(ident);
- if (schema.properties().containsKey(HiveConstants.LOCATION)) {
+ Catalog catalogObj =
+ GravitinoEnv.getInstance()
+ .catalogDispatcher()
+ .loadCatalog(
+ NameIdentifier.of(ident.namespace().level(0),
ident.namespace().level(1)));
+ Schema schema =
GravitinoEnv.getInstance().schemaDispatcher().loadSchema(ident);
+
+ switch (catalogObj.type()) {
+ case RELATIONAL:
+ LOG.info("Catalog provider is {}", catalogObj.provider());
+ if ("hive".equals(catalogObj.provider())
+ && schema.properties().containsKey(HiveConstants.LOCATION)) {
String schemaLocation =
schema.properties().get(HiveConstants.LOCATION);
- if (schemaLocation != null && schemaLocation.isEmpty()) {
+ if (StringUtils.isNotBlank(schemaLocation)) {
locations.add(schemaLocation);
} else {
- LOG.warn("Schema %s location is not found", ident);
+ LOG.warn("Schema {} location is not found", ident);
}
}
- }
- // TODO: [#6133] Supports get Fileset schema location in the
AuthorizationUtils
+ break;
+
+ case FILESET:
+ if (catalogObj instanceof HasPropertyMetadata) {
+ HasPropertyMetadata catalogObjWithProperties =
(HasPropertyMetadata) catalogObj;
+ Map<String, String> properties = schema.properties();
+ String schemaLocation =
+ (String)
+ catalogObjWithProperties
+ .schemaPropertiesMetadata()
+ .getOrDefault(properties, Constants.LOCATION);
+ if (StringUtils.isNotBlank(schemaLocation)) {
+ locations.add(normalizeFilesetLocation(schemaLocation));
+ }
+ } else {
+ FilesetCatalog filesetCatalog = catalogObj.asFilesetCatalog();
+ String catalogObjLocation =
catalogObj.properties().get(Constants.LOCATION);
+ Namespace namespace = NamespaceUtil.toFileset(ident);
+ NameIdentifier[] nameIdentifiers =
filesetCatalog.listFilesets(namespace);
+ if (nameIdentifiers.length == 0) {
+ LOG.warn(
Review Comment:
I'm unclear why we will use `catalogLocation` when is no fileset loaded?
Another point, I think we can't guarantee that `catalogObjLocation` is found
and it can be a null value.
--
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: commits-unsubscr...@gravitino.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
