[core] Support Parquet row-group copy fast path for append-only compaction

docs/docs/append-table/index.mdx

@@ -99,6 +99,22 @@ If you set `write-only` to true, the `Compact Coordinator` and `Compact Worker`
 The auto compaction is only supported in Flink engine streaming mode. You can also start a compaction job in Flink by
 Flink action in Paimon and disable all the other compactions by setting `write-only`.
 
+**Parquet RowGroup copy fast path**
+
+For Parquet files, compaction normally decodes every row and re-encodes it into new files. By setting
+`append.compaction.row-group-copy.enabled` to true, compaction instead concatenates the compressed RowGroups of the
+input files directly and only rewrites the file footer, which avoids decode/re-encode entirely and significantly
+reduces compaction cost. The fast path is applied only when all input files are eligible (same schema, same codec,
+no deletion vectors, no row tracking, no file index or bloom filter, not written with Parquet writer v2, etc.);
+otherwise compaction silently falls back to the traditional rewrite path.
+
+Two related options:
+- `append.compaction.row-group-copy.preserve-page-index` (default false): preserve the ColumnIndex/OffsetIndex of
+  input files so that page-level predicate pruning keeps working on compacted files, at the cost of reading and
+  rewriting page indexes during compaction.
+- `append.compaction.row-group-copy.footer-read.parallelism` (default 1): maximum number of footer metadata reads
+  performed in parallel when preparing one compaction batch.
+
 **Streaming Query**
 
 You can stream the Append table and use it like a Message Queue. As with primary key tables, there are two options

docs/generated/core_configuration.html

@@ -44,6 +44,24 @@
             <td>Boolean</td>
             <td>If true, it disables altering column type from null to not null. Default is true. Users can disable this option to explicitly convert null column type to not null.</td>
         </tr>
+        <tr>
+            <td><h5>append.compaction.row-group-copy.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enable Parquet RowGroup copy fast path for append-only table compaction. When enabled and all whitelist conditions are met, compaction concatenates compressed RowGroups without decoding and re-encoding rows.</td>
+        </tr>
+        <tr>
+            <td><h5>append.compaction.row-group-copy.footer-read.parallelism</h5></td>
+            <td style="word-wrap: break-word;">1</td>
+            <td>Integer</td>
+            <td>Maximum number of Parquet footer metadata reads that a single RowGroup copy compaction batch may perform in parallel. The effective parallelism is capped by the input file count and an internal hard limit of 8. The default value 1 keeps footer reads serial.</td>
+        </tr>
+        <tr>
+            <td><h5>append.compaction.row-group-copy.preserve-page-index</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether Parquet RowGroup copy compaction preserves existing ColumnIndex and OffsetIndex metadata. This keeps page-level predicate pruning at the cost of reading and rewriting the page indexes during compaction.</td>
+        </tr>
         <tr>
             <td><h5>async-file-write</h5></td>
             <td style="word-wrap: break-word;">true</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1075,6 +1075,38 @@ public InlineElement getDescription() {
                             "Ratio of the deleted rows in a data file to be forced compacted for "
                                     + "append-only table.");
 
+    public static final ConfigOption<Boolean> APPEND_COMPACTION_ROW_GROUP_COPY_ENABLED =
+            key("append.compaction.row-group-copy.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to enable Parquet RowGroup copy fast path for append-only "
+                                    + "table compaction. When enabled and all whitelist conditions "
+                                    + "are met, compaction concatenates compressed RowGroups "
+                                    + "without decoding and re-encoding rows.");
+
+    public static final ConfigOption<Boolean> APPEND_COMPACTION_ROW_GROUP_COPY_PRESERVE_PAGE_INDEX =
+            key("append.compaction.row-group-copy.preserve-page-index")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether Parquet RowGroup copy compaction preserves existing "
+                                    + "ColumnIndex and OffsetIndex metadata. This keeps "
+                                    + "page-level predicate pruning at the cost of reading "
+                                    + "and rewriting the page indexes during compaction.");
+
+    public static final ConfigOption<Integer>
+            APPEND_COMPACTION_ROW_GROUP_COPY_FOOTER_READ_PARALLELISM =
+                    key("append.compaction.row-group-copy.footer-read.parallelism")
+                            .intType()
+                            .defaultValue(1)
+                            .withDescription(
+                                    "Maximum number of Parquet footer metadata reads that a single "
+                                            + "RowGroup copy compaction batch may perform in "
+                                            + "parallel. The effective parallelism is capped by "
+                                            + "the input file count and an internal hard limit of "
+                                            + "8. The default value 1 keeps footer reads serial.");
+
     public static final ConfigOption<ChangelogProducer> CHANGELOG_PRODUCER =
             key("changelog-producer")
                     .enumType(ChangelogProducer.class)
@@ -3821,6 +3853,19 @@ public double compactionDeleteRatioThreshold() {
         return options.get(COMPACTION_DELETE_RATIO_THRESHOLD);
     }
 
+    public boolean appendCompactionRowGroupCopyEnabled() {
+        return options.get(APPEND_COMPACTION_ROW_GROUP_COPY_ENABLED);
+    }
+
+    public boolean appendCompactionRowGroupCopyPreservePageIndex() {
+        return options.get(APPEND_COMPACTION_ROW_GROUP_COPY_PRESERVE_PAGE_INDEX);
+    }
+
+    public int appendCompactionRowGroupCopyFooterReadParallelism() {
+        int configured = options.get(APPEND_COMPACTION_ROW_GROUP_COPY_FOOTER_READ_PARALLELISM);
+        return Math.max(1, Math.min(configured, 8));
+    }
+
     public long dynamicBucketTargetRowNum() {
         return options.get(DYNAMIC_BUCKET_TARGET_ROW_NUM);
     }