with as select insert - queries use incorrect sequence

[cimerren]

Hi, we are using citus in schema based sharding and are experiencing strange behavior when running the following query:

set citus.enable_schema_based_sharding to on;

create schema citus_test;

SET search_path = citus_test, public;
create table version (
	id bigserial not null,
	description varchar(255),
	primary key (id)
);

insert into version (description) values ('Version 1');

with v as (
	select * from version where description = 'Version 1'
)
insert into 
version (description) 
select description
from v;

select * from citus_test.version;

Expected:

id	description
1	Version 1
2	Version 1

Actual:

id	description
1	Version 1
562949953421313	Version 1

I found the following issue: 6798, and from the values it looks like it could be related.

[m3hm3t]

explain verbose insert into version (description) values ('Version 1');
┌──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│                                                            QUERY PLAN                                                            │
├──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)                                                                   │
│   Task Count: 1                                                                                                                  │
│   Tasks Shown: All                                                                                                               │
│   ->  Task                                                                                                                       │
│         Query: INSERT INTO citus_test.version_102008 (id, description) VALUES ('2'::bigint, 'Version 1'::character varying(255)) │
│         Node: host=localhost port=9802 dbname=citus                                                                              │
│         ->  Insert on citus_test.version_102008  (cost=0.00..0.01 rows=0 width=0)                                                │
│               ->  Result  (cost=0.00..0.01 rows=1 width=524)                                                                     │
│                     Output: '2'::bigint, 'Version 1'::character varying(255)                                                     │
│             Query Identifier: 9179276677803586298                                                                                │
│ Query Identifier: -4401786819825584319                                                                                           │
└──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
(11 rows)


