@Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        List<Row> beforeImageRows = beforeImage.getRows();
        if (CollectionUtils.isEmpty(beforeImageRows)) {
            throw new ShouldNeverHappenException("Invalid UNDO LOG"); // TODO
        }
        Row row = beforeImageRows.get(0);

        List<Field> nonPkFields = row.nonPrimaryKeys();
        // update sql undo log before image all field come from table meta. need add escape.
        // see BaseTransactionalExecutor#buildTableRecords
        String updateColumns = nonPkFields.stream().map(
            field -> {
                String addEscape = ColumnUtils.addEscape(field.getName(), JdbcConstants.MYSQL);
                return addEscape + " = " + MySQLJsonHelper.convertIfJson(field, beforeImage.getTableMeta());
            })
            .collect(Collectors.joining(", "));

        List<String> pkNameList = getOrderedPkList(beforeImage, row, JdbcConstants.MYSQL).stream().map(e -> e.getName())
            .collect(Collectors.toList());
        String whereSql = SqlGenerateUtils.buildWhereConditionByPKs(pkNameList, JdbcConstants.MYSQL);

        return String.format(UPDATE_SQL_TEMPLATE, sqlUndoLog.getTableName(), updateColumns, whereSql);
    }
@Test
    public void buildUndoSQL() {
        String sql = executor.buildUndoSQL().toLowerCase();
        Assertions.assertNotNull(sql);
        Assertions.assertTrue(sql.contains("update"));
        Assertions.assertTrue(sql.contains("id"));
        Assertions.assertTrue(sql.contains("age"));
    }