🐘🚀 JSON 페이로드에서 업데이트/삽입/일시 삭제

사용자는 자신의 "ETL"사용 사례를 다음과 같이 노출했습니다.

거래 테이블은 ID(내 예에서는 k1 int)를 보유합니다. 여러 레코드(키에 대한 k2 int 및 값에 대한 v1 int , v2 int 내 작은 예에서)

애플리케이션이 새 데이터를 설명하는 JSON 형식의 페이로드를 수신합니다

.

새 ID가 삽입됩니다

.

기존 ID가 업데이트됨

누락된 ID는 삭제된 것으로 표시됩니다( deleted boolean )

마지막 변경 시간

을 기록하기 위해 타임스탬프ts timestamptz를 추가합니다.

그들의 현재 애플리케이션은 Python Pandas를 사용하여 현재 레코드를 검색하고 새 페이로드와 비교하고 데이터베이스에 다시 씁니다. YugabyteDB와 같은 분산 데이터베이스를 사용하면 데이터베이스에서 이를 처리하는 것이 더 효율적입니다. 왜냐하면 확장이 가능하고(모든 노드에 연결할 수 있음) 하나의 서버 측 트랜잭션에서 실행될 수 있기 때문입니다(클럭 스큐의 경우 투명한 재시도를 처리하기가 더 쉽습니다). 왕복을 줄입니다(행 배치 작업). 그리고 PostgreSQL 호환성을 통해 JSON 처리 및 INSERT ... ON CONFLICT ... UPDATE 동작은 깨끗하고 간단한 SQL로 동일한 작업을 수행할 수 있습니다.

예시

다음은 내 테이블을 선언하는 간단한 예입니다.

create table demo (
 k1 int, k2 int, v1 int, v2 text
 , deleted boolean, ts timestamptz
 , primary key ( k1,k2 )
);

이것은 PostgreSQL 및 YugabyteDB에서 작동합니다. YugabyteDB에서 기본 키는 기본적으로 lsm (k1 HASH, k2 ASC)로 설정되어 있어 ID에 배포하고 레코드를 같은 위치에 배치하는 제 사용 사례에 적합합니다.

하나의 SQL 쿼리에서 전체 논리를 구현하겠습니다. WITH 절(일명 CTE - Common Table Expressions)은 가독성을 유지합니다.

payload는 $2 덕분에 레코드에 jsonb_to_recordset로 전달된 내 JSON 페이로드를 읽고 ID$1 as k1를 추가합니다.

to_upsert는 payload를 대상 형식으로 지정하고 is_deleted를 false로 설정하고 타임스탬프

를 추가합니다.

to_soft_delete는 현재 레코드를 검색하고 payload와 비교하여 is_deleted가 true

로 설정된 일시 삭제로 형식을 지정합니다.

마지막으로 to_upsert와 to_soft_delete의 결합이 INSERT ... ON CONFLICT ... UPDATE

와 병합됩니다.

다음은 준비된 진술입니다.

prepare etl (int, jsonb) as
with
payload as (
 select $1 as k1,* from jsonb_to_recordset($2) as payload( k2 int, v1 int, v2 text )
),
to_upsert as (
 select k1, k2, v1, v2, false as deleted, now() as ts from payload
),
to_soft_delete as (
 select k1, k2, v1, v2, true  as deleted, now() as ts from demo
 where  k1=$1
 and (k1, k2) not in ( select k1, k2 from payload)
)
insert into demo select * from to_upsert union select * from to_soft_delete
on conflict (k1, k2) do
update set v1=excluded.v1, v2=excluded.v2, deleted=excluded.deleted, ts=excluded.ts
;

이것은 저장 프로시저로 코딩될 수 있지만 준비된 명령문은 각 호출에 대해 이를 구문 분석하고 최적화하지 않습니다. 연결 풀 초기화 명령에서 쉽게 준비할 수 있습니다.

내 테이블이 비어 있습니다. 3개의 레코드가 있는 k1=0에 대한 페이로드를 삽입하겠습니다.

execute etl(0, $$
[
{"k2":1,"v1":1,"v2":"my first insert"},
{"k2":2,"v1":1,"v2":"my first insert"},
{"k2":3,"v1":1,"v2":"my first insert"}
]
$$::jsonb);

