[core] Introduce 'index-file-in-data-file-dir' to put index files in data dir

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -542,6 +542,12 @@
             <td>String</td>
             <td>Used to specify the end tag (inclusive), and Paimon will find an earlier tag and return changes between them. If the tag doesn't exist or the earlier tag doesn't exist, return empty. This option requires 'tag.creation-period' and 'tag.period-formatter' configured.</td>
         </tr>
+        <tr>
+            <td><h5>index-file-in-data-file-dir</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether index file in data file directory.</td>
+        </tr>
         <tr>
             <td><h5>local-merge-buffer-size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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

@@ -1943,6 +1943,13 @@ public InlineElement getDescription() {
                     .noDefaultValue()
                     .withDescription("Whether ignore empty commit.");
 
+    @Immutable
+    public static final ConfigOption<Boolean> INDEX_FILE_IN_DATA_FILE_DIR =
+            key("index-file-in-data-file-dir")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether index file in data file directory.");
+
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
@@ -2484,6 +2491,10 @@ public boolean disableExplicitTypeCasting() {
         return options.get(DISABLE_EXPLICIT_TYPE_CASTING);
     }
 
+    public boolean indexFileInDataFileDir() {
+        return options.get(INDEX_FILE_IN_DATA_FILE_DIR);
+    }
+
     public LookupStrategy lookupStrategy() {
         return LookupStrategy.from(
                 mergeEngine().equals(MergeEngine.FIRST_ROW),

paimon-common/src/main/java/org/apache/paimon/utils/IntFileUtils.java

@@ -46,15 +46,17 @@ public void close() throws IOException {
         };
     }
 
-    public static void writeInts(FileIO fileIO, Path path, IntIterator input) throws IOException {
+    public static int writeInts(FileIO fileIO, Path path, IntIterator input) throws IOException {
         try (FastBufferedOutputStream out =
                         new FastBufferedOutputStream(fileIO.newOutputStream(path, false));
                 IntIterator iterator = input) {
+            int count = 0;
             while (true) {
                 try {
                     writeInt(out, iterator.next());
+                    count++;
                 } catch (EOFException ignored) {
-                    break;
+                    return count;
                 }
             }
         }

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -131,7 +131,8 @@ protected FileStorePathFactory pathFactory(CoreOptions options, String format) {
                 options.fileSuffixIncludeCompression(),
                 options.fileCompression(),
                 options.dataFilePathDirectory(),
-                createExternalPaths());
+                createExternalPaths(),
+                options.indexFileInDataFileDir());
     }
 
     private List<Path> createExternalPaths() {

paimon-core/src/main/java/org/apache/paimon/compact/CompactDeletionFile.java

@@ -92,7 +92,7 @@ public CompactDeletionFile mergeOldFile(CompactDeletionFile old) {
         @Override
         public void clean() {
             if (deletionFile != null) {
-                dvIndexFile.delete(deletionFile.fileName());
+                dvIndexFile.delete(deletionFile);
             }
         }
     }

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionFileWriter.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.index.DeletionVectorMeta;
 import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
 
 import java.io.Closeable;
 import java.io.DataOutputStream;
@@ -35,11 +36,13 @@
 public class DeletionFileWriter implements Closeable {
 
     private final Path path;
+    private final boolean isExternalPath;
     private final DataOutputStream out;
     private final LinkedHashMap<String, DeletionVectorMeta> dvMetas;
 
-    public DeletionFileWriter(Path path, FileIO fileIO) throws IOException {
-        this.path = path;
+    public DeletionFileWriter(IndexPathFactory pathFactory, FileIO fileIO) throws IOException {
+        this.path = pathFactory.newPath();
+        this.isExternalPath = pathFactory.isExternalPath();
         this.out = new DataOutputStream(fileIO.newOutputStream(path, true));
         out.writeByte(VERSION_ID_V1);
         this.dvMetas = new LinkedHashMap<>();
@@ -63,6 +66,11 @@ public void close() throws IOException {
 
     public IndexFileMeta result() {
         return new IndexFileMeta(
-                DELETION_VECTORS_INDEX, path.getName(), getPos(), dvMetas.size(), dvMetas);
+                DELETION_VECTORS_INDEX,
+                path.getName(),
+                getPos(),
+                dvMetas.size(),
+                dvMetas,
+                isExternalPath ? path.toString() : null);
     }
 }

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileWriter.java

@@ -20,8 +20,8 @@
 
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
 import org.apache.paimon.options.MemorySize;
-import org.apache.paimon.utils.PathFactory;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -33,12 +33,12 @@
 /** Writer for deletion vector index file. */
 public class DeletionVectorIndexFileWriter {
 
-    private final PathFactory indexPathFactory;
+    private final IndexPathFactory indexPathFactory;
     private final FileIO fileIO;
     private final long targetSizeInBytes;
 
     public DeletionVectorIndexFileWriter(
-            FileIO fileIO, PathFactory pathFactory, MemorySize targetSizePerIndexFile) {
+            FileIO fileIO, IndexPathFactory pathFactory, MemorySize targetSizePerIndexFile) {
         this.indexPathFactory = pathFactory;
         this.fileIO = fileIO;
         this.targetSizeInBytes = targetSizePerIndexFile.getBytes();
@@ -53,7 +53,7 @@ public DeletionVectorIndexFileWriter(
      */
     public IndexFileMeta writeSingleFile(Map<String, DeletionVector> input) throws IOException {
 
-        DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory.newPath(), fileIO);
+        DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory, fileIO);
         try {
             for (Map.Entry<String, DeletionVector> entry : input.entrySet()) {
                 writer.write(entry.getKey(), entry.getValue());
@@ -79,7 +79,7 @@ public List<IndexFileMeta> writeWithRolling(Map<String, DeletionVector> input)
 
     private IndexFileMeta tryWriter(Iterator<Map.Entry<String, DeletionVector>> iterator)
             throws IOException {
-        DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory.newPath(), fileIO);
+        DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory, fileIO);
         try {
             while (iterator.hasNext()) {
                 Map.Entry<String, DeletionVector> entry = iterator.next();

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorsIndexFile.java

@@ -24,9 +24,9 @@
 import org.apache.paimon.index.DeletionVectorMeta;
 import org.apache.paimon.index.IndexFile;
 import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.table.source.DeletionFile;
-import org.apache.paimon.utils.PathFactory;
 
 import java.io.DataInputStream;
 import java.io.IOException;
@@ -51,7 +51,7 @@ public class DeletionVectorsIndexFile extends IndexFile {
 
     public DeletionVectorsIndexFile(
             FileIO fileIO,
-            PathFactory pathFactory,
+            IndexPathFactory pathFactory,
             MemorySize targetSizePerIndexFile,
             boolean bitmap64) {
         super(fileIO, pathFactory);
@@ -71,13 +71,11 @@ public boolean bitmap64() {
      * @throws UncheckedIOException If an I/O error occurs while reading from the file.
      */
     public Map<String, DeletionVector> readAllDeletionVectors(IndexFileMeta fileMeta) {
-        LinkedHashMap<String, DeletionVectorMeta> deletionVectorMetas =
-                fileMeta.deletionVectorMetas();
+        LinkedHashMap<String, DeletionVectorMeta> deletionVectorMetas = fileMeta.dvRanges();
         checkNotNull(deletionVectorMetas);
 
-        String indexFileName = fileMeta.fileName();
         Map<String, DeletionVector> deletionVectors = new HashMap<>();
-        Path filePath = pathFactory.toPath(indexFileName);
+        Path filePath = pathFactory.toPath(fileMeta);
         try (SeekableInputStream inputStream = fileIO.newInputStream(filePath)) {
             checkVersion(inputStream);
             DataInputStream dataInputStream = new DataInputStream(inputStream);

paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java

@@ -82,8 +82,7 @@ static AppendDeleteFileMaintainer forUnawareAppend(
                         .collect(Collectors.toList());
         Map<String, DeletionFile> deletionFiles = new HashMap<>();
         for (IndexManifestEntry file : manifestEntries) {
-            LinkedHashMap<String, DeletionVectorMeta> dvMetas =
-                    file.indexFile().deletionVectorMetas();
+            LinkedHashMap<String, DeletionVectorMeta> dvMetas = file.indexFile().dvRanges();
             checkNotNull(dvMetas);
             for (DeletionVectorMeta dvMeta : dvMetas.values()) {
                 deletionFiles.put(

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -1147,8 +1147,7 @@ private List<IcebergManifestFileMeta> createDvManifestFileMetas(Snapshot snapsho
             return Collections.emptyList();
         }
         for (IndexManifestEntry entry : newIndexes) {
-            LinkedHashMap<String, DeletionVectorMeta> dvMetas =
-                    entry.indexFile().deletionVectorMetas();
+            LinkedHashMap<String, DeletionVectorMeta> dvMetas = entry.indexFile().dvRanges();
             Path bucketPath = fileStorePathFactory.bucketPath(entry.partition(), entry.bucket());
             if (dvMetas != null) {
                 for (DeletionVectorMeta dvMeta : dvMetas.values()) {