@Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        List<Row> beforeImageRows = beforeImage.getRows();
        if (CollectionUtils.isEmpty(beforeImageRows)) {
            throw new ShouldNeverHappenException("Invalid UNDO LOG");
        }
        Row row = beforeImageRows.get(0);
        List<Field> fields = new ArrayList<>(row.nonPrimaryKeys());
        fields.addAll(getOrderedPkList(beforeImage, row, JdbcConstants.SQLSERVER));

        // delete sql undo log before image all field come from table meta, need add escape.
        // see BaseTransactionalExecutor#buildTableRecords
        String insertColumns = fields.stream()
                .map(field -> ColumnUtils.addEscape(field.getName(), JdbcConstants.SQLSERVER))
                .collect(Collectors.joining(", "));
        String insertValues = fields.stream().map(field -> "?")
                .collect(Collectors.joining(", "));

        return "SET IDENTITY_INSERT " +
                sqlUndoLog.getTableName() +
                " ON; INSERT INTO " +
                sqlUndoLog.getTableName() +
                " (" +
                insertColumns +
                ") VALUES (" +
                insertValues +
                "); SET IDENTITY_INSERT " +
                sqlUndoLog.getTableName() +
                " OFF;";
    }
@Test
    public void buildUndoSQL() {
        String sql = executor.buildUndoSQL().toUpperCase();
        Assertions.assertNotNull(sql);
        Assertions.assertTrue(sql.contains("INSERT"));
        Assertions.assertTrue(sql.contains("TABLE_NAME"));
        Assertions.assertTrue(sql.contains("ID"));
    }