¿Cómo UPSERT (COMBINAR, INSERTAR ON EN LA ACTUALIZACIÓN DUPLICADA)en PostgreSQL?


Una pregunta muy frecuente aquí es cómo hacer un upsert, que es lo que MySQL llama INSERT ... ON DUPLICATE UPDATE y el estándar soporta como parte de la operación MERGE.

Dado que PostgreSQL no lo soporta directamente (antes de pg 9.5), ¿cómo lo hace? Considere lo siguiente:

CREATE TABLE testtable (
    id integer PRIMARY KEY,
    somedata text NOT NULL
);

INSERT INTO testtable (id, somedata) VALUES
(1, 'fred'),
(2, 'bob');

Ahora imagina que quieres "upsert" las tuplas (2, 'Joe'), (3, 'Alan'), así que el nuevo contenido de la tabla sería:

(1, 'fred'),
(2, 'Joe'),    -- Changed value of existing tuple
(3, 'Alan')    -- Added new tuple

Eso es de lo que la gente está hablando cuando se habla de un upsert. Fundamentalmente, cualquier la aproximación debe ser segura en presencia de múltiples transacciones que trabajen en la misma tabla - ya sea mediante el uso de bloqueo explícito, o de otra manera defendiéndose contra las condiciones de carrera resultantes.

Este tema se discute extensamente en Insert, sobre la actualización duplicada en PostgreSQL?, pero se trata de alternativas a la sintaxis de MySQL, y ha crecido un poco de detalle sin relación con el tiempo. Estoy trabajando en respuestas definitivas.

Estas técnicas también son útiles para "insértese si no existe, de lo contrario no hacer nada", es decir, "insértese ... on duplicate key ignore"(en inglés).

Author: Community, 2013-06-24

6 answers

9.5 y posteriores:

PostgreSQL 9.5 y soporte posterior INSERT ... ON CONFLICT UPDATE (y ON CONFLICT DO NOTHING), es decir, upsert.

Comparación con ON DUPLICATE KEY UPDATE.

Explicación rápida .

Para el uso ver el manual - específicamente la cláusula conflict_action en el diagrama de sintaxis, y el texto explicativo.

A diferencia de las soluciones para 9.4 y anteriores que se dan a continuación, esta característica funciona con múltiples filas en conflicto y no requiere bloqueo exclusivo o un bucle de reintento.

El commit que agrega la característica está aquí y la discusión en torno a su desarrollo está aquí.


Si está en la versión 9.5 y no necesita ser compatible con versiones anteriores, puede dejar de leer ahora .


9.4 y anteriores:

PostgreSQL no tiene ninguna instalación incorporada UPSERT (o MERGE), y hacerlo eficientemente frente al uso concurrente es muy difícil.

Esto artículo discute el problema en detalle útil .

En general, debe elegir entre dos opciones:

  • Operaciones individuales de inserción / actualización en un bucle de reintentos; o
  • Bloquear la tabla y hacer batch merge

Bucle de reintento de fila individual

Usar upserts de fila individuales en un bucle de reintento es la opción razonable si desea que muchas conexiones intenten realizar inserciones simultáneamente.

La documentación de PostgreSQL contiene un procedimiento útil que le permitirá hacer esto en un bucle dentro de la base de datos . Protege contra las actualizaciones perdidas y las carreras de inserción, a diferencia de la mayoría de las soluciones ingenuas. Solo funcionará en modo READ COMMITTED y solo es seguro si es lo único que haces en la transacción. La función no funcionará correctamente si los disparadores o las claves únicas secundarias causan violaciones únicas.

Esta estrategia es muy ineficiente. Siempre que sea práctico, debe hacer cola para trabajar y hacer un upsert masivo como se describe a continuación en su lugar.

Muchos intentos de solución a este problema no consideran las reversiones, por lo que resultan en actualizaciones incompletas. Dos transacciones compiten entre sí; una de ellas con éxito INSERTs; la otra obtiene un error de clave duplicada y hace un UPDATE en su lugar. El UPDATE se bloquea esperando a que el INSERT se revierta o confirme. Cuando se devuelve, la condición UPDATE re-check coincide con cero filas, por lo que a pesar de que la UPDATE commits en realidad no ha hecho el upsert que esperaba. Tienes que comprobar el recuento de filas de resultados y vuelva a intentarlo cuando sea necesario.

Algunas soluciones intentadas tampoco consideran razas selectas. Si intentas lo obvio y lo simple:

-- THIS IS WRONG. DO NOT COPY IT. It's an EXAMPLE.

BEGIN;

UPDATE testtable
SET somedata = 'blah'
WHERE id = 2;

-- Remember, this is WRONG. Do NOT COPY IT.

