[core] Enable bundle writes in append-only compaction

paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRowIterator.java

@@ -115,6 +115,11 @@ public ColumnarRowIterator copy(ColumnVector[] vectors) {
 
     public ColumnarRowIterator mapping(
             @Nullable PartitionInfo partitionInfo, @Nullable int[] indexMapping) {
+        // Preserve concrete iterator capabilities when a full identity mapping changes no columns.
+        if (partitionInfo == null
+                && isFullIdentityMapping(indexMapping, row.batch().columns.length)) {
+            return this;
+        }
         if (partitionInfo != null || indexMapping != null) {
             VectorizedColumnBatch vectorizedColumnBatch = row.batch();
             ColumnVector[] vectors = vectorizedColumnBatch.columns;
@@ -129,6 +134,18 @@ public ColumnarRowIterator mapping(
         return this;
     }
 
+    private static boolean isFullIdentityMapping(@Nullable int[] indexMapping, int columnCount) {
+        if (indexMapping == null || indexMapping.length != columnCount) {
+            return false;
+        }
+        for (int i = 0; i < indexMapping.length; i++) {
+            if (indexMapping[i] != i) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     public ColumnarRowIterator assignRowTracking(
             Long firstRowId, Long snapshotId, Map<String, Integer> meta) {
         VectorizedColumnBatch vectorizedColumnBatch = row.batch();

paimon-common/src/main/java/org/apache/paimon/reader/BundleRecordIterator.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.reader;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.BundleRecords;
+
+/** A record iterator whose complete remaining logical rows can be exposed as a bundle. */
+public interface BundleRecordIterator extends RecordReader.RecordIterator<InternalRow> {
+
+    /**
+     * Returns all unconsumed logical rows as bundle records.
+     *
+     * <p>This method must be called before the first {@link #next()} invocation. Once called,
+     * {@code next()} must not be used for the same batch. The returned bundle must be consumed
+     * synchronously before {@link #releaseBatch()}.
+     */
+    BundleRecords bundleRecords();
+}

paimon-common/src/test/java/org/apache/paimon/data/columnar/ColumnarRowIteratorTest.java

@@ -18,8 +18,16 @@
 
 package org.apache.paimon.data.columnar;
 
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.PartitionInfo;
 import org.apache.paimon.data.columnar.heap.HeapIntVector;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.BundleRecords;
+import org.apache.paimon.io.VectorizedBundleRecords;
+import org.apache.paimon.reader.BundleRecordIterator;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.LongIterator;
 
 import org.junit.jupiter.api.Test;
@@ -62,4 +70,95 @@ public void testRowIterator() {
             assertThat(rowIterator.returnedPosition()).isEqualTo(positions[rowIterator.index - 1]);
         }
     }
+
+    @Test
+    public void testIdentityMappingKeepsBundleIterator() {
+        HeapIntVector heapIntVector = new HeapIntVector(1);
+        VectorizedColumnBatch vectorizedColumnBatch =
+                new VectorizedColumnBatch(new ColumnVector[] {heapIntVector});
+        vectorizedColumnBatch.setNumRows(1);
+        ColumnarRowIterator rowIterator =
+                new TestingBundleColumnarRowIterator(vectorizedColumnBatch);
+        rowIterator.reset(0);
+
+        assertThat(rowIterator.mapping(null, new int[] {0})).isSameAs(rowIterator);
+    }
+
+    @Test
+    public void testNonIdentityMappingCopiesAndReordersIterator() {
+        HeapIntVector first = new HeapIntVector(1);
+        first.setInt(0, 1);
+        HeapIntVector second = new HeapIntVector(1);
+        second.setInt(0, 2);
+        VectorizedColumnBatch vectorizedColumnBatch =
+                new VectorizedColumnBatch(new ColumnVector[] {first, second});
+        vectorizedColumnBatch.setNumRows(1);
+        ColumnarRowIterator rowIterator =
+                new ColumnarRowIterator(
+                        new Path("test"), new ColumnarRow(vectorizedColumnBatch), null);
+        rowIterator.reset(0);
+
+        ColumnarRowIterator mapped = rowIterator.mapping(null, new int[] {1, 0});
+
+        assertThat(mapped).isNotSameAs(rowIterator);
+        assertThat(mapped.next().getInt(0)).isEqualTo(2);
+        assertThat(mapped.next()).isNull();
+    }
+
+    @Test
+    public void testIdentityPrefixMappingStillCopiesIterator() {
+        HeapIntVector first = new HeapIntVector(1);
+        first.setInt(0, 1);
+        HeapIntVector second = new HeapIntVector(1);
+        second.setInt(0, 2);
+        VectorizedColumnBatch vectorizedColumnBatch =
+                new VectorizedColumnBatch(new ColumnVector[] {first, second});
+        vectorizedColumnBatch.setNumRows(1);
+        ColumnarRowIterator rowIterator =
+                new ColumnarRowIterator(
+                        new Path("test"), new ColumnarRow(vectorizedColumnBatch), null);
+        rowIterator.reset(0);
+
+        ColumnarRowIterator mapped = rowIterator.mapping(null, new int[] {0});
+
+        assertThat(mapped).isNotSameAs(rowIterator);
+        assertThat(mapped.next().getFieldCount()).isEqualTo(1);
+    }
+
+    @Test
+    public void testPartitionMappingStillCopiesIterator() {
+        HeapIntVector dataVector = new HeapIntVector(1);
+        dataVector.setInt(0, 1);
+        VectorizedColumnBatch vectorizedColumnBatch =
+                new VectorizedColumnBatch(new ColumnVector[] {dataVector});
+        vectorizedColumnBatch.setNumRows(1);
+        ColumnarRowIterator rowIterator =
+                new TestingBundleColumnarRowIterator(vectorizedColumnBatch);
+        rowIterator.reset(0);
+
+        BinaryRow partition = new BinaryRow(1);
+        BinaryRowWriter partitionWriter = new BinaryRowWriter(partition);
+        partitionWriter.writeInt(0, 9);
+        partitionWriter.complete();
+        PartitionInfo partitionInfo =
+                new PartitionInfo(new int[] {-1, 1, 0}, RowType.of(DataTypes.INT()), partition);
+
+        ColumnarRowIterator mapped = rowIterator.mapping(partitionInfo, new int[] {0});
+
+        assertThat(mapped).isNotSameAs(rowIterator);
+        assertThat(mapped.next().getInt(0)).isEqualTo(9);
+    }
+
+    private static class TestingBundleColumnarRowIterator extends ColumnarRowIterator
+            implements BundleRecordIterator {
+
+        private TestingBundleColumnarRowIterator(VectorizedColumnBatch batch) {
+            super(new Path("test"), new ColumnarRow(batch), null);
+        }
+
+        @Override
+        public BundleRecords bundleRecords() {
+            return new VectorizedBundleRecords(batch(), null);
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.AppendOnlyFileStore;
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.append.AppendOnlyWriter;
 import org.apache.paimon.append.cluster.Sorter;
 import org.apache.paimon.compact.CompactManager;
@@ -34,10 +35,13 @@
 import org.apache.paimon.io.BundleRecords;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.RollingFileWriter;
 import org.apache.paimon.io.RowDataRollingFileWriter;
 import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.metrics.BlobFetchMetrics;
+import org.apache.paimon.reader.BundleRecordIterator;
+import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.reader.RecordReaderIterator;
 import org.apache.paimon.statistics.SimpleColStatsCollector;
 import org.apache.paimon.types.DataField;
@@ -292,7 +296,9 @@ public List<DataFileMeta> compactRewrite(
             }
         }
         try {
-            rewriter.write(createFilesIterator(partition, bucket, toCompact, dvFactories));
+            writeCompactReader(
+                    readForCompact.createReader(partition, bucket, toCompact, dvFactories),
+                    rewriter);
         } catch (Exception e) {
             collectedExceptions = e;
         } finally {
@@ -308,6 +314,33 @@ public List<DataFileMeta> compactRewrite(
         return rewriter.result();
     }
 
+    private static void writeCompactReader(
+            RecordReader<InternalRow> reader, RollingFileWriter<InternalRow, ?> rewriter)
+            throws Exception {
+        try {
+            RecordReader.RecordIterator<InternalRow> batch;
+            while ((batch = reader.readBatch()) != null) {
+                try {
+                    if (batch instanceof BundleRecordIterator) {
+                        BundleRecords bundle = ((BundleRecordIterator) batch).bundleRecords();
+                        if (bundle.rowCount() > 0) {
+                            rewriter.writeBundle(bundle);
+                        }
+                    } else {
+                        InternalRow row;
+                        while ((row = batch.next()) != null) {
+                            rewriter.write(row);
+                        }
+                    }
+                } finally {
+                    batch.releaseBatch();
+                }
+            }
+        } finally {
+            reader.close();
+        }
+    }
+
     public List<DataFileMeta> clusterRewrite(
             BinaryRow partition, int bucket, List<DataFileMeta> toCluster) throws Exception {
         RecordReaderIterator<InternalRow> reader =
@@ -346,7 +379,8 @@ public List<DataFileMeta> clusterRewrite(
         return rewriter.result();
     }
 
-    private RowDataRollingFileWriter createRollingFileWriter(
+    @VisibleForTesting
+    RowDataRollingFileWriter createRollingFileWriter(
             BinaryRow partition, int bucket, Supplier<LongCounter> seqNumCounterSupplier) {
         return new RowDataRollingFileWriter(
                 fileIO,