Make INSERT-from-multiple-VALUES-rows handle domain target columns.

Commit a3c7a993d fixed some cases involving target columns that are
arrays or composites by applying transformAssignedExpr to the VALUES
entries, and then stripping off any assignment ArrayRefs or
FieldStores that the transformation added.  But I forgot about domains
over arrays or composites :-(.  Such cases would either fail with
surprising complaints about mismatched datatypes, or insert unexpected
coercions that could lead to odd results.  To fix, extend the
stripping logic to get rid of CoerceToDomain if it's atop an ArrayRef
or FieldStore.

While poking at this, I realized that there's a poorly documented and
not-at-all-tested behavior nearby: we coerce each VALUES column to
the domain type separately, and rely on the rewriter to merge those
operations so that the domain constraints are checked only once.
If that merging did not happen, it's entirely possible that we'd get
unexpected domain constraint failures due to checking a
partially-updated container value.  There's no bug there, but while
we're here let's improve the commentary about it and add some test
cases that explicitly exercise that behavior.

Per bug #18393 from Pablo Kharo.  Back-patch to all supported
branches.

Discussion: https://postgr.es/m/18393-65fedb1a0de9260d@postgresql.org

diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c
index 912aab1a6f79aebd8927e510f222208cb12c5fdf..4b238b91951489bfc9968ce35edb0459c02ad4e9 100644 (file)
--- a/src/backend/parser/analyze.c
+++ b/src/backend/parser/analyze.c
@@ -995,17 +995,28 @@ transformInsertRow(ParseState *pstate, List *exprlist,
 
                if (strip_indirection)
                {
+                       /*
+                        * We need to remove top-level FieldStores and SubscriptingRefs,
+                        * as well as any CoerceToDomain appearing above one of those ---
+                        * but not a CoerceToDomain that isn't above one of those.
+                        */
                        while (expr)
                        {
-                               if (IsA(expr, FieldStore))
+                               Expr       *subexpr = expr;
+
+                               while (IsA(subexpr, CoerceToDomain))
+                               {
+                                       subexpr = ((CoerceToDomain *) subexpr)->arg;
+                               }
+                               if (IsA(subexpr, FieldStore))
                                {
-                                       FieldStore *fstore = (FieldStore *) expr;
+                                       FieldStore *fstore = (FieldStore *) subexpr;
 
                                        expr = (Expr *) linitial(fstore->newvals);
                                }
-                               else if (IsA(expr, SubscriptingRef))
+                               else if (IsA(subexpr, SubscriptingRef))
                                {
-                                       SubscriptingRef *sbsref = (SubscriptingRef *) expr;
+                                       SubscriptingRef *sbsref = (SubscriptingRef *) subexpr;
 
                                        if (sbsref->refassgnexpr == NULL)
                                                break;

diff --git a/src/backend/parser/parse_target.c b/src/backend/parser/parse_target.c
index 43a7efeebedd620892bafb8bcf2feeb8a8b46e59..9841894d78cac8affb70d70bbe6f7d3f1d92b061 100644 (file)
--- a/src/backend/parser/parse_target.c
+++ b/src/backend/parser/parse_target.c
@@ -813,7 +813,16 @@ transformAssignmentIndirection(ParseState *pstate,
                        fstore->fieldnums = list_make1_int(attnum);
                        fstore->resulttype = baseTypeId;
 
-                       /* If target is a domain, apply constraints */
+                       /*
+                        * If target is a domain, apply constraints.  Notice that this
+                        * isn't totally right: the expression tree we build would check
+                        * the domain's constraints on a composite value with only this
+                        * one field populated or updated, possibly leading to an unwanted
+                        * failure.  The rewriter will merge together any subfield
+                        * assignments to the same table column, resulting in the domain's
+                        * constraints being checked only once after we've assigned to all
+                        * the fields that the INSERT or UPDATE means to.
+                        */
                        if (baseTypeId != targetTypeId)
                                return coerce_to_domain((Node *) fstore,
                                                                                baseTypeId, baseTypeMod,
@@ -943,7 +952,12 @@ transformAssignmentSubscripts(ParseState *pstate,
                                                                                                   subscripts,
                                                                                                   rhs);
 
-       /* If target was a domain over container, need to coerce up to the domain */
+       /*
+        * If target was a domain over container, need to coerce up to the domain.
+        * As in transformAssignmentIndirection, this coercion is premature if the
+        * query assigns to multiple elements of the container; but we'll fix that
+        * during query rewrite.
+        */
        if (containerType != targetTypeId)
        {
                Oid                     resulttype = exprType(result);

diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c
index 50948d180d0bf1e0cb652393eed3b97373a02bbc..b16cbd236c1713f6fd9fba1beb5aa15abbd553b7 100644 (file)
--- a/src/backend/rewrite/rewriteHandler.c
+++ b/src/backend/rewrite/rewriteHandler.c
@@ -1029,9 +1029,9 @@ process_matched_tle(TargetEntry *src_tle,
         * resulting in each assignment containing a CoerceToDomain node over a
         * FieldStore or SubscriptingRef.  These should have matching target
         * domains, so we strip them and reconstitute a single CoerceToDomain over
-        * the combined FieldStore/SubscriptingRef nodes.  (Notice that this has the
-        * result that the domain's checks are applied only after we do all the
-        * field or element updates, not after each one.  This is arguably desirable.)
+        * the combined FieldStore/SubscriptingRef nodes.  (Notice that this has
+        * the result that the domain's checks are applied only after we do all
+        * the field or element updates, not after each one.  This is desirable.)
         *----------
         */
        src_expr = (Node *) src_tle->expr;

diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out
index 38d6f75dcdcd2cf7741a9a23326d90b0a053b2cf..783a7967070f3ee3fdb0dd9fbc93d4bf61eb636e 100644 (file)
--- a/src/test/regress/expected/insert.out
+++ b/src/test/regress/expected/insert.out
@@ -159,7 +159,121 @@ Rules:
 
 drop table inserttest2;
 drop table inserttest;
-drop type insert_test_type;
+-- Make the same tests with domains over the array and composite fields
+create domain insert_pos_ints as int[] check (value[1] > 0);
+create domain insert_test_domain as insert_test_type
+  check ((value).if2[1] is not null);
+create table inserttesta (f1 int, f2 insert_pos_ints);
+create table inserttestb (f3 insert_test_domain, f4 insert_test_domain[]);
+insert into inserttesta (f2[1], f2[2]) values (1,2);
+insert into inserttesta (f2[1], f2[2]) values (3,4), (5,6);
+insert into inserttesta (f2[1], f2[2]) select 7,8;
+insert into inserttesta (f2[1], f2[2]) values (1,default);  -- not supported
+ERROR:  cannot set an array element to DEFAULT
+LINE 1: insert into inserttesta (f2[1], f2[2]) values (1,default);
+                                        ^
+insert into inserttesta (f2[1], f2[2]) values (0,2);
+ERROR:  value for domain insert_pos_ints violates check constraint "insert_pos_ints_check"
+insert into inserttesta (f2[1], f2[2]) values (3,4), (0,6);
+ERROR:  value for domain insert_pos_ints violates check constraint "insert_pos_ints_check"
+insert into inserttesta (f2[1], f2[2]) select 0,8;
+ERROR:  value for domain insert_pos_ints violates check constraint "insert_pos_ints_check"
+insert into inserttestb (f3.if1, f3.if2) values (1,array['foo']);
+insert into inserttestb (f3.if1, f3.if2) values (1,'{foo}'), (2,'{bar}');
+insert into inserttestb (f3.if1, f3.if2) select 3, '{baz,quux}';
+insert into inserttestb (f3.if1, f3.if2) values (1,default);  -- not supported
+ERROR:  cannot set a subfield to DEFAULT
+LINE 1: insert into inserttestb (f3.if1, f3.if2) values (1,default);
+                                         ^
+insert into inserttestb (f3.if1, f3.if2) values (1,array[null]);
+ERROR:  value for domain insert_test_domain violates check constraint "insert_test_domain_check"
+insert into inserttestb (f3.if1, f3.if2) values (1,'{null}'), (2,'{bar}');
+ERROR:  value for domain insert_test_domain violates check constraint "insert_test_domain_check"
+insert into inserttestb (f3.if1, f3.if2) select 3, '{null,quux}';
+ERROR:  value for domain insert_test_domain violates check constraint "insert_test_domain_check"
+insert into inserttestb (f3.if2[1], f3.if2[2]) values ('foo', 'bar');
+insert into inserttestb (f3.if2[1], f3.if2[2]) values ('foo', 'bar'), ('baz', 'quux');
+insert into inserttestb (f3.if2[1], f3.if2[2]) select 'bear', 'beer';
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) values (row(1,'{x}'), 'foo', 'bar');
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) values (row(1,'{x}'), 'foo', 'bar'), (row(2,'{y}'), 'baz', 'quux');
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) select row(1,'{x}')::insert_test_domain, 'bear', 'beer';
+select * from inserttesta;
+ f1 |  f2   
+----+-------
+    | {1,2}
+    | {3,4}
+    | {5,6}
+    | {7,8}
+(4 rows)
+
+select * from inserttestb;
+        f3        |           f4           
+------------------+------------------------
+ (1,{foo})        | 
+ (1,{foo})        | 
+ (2,{bar})        | 
+ (3,"{baz,quux}") | 
+ (,"{foo,bar}")   | 
+ (,"{foo,bar}")   | 
+ (,"{baz,quux}")  | 
+ (,"{bear,beer}") | 
+ (1,{x})          | {"(,\"{foo,bar}\")"}
+ (1,{x})          | {"(,\"{foo,bar}\")"}
+ (2,{y})          | {"(,\"{baz,quux}\")"}
+ (1,{x})          | {"(,\"{bear,beer}\")"}
+(12 rows)
+
+-- also check reverse-listing
+create table inserttest2 (f1 bigint, f2 text);
+create rule irule1 as on insert to inserttest2 do also
+  insert into inserttestb (f3.if2[1], f3.if2[2])
+  values (new.f1,new.f2);
+create rule irule2 as on insert to inserttest2 do also
+  insert into inserttestb (f4[1].if1, f4[1].if2[2])
+  values (1,'fool'),(new.f1,new.f2);
+create rule irule3 as on insert to inserttest2 do also
+  insert into inserttestb (f4[1].if1, f4[1].if2[2])
+  select new.f1, new.f2;
+\d+ inserttest2
+                                Table "public.inserttest2"
+ Column |  Type  | Collation | Nullable | Default | Storage  | Stats target | Description 
+--------+--------+-----------+----------+---------+----------+--------------+-------------
+ f1     | bigint |           |          |         | plain    |              | 
+ f2     | text   |           |          |         | extended |              | 
+Rules:
+    irule1 AS
+    ON INSERT TO inserttest2 DO  INSERT INTO inserttestb (f3.if2[1], f3.if2[2])
+  VALUES (new.f1, new.f2)
+    irule2 AS
+    ON INSERT TO inserttest2 DO  INSERT INTO inserttestb (f4[1].if1, f4[1].if2[2]) VALUES (1,'fool'::text), (new.f1,new.f2)
+    irule3 AS
+    ON INSERT TO inserttest2 DO  INSERT INTO inserttestb (f4[1].if1, f4[1].if2[2])  SELECT new.f1,
+            new.f2
+
+drop table inserttest2;
+drop table inserttesta;
+drop table inserttestb;
+drop domain insert_pos_ints;
+drop domain insert_test_domain;
+-- Verify that multiple inserts to subfields of a domain-over-container
+-- check the domain constraints only on the finished value
+create domain insert_nnarray as int[]
+  check (value[1] is not null and value[2] is not null);
+create domain insert_test_domain as insert_test_type
+  check ((value).if1 is not null and (value).if2 is not null);
+create table inserttesta (f1 insert_nnarray);
+insert into inserttesta (f1[1]) values (1);  -- fail
+ERROR:  value for domain insert_nnarray violates check constraint "insert_nnarray_check"
+insert into inserttesta (f1[1], f1[2]) values (1, 2);
+create table inserttestb (f1 insert_test_domain);
+insert into inserttestb (f1.if1) values (1);  -- fail
+ERROR:  value for domain insert_test_domain violates check constraint "insert_test_domain_check"
+insert into inserttestb (f1.if1, f1.if2) values (1, '{foo}');
+drop table inserttesta;
+drop table inserttestb;
+drop domain insert_nnarray;
+drop type insert_test_type cascade;
+NOTICE:  drop cascades to type insert_test_domain
 -- direct partition inserts should check partition bound constraint
 create table range_parted (
        a text,

diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql
index ffd4aacbc48b1ce180e5c7270e19d591087b7b03..b3460e083bc4ee970671c9a81170f6e2d321920c 100644 (file)
--- a/src/test/regress/sql/insert.sql
+++ b/src/test/regress/sql/insert.sql
@@ -83,7 +83,84 @@ create rule irule3 as on insert to inserttest2 do also
 
 drop table inserttest2;
 drop table inserttest;
-drop type insert_test_type;
+
+-- Make the same tests with domains over the array and composite fields
+
+create domain insert_pos_ints as int[] check (value[1] > 0);
+
+create domain insert_test_domain as insert_test_type
+  check ((value).if2[1] is not null);
+
+create table inserttesta (f1 int, f2 insert_pos_ints);
+create table inserttestb (f3 insert_test_domain, f4 insert_test_domain[]);
+
+insert into inserttesta (f2[1], f2[2]) values (1,2);
+insert into inserttesta (f2[1], f2[2]) values (3,4), (5,6);
+insert into inserttesta (f2[1], f2[2]) select 7,8;
+insert into inserttesta (f2[1], f2[2]) values (1,default);  -- not supported
+insert into inserttesta (f2[1], f2[2]) values (0,2);
+insert into inserttesta (f2[1], f2[2]) values (3,4), (0,6);
+insert into inserttesta (f2[1], f2[2]) select 0,8;
+
+insert into inserttestb (f3.if1, f3.if2) values (1,array['foo']);
+insert into inserttestb (f3.if1, f3.if2) values (1,'{foo}'), (2,'{bar}');
+insert into inserttestb (f3.if1, f3.if2) select 3, '{baz,quux}';
+insert into inserttestb (f3.if1, f3.if2) values (1,default);  -- not supported
+insert into inserttestb (f3.if1, f3.if2) values (1,array[null]);
+insert into inserttestb (f3.if1, f3.if2) values (1,'{null}'), (2,'{bar}');
+insert into inserttestb (f3.if1, f3.if2) select 3, '{null,quux}';
+
+insert into inserttestb (f3.if2[1], f3.if2[2]) values ('foo', 'bar');
+insert into inserttestb (f3.if2[1], f3.if2[2]) values ('foo', 'bar'), ('baz', 'quux');
+insert into inserttestb (f3.if2[1], f3.if2[2]) select 'bear', 'beer';
+
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) values (row(1,'{x}'), 'foo', 'bar');
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) values (row(1,'{x}'), 'foo', 'bar'), (row(2,'{y}'), 'baz', 'quux');
+insert into inserttestb (f3, f4[1].if2[1], f4[1].if2[2]) select row(1,'{x}')::insert_test_domain, 'bear', 'beer';
+
+select * from inserttesta;
+select * from inserttestb;
+
+-- also check reverse-listing
+create table inserttest2 (f1 bigint, f2 text);
+create rule irule1 as on insert to inserttest2 do also
+  insert into inserttestb (f3.if2[1], f3.if2[2])
+  values (new.f1,new.f2);
+create rule irule2 as on insert to inserttest2 do also
+  insert into inserttestb (f4[1].if1, f4[1].if2[2])
+  values (1,'fool'),(new.f1,new.f2);
+create rule irule3 as on insert to inserttest2 do also
+  insert into inserttestb (f4[1].if1, f4[1].if2[2])
+  select new.f1, new.f2;
+\d+ inserttest2
+
+drop table inserttest2;
+drop table inserttesta;
+drop table inserttestb;
+drop domain insert_pos_ints;
+drop domain insert_test_domain;
+
+-- Verify that multiple inserts to subfields of a domain-over-container
+-- check the domain constraints only on the finished value
+
+create domain insert_nnarray as int[]
+  check (value[1] is not null and value[2] is not null);
+
+create domain insert_test_domain as insert_test_type
+  check ((value).if1 is not null and (value).if2 is not null);
+
+create table inserttesta (f1 insert_nnarray);
+insert into inserttesta (f1[1]) values (1);  -- fail
+insert into inserttesta (f1[1], f1[2]) values (1, 2);
+
+create table inserttestb (f1 insert_test_domain);
+insert into inserttestb (f1.if1) values (1);  -- fail
+insert into inserttestb (f1.if1, f1.if2) values (1, '{foo}');
+
+drop table inserttesta;
+drop table inserttestb;
+drop domain insert_nnarray;
+drop type insert_test_type cascade;
 
 -- direct partition inserts should check partition bound constraint
 create table range_parted (