Bardzo często zadawanym pytaniem jest tutaj, jak zrobić upsert, czyli to, co wywołuje MySQL, INSERT ... ON DUPLICATE UPDATEa standardowe obsługuje jako częśćMERGE operacji.

Biorąc pod uwagę, że PostgreSQL nie obsługuje go bezpośrednio (przed pg 9.5), jak to zrobić? Rozważ następujące:

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

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

Teraz wyobraź sobie, że chcesz „upsert” krotek (2, 'Joe'), (3, 'Alan')tak nowe zawartość tabeli będzie:

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

O tym ludzie rozmawiają podczas dyskusji upsert. Co najważniejsze, każde podejście musi być bezpieczne w obecności wielu transakcji pracujących przy tym samym stole - albo przez jawne blokowanie, albo w inny sposób bronić się przed wynikowymi warunkami wyścigu.

Ten temat jest obszernie omawiany na stronie Insert, przy zduplikowanej aktualizacji w PostgreSQL?, ale chodzi o alternatywy dla składni MySQL, az czasem pojawiło się sporo niepowiązanych szczegółów. Pracuję nad ostatecznymi odpowiedziami.

Techniki te są również przydatne do „wstaw, jeśli nie istnieje, w przeciwnym razie nic nie rób”, tj. „Wstaw ... przy zduplikowanym kluczu ignoruj”.

9.5 i nowsze:

PostgreSQL 9.5 i nowsze wsparcie INSERT ... ON CONFLICT UPDATE(i ON CONFLICT DO NOTHING), tj. Upsert.

Porównanie zON DUPLICATE KEY UPDATE .

Szybkie wyjaśnienie .

Aby zapoznać się z użytkowaniem, zobacz instrukcję - w szczególności klauzulę o konfliktach działań na schemacie składni oraz tekst wyjaśniający .

W przeciwieństwie do rozwiązań dla wersji 9.4 i starszych, które podano poniżej, ta funkcja działa z wieloma sprzecznymi wierszami i nie wymaga wyłącznego blokowania ani ponownej próby.

Zatwierdzenie dodające funkcję jest tutaj, a dyskusja wokół jej rozwoju jest tutaj .

Jeśli korzystasz z wersji 9.5 i nie musisz być kompatybilny wstecz, możesz teraz przestać czytać .

9.4 i starsze:

PostgreSQL nie ma żadnej wbudowanej UPSERT(lub MERGE) funkcji, a robienie tego skutecznie w obliczu równoczesnego użycia jest bardzo trudne.

W tym artykule szczegółowo omówiono problem .

Ogólnie rzecz biorąc, musisz wybrać jedną z dwóch opcji:

• Indywidualne operacje wstawiania / aktualizacji w pętli ponownych prób; lub
• Blokowanie stołu i łączenie partii

Pętla ponawiania pojedynczych wierszy

Użycie pojedynczych poprawek wierszy w pętli ponownej próby jest rozsądną opcją, jeśli chcesz, aby wiele połączeń jednocześnie próbowało wykonać wstawienia.

Dokumentacja PostgreSQL zawiera przydatną procedurę, która pozwoli ci to zrobić w pętli wewnątrz bazy danych . Chroni przed utraconymi aktualizacjami i wstawia rasy, w przeciwieństwie do większości naiwnych rozwiązań. Działa tylko w READ COMMITTEDtrybie i jest bezpieczny tylko wtedy, gdy jest to jedyna rzecz, którą robisz w transakcji. Funkcja nie będzie działać poprawnie, jeśli wyzwalacze lub dodatkowe unikalne klucze powodują unikalne naruszenia.

Ta strategia jest bardzo nieefektywna. Kiedy tylko jest to praktyczne, powinieneś ustawiać w kolejce prace i wykonać zbiorczy upsert, jak opisano poniżej.

Wiele prób rozwiązania tego problemu nie uwzględnia wycofania, więc powodują niekompletne aktualizacje. Dwie transakcje ścigają się ze sobą; jeden z nich pomyślnie INSERT; drugi otrzymuje duplikat błędu klucza i UPDATEzamiast tego robi błąd . Te UPDATEbloki czekają na INSERTwycofywania lub zatwierdzenia. Kiedy się wycofuje, UPDATEwarunek ponownego sprawdzania dopasowuje zero wierszy, więc nawet jeśli UPDATEzatwierdzenia nie wykonały oczekiwanego upsert. Musisz sprawdzić liczbę wierszy wyników i spróbować ponownie w razie potrzeby.

Niektóre próby rozwiązania nie uwzględniają również wyścigów SELECT. Jeśli spróbujesz oczywistego i prostego:

-- 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;

wtedy, gdy dwa działają jednocześnie, istnieje kilka trybów awarii. Jednym z nich jest już omówiony problem z ponownym sprawdzaniem aktualizacji. Innym jest, gdy oba UPDATEjednocześnie, dopasowując zero wierszy i kontynuując. Potem oboje zrobić EXISTStest, który dzieje się przedINSERT . Oba mają zero wierszy, więc oba robią INSERT. Jeden kończy się niepowodzeniem z duplikatem błędu klucza.

Dlatego potrzebujesz pętli ponownej próby. Możesz pomyśleć, że możesz zapobiec zduplikowanym kluczowym błędom lub utracie aktualizacji dzięki sprytnemu SQL, ale nie możesz. Musisz sprawdzić liczbę wierszy lub obsłużyć zduplikowane błędy klucza (w zależności od wybranego podejścia) i spróbować ponownie.

