Merge pull request #337 from rajyan/swap-table-replace-2

Swap name + drop instead of drop + rename in MySQL replace mode

Author: dmikurube

embulk-output-mysql/README.md

@@ -53,8 +53,8 @@ MySQL output plugin for Embulk loads records to MySQL.
   * Transactional: Yes.
   * Resumable: No.
 * **replace**:
-  * Behavior: This mode writes rows to an intermediate table first. If all those tasks run correctly, drops the target table and alters the name of the intermediate table into the target table name.
-  * Transactional: No. If fails, the target table could be dropped (because MySQL can't rollback DDL).
+  * Behavior: This mode writes rows to an intermediate table first. If all those tasks run correctly, swaps the intermediate table with target table by altering the table names atomically, and then drops the intermediate table.
+  * Transactional: No. If fails, the intermediate table could remain (because MySQL can't rollback DDL)
   * Resumable: No.
 * **merge**:
   * Behavior: This mode writes rows to some intermediate tables first. If all those tasks run correctly, runs `INSERT INTO <target_table> SELECT * FROM <intermediate_table_1> UNION ALL SELECT * FROM <intermediate_table_2> UNION ALL ... ON DUPLICATE KEY UPDATE ...` query. Namely, if primary keys of a record in the intermediate tables already exist in the target table, the target record is updated by the intermediate record, otherwise the intermediate record is inserted. If the target table doesn't exist, it is created automatically.

embulk-output-mysql/src/main/java/org/embulk/output/mysql/MySQLOutputConnection.java

@@ -1,8 +1,10 @@
 package org.embulk.output.mysql;
 
+import java.sql.Statement;
 import java.util.List;
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.util.Optional;
 
 import org.embulk.output.MySQLTimeZoneComparison;
 import org.embulk.output.jdbc.JdbcColumn;
@@ -101,6 +103,56 @@ protected String buildCollectMergeSql(List<TableIdentifier> fromTables, JdbcSche
         return sb.toString();
     }
 
+    private String buildSwapTableSql(TableIdentifier fromTable, TableIdentifier toTable)
+    {
+        String suffix = "_embulk_swap_tmp";
+        String uniqueName = String.format("%016x", System.currentTimeMillis()) + suffix;
+        // NOTE: The table name should be always shorter than 64 characters
+        // See also: https://dev.mysql.com/doc/refman/8.0/en/identifier-length.html
+        TableIdentifier tmpTable = new TableIdentifier(fromTable.getDatabase(), fromTable.getSchemaName(), uniqueName);
+
+        StringBuilder sb = new StringBuilder();
+        sb.append("RENAME TABLE ");
+        quoteTableIdentifier(sb, fromTable);
+        sb.append(" TO ");
+        quoteTableIdentifier(sb, tmpTable);
+
+        sb.append(", ");
+        quoteTableIdentifier(sb, toTable);
+        sb.append(" TO ");
+        quoteTableIdentifier(sb, fromTable);
+
+        sb.append(", ");
+        quoteTableIdentifier(sb, tmpTable);
+        sb.append(" TO ");
+        quoteTableIdentifier(sb, toTable);
+
+        return sb.toString();
+    }
+
+    @Override
+    public void replaceTable(TableIdentifier fromTable, JdbcSchema schema, TableIdentifier toTable, Optional<String> postSql) throws SQLException
+    {
+        Statement stmt = connection.createStatement();
+        try {
+            // "DROP TABLE" causes an implicit commit in MySQL, so we rename the table at first.
+            // See also: https://dev.mysql.com/doc/refman/8.0/en/implicit-commit.html
+            executeUpdate(stmt, buildSwapTableSql(fromTable, toTable));
+
+            dropTableIfExists(stmt, fromTable);
+
+            if (postSql.isPresent()) {
+                execute(stmt, postSql.get());
+            }
+
+            commitIfNecessary(connection);
+        } catch (SQLException ex) {
+            throw safeRollback(connection, ex);
+        } finally {
+            stmt.close();
+        }
+    }
+
     @Override
     protected String buildColumnTypeName(JdbcColumn c)
     {