[#7190] fix(catalogs): Validate and filter empty storageLocations in HadoopCatalogOperations

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java

@@ -245,6 +245,9 @@ public Fileset createMultipleLocationFileset(
       Map<String, String> storageLocations,
       Map<String, String> properties)
       throws NoSuchSchemaException, FilesetAlreadyExistsException {
+    // remove the entries with empty value and check the default-location-names
+    storageLocations = filterAndValidateStorageLocations(storageLocations, properties);
+
     storageLocations.forEach(
         (name, path) -> {
           if (StringUtils.isBlank(name)) {
@@ -1128,6 +1131,35 @@ private boolean containsPlaceholder(String location) {
         && LOCATION_PLACEHOLDER_PATTERN.matcher(location).find();
   }
 
+  private Map<String, String> filterAndValidateStorageLocations(
+      Map<String, String> storageLocations, Map<String, String> properties) {
+    if (storageLocations == null) {
+      return new HashMap<>();
+    }
+    String defaultLocationName =
+        properties != null ? properties.get(PROPERTY_DEFAULT_LOCATION_NAME) : null;
+    Set<String> removedKeys = new HashSet<>();
+    Map<String, String> filtered = new HashMap<>();
+    // if storageLocations is null or empty, remove the entries
+    for (Map.Entry<String, String> e : storageLocations.entrySet()) {
+      String key = e.getKey();
+      String val = e.getValue();
+      if (StringUtils.isNotBlank(val)) {
+        filtered.put(key, val);
+      } else {
+        removedKeys.add(key);
+      }
+    }
+    // check the default-location-name is not in removedKeys
+    if (defaultLocationName != null && removedKeys.contains(defaultLocationName)) {
+      throw new IllegalArgumentException(
+          String.format(
+              "default-location-name '%s' is invalid; please verify that the specified storageLocations are correct.",
+              defaultLocationName));
+    }
+    return filtered;
+  }
+
   private Map<String, Path> calculateFilesetPaths(
       String schemaName,
       String filesetName,

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/TestHadoopCatalogOperations.java

@@ -1519,6 +1519,95 @@ public void testCreateMultipleLocationsWithExceptions() throws IOException {
               .contains(
                   "Default location name must be set and must be one of the fileset locations"),
           "Exception message: " + exception.getMessage());
+
+      Schema emptySchema =
+          createMultiLocationSchema(
+              "emptySchema", "no-location", ImmutableMap.of(), ImmutableMap.of());
+
+      // only one storage location but value is null
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  createMultiLocationFileset(
+                      "fs1",
+                      emptySchema.name(),
+                      null,
+                      Fileset.Type.MANAGED,
+                      ImmutableMap.of(),
+                      null,
+                      null));
+      Assertions.assertEquals(
+          "Storage location must be set for fileset m1.c1.emptySchema.fs1 when it's catalog and schema location are not set",
+          exception.getMessage());
+
+      // only one storage location but value is ""
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  createMultiLocationFileset(
+                      "fs2",
+                      emptySchema.name(),
+                      null,
+                      Fileset.Type.MANAGED,
+                      ImmutableMap.of(),
+                      ImmutableMap.of("v1", ""),
+                      ImmutableMap.of()));
+      Assertions.assertEquals(
+          "Storage location must be set for fileset m1.c1.emptySchema.fs2 when it's catalog and schema location are not set",
+          exception.getMessage());
+
+      // only one storage location but value is " "
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  createMultiLocationFileset(
+                      "fs3",
+                      emptySchema.name(),
+                      null,
+                      Fileset.Type.MANAGED,
+                      ImmutableMap.of(),
+                      ImmutableMap.of("v1", " "),
+                      ImmutableMap.of()));
+      Assertions.assertEquals(
+          "Storage location must be set for fileset m1.c1.emptySchema.fs3 when it's catalog and schema location are not set",
+          exception.getMessage());
+
+      // single storage location is " " and default-location-name points to it
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  createMultiLocationFileset(
+                      "fs4",
+                      emptySchema.name(),
+                      null,
+                      Fileset.Type.MANAGED,
+                      ImmutableMap.of(),
+                      ImmutableMap.of("loc1", " "),
+                      ImmutableMap.of(PROPERTY_DEFAULT_LOCATION_NAME, "loc1")));
+      Assertions.assertEquals(
+          "default-location-name 'loc1' is invalid; please verify that the specified storageLocations are correct.",
+          exception.getMessage());
+
+      // two storage locations, one valid and one "", default-location-name points to ""
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  createMultiLocationFileset(
+                      "fs5",
+                      emptySchema.name(),
+                      null,
+                      Fileset.Type.MANAGED,
+                      ImmutableMap.of(),
+                      ImmutableMap.of("v1", TEST_ROOT_PATH + "/valid", "v2", ""),
+                      ImmutableMap.of(PROPERTY_DEFAULT_LOCATION_NAME, "v2")));
+      Assertions.assertEquals(
+          "default-location-name 'v2' is invalid; please verify that the specified storageLocations are correct.",
+          exception.getMessage());
     }
   }
 
@@ -2047,7 +2136,29 @@ private static Stream<Arguments> multipleLocationsArguments() {
                 "v1",
                 TEST_ROOT_PATH + "/fileset509_1",
                 "v2",
-                TEST_ROOT_PATH + "/fileset509_2")));
+                TEST_ROOT_PATH + "/fileset509_2")),
+        // storageLocations == null, use schemaPaths
+        Arguments.of(
+            "fileset520",
+            Fileset.Type.MANAGED,
+            ImmutableMap.of(),
+            ImmutableMap.of(
+                PROPERTY_MULTIPLE_LOCATIONS_PREFIX + "s",
+                TEST_ROOT_PATH + "/s1/{{schema}}/{{fileset}}"),
+            null,
+            null,
+            ImmutableMap.of("s", TEST_ROOT_PATH + "/s1/s1_fileset520/fileset520")),
+        // storageLocations == null, use catalogPaths
+        Arguments.of(
+            "fileset521",
+            Fileset.Type.MANAGED,
+            ImmutableMap.of(
+                PROPERTY_MULTIPLE_LOCATIONS_PREFIX + "c",
+                TEST_ROOT_PATH + "/catalog1/{{schema}}/{{fileset}}"),
+            ImmutableMap.of(),
+            null,
+            null,
+            ImmutableMap.of("c", TEST_ROOT_PATH + "/catalog1/s1_fileset521/fileset521")));
   }
 
   private static Stream<Arguments> locationWithPlaceholdersArguments() {

web/web/src/app/metalakes/metalake/rightContent/CreateFilesetDialog.js

@@ -234,10 +234,12 @@ const CreateFilesetDialog = props => {
           name: data.name,
           type: data.type,
           storageLocations: data.storageLocations.reduce((acc, item) => {
-            if (item.name && item.location) {
-              acc[item.name] = item.location
+            const trimmedName = item.name?.trim()
+            const trimmedLocation = item.location?.trim()
+            if (trimmedName && trimmedLocation) {
+              acc[trimmedName] = trimmedLocation
               if (item.defaultLocation && !properties['default-location-name']) {
-                properties['default-location-name'] = item.name
+                properties['default-location-name'] = trimmedName
               }
             }