Skip to content

Commit 99cea49

Browse files
tvondratvondra
committed
Fix copying data into slots with FDW batching
Commit b676ac4 optimized handling of tuple slots with bulk inserts into foreign tables, so that the slots are initialized only once and reused for all batches. The data was however copied into the slots only after the initialization, inserting duplicate values when the slot gets reused. Fixed by moving the ExecCopySlot outside the init branch. The existing postgres_fdw tests failed to catch this due to inserting data into foreign tables without unique indexes, and then checking only the number of inserted rows. This adds a new test with both a unique index and a check of inserted values. Reported-by: Alexander Pyhalov Discussion: https://postgr.es/m/7a8cf8d56b3d18e5c0bccd6cd42d04ac%40postgrespro.ru
1 parent 6b787d9 commit 99cea49

File tree

3 files changed

+115
-6
lines changed

3 files changed

+115
-6
lines changed

‎contrib/postgres_fdw/expected/postgres_fdw.out

Lines changed: 81 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9761,7 +9761,87 @@ SELECT tableoid::regclass, * FROM batch_cp_upd_test;
97619761
(2 rows)
97629762

97639763
-- Clean up
9764-
DROP TABLE batch_table, batch_cp_upd_test CASCADE;
9764+
DROP TABLE batch_table, batch_cp_upd_test, batch_table_p0, batch_table_p1 CASCADE;
9765+
-- Use partitioning
9766+
ALTER SERVER loopback OPTIONS (ADD batch_size '10');
9767+
CREATE TABLE batch_table ( x int, field1 text, field2 text) PARTITION BY HASH (x);
9768+
CREATE TABLE batch_table_p0 (LIKE batch_table);
9769+
ALTER TABLE batch_table_p0 ADD CONSTRAINT p0_pkey PRIMARY KEY (x);
9770+
CREATE FOREIGN TABLE batch_table_p0f
9771+
PARTITION OF batch_table
9772+
FOR VALUES WITH (MODULUS 2, REMAINDER 0)
9773+
SERVER loopback
9774+
OPTIONS (table_name 'batch_table_p0');
9775+
CREATE TABLE batch_table_p1 (LIKE batch_table);
9776+
ALTER TABLE batch_table_p1 ADD CONSTRAINT p1_pkey PRIMARY KEY (x);
9777+
CREATE FOREIGN TABLE batch_table_p1f
9778+
PARTITION OF batch_table
9779+
FOR VALUES WITH (MODULUS 2, REMAINDER 1)
9780+
SERVER loopback
9781+
OPTIONS (table_name 'batch_table_p1');
9782+
INSERT INTO batch_table SELECT i, 'test'||i, 'test'|| i FROM generate_series(1, 50) i;
9783+
SELECT COUNT(*) FROM batch_table;
9784+
count
9785+
-------
9786+
50
9787+
(1 row)
9788+
9789+
SELECT * FROM batch_table ORDER BY x;
9790+
x | field1 | field2
9791+
----+--------+--------
9792+
1 | test1 | test1
9793+
2 | test2 | test2
9794+
3 | test3 | test3
9795+
4 | test4 | test4
9796+
5 | test5 | test5
9797+
6 | test6 | test6
9798+
7 | test7 | test7
9799+
8 | test8 | test8
9800+
9 | test9 | test9
9801+
10 | test10 | test10
9802+
11 | test11 | test11
9803+
12 | test12 | test12
9804+
13 | test13 | test13
9805+
14 | test14 | test14
9806+
15 | test15 | test15
9807+
16 | test16 | test16
9808+
17 | test17 | test17
9809+
18 | test18 | test18
9810+
19 | test19 | test19
9811+
20 | test20 | test20
9812+
21 | test21 | test21
9813+
22 | test22 | test22
9814+
23 | test23 | test23
9815+
24 | test24 | test24
9816+
25 | test25 | test25
9817+
26 | test26 | test26
9818+
27 | test27 | test27
9819+
28 | test28 | test28
9820+
29 | test29 | test29
9821+
30 | test30 | test30
9822+
31 | test31 | test31
9823+
32 | test32 | test32
9824+
33 | test33 | test33
9825+
34 | test34 | test34
9826+
35 | test35 | test35
9827+
36 | test36 | test36
9828+
37 | test37 | test37
9829+
38 | test38 | test38
9830+
39 | test39 | test39
9831+
40 | test40 | test40
9832+
41 | test41 | test41
9833+
42 | test42 | test42
9834+
43 | test43 | test43
9835+
44 | test44 | test44
9836+
45 | test45 | test45
9837+
46 | test46 | test46
9838+
47 | test47 | test47
9839+
48 | test48 | test48
9840+
49 | test49 | test49
9841+
50 | test50 | test50
9842+
(50 rows)
9843+
9844+
ALTER SERVER loopback OPTIONS (DROP batch_size);
97659845
-- ===================================================================
97669846
-- test asynchronous execution
97679847
-- ===================================================================

