diff --git a/.github/workflows/paimon1.yml b/.github/workflows/paimon1.yml index e165f0786..21901bd29 100644 --- a/.github/workflows/paimon1.yml +++ b/.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 diff --git a/docs/connectors/paimon.md b/docs/connectors/paimon.md index 1e014a391..8f2ca032a 100644 --- a/docs/connectors/paimon.md +++ b/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: diff --git a/streamfusion-paimon1/src/test/java/tech/streamfusion/paimon/PaimonFullCompactionRecoveryReproducerTest.java b/streamfusion-paimon1/src/test/java/tech/streamfusion/paimon/PaimonFullCompactionRecoveryReproducerTest.java new file mode 100644 index 000000000..7295b1152 --- /dev/null +++ b/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 batch = values(table, table.newScan().plan()); + assertTrue(batch.contains(20), "Latest value must be durable in the batch scan: " + batch); + List 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 values(FileStoreTable table, TableScan.Plan plan) throws Exception { + List 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; + } +}