使用jOOQ在PostgreSQL中进行UPSERT

Question

我正在尝试使用jOOQ库在PostgreSQL中执行UPSERT操作。

为此，我目前正在尝试在jOOQ中实现以下SQL语句： https://dev59.com/uHNA5IYBdhLWcg3wC5Xh#6527838

到目前为止，我的代码看起来像这样：

public class UpsertExecutor {

    private static final Logger logger = LoggerFactory.getLogger(UpsertExecutor.class);

    private final JOOQContextProvider jooqProvider;

    @Inject
    public UpsertExecutor(JOOQContextProvider jooqProvider) {
        Preconditions.checkNotNull(jooqProvider);

        this.jooqProvider = jooqProvider;
    }

    @Transactional
    public <T extends Record> void executeUpsert(Table<T> table, Condition condition, Map<? extends Field<?>, ?> recordValues) {
        /*
         * All of this is for trying to do an UPSERT on PostgreSQL. See:
         * https://dev59.com/uHNA5IYBdhLWcg3wC5Xh#6527838
         */

        SelectConditionStep<Record1<Integer>> notExistsSelect = jooqProvider.getDSLContext().selectOne().from(table).where(condition);
        SelectConditionStep<Record> insertIntoSelect = jooqProvider.getDSLContext().select(recordValues).whereNotExists(notExistsSelect);

        try {
            int[] result = jooqProvider.getDSLContext().batch(
                jooqProvider.getDSLContext().update(table).set(recordValues).where(condition),
                jooqProvider.getDSLContext().insertInto(table).select(insertIntoSelect)
            ).execute();

            long rowsAffectedTotal = 0;
            for (int rowsAffected : result) {
                rowsAffectedTotal += rowsAffected;
            }

            if (rowsAffectedTotal != 1) {
                throw new RuntimeException("Upsert must only affect 1 row. Affected: " + rowsAffectedTotal + ". Table: " + table + ". Condition: " + condition);
            }
        } catch (DataAccessException e) {
            if (e.getCause() instanceof BatchUpdateException) {
                BatchUpdateException cause = (BatchUpdateException)e.getCause();

                logger.error("Batch update error in upsert.", cause.getNextException());
            }

            throw e;
        }
    }
}

这段代码无法编译，因为select()不支持值的映射：

SelectConditionStep<Record> insertIntoSelect = jooqProvider.getDSLContext().select(recordValues).whereNotExists(notExistsSelect);

问题

如何向select()提供预定义的值集，例如：SELECT 3, 'C', 'Z'？

更新1

我设法让代码正常工作了。这是完整的类：

public class UpsertExecutor {

    private static final Logger logger = LoggerFactory.getLogger(UpsertExecutor.class);

    private final JOOQContextProvider jooqProvider;

    @Inject
    public UpsertExecutor(JOOQContextProvider jooqProvider) {
        Preconditions.checkNotNull(jooqProvider);

        this.jooqProvider = jooqProvider;
    }

    @Transactional
    public <T extends Record> void executeUpsert(Table<T> table, Condition condition, List<FieldValue<Field<?>, ?>> recordValues) {
        /*
         * All of this is for trying to do an UPSERT on PostgreSQL. See:
         * https://dev59.com/uHNA5IYBdhLWcg3wC5Xh#6527838
         */

        Map<Field<?>, Object> recordValuesMap = new HashMap<Field<?>, Object>();
        for (FieldValue<Field<?>, ?> entry : recordValues) {
            recordValuesMap.put(entry.getFieldName(), entry.getFieldValue());
        }

        List<Param<?>> params = new LinkedList<Param<?>>();
        for (FieldValue<Field<?>, ?> entry : recordValues) {
            params.add(val(entry.getFieldValue()));
        }

        List<Field<?>> fields = new LinkedList<Field<?>>();
        for (FieldValue<Field<?>, ?> entry : recordValues) {
            fields.add(entry.getFieldName());
        }

        SelectConditionStep<Record1<Integer>> notExistsSelect = jooqProvider.getDSLContext().selectOne().from(table).where(condition);
        SelectConditionStep<Record> insertIntoSelect = jooqProvider.getDSLContext().select(params).whereNotExists(notExistsSelect);

        try {
            int[] result = jooqProvider.getDSLContext().batch(
                jooqProvider.getDSLContext().update(table).set(recordValuesMap).where(condition),
                jooqProvider.getDSLContext().insertInto(table, fields).select(insertIntoSelect)
            ).execute();

            long rowsAffectedTotal = 0;
            for (int rowsAffected : result) {
                rowsAffectedTotal += rowsAffected;
            }

            if (rowsAffectedTotal != 1) {
                throw new RuntimeException("Upsert must only affect 1 row. Affected: " + rowsAffectedTotal + ". Table: " + table + ". Condition: " + condition);
            }
        } catch (DataAccessException e) {
            if (e.getCause() instanceof BatchUpdateException) {
                BatchUpdateException cause = (BatchUpdateException)e.getCause();

                logger.error("Batch update error in upsert.", cause.getNextException());
            }

            throw e;
        }
    }
}

