[core] Generate sidecars for newly written manifests

docs/docs/concepts/spec/manifest.md

@@ -71,11 +71,14 @@ using independent partition, row-ID and bucket coverage. A sidecar uses the
 `.avro.sidecar` reference in the manifest metadata's `_EXTRA_FILES`, without probing a
 derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged.
 
-The utility includes construction, validation, block selection and optional caching. Table
-writers and scans do not yet invoke it automatically. Callers are responsible for publishing
-sidecar references, managing file ownership, applying entry filters and reconciling ADD/DELETE
-entries after block selection. `build` reads the completed physical manifest and returns
-sidecar bytes; it does not write or publish another file.
+The utility includes construction, validation, block selection and optional caching. Java table
+writers generate sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits
+`manifest-sort.enabled`. Both ordinary writes and raw manifest rewrites build the sidecar from
+the completed output manifest and publish its `_EXTRA_FILES` reference only after both files
+close successfully. Failed writes and aborted writers clean up their own manifest/sidecar pairs.
+Scans do not yet invoke sidecar pruning automatically. Callers remain responsible for applying
+entry filters and reconciling ADD/DELETE entries after block selection. The low-level `build`
+method returns sidecar bytes without writing or publishing another file.
 
 Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches.
 `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent

docs/generated/core_configuration.html

@@ -1113,6 +1113,12 @@
             <td>Boolean</td>
             <td>Whether to skip automatic manifest merging during commit when write-only is true. This also skips automatic manifest sort rewrite. Explicit manifest compaction is not affected.</td>
         </tr>
+        <tr>
+            <td><h5>manifest.sidecar.enabled</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Boolean</td>
+            <td>Whether to enable manifest sidecars with independent partition, row-id and bucket coverage. Defaults to manifest-sort.enabled when unset.</td>
+        </tr>
         <tr>
             <td><h5>manifest.target-file-size</h5></td>
             <td style="word-wrap: break-word;">8 mb</td>

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

@@ -522,6 +522,13 @@ public InlineElement getDescription() {
                     .defaultValue(MemorySize.ofMebiBytes(8))
                     .withDescription("Suggested file size of a manifest file.");
 
+    public static final ConfigOption<Boolean> MANIFEST_SIDECAR_ENABLED =
+            key("manifest.sidecar.enabled")
+                    .booleanType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Whether to enable manifest sidecars with independent partition, row-id and bucket coverage. Defaults to manifest-sort.enabled when unset.");
+
     public static final ConfigOption<MemorySize> MANIFEST_FULL_COMPACTION_FILE_SIZE =
             key("manifest.full-compaction-threshold-size")
                     .memoryType()
@@ -3217,6 +3224,10 @@ public MemorySize manifestTargetSize() {
         return options.get(MANIFEST_TARGET_FILE_SIZE);
     }
 
+    public boolean manifestSidecarEnabled() {
+        return options.getOptional(MANIFEST_SIDECAR_ENABLED).orElseGet(this::manifestSortEnabled);
+    }
+
     public MemorySize manifestFullCompactionThresholdSize() {
         return options.get(MANIFEST_FULL_COMPACTION_FILE_SIZE);
     }

paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java

@@ -300,7 +300,8 @@ private ManifestFile createManifestFile() {
                         "zstd",
                         pathFactory,
                         TARGET_MANIFEST_SIZE,
-                        null)
+                        null,
+                        new CoreOptions(new Options()))
                 .create();
     }
 

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

@@ -211,7 +211,8 @@ public ManifestFile.Factory manifestFileFactory() {
                 options.manifestCompression(),
                 pathFactory(),
                 options.manifestTargetSize().getBytes(),
-                readManifestCache);
+                readManifestCache,
+                options);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.manifest;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.SimpleColStats;
@@ -43,6 +44,7 @@
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
@@ -68,6 +70,7 @@ public final class ManifestAvroWriter implements AutoCloseable {
     private final String compression;
     private final PathFactory pathFactory;
     private final long targetFileSize;
+    private final CoreOptions options;
 
     private final List<ManifestFileMeta> results = new ArrayList<>();
     private final List<Path> completedPaths = new ArrayList<>();
@@ -83,7 +86,8 @@ public final class ManifestAvroWriter implements AutoCloseable {
             ObjectSerializer<ManifestEntry> serializer,
             String compression,
             PathFactory pathFactory,
-            long targetFileSize) {
+            long targetFileSize,
+            CoreOptions options) {
         this.fileIO = fileIO;
         this.schemaManager = schemaManager;
         this.partitionType = partitionType;
@@ -92,6 +96,7 @@ public final class ManifestAvroWriter implements AutoCloseable {
         this.compression = compression;
         this.pathFactory = pathFactory;
         this.targetFileSize = targetFileSize;
+        this.options = options;
     }
 
     public void write(ManifestEntry entry) throws IOException {
@@ -218,6 +223,9 @@ private void closeCurrentWriter() throws IOException {
         currentWriter.close();
         ManifestFileMeta result = currentWriter.result();
         completedPaths.add(currentWriter.path);
+        if (currentWriter.sidecarCreated) {
+            completedPaths.add(ManifestSidecar.path(currentWriter.path));
+        }
         results.add(result);
         currentWriter = null;
     }
@@ -413,6 +421,7 @@ private final class FileWriter {
         private @Nullable RowIdStats rowIdStats = new RowIdStats();
         private boolean closed;
         private boolean aborted;
+        private boolean sidecarCreated;
 
         private FileWriter(Path path) {
             this.path = path;
@@ -488,7 +497,7 @@ private void collectStats(ManifestEntry entry) {
             maxLevel = Math.max(maxLevel, entry.level());
             if (rowIdStats != null) {
                 Long firstRowId = entry.file().firstRowId();
-                if (firstRowId == null) {
+                if (!validRowIdRange(firstRowId, entry.file().rowCount())) {
                     rowIdStats = null;
                 } else {
                     rowIdStats.collect(firstRowId, entry.file().rowCount());
@@ -515,7 +524,7 @@ private void collectStats(EncodedEntry entry) {
             minLevel = Math.min(minLevel, entry.level);
             maxLevel = Math.max(maxLevel, entry.level);
             if (rowIdStats != null) {
-                if (!entry.hasRowId) {
+                if (!entry.hasRowId || !validRowIdRange(entry.firstRowId, entry.rowCount)) {
                     rowIdStats = null;
                 } else {
                     rowIdStats.collect(entry.firstRowId, entry.rowCount);
@@ -697,6 +706,14 @@ private Throwable abortCollecting(@Nullable Throwable primaryFailure, boolean de
                             ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
                 }
             }
+            if (sidecarCreated) {
+                try {
+                    fileIO.deleteQuietly(ManifestSidecar.path(path));
+                } catch (Throwable cleanupFailure) {
+                    primaryFailure =
+                            ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+                }
+            }
             return primaryFailure;
         }
 
@@ -711,6 +728,7 @@ private void close() throws IOException {
                 outputBytes = out.getPos();
                 out.close();
                 out = null;
+                writeSidecar();
             } catch (IOException | RuntimeException | Error failure) {
                 abortCollecting(failure, true);
                 throw failure;
@@ -719,6 +737,26 @@ private void close() throws IOException {
             }
         }
 
+        private void writeSidecar() throws IOException {
+            if (!options.manifestSidecarEnabled()) {
+                return;
+            }
+            byte[] bytes =
+                    ManifestSidecar.build(
+                            fileIO,
+                            path,
+                            outputBytes,
+                            Math.addExact(numAddedFiles, numDeletedFiles),
+                            options.dataEvolutionEnabled(),
+                            options.bucket() != -1);
+            // Publish result() only after both immutable objects have closed. No rename.
+            try (PositionOutputStream sidecarOut =
+                    fileIO.newOutputStream(ManifestSidecar.path(path), false)) {
+                sidecarCreated = true;
+                sidecarOut.write(bytes);
+            }
+        }
+
         private ManifestFileMeta result() {
             if (!closed || outputBytes == null) {
                 throw new IllegalStateException(
@@ -740,10 +778,16 @@ private ManifestFileMeta result() {
                     rowIdStats == null ? null : rowIdStats.minRowId,
                     rowIdStats == null ? null : rowIdStats.maxRowId,
                     totalBucketsKnown ? totalBuckets : null,
-                    null);
+                    sidecarCreated
+                            ? Collections.singletonList(ManifestSidecar.path(path).getName())
+                            : null);
         }
     }
 
+    private static boolean validRowIdRange(@Nullable Long first, long count) {
+        return first != null && first >= 0 && count > 0 && count - 1 <= Long.MAX_VALUE - first;
+    }
+
     private static class RowIdStats {
 
         private long minRowId = Long.MAX_VALUE;

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.manifest;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.FileFormat;
@@ -59,6 +60,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
     private final RowType partitionType;
     private final AvroFileFormat avroFileFormat;
     private final long suggestedFileSize;
+    private final CoreOptions options;
 
     private ManifestFile(
             FileIO fileIO,
@@ -69,7 +71,8 @@ private ManifestFile(
             String compression,
             PathFactory pathFactory,
             long suggestedFileSize,
-            @Nullable SegmentsCache<Path> cache) {
+            @Nullable SegmentsCache<Path> cache,
+            CoreOptions options) {
         super(
                 fileIO,
                 serializer,
@@ -85,6 +88,7 @@ private ManifestFile(
         this.partitionType = partitionType;
         this.avroFileFormat = avroFileFormat;
         this.suggestedFileSize = suggestedFileSize;
+        this.options = options;
     }
 
     @Override
@@ -301,7 +305,8 @@ public ManifestAvroWriter createAvroWriter() {
                 serializer,
                 compression,
                 pathFactory,
-                suggestedFileSize);
+                suggestedFileSize,
+                options);
     }
 
     /** Creates an Avro manifest writer for one explicit path. */
@@ -314,7 +319,8 @@ public ManifestAvroWriter createAvroWriter(Path manifestPath) {
                 serializer,
                 compression,
                 singlePathFactory(manifestPath),
-                Long.MAX_VALUE);
+                Long.MAX_VALUE,
+                options);
     }
 
     private PathFactory singlePathFactory(Path manifestPath) {
@@ -357,6 +363,7 @@ public static class Factory {
         private final String compression;
         private final FileStorePathFactory pathFactory;
         private final long suggestedFileSize;
+        private final CoreOptions options;
         @Nullable private final SegmentsCache<Path> cache;
 
         public Factory(
@@ -367,7 +374,8 @@ public Factory(
                 String compression,
                 FileStorePathFactory pathFactory,
                 long suggestedFileSize,
-                @Nullable SegmentsCache<Path> cache) {
+                @Nullable SegmentsCache<Path> cache,
+                CoreOptions options) {
             this.fileIO = fileIO;
             this.schemaManager = schemaManager;
             this.partitionType = partitionType;
@@ -376,6 +384,7 @@ public Factory(
             this.pathFactory = pathFactory;
             this.suggestedFileSize = suggestedFileSize;
             this.cache = cache;
+            this.options = options;
         }
 
         public boolean isCacheEnabled() {
@@ -392,7 +401,8 @@ public ManifestFile create() {
                     compression,
                     pathFactory.manifestFileFactory(),
                     suggestedFileSize,
-                    cache);
+                    cache,
+                    options);
         }
     }
 }

paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java

@@ -29,6 +29,24 @@
 /** Tests for {@link org.apache.paimon.CoreOptions}. */
 public class CoreOptionsTest {
 
+    @Test
+    void testManifestSidecarDefaultsToManifestSort() {
+        assertThat(CoreOptions.MANIFEST_SIDECAR_ENABLED.defaultValue()).isNull();
+        for (Boolean sort : new Boolean[] {null, false, true}) {
+            for (Boolean configured : new Boolean[] {null, false, true}) {
+                Options options = new Options();
+                if (sort != null) {
+                    options.set(CoreOptions.MANIFEST_SORT_ENABLED, sort);
+                }
+                if (configured != null) {
+                    options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, configured);
+                }
+                assertThat(new CoreOptions(options).manifestSidecarEnabled())
+                        .isEqualTo(configured == null ? Boolean.TRUE.equals(sort) : configured);
+            }
+        }
+    }
+
     @Test
     public void testDefaultStartupMode() {
         Options conf = new Options();

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java

@@ -2744,7 +2744,8 @@ public void testManifestSortWithMultiplePartitions() {
                                         false,
                                         null),
                                 Long.MAX_VALUE,
-                                null)
+                                null,
+                                new CoreOptions(new Options()))
                         .create();
 
         List<ManifestFileMeta> input = new ArrayList<>();
@@ -3204,7 +3205,8 @@ private ManifestFile createManifestFileForPartitionType(RowType partitionType) {
                                 false,
                                 null),
                         Long.MAX_VALUE,
-                        null)
+                        null,
+                        new CoreOptions(new Options()))
                 .create();
     }
 

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java

@@ -166,7 +166,8 @@ protected ManifestFile createManifestFile(String pathStr, FileIO fileIO) {
                                 false,
                                 null),
                         Long.MAX_VALUE,
-                        null)
+                        null,
+                        new CoreOptions(new Options()))
                 .create();
     }
 

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java

@@ -1442,7 +1442,8 @@ private ManifestFile createManifestFile(
                         "zstd",
                         pathFactory,
                         suggestedFileSize,
-                        cache)
+                        cache,
+                        new CoreOptions(new Options()))
                 .create();
     }
 

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java

@@ -35,6 +35,7 @@
 
 /** Synthetic index references for manifest serialization and lifecycle tests. */
 public final class ManifestIndexTestUtils {
+
     private ManifestIndexTestUtils() {}
 
     public static ManifestFileMeta withIndexFileName(ManifestFileMeta meta, String indexFileName) {