3개의 행이 삽입되었습니다.

yugabyte=# select * from demo;
 k1 | k2 | v1 |       v2        | deleted |              ts
----+----+----+-----------------+---------+-------------------------------
  0 |  1 |  1 | my first insert | f       | 2022-04-29 14:39:08.467744+00
  0 |  2 |  1 | my first insert | f       | 2022-04-29 14:39:08.467744+00
  0 |  3 |  1 | my first insert | f       | 2022-04-29 14:39:08.467744+00
(3 rows)

이제 동일한k1=0에서 새 페이로드가 2개의 행을 업데이트하고 다른 행을 제거하고 새 행을 추가합니다.

execute etl(0,$$
[
{"k2":1,"v1":1,"v2":"my update"},
{"k2":2,"v1":1,"v2":"my update"},
{"k2":4,"v1":1,"v2":"my second insert"}
]
$$::jsonb);

결과는 다음과 같습니다.

yugabyte=# select * from demo;
 k1 | k2 | v1 |        v2        | deleted |              ts
----+----+----+------------------+---------+-------------------------------
  0 |  1 |  1 | my update        | f       | 2022-04-29 14:41:00.109561+00
  0 |  2 |  1 | my update        | f       | 2022-04-29 14:41:00.109561+00
  0 |  3 |  1 | my first insert  | t       | 2022-04-29 14:41:00.109561+00
  0 |  4 |  1 | my second insert | f       | 2022-04-29 14:41:00.109561+00
(4 rows)

성능

항상 실행 계획을 확인하십시오.

yugabyte=# explain (costs off, analyze) execute etl(0,$$                                                          [9/3108]
[
{"k2":1,"v1":1,"v2":"my update"},
{"k2":2,"v1":1,"v2":"my update"},
{"k2":4,"v1":1,"v2":"my second insert"}
]
$$::jsonb);
                                                 QUERY PLAN
------------------------------------------------------------------------------------------------------------
 Insert on demo (actual time=11.787..11.787 rows=0 loops=1)
   Conflict Resolution: UPDATE
   Conflict Arbiter Indexes: demo_pkey
   Tuples Inserted: 0
   Conflicting Tuples: 4
   CTE payload
     ->  Function Scan on jsonb_to_recordset payload (actual time=0.011..0.012 rows=3 loops=1)
   CTE to_upsert
     ->  CTE Scan on payload payload_1 (actual time=0.013..0.014 rows=3 loops=1)
   CTE to_soft_delete
     ->  Index Scan using demo_pkey on demo demo_1 (actual time=1.079..1.081 rows=1 loops=1)
           Index Cond: (k1 = 0)
           Filter: (NOT (hashed SubPlan 3))
           Rows Removed by Filter: 3
           SubPlan 3
             ->  CTE Scan on payload payload_2 (actual time=0.001..0.001 rows=3 loops=1)
   ->  HashAggregate (actual time=1.105..1.110 rows=4 loops=1)
         Group Key: to_upsert.k1, to_upsert.k2, to_upsert.v1, to_upsert.v2, to_upsert.deleted, to_upsert.ts
         ->  Append (actual time=0.014..1.099 rows=4 loops=1)
               ->  CTE Scan on to_upsert (actual time=0.014..0.016 rows=3 loops=1)
               ->  CTE Scan on to_soft_delete (actual time=1.080..1.082 rows=1 loops=1)
 Planning Time: 0.262 ms
 Execution Time: 11.864 ms
(23 rows)

이것은 Index Cond: (k1 = 0) 에 대한 하나의 태블릿만 읽고 모든 행이 함께 배치된 기본 키에 Index Scan 가 있습니다. 여기서 (k1 HASH, k2 ASC)가 중요합니다.

물론 이것은 동일한 값을 업데이트하지 않는 것과 같이 최적화될 수 있습니다. 그러나 코드를 단순하게 유지하는 것이 최선의 선택일 것입니다. 최적화는 보다 정확한 사용 사례에 따라 달라집니다.

이 쿼리는 정보 스키마에서 쉽게 생성할 수도 있습니다. 열 정의( k1 HASH, k2 ASC ), 열 목록( k1, k2, v1, v2 ), 키 열( k1, k2 ) 및 집합 절( v1=excluded.v1, v2=excluded.v2 )이 필요합니다.