然而，List<FieldValue<Field<?>, ?>> recordValues 参数并不太清晰。有没有更好的想法来解决这个问题？

- uldall

---

看起来val()静态方法可能是我正在寻找的：http://www.jooq.org/javadoc/3.3.x/org/jooq/impl/DSL.html#val%28java.lang.Object,%20java.lang.Class%29 - uldall

你知道为了让它可靠地工作，你必须在重试循环中使用SERIALIZABLE事务或锁定表格，对吧？ - Craig Ringer

@CraigRinger 是的，我知道。你能推荐一个更好的方法吗？我对这个问题的更好解决方案非常感兴趣。 - uldall

SQL语句 select 3,'C','Z'; 和 values (3,'C','Z'); 差别不大，因此使用 val() 方法是正确的选择。 - Jasen

FYI，PostgreSQL 9.5将拥有UPSERT功能。https://wiki.postgresql.org/wiki/UPSERT - Roger

Answer 1

• https://github.com/jOOQ/jOOQ/issues/4299
• http://www.postgresql.org/docs/9.5/static/sql-insert.html

MySQL的INSERT .. ON DUPLICATE KEY UPDATE：

DSL.using(configuration)
   .insertInto(TABLE)
   .columns(ID, A, B)
   .values(1, "a", "b")
   .onDuplicateKeyUpdate()
   .set(A, "a")
   .set(B, "b")
   .execute();

H2 MERGE INTO ..

->

H2 MERGE INTO ..

DSL.using(configuration)
   .mergeInto(TABLE, A, B, C)
   .values(1, "a", "b")
   .execute();

- Lukas Eder

---

基于Lucas解决方案的TableRecord对象实用程序： return dslContext.insertInto(record.getTable()).set(record).onDuplicateKeyUpdate().set(record).execute(); }``` - ud3sh

@ud3sh：嘿，谢谢提及。随意将其作为完整答案提供。我相信这对其他人可能也有用！ - Lukas Eder

Answer 2

这是一个针对可更新记录对象的upsert实用方法，基于Lucas上面提供的解决方案：

public static int upsert(final DSLContext dslContext, final UpdatableRecord record) {
    return dslContext.insertInto(record.getTable())
                     .set(record)
                     .onDuplicateKeyUpdate()
                     .set(record)
                     .execute();
}

- ud3sh

---

不适用于PostrgreSQL（实际上是在PostgreSQL模式下的H2） - pstobiecki

Answer 3

受 @ud3sh 有关 JOOQ 3.11、Kotlin 和 PostgreSQL DSL 的评论启发。

这是一个扩展函数，可以直接在 UpdatableRecord 对象上调用 upsert。

import org.jooq.UpdatableRecord

internal fun UpdatableRecord<*>.upsert(): Int {
    if(this.configuration() == null) {
        throw NullPointerException("Attach configuration to record before calling upsert")
    }
    return this.configuration().dsl().insertInto(this.getTable()).set(this).onConflict().doUpdate().set(this).execute()
}

- frno

Answer 4

似乎这是一种实现目标的复杂方式。为什么不使用一个简单的存储函数？如何创建一个upsert函数在postgresql手册中有描述，然后只需从您的Java代码调用它即可。

- e4c5

---

现在（直到9.5发布），实现upsert的唯一正确方法是使用循环存储函数，如链接中所述。更多信息请参见http://www.depesz.com/2012/06/10/why-is-upsert-so-complicated/。 - alexius