INSERT INTO testtable (id, somedata)
SELECT 2, 'blah'
WHERE NOT EXISTS (SELECT 1 FROM testtable WHERE testtable.id = 2);

COMMIT;

Luego, cuando dos se ejecutan a la vez, hay varios modos de falla. Uno es el problema ya discutido con una revisión de actualización. Otro es donde ambos UPDATE al mismo tiempo, coincidiendo con cero filas y continuando. Luego ambos hacen la prueba EXISTS, que sucede antes de el INSERT. Ambos obtienen cero filas, por lo que ambos hacen el INSERT. Uno falla con un error de clave duplicada.

Esta es la razón por la que necesita un bucle de reinicio. Puede pensar que puede evitar errores de clave duplicada o actualizaciones perdidas con clever SQL, pero no puede. Debe verificar el recuento de filas o manejar errores de clave duplicada (dependiendo del enfoque elegido) y volver a intentarlo.

Por favor, no ruede su propia solución para esto. Como con la cola de mensajes, probablemente esté mal.

Upsert a granel con bloqueo

A veces quieres hacer un upsert a granel, donde tiene un nuevo conjunto de datos que desea fusionar en un conjunto de datos existente más antiguo. Esto es enormemente más eficiente que los upserts de fila individuales y debe ser preferido siempre que sea práctico.

En este caso, normalmente se sigue el siguiente proceso:

  • CREATE a TEMPORARY cuadro

  • COPY o bulk-inserte los nuevos datos en la tabla temp

  • LOCK la tabla de destino IN EXCLUSIVE MODE. Esto permite otras transacciones a SELECT, pero no hacer ningún cambio en la tabla.

  • Haga un UPDATE ... FROM de registros existentes usando los valores en la tabla temp;

  • Haga un INSERT de filas que no existen ya en la tabla de destino;

  • COMMIT, liberando la cerradura.

Por ejemplo, para el ejemplo dado en la pregunta, usando multi-valued INSERT para llenar la tabla de temp:

BEGIN;

CREATE TEMPORARY TABLE newvals(id integer, somedata text);

INSERT INTO newvals(id, somedata) VALUES (2, 'Joe'), (3, 'Alan');

LOCK TABLE testtable IN EXCLUSIVE MODE;

UPDATE testtable
SET somedata = newvals.somedata
FROM newvals
WHERE newvals.id = testtable.id;

INSERT INTO testtable
SELECT newvals.id, newvals.somedata
FROM newvals
LEFT OUTER JOIN testtable ON (testtable.id = newvals.id)
WHERE testtable.id IS NULL;

COMMIT;

Lectura relacionada

¿Qué pasa con MERGE?

SQL-standard MERGE en realidad tiene una semántica de concurrencia mal definida y no es adecuada para upserting sin bloquear una tabla primero.

Es una declaración OLAP realmente útil para la fusión de datos, pero en realidad no es una solución útil para upsert seguro de concurrencia. Hay muchos consejos para las personas que usan otros DBMSes para usar MERGE para upserts, pero en realidad está mal.

Otras ECD:

 335
Author: Craig Ringer,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2017-05-31 14:28:18

Estoy tratando de contribuir con otra solución para el problema de inserción única con las versiones pre-9.5 de PostgreSQL. La idea es simplemente intentar realizar primero la inserción, y en caso de que el registro ya esté presente, actualizarlo:

do $$
begin 
  insert into testtable(id, somedata) values(2,'Joe');
exception when unique_violation then
  update testtable set somedata = 'Joe' where id = 2;
end $$;

Tenga en cuenta que esta solución se puede aplicar solo si no hay supresiones de filas de la tabla.

No conozco la eficacia de esta solución, pero me parece bastante razonable.

 25
Author: Renzo,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2016-08-22 08:11:02

Aquí hay algunos ejemplos para insert ... on conflict ... (pg 9.5+) :

  • Insértese, on conflict - do nothing.
    insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict do nothing;

  • Insert, on conflict - do update, specify conflict target via column.
    insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict(id) do update set name = 'new_name', size = 3;

  • Insert, on conflict - do update, especifique el destino del conflicto a través de nombre de la restricción.
    insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict on constraint dummy_pkey do update set name = 'new_name', size = 4;

 4
Author: Eric Wang,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2018-01-30 11:22:14
WITH UPD AS (UPDATE TEST_TABLE SET SOME_DATA = 'Joe' WHERE ID = 2 
RETURNING ID),
INS AS (SELECT '2', 'Joe' WHERE NOT EXISTS (SELECT * FROM UPD))
INSERT INTO TEST_TABLE(ID, SOME_DATA) SELECT * FROM INS

Probado en Postgresql 9.3

 3
