[core] Require write-only mode for snapshot-sequence-ordering

Snapshot ordering relies on snapshot id to determine record order,
but inline compaction happens before snapshot creation — files have
not been stamped with the correct snapshot id yet. Compaction would
propagate incorrect values to KV records.

Enforce write-only=true at schema validation time so that writers
use NoopCompactManager. Dedicated compact jobs override write-only
at runtime via table.copy(), which passes dynamicOptionKeys to skip
this specific check.

Author: JunRuiLee

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1305,7 +1305,7 @@
             <td><h5>sequence.snapshot-ordering</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>When enabled, merge uses the commit snapshot id as the primary tiebreaker for primary-key conflicts: records from later snapshots always win, with the existing sequence number used as a secondary tiebreaker. Designed for multi-writer scenarios on the same primary-key table where wall-clock sequence numbers cannot be globally ordered. Mutually exclusive with sequence.field. Requires a primary-key table.</td>
+            <td>When enabled, merge uses the commit snapshot id as the primary tiebreaker for primary-key conflicts: records from later snapshots always win, with the existing sequence number used as a secondary tiebreaker. Designed for multi-writer scenarios on the same primary-key table where wall-clock sequence numbers cannot be globally ordered. Mutually exclusive with sequence.field. Requires a primary-key table with write-only=true. Inline compaction is not allowed because snapshot ids are assigned only after commit. To compact such tables, run a dedicated compaction job/action with write-only=false.</td>
         </tr>
         <tr>
             <td><h5>sink.process-time-zone</h5></td>

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

@@ -977,7 +977,11 @@ public InlineElement getDescription() {
                                     + "as a secondary tiebreaker. Designed for multi-writer "
                                     + "scenarios on the same primary-key table where wall-clock "
                                     + "sequence numbers cannot be globally ordered. Mutually "
-                                    + "exclusive with sequence.field. Requires a primary-key table.");
+                                    + "exclusive with sequence.field. Requires a primary-key table "
+                                    + "with write-only=true. Inline compaction is not allowed "
+                                    + "because snapshot ids are assigned only after commit. To "
+                                    + "compact such tables, run a dedicated compaction job/action "
+                                    + "with write-only=false.");
 
     @Immutable
     public static final ConfigOption<Boolean> AGGREGATION_REMOVE_RECORD_ON_DELETE =

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

@@ -95,6 +95,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
     private boolean closeCompactExecutorWhenLeaving = true;
     private boolean ignorePreviousFiles = false;
     private boolean ignoreNumBucketCheck = false;
+    private final boolean compactOnly;
 
     protected CompactionMetrics compactionMetrics = null;
     protected final String tableName;
@@ -123,6 +124,7 @@ protected AbstractFileStoreWrite(
         this.tableName = tableName;
         this.writerNumberMax = options.writeMaxWritersToSpill();
         this.legacyPartitionName = options.legacyPartitionName();
+        this.compactOnly = options.snapshotSequenceOrdering() && !options.writeOnly();
         this.partitionTimestampValidator =
                 PartitionTimestampValidator.create(options, partitionType);
     }
@@ -167,6 +169,15 @@ public void withCompactExecutor(ExecutorService compactExecutor) {
 
     @Override
     public void write(BinaryRow partition, int bucket, T data) throws Exception {
+        if (compactOnly) {
+            throw new IllegalStateException(
+                    CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key()
+                            + " is enabled with "
+                            + CoreOptions.WRITE_ONLY.key()
+                            + " = false. This writer only allows compaction, not data writes. "
+                            + "Inline compaction is not supported because snapshot ids are "
+                            + "assigned only after commit.");
+        }
         WriterContainer<T> container = getWriterWrapper(partition, bucket);
         container.writer.write(data);
         if (container.dynamicBucketMaintainer != null) {

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -104,6 +104,10 @@ public class SchemaValidation {
      * @param schema the schema to be validated
      */
     public static void validateTableSchema(TableSchema schema) {
+        validateTableSchema(schema, Collections.emptySet());
+    }
+
+    public static void validateTableSchema(TableSchema schema, Set<String> dynamicOptionKeys) {
         CoreOptions options = new CoreOptions(schema.options());
 
         validateOnlyContainPrimitiveType(schema.fields(), schema.primaryKeys(), "primary key");
@@ -255,7 +259,7 @@ public static void validateTableSchema(TableSchema schema) {
         }
 
         if (options.snapshotSequenceOrdering()) {
-            validateSnapshotSequenceOrdering(schema, options);
+            validateSnapshotSequenceOrdering(schema, options, dynamicOptionKeys);
         }
 
         // vector field names must point to vector type
@@ -582,7 +586,8 @@ private static void validateFileIndex(TableSchema schema) {
         }
     }
 
-    private static void validateSnapshotSequenceOrdering(TableSchema schema, CoreOptions options) {
+    private static void validateSnapshotSequenceOrdering(
+            TableSchema schema, CoreOptions options, Set<String> dynamicOptionKeys) {
         checkArgument(
                 !schema.primaryKeys().isEmpty(),
                 "%s = true requires a primary-key table; append-only tables cannot use "
@@ -593,6 +598,18 @@ private static void validateSnapshotSequenceOrdering(TableSchema schema, CoreOpt
                 "%s = true is mutually exclusive with %s; the snapshot id is the sole tiebreaker.",
                 CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(),
                 CoreOptions.SEQUENCE_FIELD.key());
+        // Skip writeOnly check when write-only is dynamically overridden (e.g. by dedicated
+        // compact jobs that override write-only=false at runtime).
+        if (!dynamicOptionKeys.contains(CoreOptions.WRITE_ONLY.key())) {
+            checkArgument(
+                    options.writeOnly(),
+                    "%s = true requires %s = true. Snapshot ordering relies on snapshot id to "
+                            + "determine record order, but inline compaction happens before "
+                            + "snapshot creation — files have not been stamped with the correct "
+                            + "snapshot id yet. Use dedicated compaction job instead.",
+                    CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(),
+                    CoreOptions.WRITE_ONLY.key());
+        }
     }
 
     private static void validateForDeletionVectors(CoreOptions options) {

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -373,7 +373,7 @@ protected FileStoreTable copyInternal(
         }
 
         // validate schema with new options
-        SchemaValidation.validateTableSchema(newTableSchema);
+        SchemaValidation.validateTableSchema(newTableSchema, dynamicOptions.keySet());
 
         return copy(newTableSchema);
     }

paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java

@@ -449,13 +449,23 @@ private void validateTableSchemaWithMapField(Map<String, String> options) {
     public void testSnapshotSequenceOrderingHappyPath() {
         Map<String, String> options = new HashMap<>();
         options.put(CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(), "true");
+        options.put(CoreOptions.WRITE_ONLY.key(), "true");
         assertThatNoException().isThrownBy(() -> validateTableSchemaExec(options));
     }
 
+    @Test
+    public void testSnapshotSequenceOrderingRejectsNonWriteOnly() {
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(), "true");
+        assertThatThrownBy(() -> validateTableSchemaExec(options))
+                .hasMessageContaining(CoreOptions.WRITE_ONLY.key());
+    }
+
     @Test
     public void testSnapshotSequenceOrderingRejectsSequenceField() {
         Map<String, String> options = new HashMap<>();
         options.put(CoreOptions.SEQUENCE_SNAPSHOT_ORDERING.key(), "true");
+        options.put(CoreOptions.WRITE_ONLY.key(), "true");
         options.put(CoreOptions.SEQUENCE_FIELD.key(), "f2");
         assertThatThrownBy(() -> validateTableSchemaExec(options))
                 .hasMessageContaining("sequence.field");