explain verbose with v as (
        select * from version where description = 'Version 1'
)
insert into 
version (description) 
select description
from v;
DEBUG:  Creating router plan
DEBUG:  distributed INSERT ... SELECT can only select from distributed tables
DEBUG:  CTE v is going to be inlined via distributed planning
DEBUG:  Creating router plan
DEBUG:  Collecting INSERT ... SELECT results on coordinator
DEBUG:  opening 1 new connections to localhost:9802
DEBUG:  established connection to localhost:9802 for session 11 in 7715 microseconds
DEBUG:  task execution (0) for placement (1) on anchor shard (102008) finished in 719 microseconds on worker node localhost:9802
DEBUG:  Total number of commands sent over the session 11: 1 to node localhost:9802
┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
│                                                                                                                                                           QUERY PLAN                                                                                                                        
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
│ Custom Scan (Citus INSERT ... SELECT)  (cost=0.00..0.00 rows=0 width=0)                                                                                                                                                                                                                     
│   INSERT/SELECT method: pull to coordinator                                                                                                                                                                                                                                                 
│   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)                                                                                                                                                                                                                        
│         Output: remote_scan.id, remote_scan.description                                                                                                                                                                                                                                     
│         Task Count: 1                                                                                                                                                                                                                                                                       
│         Tasks Shown: All                                                                                                                                                                                                                                                                    
│         ->  Task                                                                                                                                                                                                                                                                            
│               Query: SELECT id, description FROM (SELECT nextval('citus_test.version_id_seq'::regclass) AS id, v.description FROM (SELECT version.id, version.description FROM citus_test.version_102008 version WHERE ((version.description)::text OPERATOR(pg_catalog.=) 'Version 1'::text
│               Node: host=localhost port=9802 dbname=citus                                                                                                                                                                                                                                   
│               ->  Seq Scan on citus_test.version_102008 version  (cost=0.00..11.75 rows=1 width=524)                                                                                                                                                                                        
│                     Output: nextval('citus_test.version_id_seq'::regclass), version.description                                                                                                                                                                                             
│                     Filter: ((version.description)::text = 'Version 1'::text)                                                                                                                                                                                                               
│                   Query Identifier: 9147789847994792390                                                                                                                                                                                                                                     
│ Query Identifier: -4422929835560746835                                                                                                                                                                                                                                                      
└─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
(14 rows)

[m3hm3t]

explain insert into 
version (description) 
select description
from (select * from version where description = 'Version 1');
DEBUG:  Creating router plan
DEBUG:  volatile functions are not allowed in distributed INSERT ... SELECT queries
DEBUG:  Sequences cannot be used in router queries
DEBUG:  generated sql query for task 1
DETAIL:  query string: "SELECT description FROM citus_test.version_102008 version WHERE ((description)::text OPERATOR(pg_catalog.=) 'Version 1'::text)"
DEBUG:  assigned task 1 to node localhost:9802
DEBUG:  combine query: SELECT nextval('citus_test.version_id_seq'::regclass) AS id, description FROM pg_catalog.citus_extradata_container(10, NULL::cstring(0), NULL::cstring(0), '(i 2)'::cstring(0)) remote_scan(description character varying(255))
DEBUG:  Sequences cannot be used in router queries
DEBUG:  generated sql query for task 1
DETAIL:  query string: "SELECT description FROM citus_test.version_102008 version WHERE ((description)::text OPERATOR(pg_catalog.=) 'Version 1'::text)"
DEBUG:  assigned task 1 to node localhost:9802
DEBUG:  combine query: SELECT nextval('citus_test.version_id_seq'::regclass) AS id, description FROM pg_catalog.citus_extradata_container(10, NULL::cstring(0), NULL::cstring(0), '(i 2)'::cstring(0)) remote_scan(description character varying(255))
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│                                        QUERY PLAN                                         │
├───────────────────────────────────────────────────────────────────────────────────────────┤
│ Custom Scan (Citus INSERT ... SELECT)  (cost=0.00..0.00 rows=0 width=0)                   │
│   INSERT/SELECT method: pull to coordinator                                               │
│   ->  Custom Scan (Citus Adaptive)  (cost=0.00..250.00 rows=100000 width=524)             │
│         Task Count: 1                                                                     │
│         Tasks Shown: All                                                                  │
│         ->  Task                                                                          │
│               Node: host=localhost port=9802 dbname=citus                                 │
│               ->  Seq Scan on version_102008 version  (cost=0.00..11.75 rows=1 width=516) │
│                     Filter: ((description)::text = 'Version 1'::text)                     │
└───────────────────────────────────────────────────────────────────────────────────────────┘
(9 rows)

[m3hm3t]

-- Create the table
CREATE TABLE version (
    id bigserial NOT NULL,
    description varchar(255),
    PRIMARY KEY (id)
);

-- Register the table as distributed.
SELECT create_distributed_table('version', 'id');

-- Insert an initial row.
INSERT INTO version (description) VALUES ('Version 1');

-- Use a CTE and perform an INSERT ... SELECT.
WITH v AS (
    SELECT * FROM version WHERE description = 'Version 1'
)
INSERT INTO version (description)
SELECT description
FROM v;


-- Query the table.
SELECT * FROM version;


explain verbose WITH v AS (
    SELECT * FROM version WHERE description = 'Version 1'
)
INSERT INTO version (description)
SELECT description
FROM v;
DEBUG:  distributed INSERT ... SELECT can only select from distributed tables
DEBUG:  CTE v is going to be inlined via distributed planning
DEBUG:  Router planner cannot handle multi-shard select queries
DEBUG:  CTE v is going to be inlined via distributed planning
DEBUG:  Router planner cannot handle multi-shard select queries
┌───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
│                                                                                                                                     QUERY PLAN
├───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
│ Custom Scan (Citus INSERT ... SELECT)  (cost=0.00..0.00 rows=0 width=0)
│   INSERT/SELECT method: repartition
│   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=524)
│         Output: remote_scan.id, remote_scan.description
│         Task Count: 32
│         Tasks Shown: One of 32
│         ->  Task
│               Query: SELECT worker_column_1 AS id, worker_column_2 AS description FROM (SELECT citus_insert_select_subquery.id AS worker_column_1, citus_insert_select_subquery.description AS worker_column_2 FROM (SELECT nextval('citus.version_id_seq'::regclass) AS id, v.des
│               Node: host=localhost port=9701 dbname=citus
│               ->  Seq Scan on citus.version_102008 version  (cost=0.00..11.75 rows=1 width=524)
│                     Output: nextval('version_id_seq'::regclass), version.description
│                     Filter: ((version.description)::text = 'Version 1'::text)
│                   Query Identifier: -4194227824661966589
│ Query Identifier: -8734172042941213464
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
(14 rows)


┌─────────────────┬─────────────┐
│       id        │ description │
├─────────────────┼─────────────┤
│               1 │ Version 1   │
│ 562949953421313 │ Version 1   │
└─────────────────┴─────────────┘
(2 rows)

[m3hm3t]

WITH v AS (
    SELECT nextval('version_id_seq') as id, description
    FROM version
    WHERE description = 'Version 1'
)
INSERT INTO version (id, description)
SELECT id, description
FROM v;