Reproduce Paimon full-compaction recovery loss and retain failed CI state

.github/workflows/paimon1.yml

@@ -83,7 +83,9 @@ jobs:
       - name: Upstream SQL tests (reuse release libraries)
         env:
           SF_PAIMON1_TEST_FORKS: '1'
-        run: bin/paimon1-suite.sh -Pbench -Dnative.build.skip=true
+        run: >-
+          bin/paimon1-suite.sh -Pbench -Dnative.build.skip=true
+          -Dstreamfusion.flink-suite.retain-failed-warehouse=true
       - name: Retain test evidence
         if: always()
         uses: actions/upload-artifact@v4

docs/connectors/paimon.md

@@ -1245,6 +1245,33 @@ this durable snapshot state. A passing randomized run does not establish absence
 StreamFusion remains on released Paimon 1.0.0; fixing the split-commit recovery requires a
 verified upstream release, not a planner workaround or a forked dependency.
 
+The same incomplete commit also has a deterministic full-compaction changelog reproducer,
+without deletion vectors: commit `10`, consume the initial stream snapshot, then write `20`
+and inject a manifest-read failure after APPEND and before COMPACT. After recovery a fresh
+batch reader returns `20`, but the incremental stream scan returns no final changelog.
+The no-failure control returns the update. This uses released Paimon writer/commit/read APIs
+without a Flink job or native operators. Standalone runs excluding StreamFusion production
+classes reproduce the result on both JDK 11 and 17. The explicit recovery assertion fails on 1.0.0;
+it is an opt-in diagnostic, not a passing recovery test:
+
+```bash
+mvn -Pflink-1.18,paimon1 -pl streamfusion-paimon1 -am \
+  -Dtest=PaimonFullCompactionRecoveryReproducerTest \
+  -Dsurefire.failIfNoSpecifiedTests=false \
+  -Dsf.paimon1.reproduceFullCompactionCommit=true test
+```
+
+[#318](https://github.com/datafusion-contrib/StreamFusion/issues/318) tracks this defect
+and the remaining CI investigation. In [run 38020828273](https://github.com/datafusion-contrib/StreamFusion/actions/runs/38020828273/job/114121230011),
+241 of 242 upstream cases passed; the unchanged full-compaction streaming test timed out
+after injected-failure recovery. The writer finished while its consumer kept waiting.
+That invocation did not retain the warehouse, so the standalone defect does not prove
+its exact cause. The Paimon 1.0 workflow now enables the existing failed-warehouse capture
+and uploads snapshots, manifests and data files with its diagnostics. Upstream assertions,
+randomized inputs and deadlines remain authoritative; the capture is diagnostic rather
+than an atomic backup. Compare stock/native replay over that retained state before
+claiming the timeout resolved.
+
 Run the opt-in diagnostic with JDK 17. **The recovery case is expected to fail on 1.0.0**
 (`expected: 20 but was: 10`); the control passes. It is separate from the regular test gate:
 

streamfusion-paimon1/src/test/java/tech/streamfusion/paimon/PaimonFullCompactionRecoveryReproducerTest.java

@@ -0,0 +1,137 @@
+package tech.streamfusion.paimon;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Opt-in diagnostic using released Paimon commits and readers without a Flink job. */
+@org.junit.jupiter.api.condition.EnabledIfSystemProperty(
+    named = "sf.paimon1.reproduceFullCompactionCommit",
+    matches = "true")
+class PaimonFullCompactionRecoveryReproducerTest {
+  @TempDir java.nio.file.Path directory;
+
+  @Test
+  void completeCommitProducesFinalChangelog() throws Exception {
+    verify(false);
+  }
+
+  @Test
+  void recoveredCommitMustProduceFinalChangelog() throws Exception {
+    verify(true);
+  }
+
+  private void verify(boolean fail) throws Exception {
+    Path root = new Path(directory.toUri());
+    class FailBetweenSnapshots extends LocalFileIO {
+      Path appendSnapshot;
+
+      @Override
+      public SeekableInputStream newInputStream(Path path) throws IOException {
+        if (appendSnapshot != null
+            && path.toString().contains("/manifest/")
+            && super.exists(appendSnapshot)) {
+          appendSnapshot = null;
+          throw new IOException("Injected failure after APPEND before COMPACT");
+        }
+        return super.newInputStream(path);
+      }
+    }
+    var io = new FailBetweenSnapshots();
+    new SchemaManager(io, root)
+        .createTable(
+            Schema.newBuilder()
+                .column("k", DataTypes.INT())
+                .column("v", DataTypes.INT())
+                .primaryKey("k")
+                .option("bucket", "1")
+                .option("changelog-producer", "full-compaction")
+                .build());
+    var table = FileStoreTableFactory.create(io, root);
+    try (var disk = org.apache.paimon.disk.IOManager.create(directory.toString());
+        var write = table.newWrite("repro").withIOManager(disk);
+        var commit = table.newCommit("repro")) {
+      var first = GenericRow.of(1, 10);
+      write.write(first);
+      write.compact(write.getPartition(first), write.getBucket(first), true);
+      commit.commit(1, write.prepareCommit(true, 1));
+      var scan = table.newStreamScan();
+      assertTrue(values(table, scan.plan()).contains(10));
+
+      var last = GenericRow.of(1, 20);
+      write.write(last);
+      write.compact(write.getPartition(last), write.getBucket(last), true);
+      var messages = write.prepareCommit(true, Long.MAX_VALUE);
+      assertTrue(
+          messages.stream()
+              .anyMatch(
+                  message ->
+                      !((CommitMessageImpl) message).compactIncrement().changelogFiles().isEmpty()),
+          "The prepared commit must contain the final full-compaction changelog");
+      if (fail) {
+        io.appendSnapshot =
+            table.snapshotManager().snapshotPath(table.snapshotManager().latestSnapshotId() + 1);
+        assertThrows(Exception.class, () -> commit.commit(Long.MAX_VALUE, messages));
+        assertNull(io.appendSnapshot, "Failure must occur after APPEND");
+        assertEquals(
+            org.apache.paimon.Snapshot.CommitKind.APPEND,
+            table.snapshotManager().latestSnapshot().commitKind());
+        try (var recovered = table.newCommit("repro")) {
+          recovered.filterAndCommit(Collections.singletonMap(Long.MAX_VALUE, messages));
+        }
+      } else {
+        commit.commit(Long.MAX_VALUE, messages);
+      }
+      List<Integer> batch = values(table, table.newScan().plan());
+      assertTrue(batch.contains(20), "Latest value must be durable in the batch scan: " + batch);
+      List<Integer> changelog = new ArrayList<>();
+      for (int attempt = 0; attempt < 4; attempt++) {
+        changelog.addAll(values(table, scan.plan()));
+      }
+      assertTrue(
+          changelog.contains(20),
+          "Final changelog missing: latest snapshot="
+              + table.snapshotManager().latestSnapshot().commitKind()
+              + ", batch="
+              + batch
+              + ", changelog="
+              + changelog);
+    }
+  }
+
+  private static List<Integer> values(FileStoreTable table, TableScan.Plan plan) throws Exception {
+    List<Integer> values = new ArrayList<>();
+    try (var reader = table.newRead().createReader(plan)) {
+      var batch = reader.readBatch();
+      while (batch != null) {
+        try {
+          var row = batch.next();
+          while (row != null) {
+            values.add(row.getInt(1));
+            row = batch.next();
+          }
+        } finally {
+          batch.releaseBatch();
+        }
+        batch = reader.readBatch();
+      }
+    }
+    return values;
+  }
+}