[#14123] feat(fileset): Support managed and external type conversion

api/src/main/java/org/apache/gravitino/file/FilesetChange.java

@@ -25,7 +25,8 @@
 
 /**
  * A fileset change is a change to a fileset. It can be used to rename a fileset, update the comment
- * of a fileset, set a property and value pair for a fileset, or remove a property from a fileset.
+ * of a fileset, update the type of a fileset, set a property and value pair for a fileset, or
+ * remove a property from a fileset.
  */
 @Evolving
 public interface FilesetChange {
@@ -50,6 +51,22 @@ static FilesetChange updateComment(String newComment) {
     return new UpdateFilesetComment(newComment);
   }
 
+  /**
+   * Creates a new fileset change to update the fileset type.
+   *
+   * <p>Changing the type preserves the fileset's identity, storage locations, and existing files.
+   * Changing to {@link Fileset.Type#EXTERNAL} relinquishes management of the data, so dropping the
+   * fileset retains it. Changing to {@link Fileset.Type#MANAGED} accepts management of the data,
+   * including deleting it when the fileset is dropped. Setting the current type is allowed.
+   *
+   * @param newType The new type of the fileset.
+   * @return The fileset change.
+   * @throws IllegalArgumentException if the new type is null.
+   */
+  static FilesetChange updateType(Fileset.Type newType) {
+    return new UpdateFilesetType(newType);
+  }
+
   /**
    * Creates a new fileset change to set the property and value for the fileset.
    *
@@ -215,6 +232,48 @@ public String toString() {
     }
   }
 
+  /** A fileset change to update the fileset type without changing its data or locations. */
+  final class UpdateFilesetType implements FilesetChange {
+    private final Fileset.Type newType;
+
+    private UpdateFilesetType(Fileset.Type newType) {
+      if (newType == null) {
+        throw new IllegalArgumentException("New fileset type cannot be null");
+      }
+      this.newType = newType;
+    }
+
+    /**
+     * Returns the new type of the fileset.
+     *
+     * @return The new fileset type.
+     */
+    public Fileset.Type getNewType() {
+      return newType;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      UpdateFilesetType that = (UpdateFilesetType) o;
+      return newType == that.newType;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public int hashCode() {
+      return Objects.hash(newType);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public String toString() {
+      return "UPDATEFILESETTYPE " + newType;
+    }
+  }
+
   /** A fileset change to set the property and value for the fileset. */
   final class SetProperty implements FilesetChange {
     private final String property;

api/src/test/java/org/apache/gravitino/file/TestFilesetChange.java

@@ -23,6 +23,24 @@
 import org.junit.jupiter.api.Test;
 
 public class TestFilesetChange {
+  @Test
+  void testUpdateTypeChange() {
+    for (Fileset.Type type : Fileset.Type.values()) {
+      FilesetChange.UpdateFilesetType change =
+          (FilesetChange.UpdateFilesetType) FilesetChange.updateType(type);
+      Assertions.assertEquals(type, change.getNewType());
+      Assertions.assertEquals(change, FilesetChange.updateType(type));
+      Assertions.assertEquals(change.hashCode(), FilesetChange.updateType(type).hashCode());
+      Assertions.assertNotEquals(change, null);
+      Assertions.assertNotEquals(change, FilesetChange.rename("fileset"));
+      Assertions.assertEquals("UPDATEFILESETTYPE " + type, change.toString());
+    }
+    Assertions.assertNotEquals(
+        FilesetChange.updateType(Fileset.Type.MANAGED),
+        FilesetChange.updateType(Fileset.Type.EXTERNAL));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> FilesetChange.updateType(null));
+  }
+
   @Test
   void testRemovePropertyChange() {
     String property = "property1";

catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java

@@ -1292,6 +1292,7 @@ private FilesetEntity updateFilesetEntity(
             : Maps.newHashMap(filesetEntity.properties());
     String newName = ident.name();
     String newComment = filesetEntity.comment();
+    Fileset.Type newType = filesetEntity.filesetType();
 
     for (FilesetChange change : changes) {
       if (change instanceof FilesetChange.SetProperty) {
@@ -1304,6 +1305,8 @@ private FilesetEntity updateFilesetEntity(
         newName = ((FilesetChange.RenameFileset) change).getNewName();
       } else if (change instanceof FilesetChange.UpdateFilesetComment) {
         newComment = ((FilesetChange.UpdateFilesetComment) change).getNewComment();
+      } else if (change instanceof FilesetChange.UpdateFilesetType) {
+        newType = ((FilesetChange.UpdateFilesetType) change).getNewType();
       } else if (change instanceof FilesetChange.RemoveComment) {
         newComment = null;
       } else {
@@ -1317,7 +1320,7 @@ private FilesetEntity updateFilesetEntity(
         .withNamespace(ident.namespace())
         .withId(filesetEntity.id())
         .withComment(newComment)
-        .withFilesetType(filesetEntity.filesetType())
+        .withFilesetType(newType)
         .withStorageLocations(filesetEntity.storageLocations())
         .withProperties(props)
         .withAuditInfo(

catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java

@@ -163,6 +163,7 @@
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -1398,6 +1399,123 @@ public void testAlterFilesetProperties() throws IOException {
     }
   }
 
+  @ParameterizedTest
+  @CsvSource({"EXTERNAL,false", "MANAGED,false", "EXTERNAL,true", "MANAGED,true"})
+  void testUpdateFilesetTypePreservesDataAndControlsDrop(
+      Fileset.Type targetType, boolean cascade, @TempDir File root) throws IOException {
+    String schemaName = "schema" + generateTestId();
+    createSchema(schemaName, "schema comment", null, null);
+    NameIdentifier ident = NameIdentifier.of("m1", "c1", schemaName, "fileset");
+    Fileset.Type originalType =
+        targetType == Fileset.Type.MANAGED ? Fileset.Type.EXTERNAL : Fileset.Type.MANAGED;
+    File primary = new File(root, "primary");
+    File backup = new File(root, "backup");
+    Files.createDirectories(primary.toPath());
+    Files.createDirectories(backup.toPath());
+    Map<String, String> locations =
+        ImmutableMap.of("primary", primary.toURI().toString(), "backup", backup.toURI().toString());
+    Map<String, String> props =
+        StringIdentifier.newPropertiesWithId(
+            StringIdentifier.fromId(idGenerator.nextId()),
+            ImmutableMap.of(PROPERTY_DEFAULT_LOCATION_NAME, "primary", "k1", "v1"));
+
+    try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) {
+      ops.initialize(Maps.newHashMap(), randomCatalogInfo(), FILESET_PROPERTIES_METADATA);
+      ops.createMultipleLocationFileset(ident, "comment", originalType, locations, props);
+      for (File directory : Arrays.asList(primary, backup)) {
+        Files.writeString(new File(directory, "data.txt").toPath(), "existing data");
+      }
+      FilesetEntity before = store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
+
+      // Repeating the conversion must preserve both metadata and existing data.
+      for (int attempt = 0; attempt < 2; attempt++) {
+        Fileset updated = ops.alterFileset(ident, FilesetChange.updateType(targetType));
+        Assertions.assertEquals(targetType, updated.type());
+        Assertions.assertEquals(before.name(), updated.name());
+        Assertions.assertEquals(before.comment(), updated.comment());
+        Assertions.assertEquals(before.storageLocations(), updated.storageLocations());
+        Assertions.assertEquals(before.properties(), updated.properties());
+        FilesetEntity persisted = store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
+        Assertions.assertEquals(before.id(), persisted.id());
+        Assertions.assertEquals(before.namespace(), persisted.namespace());
+        Assertions.assertEquals(targetType, persisted.filesetType());
+        Assertions.assertEquals(before.auditInfo().creator(), persisted.auditInfo().creator());
+        Assertions.assertEquals(
+            before.auditInfo().createTime(), persisted.auditInfo().createTime());
+        Assertions.assertNotNull(persisted.auditInfo().lastModifiedTime());
+        for (File directory : Arrays.asList(primary, backup)) {
+          Assertions.assertEquals(
+              "existing data", Files.readString(new File(directory, "data.txt").toPath()));
+        }
+      }
+
+      if (cascade) {
+        Assertions.assertTrue(ops.dropSchema(NameIdentifier.of("m1", "c1", schemaName), true));
+      } else {
+        Assertions.assertTrue(ops.dropFileset(ident));
+      }
+      Assertions.assertFalse(store.exists(ident, Entity.EntityType.FILESET));
+      for (File directory : Arrays.asList(primary, backup)) {
+        if (targetType == Fileset.Type.EXTERNAL) {
+          Assertions.assertEquals(
+              "existing data", Files.readString(new File(directory, "data.txt").toPath()));
+        } else {
+          Assertions.assertFalse(directory.exists());
+        }
+      }
+    }
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  void testUpdateFilesetTypeWithOtherChanges(boolean disableFsOps) throws IOException {
+    String schemaName = "schema" + generateTestId();
+    createSchema(schemaName, "comment", null, null, true);
+    NameIdentifier ident = NameIdentifier.of("m1", "c1", schemaName, "fileset");
+    File missingLocation = new File(UNFORMALIZED_TEST_ROOT_PATH, "missing-" + schemaName);
+    createFileset(
+        "fileset",
+        schemaName,
+        "comment",
+        Fileset.Type.EXTERNAL,
+        null,
+        missingLocation.toURI().toString(),
+        true);
+    FilesetEntity before = store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
+    try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) {
+      ops.initialize(
+          ImmutableMap.of(DISABLE_FILESYSTEM_OPS, Boolean.toString(disableFsOps)),
+          randomCatalogInfo(),
+          FILESET_PROPERTIES_METADATA);
+      Fileset updated =
+          ops.alterFileset(
+              ident,
+              FilesetChange.updateType(Fileset.Type.MANAGED),
+              FilesetChange.rename("renamed"),
+              FilesetChange.updateComment("updated"),
+              FilesetChange.setProperty("k1", "v1"));
+      NameIdentifier renamed = NameIdentifier.of(ident.namespace(), "renamed");
+      Assertions.assertEquals(Fileset.Type.MANAGED, updated.type());
+      Assertions.assertEquals("renamed", updated.name());
+      Assertions.assertEquals("updated", updated.comment());
+      Assertions.assertEquals("v1", updated.properties().get("k1"));
+      Assertions.assertFalse(store.exists(ident, Entity.EntityType.FILESET));
+      Assertions.assertEquals(
+          before.id(), store.get(renamed, Entity.EntityType.FILESET, FilesetEntity.class).id());
+      Assertions.assertEquals(before.storageLocations(), updated.storageLocations());
+      Assertions.assertFalse(missingLocation.exists());
+
+      // Multiple type changes follow request order, like other fileset changes.
+      Fileset reverted =
+          ops.alterFileset(
+              renamed,
+              FilesetChange.updateType(Fileset.Type.MANAGED),
+              FilesetChange.updateType(Fileset.Type.EXTERNAL));
+      Assertions.assertEquals(Fileset.Type.EXTERNAL, reverted.type());
+      Assertions.assertTrue(ops.dropFileset(renamed));
+    }
+  }
+
   @Test
   public void testFormalizePath() throws IOException, IllegalAccessException, InterruptedException {
 

clients/client-java/src/main/java/org/apache/gravitino/client/DTOConverters.java

@@ -286,6 +286,9 @@ static FilesetUpdateRequest toFilesetUpdateRequest(FilesetChange change) {
     } else if (change instanceof FilesetChange.UpdateFilesetComment) {
       return new FilesetUpdateRequest.UpdateFilesetCommentRequest(
           ((FilesetChange.UpdateFilesetComment) change).getNewComment());
+    } else if (change instanceof FilesetChange.UpdateFilesetType) {
+      return new FilesetUpdateRequest.UpdateFilesetTypeRequest(
+          ((FilesetChange.UpdateFilesetType) change).getNewType());
     } else if (change instanceof FilesetChange.RemoveComment) {
       return new FilesetUpdateRequest.RemoveFilesetCommentRequest();
     } else if (change instanceof FilesetChange.SetProperty) {

clients/client-java/src/test/java/org/apache/gravitino/client/TestFilesetCatalog.java

@@ -62,6 +62,7 @@
 import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.exceptions.NotFoundException;
 import org.apache.gravitino.file.Fileset;
+import org.apache.gravitino.file.FilesetChange;
 import org.apache.hc.core5.http.Method;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
@@ -429,6 +430,30 @@ public void testAlterFileset() throws JsonProcessingException {
         "internal error");
   }
 
+  @Test
+  public void testAlterFilesetType() throws JsonProcessingException {
+    NameIdentifier ident = NameIdentifier.of("schema1", "fileset1");
+    String path =
+        withSlash(
+            FilesetCatalog.formatFilesetRequestPath(
+                    Namespace.of(metalakeName, catalogName, "schema1"))
+                + "/fileset1");
+    for (Fileset.Type type : Fileset.Type.values()) {
+      FilesetDTO expected =
+          mockFilesetDTO("fileset1", type, "comment", "mock location", ImmutableMap.of("k1", "v1"));
+      FilesetUpdateRequest request = new FilesetUpdateRequest.UpdateFilesetTypeRequest(type);
+      buildMockResource(
+          Method.PUT,
+          path,
+          new FilesetUpdatesRequest(ImmutableList.of(request)),
+          new FilesetResponse(expected),
+          SC_OK);
+      Fileset actual =
+          catalog.asFilesetCatalog().alterFileset(ident, FilesetChange.updateType(type));
+      assertFileset(expected, actual);
+    }
+  }
+
   @Test
   public void testGetFileLocation() throws JsonProcessingException {
     NameIdentifier fileset = NameIdentifier.of(metalakeName, catalogName, "schema1", "fileset1");

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/FilesetAuthorizationIT.java

@@ -438,6 +438,56 @@ public void testUseSecretsPrivilegeForGetSecrets() {
     adminFilesetCatalog.dropFileset(filesetIdent);
   }
 
+  /** Verifies that writers can change the fileset lifecycle without gaining drop permission. */
+  @Test
+  @Order(9)
+  public void testWriterCanConvertFilesetTypeButCannotDrop() {
+    String filesetName = "type_conversion_fileset";
+    NameIdentifier filesetIdent = NameIdentifier.of(SCHEMA, filesetName);
+    MetadataObject filesetObject =
+        MetadataObjects.of(
+            ImmutableList.of(CATALOG, SCHEMA, filesetName), MetadataObject.Type.FILESET);
+    GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+    FilesetCatalog ownerFilesetCatalog = metalake.loadCatalog(CATALOG).asFilesetCatalog();
+    FilesetCatalog writerFilesetCatalog =
+        normalUserClient.loadMetalake(METALAKE).loadCatalog(CATALOG).asFilesetCatalog();
+
+    ownerFilesetCatalog.createFileset(
+        filesetIdent,
+        "comment",
+        Fileset.Type.EXTERNAL,
+        storageLocation(GravitinoITUtils.genRandomName("FilesetAuthorizationIT_type_conversion")),
+        new HashMap<>());
+    try {
+      metalake.grantPrivilegesToRole(
+          role, filesetObject, ImmutableList.of(Privileges.ReadFileset.allow()));
+      assertThrows(
+          ForbiddenException.class,
+          () ->
+              writerFilesetCatalog.alterFileset(
+                  filesetIdent, FilesetChange.updateType(Fileset.Type.MANAGED)));
+      assertEquals(Fileset.Type.EXTERNAL, ownerFilesetCatalog.loadFileset(filesetIdent).type());
+
+      metalake.grantPrivilegesToRole(
+          role, filesetObject, ImmutableList.of(Privileges.WriteFileset.allow()));
+      Fileset managedFileset =
+          writerFilesetCatalog.alterFileset(
+              filesetIdent, FilesetChange.updateType(Fileset.Type.MANAGED));
+      assertEquals(Fileset.Type.MANAGED, managedFileset.type());
+      assertThrows(ForbiddenException.class, () -> writerFilesetCatalog.dropFileset(filesetIdent));
+      assertEquals(Fileset.Type.MANAGED, writerFilesetCatalog.loadFileset(filesetIdent).type());
+
+      Fileset externalFileset =
+          writerFilesetCatalog.alterFileset(
+              filesetIdent, FilesetChange.updateType(Fileset.Type.EXTERNAL));
+      assertEquals(Fileset.Type.EXTERNAL, externalFileset.type());
+      assertThrows(ForbiddenException.class, () -> writerFilesetCatalog.dropFileset(filesetIdent));
+      assertEquals(Fileset.Type.EXTERNAL, writerFilesetCatalog.loadFileset(filesetIdent).type());
+    } finally {
+      ownerFilesetCatalog.dropFileset(filesetIdent);
+    }
+  }
+
   private String defaultBaseLocation() {
     if (defaultBaseLocation == null) {
       defaultBaseLocation =