[improvement] When dropping table, data should be moved to trash instead of being deleted directly

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -226,6 +226,13 @@ default FileStatus[] listDirectories(Path path) throws IOException {
      */
     boolean delete(Path path, boolean recursive) throws IOException;
 
+    /**
+     * Move the path to trash.
+     *
+     * @param path the path to move to trash.
+     */
+    boolean moveToTrash(Path path) throws IOException;
+
     /**
      * Make the given file and all non-existent parents into directories. Has the semantics of Unix
      * 'mkdir -p'. Existence of the directory hierarchy is not an error.

paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java

@@ -97,6 +97,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
         return wrap(() -> fileIO(path).delete(path, recursive));
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean mkdirs(Path path) throws IOException {
         return wrap(() -> fileIO(path).mkdirs(path));

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java

@@ -37,6 +37,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Trash;
 
 import java.io.IOException;
 import java.io.OutputStreamWriter;
@@ -153,6 +154,19 @@ public boolean delete(Path path, boolean recursive) throws IOException {
         return getFileSystem(hadoopPath).delete(hadoopPath, recursive);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        org.apache.hadoop.fs.Path hadoopPath = path(path);
+        FileSystem fs = getFileSystem(hadoopPath);
+        Configuration conf = hadoopConf.get();
+
+        org.apache.hadoop.fs.Path fullyResolvedPath = fs.resolvePath(hadoopPath);
+        FileSystem fullyResolvedFs = FileSystem.get(fullyResolvedPath.toUri(), conf);
+
+        Trash trash = new Trash(fullyResolvedFs, conf);
+        return trash.moveToTrash(fullyResolvedPath);
+    }
+
     @Override
     public boolean mkdirs(Path path) throws IOException {
         org.apache.hadoop.fs.Path hadoopPath = path(path);

paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java

@@ -165,6 +165,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
         return delete(file);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     private boolean delete(final File f) {
         if (f.isDirectory()) {
             final File[] files = f.listFiles();

paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -132,6 +132,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
         return fileIO().delete(path, recursive);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean mkdirs(Path path) throws IOException {
         return fileIO().mkdirs(path);

paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java

@@ -235,6 +235,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
             return delete(file);
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         private boolean delete(final File f) {
             if (f.isDirectory()) {
                 final File[] files = f.listFiles();

paimon-core/src/test/java/org/apache/paimon/utils/TraceableFileIO.java

@@ -112,6 +112,11 @@ public boolean delete(Path f, boolean recursive) throws IOException {
         return originalFs.delete(f, recursive);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean mkdirs(Path f) throws IOException {
         return originalFs.mkdirs(f);

paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/AzureFileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
@@ -66,6 +67,11 @@ public void configure(CatalogContext context) {
         this.hadoopOptions = mirrorCertainHadoopConfig(loadHadoopConfigFromContext(context));
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     // add additional config entries from the IO config to the Hadoop config
     private Options loadHadoopConfigFromContext(CatalogContext context) {
         Options hadoopConfig = new Options();

paimon-filesystems/paimon-azure/src/main/java/org/apache/paimon/azure/AzureLoader.java

@@ -25,6 +25,8 @@
 import org.apache.paimon.fs.PluginFileIO;
 import org.apache.paimon.plugin.PluginLoader;
 
+import java.io.IOException;
+
 /** Azure Blob Storage {@link FileIOLoader}. */
 public class AzureLoader implements FileIOLoader {
 
@@ -62,6 +64,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(AZURE_CLASS);

paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/COSNFileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
@@ -89,6 +90,11 @@ public void configure(CatalogContext context) {
         }
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) {
         final String scheme = path.toUri().getScheme();

paimon-filesystems/paimon-cosn/src/main/java/org/apache/paimon/cosn/COSNLoader.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.fs.PluginFileIO;
 import org.apache.paimon.plugin.PluginLoader;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -70,6 +71,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(COSN_CLASS);

paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java

@@ -76,6 +76,11 @@ public void configure(CatalogContext context) {
         }
     }
 
+    @Override
+    public boolean moveToTrash(org.apache.paimon.fs.Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     protected FileSystem createFileSystem(Path path) throws IOException {
         final String scheme = path.toUri().getScheme();

paimon-filesystems/paimon-gs/src/main/java/org/apache/paimon/gs/GSLoader.java

@@ -25,6 +25,8 @@
 import org.apache.paimon.fs.PluginFileIO;
 import org.apache.paimon.plugin.PluginLoader;
 
+import java.io.IOException;
+
 /** A {@link PluginLoader} to load gcs. */
 public class GSLoader implements FileIOLoader {
 
@@ -63,6 +65,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(GCS_CLASS);

paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java

@@ -136,6 +136,11 @@ public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite
         return new JindoTwoPhaseOutputStream(new JindoMultiPartUpload(fs, hadoopPath), hadoopPath);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     protected Pair<JindoHadoopSystem, String> createFileSystem(org.apache.hadoop.fs.Path path) {
         final String scheme = path.toUri().getScheme();

paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/OBSFileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
@@ -105,6 +106,11 @@ public void configure(CatalogContext context) {
         }
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) {
         final String scheme = path.toUri().getScheme();

paimon-filesystems/paimon-obs/src/main/java/org/apache/paimon/obs/OBSLoader.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.fs.PluginFileIO;
 import org.apache.paimon.plugin.PluginLoader;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -76,6 +77,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(OBS_CLASS);

paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java

@@ -113,6 +113,11 @@ public void configure(CatalogContext context) {
         }
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite)
             throws IOException {

paimon-filesystems/paimon-oss/src/main/java/org/apache/paimon/oss/OSSLoader.java

@@ -77,6 +77,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(OSS_CLASS);

paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3FileIO.java

@@ -84,6 +84,11 @@ public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite
         return new S3TwoPhaseOutputStream(new S3MultiPartUpload(fs, fs.getConf()), hadoopPath);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     // add additional config entries from the IO config to the Hadoop config
     private Options loadHadoopConfigFromContext(CatalogContext context) {
         Options hadoopConfig = new Options();

paimon-filesystems/paimon-s3/src/main/java/org/apache/paimon/s3/S3Loader.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.fs.PluginFileIO;
 import org.apache.paimon.plugin.PluginLoader;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -76,6 +77,11 @@ public boolean isObjectStore() {
             return true;
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         protected FileIO createFileIO(Path path) {
             FileIO fileIO = getLoader().newInstance(S3_CLASS);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java

@@ -109,6 +109,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
         return getFileSystem(flinkPath).delete(flinkPath, recursive);
     }
 
+    @Override
+    public boolean moveToTrash(Path path) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean mkdirs(Path path) throws IOException {
         org.apache.flink.core.fs.Path flinkPath = path(path);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/format/CompactedChangelogFormatReaderFactory.java

@@ -156,6 +156,11 @@ public boolean delete(Path path, boolean recursive) throws IOException {
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public boolean moveToTrash(Path path) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public boolean mkdirs(Path path) throws IOException {
             throw new UnsupportedOperationException();

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -961,7 +961,7 @@ protected void dropTableImpl(Identifier identifier, List<Path> externalPaths) {
                                     identifier.getTableName(),
                                     !externalTable,
                                     false,
-                                    true));
+                                    false));
 
             // When drop a Hive external table, only the hive metadata is deleted and the data files
             // are not deleted.
@@ -975,7 +975,15 @@ protected void dropTableImpl(Identifier identifier, List<Path> externalPaths) {
             Path path = getTableLocation(identifier);
             try {
                 if (fileIO.exists(path)) {
-                    fileIO.deleteDirectoryQuietly(path);
+                    try {
+                        LOG.info("Move to trash, path: {}", path);
+                        fileIO.moveToTrash(path);
+                    } catch (UnsupportedOperationException e) {
+                        LOG.warn(
+                                "Catch UnsupportedOperationException and delete directory quietly, path: {}",
+                                path);
+                        fileIO.deleteDirectoryQuietly(path);
+                    }
                 }
                 for (Path externalPath : externalPaths) {
                     if (fileIO.exists(externalPath)) {