fix

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/format/FormatWriter.java

@@ -60,7 +60,7 @@ public interface FormatWriter extends Closeable {
      *
      * @throws UnsupportedOperationException if the format does not support appending files.
      */
-    default void appendFile(FileIO fileIO, Path sourcePath) throws IOException {
+    default void appendFile(FileIO fileIO, Path sourcePath, long fileLength) throws IOException {
         throw new UnsupportedOperationException(
                 getClass().getSimpleName() + " does not support appendFile");
     }

paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java

@@ -403,7 +403,7 @@ public void writeBundle(BundleRecords bundle) throws IOException {
     }
 
     @Override
-    public void appendFile(FileIO fileIO, Path sourcePath, long recordCount) {
+    public void appendFile(FileIO fileIO, Path sourcePath, long fileLength, long recordCount) {
         throw new UnsupportedOperationException(
                 getClass().getSimpleName() + " does not support appendFile");
     }

paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriter.java

@@ -46,7 +46,8 @@ public interface RollingFileWriter<T, R> extends FileWriter<T, List<R>> {
 
     void writeBundle(BundleRecords records) throws IOException;
 
-    void appendFile(FileIO fileIO, Path sourcePath, long recordCount) throws IOException;
+    void appendFile(FileIO fileIO, Path sourcePath, long fileLength, long recordCount)
+            throws IOException;
 
     @VisibleForTesting
     static FileWriterContext createFileWriterContext(

paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java

@@ -119,14 +119,14 @@ public void writeBundle(BundleRecords bundle) throws IOException {
     }
 
     @Override
-    public void appendFile(FileIO fileIO, Path sourcePath, long sourceRecordCount)
+    public void appendFile(FileIO fileIO, Path sourcePath, long fileLength, long sourceRecordCount)
             throws IOException {
         try {
             if (currentWriter == null) {
                 openCurrentWriter();
             }
 
-            currentWriter.appendFile(fileIO, sourcePath, sourceRecordCount);
+            currentWriter.appendFile(fileIO, sourcePath, fileLength, sourceRecordCount);
             recordCount += sourceRecordCount;
 
             if (rollingFile(true)) {

paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java

@@ -89,9 +89,10 @@ public void write(InternalRow row) throws IOException {
     }
 
     @Override
-    public void appendFile(FileIO sourceFileIO, Path sourcePath, long sourceRecordCount)
+    public void appendFile(
+            FileIO sourceFileIO, Path sourcePath, long fileLength, long sourceRecordCount)
             throws IOException {
-        super.appendFile(sourceFileIO, sourcePath, sourceRecordCount);
+        super.appendFile(sourceFileIO, sourcePath, fileLength, sourceRecordCount);
         seqNumCounter.add(sourceRecordCount);
     }
 

paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java

@@ -158,14 +158,15 @@ public long recordCount() {
         return recordCount;
     }
 
-    public void appendFile(FileIO sourceFileIO, Path sourcePath, long sourceRecordCount)
+    public void appendFile(
+            FileIO sourceFileIO, Path sourcePath, long fileLength, long sourceRecordCount)
             throws IOException {
         if (closed) {
             throw new RuntimeException("Writer has already closed!");
         }
 
         try {
-            writer.appendFile(sourceFileIO, sourcePath);
+            writer.appendFile(sourceFileIO, sourcePath, fileLength);
             recordCount += sourceRecordCount;
         } catch (Throwable e) {
             LOG.warn("Exception occurs when writing file {}. Cleaning up.", path, e);

paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java

@@ -220,7 +220,8 @@ public void testAppendFile() throws IOException {
                         Long.MAX_VALUE);
 
         for (int f = 0; f < 3; f++) {
-            writer.appendFile(fileIO, sourcePaths[f], recordsPerFile);
+            long fileLength = fileIO.getFileSize(sourcePaths[f]);
+            writer.appendFile(fileIO, sourcePaths[f], fileLength, recordsPerFile);
         }
         writer.close();
 

paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetBulkWriter.java

@@ -59,9 +59,8 @@ public void addElement(InternalRow datum) throws IOException {
     }
 
     @Override
-    public void appendFile(FileIO fileIO, Path sourcePath) throws IOException {
-        long fileLen = fileIO.getFileSize(sourcePath);
-        ParquetInputFile inputFile = ParquetInputFile.fromPath(fileIO, sourcePath, fileLen);
+    public void appendFile(FileIO fileIO, Path sourcePath, long fileLength) throws IOException {
+        ParquetInputFile inputFile = ParquetInputFile.fromPath(fileIO, sourcePath, fileLength);
         try (ParquetFileReader reader =
                 new ParquetFileReader(inputFile, ParquetReadOptions.builder().build(), null)) {
             reader.appendTo(parquetWriter.getFileWriter());