Nie rzucaj na to własnym rozwiązaniem. Podobnie jak w przypadku kolejkowania wiadomości, prawdopodobnie jest to złe.

Masywny upsert z zamkiem

Czasami chcesz zrobić zbiorczy upsert, w którym masz nowy zestaw danych, który chcesz scalić ze starszym istniejącym zestawem danych. Jest to znacznie bardziej wydajne niż wstawianie pojedynczych rzędów i powinno być preferowane, gdy jest to praktyczne.

W takim przypadku zazwyczaj postępujesz według następującego procesu:

• CREATETEMPORARYstół

• COPY lub wstaw zbiorczo nowe dane do tabeli temp

• LOCKtabela docelowa IN EXCLUSIVE MODE. Pozwala to na inne transakcje SELECT, ale nie wprowadza żadnych zmian w tabeli.

• Wykonaj jeden UPDATE ... FROMz istniejących rekordów, używając wartości z tabeli temp;

• Wykonaj jeden INSERTz wierszy, które jeszcze nie istnieją w tabeli docelowej;

• COMMIT, zwalniając blokadę.

Na przykład w przykładzie podanym w pytaniu, używając wielowartościowego INSERTdo wypełnienia tabeli 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;

Powiązane czytanie

• Strona wiki UPSERT
• UPSERTisms in Postgres
• Wstawić po powieleniu aktualizacji w PostgreSQL?
• http://petereisentraut.blogspot.com/2010/05/merge-syntax.html
• Upsert z transakcją
• Czy funkcja SELECT lub INSERT ma funkcję podatną na warunki wyścigowe?
• SQL MERGEna wiki PostgreSQL
• Najbardziej idiomatyczny sposób na wdrożenie UPSERT w Postgresql w dzisiejszych czasach

Co MERGE?

Standard SQL MERGEma właściwie źle zdefiniowaną semantykę współbieżności i nie nadaje się do upserowania bez uprzedniego zablokowania tabeli.

To naprawdę przydatna instrukcja OLAP do łączenia danych, ale tak naprawdę nie jest to przydatne rozwiązanie dla bezpiecznego współbieżności. Istnieje wiele porad dla osób korzystających z innych DBMS do używania MERGEupserts, ale tak naprawdę jest to złe.

Inne bazy danych:

• INSERT ... ON DUPLICATE KEY UPDATE w MySQL
• MERGEz MS SQL Server (ale patrz wyżej o MERGEproblemach)
• MERGEz Oracle (ale patrz wyżej o MERGEproblemach)

Próbuję wnieść wkład w inne rozwiązanie problemu pojedynczego wstawiania w wersjach PostgreSQL wcześniejszych niż 9.5. Chodzi o to, aby po prostu spróbować wykonać najpierw wstawienie, a jeśli rekord już istnieje, zaktualizuj go:

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

Pamiętaj, że to rozwiązanie można zastosować tylko wtedy, gdy nie zostaną usunięte wiersze tabeli .

Nie wiem o skuteczności tego rozwiązania, ale wydaje mi się to dość rozsądne.

Oto kilka przykładów insert ... on conflict ...( str. 9.5+ ):

• Wstaw w razie konfliktu - nie rób nic .

  insert into dummy(id, name, size) values(1, 'new_name', 3)
  on conflict do nothing;`  

• Wstaw, w przypadku konfliktu - wykonaj aktualizację , określ konflikt docelowy za pomocą kolumny .

  insert into dummy(id, name, size) values(1, 'new_name', 3)
  on conflict(id)
  do update set name = 'new_name', size = 3;  

• Wstaw, w przypadku konfliktu - wykonaj aktualizację , określ cel konfliktu za pomocą nazwy ograniczenia .

  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;

Wprowadzenie SQLAlchemy dla Postgres> = 9.5

Ponieważ powyższy duży post obejmuje wiele różnych podejść SQL do wersji Postgres (nie tylko nie-9.5 jak w pytaniu), chciałbym dodać, jak to zrobić w SQLAlchemy, jeśli używasz Postgres 9.5. Zamiast implementować własny upsert, możesz także użyć funkcji SQLAlchemy (które zostały dodane w SQLAlchemy 1.1). Osobiście polecam korzystanie z nich, jeśli to możliwe. Nie tylko ze względu na wygodę, ale także dlatego, że pozwala PostgreSQLowi obsługiwać wszelkie warunki wyścigowe, które mogą wystąpić.

Cross-posting z innej odpowiedzi, którą wczoraj podałem ( https://stackoverflow.com/a/44395983/2156909 )

SQLAlchemy obsługuje ON CONFLICTteraz dwie metody on_conflict_do_update()i on_conflict_do_nothing():

Kopiowanie z dokumentacji:

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

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

Testowane na Postgresql 9.3

Ponieważ to pytanie zostało zamknięte, piszę tutaj o tym, jak to zrobić za pomocą SQLAlchemy. Poprzez rekurencję ponawia próbę włożenia lub aktualizacji zbiorczej w celu zwalczania warunków wyścigu i błędów walidacji.

Najpierw import

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

Teraz działa kilka pomocników

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

I wreszcie funkcja 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]))

Oto jak z niego korzystasz

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

Zaletą tego jest bulk_save_objectsto, że może obsługiwać relacje, sprawdzanie błędów itp. Podczas wstawiania (w przeciwieństwie do operacji masowych ).