remove partial snapshot test

This test provides questionable value and requires modifying code
behavior deep in the dataflow to trigger the condition it wants to test.
Hence it is removed. I put this as a separate commit in case we change
our minds during review and want to keep it.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

Author: petrosagg

src/storage/src/render/persist_sink.rs

@@ -1026,15 +1026,6 @@ where
         let mut batch_description_frontier = Antichain::from_elem(Timestamp::minimum());
         let mut batches_frontier = Antichain::from_elem(Timestamp::minimum());
 
-        // Pause the source to prevent committing the snapshot,
-        // if the failpoint is configured
-        let mut pg_snapshot_pause = false;
-        (|| {
-            fail::fail_point!("pg_snapshot_pause", |_| {
-                pg_snapshot_pause = true;
-            });
-        })();
-
         loop {
             tokio::select! {
                 Some(event) = descriptions_input.next() => {
@@ -1168,18 +1159,6 @@ where
 
                 let mut to_append = batches.iter_mut().map(|b| &mut b.batch).collect::<Vec<_>>();
 
-                // We evaluate this above to avoid checking an environment variable
-                // in a hot loop. Note that we only pause before we emit
-                // non-empty batches, because we do want to bump the upper
-                // with empty ones before we start ingesting the snapshot.
-                //
-                // This is a fairly complex failure case we need to check
-                // see `test/cluster/pg-snapshot-partial-failure` for more
-                // information.
-                if pg_snapshot_pause && !to_append.is_empty() && !batch_metrics.is_empty() {
-                    futures::future::pending().await
-                }
-
                 let result = {
                     let maybe_err = if *read_only_rx.borrow() {
 

test/cluster/mzcompose.py

@@ -1716,39 +1716,6 @@ def kill_replica_with_delay() -> None:
     killer.join()
 
 
-def workflow_pg_snapshot_partial_failure(c: Composition) -> None:
-    """Test PostgreSQL snapshot partial failure"""
-
-    c.down(destroy_volumes=True)
-
-    with c.override(
-        # Start postgres for the pg source
-        Testdrive(no_reset=True),
-        Clusterd(
-            name="clusterd1",
-            environment_extra=["FAILPOINTS=pg_snapshot_pause=return(2)"],
-        ),
-    ):
-        c.up("materialized", "postgres", "clusterd1")
-
-        c.run_testdrive_files("pg-snapshot-partial-failure/01-configure-postgres.td")
-        c.run_testdrive_files("pg-snapshot-partial-failure/02-create-sources.td")
-
-        c.run_testdrive_files(
-            "pg-snapshot-partial-failure/03-verify-good-sub-source.td"
-        )
-
-        c.kill("clusterd1")
-        # Restart the storage instance with the failpoint off...
-        with c.override(
-            # turn off the failpoint
-            Clusterd(name="clusterd1")
-        ):
-            c.run_testdrive_files("pg-snapshot-partial-failure/04-add-more-data.td")
-            c.up("clusterd1")
-            c.run_testdrive_files("pg-snapshot-partial-failure/05-verify-data.td")
-
-
 def workflow_test_compute_reconciliation_reuse(c: Composition) -> None:
     """
     Test that compute reconciliation reuses existing dataflows.