‎contrib/postgres_fdw/sql/postgres_fdw.sql

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3083,7 +3083,34 @@ UPDATE batch_cp_upd_test t SET a = 1 FROM (VALUES (1), (2)) s(a) WHERE t.a = s.a
30833083
SELECT tableoid::regclass, * FROM batch_cp_upd_test;
30843084

30853085
-- Clean up
3086-
DROP TABLE batch_table, batch_cp_upd_test CASCADE;
3086+
DROP TABLE batch_table, batch_cp_upd_test, batch_table_p0, batch_table_p1 CASCADE;
3087+
3088+
-- Use partitioning
3089+
ALTER SERVER loopback OPTIONS (ADD batch_size '10');
3090+
3091+
CREATE TABLE batch_table ( x int, field1 text, field2 text) PARTITION BY HASH (x);
3092+
3093+
CREATE TABLE batch_table_p0 (LIKE batch_table);
3094+
ALTER TABLE batch_table_p0 ADD CONSTRAINT p0_pkey PRIMARY KEY (x);
3095+
CREATE FOREIGN TABLE batch_table_p0f
3096+
PARTITION OF batch_table
3097+
FOR VALUES WITH (MODULUS 2, REMAINDER 0)
3098+
SERVER loopback
3099+
OPTIONS (table_name 'batch_table_p0');
3100+
3101+
CREATE TABLE batch_table_p1 (LIKE batch_table);
3102+
ALTER TABLE batch_table_p1 ADD CONSTRAINT p1_pkey PRIMARY KEY (x);
3103+
CREATE FOREIGN TABLE batch_table_p1f
3104+
PARTITION OF batch_table
3105+
FOR VALUES WITH (MODULUS 2, REMAINDER 1)
3106+
SERVER loopback
3107+
OPTIONS (table_name 'batch_table_p1');
3108+
3109+
INSERT INTO batch_table SELECT i, 'test'||i, 'test'|| i FROM generate_series(1, 50) i;
3110+
SELECT COUNT(*) FROM batch_table;
3111+
SELECT * FROM batch_table ORDER BY x;
3112+
3113+
ALTER SERVER loopback OPTIONS (DROP batch_size);
30873114

30883115
-- ===================================================================
30893116
-- test asynchronous execution

‎src/backend/executor/nodeModifyTable.c

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -717,18 +717,20 @@ ExecInsert(ModifyTableState *mtstate,
717717

718718
resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots] =
719719
MakeSingleTupleTableSlot(tdesc, slot->tts_ops);
720-
ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
721-
slot);
722720

723721
resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots] =
724722
MakeSingleTupleTableSlot(tdesc, planSlot->tts_ops);
725-
ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
726-
planSlot);
727723

728724
/* remember how many batch slots we initialized */
729725
resultRelInfo->ri_NumSlotsInitialized++;
730726
}
731727

728+
ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
729+
slot);
730+
731+
ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
732+
planSlot);
733+
732734
resultRelInfo->ri_NumSlots++;
733735

734736
MemoryContextSwitchTo(oldContext);

0 commit comments

Comments
 (0)