Author: aristar,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2017-03-02 05:25:13

SQLAlchemy upsert para Postgres > = 9.5

Dado que la gran publicación anterior cubre muchos enfoques SQL diferentes para las versiones de Postgres (no solo no-9.5 como en la pregunta), me gustaría agregar cómo hacerlo en SQLAlchemy si está utilizando Postgres 9.5. En lugar de implementar su propio upsert, también puede usar las funciones de SQLAlchemy (que se agregaron en SQLAlchemy 1.1). Personalmente, recomendaría usar estos, si es posible. No solo por conveniencia, sino también porque permite PostgreSQL maneja cualquier condición de carrera que pueda ocurrir.

Publicación cruzada de otra respuesta que di ayer ( https://stackoverflow.com/a/44395983/2156909 )

SQLAlchemy soporta ON CONFLICT ahora con dos métodos on_conflict_do_update() y on_conflict_do_nothing():

Copiando de la documentación:

from sqlalchemy.dialects.postgresql import insert

stmt = insert(my_table).values(user_email='[email protected]', data='inserted data')
stmt = stmt.on_conflict_do_update(
    index_elements=[my_table.c.user_email],
    index_where=my_table.c.user_email.like('%@gmail.com'),
    set_=dict(data=stmt.excluded.data)
    )
conn.execute(stmt)

Http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html?highlight=conflict#insert-on-conflict-upsert

 2
Author: P.R.,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2017-06-07 09:38:49

Dado que esta pregunta se cerró, estoy publicando aquí para ver cómo se hace usando SQLAlchemy. Mediante recursión, reintenta una inserción o actualización masiva para combatir condiciones de carrera y errores de validación.

Primero las importaciones

import itertools as it

from functools import partial
from operator import itemgetter

from sqlalchemy.exc import IntegrityError
from app import session
from models import Posts

Ahora un par de funciones auxiliares

def chunk(content, chunksize=None):
    """Groups data into chunks each with (at most) `chunksize` items.
    https://stackoverflow.com/a/22919323/408556
    """
    if chunksize:
        i = iter(content)
        generator = (list(it.islice(i, chunksize)) for _ in it.count())
    else:
        generator = iter([content])

    return it.takewhile(bool, generator)


def gen_resources(records):
    """Yields a dictionary if the record's id already exists, a row object 
    otherwise.
    """
    ids = {item[0] for item in session.query(Posts.id)}

    for record in records:
        is_row = hasattr(record, 'to_dict')

        if is_row and record.id in ids:
            # It's a row but the id already exists, so we need to convert it 
            # to a dict that updates the existing record. Since it is duplicate,
            # also yield True
            yield record.to_dict(), True
        elif is_row:
            # It's a row and the id doesn't exist, so no conversion needed. 
            # Since it's not a duplicate, also yield False
            yield record, False
        elif record['id'] in ids:
            # It's a dict and the id already exists, so no conversion needed. 
            # Since it is duplicate, also yield True
            yield record, True
        else:
            # It's a dict and the id doesn't exist, so we need to convert it. 
            # Since it's not a duplicate, also yield False
            yield Posts(**record), False

Y finalmente la función upsert

def upsert(data, chunksize=None):
    for records in chunk(data, chunksize):
        resources = gen_resources(records)
        sorted_resources = sorted(resources, key=itemgetter(1))

        for dupe, group in it.groupby(sorted_resources, itemgetter(1)):
            items = [g[0] for g in group]

            if dupe:
                _upsert = partial(session.bulk_update_mappings, Posts)
            else:
                _upsert = session.add_all

            try:
                _upsert(items)
                session.commit()
            except IntegrityError:
                # A record was added or deleted after we checked, so retry
                # 
                # modify accordingly by adding additional exceptions, e.g.,
                # except (IntegrityError, ValidationError, ValueError)
                db.session.rollback()
                upsert(items)
            except Exception as e:
                # Some other error occurred so reduce chunksize to isolate the 
                # offending row(s)
                db.session.rollback()
                num_items = len(items)

                if num_items > 1:
                    upsert(items, num_items // 2)
                else:
                    print('Error adding record {}'.format(items[0]))

Así es como lo usas

>>> data = [
...     {'id': 1, 'text': 'updated post1'}, 
...     {'id': 5, 'text': 'updated post5'}, 
...     {'id': 1000, 'text': 'new post1000'}]
... 
>>> upsert(data)

La ventaja que esto tiene sobre bulk_save_objects es que puede manejar relaciones, error comprobación, etc. en insert (a diferencia de operaciones masivas).

 0
Author: reubano,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/ajaxhispano.com/template/agent.layouts/content.php on line 61
2017-05-23 12:02:50