Made it possible to redo a clone operation

Author: HHMagnus

core/src/main/java/dev/mhh/cloner/api/CloneError.java

@@ -22,6 +22,12 @@ public enum CloneError {
     CREATE_TEMP_TABLE_FAILED("Failed to create a temporary table for data transfer"),
     COPY_IN_FAILED("Failed to copy data into the temporary table"),
     MOVE_DATA_FAILED("Failed to move data from temporary table to the target table"),
+    DELETE_STALE_DATA_FAILED("Failed to delete stale data from the target table"),
+    GET_CONFLICT_COLUMNS_FAILED("Failed to get columns for the on conflict clause"),
+    GET_CONFLICT_COLUMNS_RETURNED_NOTHING("The on conflict clause returned nothing"),
+    CONFIG_FINGERPRINT_WAS_NOT_FIRST_ENTRY("The configuration fingerprint was not the first entry in the ZIP"),
+    CONFIG_FINGERPRINT_READ_FAILED("Failed to read the configuration fingerprint from the ZIP"),
+    CONFIG_FINGERPRINT_MISMATCH("The configuration fingerprint in the ZIP does not match the provided configuration"),
     ;
 
     private final String message;
@@ -34,6 +40,11 @@ public String message(final Object... args) {
         return String.format(message, args);
     }
 
+
+    public CloneException toException() {
+        return new CloneException(this);
+    }
+
     public Supplier<CloneException> toException(final Object... args) {
         return () -> new CloneException(this, args);
     }

core/src/main/java/dev/mhh/cloner/api/CloneException.java

@@ -22,7 +22,7 @@ public CloneError error() {
 
     @FunctionalInterface
     public interface ThrowableSupplier<T> {
-        T get() throws SQLException, IOException;
+        T get() throws SQLException, IOException, CloneException;
     }
 
     public static <T> T wrapKnownExceptions(

core/src/main/java/dev/mhh/cloner/api/Cloner.java

@@ -15,6 +15,7 @@ void exportClone(
     ) throws CloneException;
     void importClone(
             InputStream inputStream,
-            Connection connection
+            Connection connection,
+            CloneConfiguration<T> configuration
     ) throws CloneException;
 }

core/src/main/java/dev/mhh/cloner/impl/ClonerImpl.java

@@ -6,12 +6,14 @@
 import dev.mhh.cloner.api.Cloner;
 import dev.mhh.cloner.api.verification.ConfigurationVerifier;
 import dev.mhh.cloner.impl.select.CloneConfigurationTransformer;
+import dev.mhh.cloner.impl.select.DeletePreviousTransformer;
 import org.postgresql.PGConnection;
 import org.postgresql.copy.CopyManager;
 
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
@@ -20,6 +22,8 @@
 import java.util.zip.ZipOutputStream;
 
 public class ClonerImpl<T extends Serializable> implements Cloner<T> {
+    private static final String CONFIG_ENTRY_NAME = "__config__.txt";
+
     @Override
     public void exportClone(
             final OutputStream outputStream,
@@ -35,8 +39,21 @@ public void exportClone(
 
         final var zos = new ZipOutputStream(outputStream);
 
-        for (final var select : selects) {
+        final var configEntry = new ZipEntry(CONFIG_ENTRY_NAME);
+        CloneException.wrapKnownExceptions(
+                () -> zos.putNextEntry(configEntry),
+                CloneError.ZIP_ENTRY_CREATION_FAILED
+        );
+        CloneException.wrapKnownExceptions(
+                () -> zos.write(configuration.toString().getBytes(StandardCharsets.UTF_8)),
+                CloneError.ZIP_ENTRY_CREATION_FAILED
+        );
+        CloneException.wrapKnownExceptions(
+                zos::closeEntry,
+                CloneError.ZIP_ENTRY_CLOSE_FAILED
+        );
 
+        for (final var select : selects) {
             final var zipEntry = new ZipEntry(select.tableName() + ".csv");
             CloneException.wrapKnownExceptions(
                     () -> zos.putNextEntry(zipEntry),
@@ -73,12 +90,28 @@ private static CopyManager getCopyManager(final Connection connection) throws Cl
     }
 
     @Override
-    public void importClone(final InputStream inputStream, final Connection connection) throws CloneException {
+    public void importClone(
+            final InputStream inputStream,
+            final Connection connection,
+            final CloneConfiguration<T> configuration
+    ) throws CloneException {
         final var copyManager = getCopyManager(connection);
 
+        final var deleteScopeByTable = DeletePreviousTransformer.transform(configuration);
+
         final var zis = new ZipInputStream(inputStream);
 
-        ZipEntry entry;
+        var entry = CloneException.wrapKnownExceptions(
+                zis::getNextEntry,
+                CloneError.ZIP_ENTRY_NEXT_FAILED
+        );
+
+        if (entry.getName().equals(CONFIG_ENTRY_NAME)) {
+            verifyConfigFingerprint(zis, configuration);
+            CloneException.wrapKnownExceptions(zis::closeEntry, CloneError.ZIP_ENTRY_CLOSE_FAILED);
+        } else {
+            throw CloneError.CONFIG_FINGERPRINT_WAS_NOT_FIRST_ENTRY.toException();
+        }
 
         while ((entry = CloneException.wrapKnownExceptions(
                 zis::getNextEntry,
@@ -88,7 +121,7 @@ public void importClone(final InputStream inputStream, final Connection connecti
 
             final var tableName = fileName.substring(0, fileName.length() - 4);
 
-            final var tempTableSql = "create temp table temp_" + tableName + " (like " + tableName + " including defaults)";
+            final var tempTableSql = "create temp table temp_" + tableName + " (like " + tableName + " including defaults) on commit drop";
             CloneException.wrapKnownExceptions(
                     () -> execute(connection, tempTableSql),
                     CloneError.CREATE_TEMP_TABLE_FAILED
@@ -100,7 +133,20 @@ public void importClone(final InputStream inputStream, final Connection connecti
                     CloneError.COPY_IN_FAILED
             );
 
-            final var moveSql = "insert into " + tableName + " select * from temp_" + tableName;
+            final var deleteScope = deleteScopeByTable.get(tableName);
+            if (deleteScope != null) {
+                final var idColumn = configuration.idColumnName();
+                final var deleteSql = "delete from " + tableName
+                        + " where " + idColumn + " not in (select " + idColumn + " from temp_" + tableName + ")"
+                        + " and " + deleteScope;
+                CloneException.wrapKnownExceptions(
+                        () -> execute(connection, deleteSql),
+                        CloneError.DELETE_STALE_DATA_FAILED
+                );
+            }
+
+            final var conflictClause = buildConflictClause(connection, tableName, configuration);
+            final var moveSql = "insert into " + tableName + " select * from temp_" + tableName + " " + conflictClause;
             CloneException.wrapKnownExceptions(
                     () -> execute(connection, moveSql),
                     CloneError.MOVE_DATA_FAILED
@@ -113,6 +159,41 @@ public void importClone(final InputStream inputStream, final Connection connecti
         }
     }
 
+    private String buildConflictClause(final Connection connection, final String tableName, final CloneConfiguration<T> configuration) throws CloneException {
+        final var idColumn = configuration.idColumnName();
+        final var schema = configuration.schema();
+        final var sql = "select 'on conflict (" + idColumn + ") do update set ' || "
+                + "string_agg(column_name || ' = excluded.' || column_name, ', ') "
+                + "from information_schema.columns "
+                + "where table_schema = '" + schema + "' "
+                + "and table_name = '" + tableName + "' "
+                + "and column_name != '" + idColumn + "'";
+
+        return CloneException.wrapKnownExceptions(() -> {
+            try (final var stmt = connection.createStatement();
+                 final var rs = stmt.executeQuery(sql)) {
+                if (rs.next()) {
+                    final var result = rs.getString(1);
+                    if (result != null) {
+                        return result;
+                    }
+                }
+                throw CloneError.GET_CONFLICT_COLUMNS_RETURNED_NOTHING.toException();
+            }
+        }, CloneError.GET_CONFLICT_COLUMNS_FAILED);
+    }
+
+    private void verifyConfigFingerprint(final ZipInputStream zis, final CloneConfiguration<T> configuration) throws CloneException {
+        final var storedFingerprint = CloneException.wrapKnownExceptions(
+                () -> new String(zis.readAllBytes(), StandardCharsets.UTF_8),
+                CloneError.CONFIG_FINGERPRINT_READ_FAILED
+        );
+
+        if (!configuration.toString().equals(storedFingerprint)) {
+            throw new CloneException(CloneError.CONFIG_FINGERPRINT_MISMATCH);
+        }
+    }
+
     private void execute(final Connection connection, final String sql) throws SQLException {
         try (final var stmt = connection.createStatement()) {
             stmt.execute(sql);

core/src/main/java/dev/mhh/cloner/impl/select/DeletePreviousTransformer.java

@@ -0,0 +1,59 @@
+package dev.mhh.cloner.impl.select;
+
+import dev.mhh.cloner.api.CloneConfiguration;
+import dev.mhh.cloner.api.table.JoinByJoinTableFk;
+import dev.mhh.cloner.api.table.JoinByMatchingColumns;
+import dev.mhh.cloner.api.table.JoinConfig;
+import dev.mhh.cloner.api.table.TableConfig;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Produces a map of table name to delete scope condition fragment.
+ * The condition is used to scope the DELETE statement during recloning:
+ *   DELETE FROM table WHERE id NOT IN (SELECT id FROM temp_table) AND <condition>
+ *
+ * Tables joined via joinByMainTableFk are excluded (null) because the FK lives in
+ * the parent table, which hasn't been processed yet when the child is imported.
+ */
+public final class DeletePreviousTransformer<T extends Serializable> {
+    private final CloneConfiguration<T> configuration;
+
+    private DeletePreviousTransformer(final CloneConfiguration<T> configuration) {
+        this.configuration = configuration;
+    }
+
+    public static <T extends Serializable> Map<String, String> transform(final CloneConfiguration<T> configuration) {
+        return new DeletePreviousTransformer<>(configuration).transform();
+    }
+
+    private Map<String, String> transform() {
+        final var result = new HashMap<String, String>();
+        transformRoot(configuration.rootTable(), result);
+        return result;
+    }
+
+    private void transformRoot(final TableConfig<?> table, final Map<String, String> result) {
+        table.joins().forEach(join -> transformJoin(join, result));
+    }
+
+    private void transformJoin(final JoinConfig join, final Map<String, String> result) {
+        final var newIn = switch (join) {
+            case final JoinByJoinTableFk j -> j.foreignKey() + " in (select " + j.foreignKey() + " from temp_" + j.tableName() + ")";
+            case final JoinByMatchingColumns j -> j.joinedTableColumn() + " in (select " + j.joinedTableColumn() + " from temp_" + j.tableName() + ")";
+            default -> null;
+        };
+
+        if (newIn != null) {
+            final var previous = result.get(join.tableName());
+            final var condition = previous == null
+                    ? newIn
+                    : ("(" + previous + " or " + newIn + ")");
+            result.put(join.tableName(), condition);
+        }
+
+        join.joins().forEach(j -> transformJoin(j, result));
+    }
+}

core/src/test/java/dev/mhh/cloner/OverwriteDataTest.java

@@ -53,7 +53,7 @@ void runOverwriteTestCase(final Connection exportConn, final Connection importCo
         }
 
         try (final var in = new ByteArrayInputStream(zipBytes)) {
-            cloner.importClone(in, importConn);
+            cloner.importClone(in, importConn, CLONE_CONFIGURATION);
         }
 
         // Assert — stale body values should be gone, replaced by 'relevant'

core/src/test/java/dev/mhh/cloner/RecloneTest.java

@@ -0,0 +1,128 @@
+package dev.mhh.cloner;
+
+import dev.mhh.cloner.api.CloneException;
+import dev.mhh.cloner.api.Cloner;
+import dev.mhh.cloner.impl.ClonerImpl;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+
+import static dev.mhh.cloner.TestDatabase.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class RecloneTest {
+    Cloner<Long> cloner = new ClonerImpl<>();
+
+    @Test
+    void recloningSameDataTwiceSucceeds() throws SQLException, IOException, CloneException {
+        runTestWithScripts(
+                List.of("V1__schema.sql", "V2__data.sql"),
+                List.of("V1__schema.sql"),
+                this::runRecloneTestCase
+        );
+    }
+
+    @Test
+    void recloningSameDataTwiceSucceedsWithRLS() throws SQLException, IOException, CloneException {
+        runTestWithScripts(
+                List.of("V1__schema.sql", "V2__data.sql", "V3__rls.sql"),
+                List.of("V1__schema.sql", "V3__rls.sql"),
+                this::runRecloneTestCase
+        );
+    }
+
+    @Test
+    void recloneDeletesStaleScopedRows() throws SQLException, IOException, CloneException {
+        runTestWithScripts(
+                List.of("V1__schema.sql", "V2__data.sql"),
+                List.of("V1__schema.sql"),
+                this::runDeleteStaleScopedRowsTestCase
+        );
+    }
+
+    void runRecloneTestCase(final Connection exportConn, final Connection importConn) throws SQLException, IOException, CloneException {
+        final byte[] zipBytes;
+        try (final var out = new ByteArrayOutputStream()) {
+            cloner.exportClone(out, exportConn, CLONE_CONFIGURATION, List.of(1L));
+            zipBytes = out.toByteArray();
+        }
+
+        // First import
+        try (final var in = new ByteArrayInputStream(zipBytes)) {
+            cloner.importClone(in, importConn, CLONE_CONFIGURATION);
+            importConn.commit();
+        }
+
+        assertCount(importConn, MAIN_TABLE, RELEVANT, 1);
+        assertCount(importConn, SUB_TABLE, RELEVANT, 2);
+
+        // Second import — must not fail with conflict errors
+        try (final var in = new ByteArrayInputStream(zipBytes)) {
+            cloner.importClone(in, importConn, CLONE_CONFIGURATION);
+        }
+
+        // Data should be identical after reclone — no duplicates
+        assertCount(importConn, MAIN_TABLE, RELEVANT, 1);
+        assertCount(importConn, SUB_TABLE, RELEVANT, 2);
+        assertCount(importConn, SUB_EQUALITY_TABLE, RELEVANT, 2);
+        assertCount(importConn, SUB_SUB_TABLE, RELEVANT, 6);
+        assertCount(importConn, SUB_SUB_SUB_TABLE, RELEVANT, 2);
+        assertCount(importConn, SUB_SUB_POINTER_TABLE, RELEVANT, 3);
+    }
+
+    void runDeleteStaleScopedRowsTestCase(final Connection exportConn, final Connection importConn) throws SQLException, IOException, CloneException {
+        final byte[] zipBytes;
+        try (final var out = new ByteArrayOutputStream()) {
+            cloner.exportClone(out, exportConn, CLONE_CONFIGURATION, List.of(1L));
+            zipBytes = out.toByteArray();
+        }
+
+        // First import
+        try (final var in = new ByteArrayInputStream(zipBytes)) {
+            cloner.importClone(in, importConn, CLONE_CONFIGURATION);
+            importConn.commit();
+        }
+
+        assertCount(importConn, SUB_TABLE, RELEVANT, 2);
+
+        // Insert a stale row that is within the clone scope (main_id=1 is cloned)
+        // but whose id won't appear in any future export — simulates a row deleted from source
+        try (final var stmt = importConn.createStatement()) {
+            stmt.execute("insert into " + SUB_TABLE + "(id, main_id, common_value, body) values (99, 1, 'deleted_value', 'stale_scoped')");
+        }
+
+        assertCount(importConn, SUB_TABLE, "stale_scoped", 1);
+
+        // Reclone — the stale_scoped row (id=99, main_id=1) should be deleted
+        // because main_id=1 is in scope but id=99 is not in the export
+        try (final var in = new ByteArrayInputStream(zipBytes)) {
+            cloner.importClone(in, importConn, CLONE_CONFIGURATION);
+        }
+
+        assertCount(importConn, SUB_TABLE, "stale_scoped", 0);
+        assertCount(importConn, SUB_TABLE, RELEVANT, 2);
+    }
+
+    void assertCount(final Connection conn, final String table, final String body, final int expectedCount) throws SQLException {
+        assertEquals(
+                expectedCount,
+                count(conn, "select count(*) from " + table + " where body = '" + body + "'"),
+                "Counts of row in table '" + table + "' with body '" + body + "' do not match"
+        );
+    }
+
+    @SuppressWarnings("SqlSourceToSinkFlow")
+    int count(final Connection conn, final String sql) throws SQLException {
+        try (final var stmt = conn.prepareStatement(sql)) {
+            final var rs = stmt.executeQuery();
+            assertTrue(rs.next(), "Query returned no rows: " + sql);
+            return rs.getInt(1);
+        }
+    }
+}