[core] Skip file index for changelog files (apache#7697)

Changelog files are sequentially consumed by streaming readers, file
index (bloom filter/bitmap) provides no benefit. Skip generating .index
files for changelog to avoid unnecessary storage overhead and potential
orphan files after changelog expiration. These orphan .index files
increase I/O pressure on the orphan file clean job.

Author: XiaoHongbo-Hope

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

@@ -149,6 +149,8 @@ private KeyValueClusteringFileWriter createKvSeparatedFileWriter(
 
     private KeyValueDataFileWriter createDataFileWriter(
             Path path, WriteFormatKey key, FileSource fileSource, boolean isExternalPath) {
+        // Changelog is sequentially consumed, file index is unnecessary.
+        FileIndexOptions indexOptions = key.isChangelog ? new FileIndexOptions() : fileIndexOptions;
         return formatContext.thinModeEnabled
                 ? new KeyValueThinDataFileWriterImpl(
                         fileIO,
@@ -161,7 +163,7 @@ private KeyValueDataFileWriter createDataFileWriter(
                         key.level,
                         options,
                         fileSource,
-                        fileIndexOptions,
+                        indexOptions,
                         isExternalPath)
                 : new KeyValueDataFileWriterImpl(
                         fileIO,
@@ -174,7 +176,7 @@ private KeyValueDataFileWriter createDataFileWriter(
                         key.level,
                         options,
                         fileSource,
-                        fileIndexOptions,
+                        indexOptions,
                         isExternalPath);
     }
 

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

@@ -308,11 +308,16 @@ public void testFileSuffix(@TempDir java.nio.file.Path tempDir) throws Exception
     }
 
     protected KeyValueFileWriterFactory createWriterFactory(String pathStr, String format) {
+        Options options = new Options();
+        options.set(CoreOptions.METADATA_STATS_MODE, "FULL");
+        return createWriterFactory(pathStr, format, options);
+    }
+
+    protected KeyValueFileWriterFactory createWriterFactory(
+            String pathStr, String format, Options options) {
         Path path = new Path(pathStr);
         int suggestedFileSize = ThreadLocalRandom.current().nextInt(8192) + 1024;
         FileIO fileIO = FileIOFinder.find(path);
-        Options options = new Options();
-        options.set(CoreOptions.METADATA_STATS_MODE, "FULL");
 
         Function<String, FileStorePathFactory> pathFactoryMap =
                 format1 ->
@@ -455,6 +460,49 @@ private void checkRollingFiles(
         }
     }
 
+    @Test
+    void testChangelogFile() throws Exception {
+        Options options = new Options();
+        options.set(CoreOptions.METADATA_STATS_MODE, "FULL");
+        options.setString("file-index.bloom-filter.columns", "comment");
+        options.setString("file-index.in-manifest-threshold", "1B");
+
+        KeyValueFileWriterFactory writerFactory =
+                createWriterFactory(tempDir.toString(), "avro", options);
+
+        DataFileTestDataGenerator.Data data = gen.next();
+        RollingFileWriter<KeyValue, DataFileMeta> dataWriter =
+                writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
+        dataWriter.write(CloseableIterator.fromList(data.content, kv -> {}));
+        dataWriter.close();
+        List<DataFileMeta> dataFileMetas = dataWriter.result();
+
+        assertThat(dataFileMetas).isNotEmpty();
+        assertThat(
+                        dataFileMetas.stream()
+                                .anyMatch(
+                                        meta ->
+                                                meta.extraFiles().stream()
+                                                        .anyMatch(
+                                                                f ->
+                                                                        f.endsWith(
+                                                                                DataFilePathFactory
+                                                                                        .INDEX_PATH_SUFFIX))))
+                .isTrue();
+
+        RollingFileWriter<KeyValue, DataFileMeta> changelogWriter =
+                writerFactory.createRollingChangelogFileWriter(0);
+        changelogWriter.write(CloseableIterator.fromList(data.content, kv -> {}));
+        changelogWriter.close();
+        List<DataFileMeta> changelogMetas = changelogWriter.result();
+
+        assertThat(changelogMetas).isNotEmpty();
+        for (DataFileMeta meta : changelogMetas) {
+            assertThat(meta.extraFiles())
+                    .noneMatch(f -> f.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX));
+        }
+    }
+
     @ParameterizedTest
     @ValueSource(strings = {"parquet", "orc", "avro"})
     public void testReaderUseFileSizeFromMetadata(String format) throws Exception {