Fix wrong result in parallel ChunkAppend

src/nodes/chunk_append/planner.c

@@ -255,10 +255,12 @@ ts_chunk_append_plan_create(PlannerInfo *root, RelOptInfo *rel, CustomPath *path
 	cscan->custom_plans = custom_plans;
 
 	/*
-	 * If we do either startup or runtime exclusion, we need to pass restrictinfo
-	 * clauses into executor.
+	 * Always pass per-chunk restrictinfo clauses and RT indexes into the
+	 * executor so that the three per-chunk lists (subplans, ri_clauses,
+	 * constraints) are uniformly populated.  This is required for parallel
+	 * workers that index into these lists, and keeps startup/runtime
+	 * exclusion from having to cope with missing data.
 	 */
-	if (capath->startup_exclusion || capath->runtime_exclusion_children)
 	{
 		foreach (lc_child, cscan->custom_plans)
 		{

test/expected/parallel-16.out

@@ -605,6 +605,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=2);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_2_chunk
 
+SELECT count(*) FROM "test" WHERE i > 400000 AND length(version()) > 0;
+ count 
+-------
+ 59999
+
 ALTER TABLE :CHUNK1 SET (parallel_workers=2);
 ALTER TABLE :CHUNK2 SET (parallel_workers=0);
 :PREFIX SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
@@ -625,6 +630,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=0);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_1_chunk
 
+SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
+ count 
+-------
+ 60000
+
 -- Verify result correctness with mixed partial/non-partial subplans.
 -- chunk1 is partial (parallel_workers=2), chunk2 is non-partial (parallel_workers=0).
 SET max_parallel_workers_per_gather TO 0;
@@ -759,3 +769,36 @@ RESET max_parallel_workers_per_gather;
          ->  Seq Scan on _hyper_1_2_chunk
                Filter: (ts < now())
 
+-- test parallel ChunkAppend with InitPlan params (runtime exclusion)
+-- Filter on j (non-partitioning column) so that the planner produces a
+-- parallel-aware ChunkAppend with runtime_exclusion_parent but without
+-- startup_exclusion. Filtering on the partitioning column i would enable
+-- runtime_exclusion_children, making the planner pick Single Copy mode
+-- (non-parallel ChunkAppend) which does not exercise the shared-memory
+-- subplan coordination path.
+SET max_parallel_workers_per_gather = 2;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET min_parallel_table_scan_size = 0;
+-- get the sequential result for comparison
+SET max_parallel_workers_per_gather = 0;
+SELECT count(*) AS expected FROM "test" WHERE j = (SELECT max(j) FROM "test") \gset
+-- parallel with leader participation
+SET max_parallel_workers_per_gather = 2;
+SELECT count(*) = :expected AS leader_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ leader_ok 
+-----------
+ t
+
+-- parallel with only workers (leader does not participate)
+SET parallel_leader_participation = off;
+SELECT count(*) = :expected AS workers_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ workers_ok 
+------------
+ t
+
+RESET parallel_leader_participation;
+RESET max_parallel_workers_per_gather;
+RESET parallel_tuple_cost;
+RESET parallel_setup_cost;
+RESET min_parallel_table_scan_size;

test/expected/parallel-17.out

@@ -605,6 +605,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=2);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_2_chunk
 
+SELECT count(*) FROM "test" WHERE i > 400000 AND length(version()) > 0;
+ count 
+-------
+ 59999
+
 ALTER TABLE :CHUNK1 SET (parallel_workers=2);
 ALTER TABLE :CHUNK2 SET (parallel_workers=0);
 :PREFIX SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
@@ -625,6 +630,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=0);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_1_chunk
 
+SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
+ count 
+-------
+ 60000
+
 -- Verify result correctness with mixed partial/non-partial subplans.
 -- chunk1 is partial (parallel_workers=2), chunk2 is non-partial (parallel_workers=0).
 SET max_parallel_workers_per_gather TO 0;
@@ -759,3 +769,36 @@ RESET max_parallel_workers_per_gather;
          ->  Seq Scan on _hyper_1_2_chunk
                Filter: (ts < now())
 
+-- test parallel ChunkAppend with InitPlan params (runtime exclusion)
+-- Filter on j (non-partitioning column) so that the planner produces a
+-- parallel-aware ChunkAppend with runtime_exclusion_parent but without
+-- startup_exclusion. Filtering on the partitioning column i would enable
+-- runtime_exclusion_children, making the planner pick Single Copy mode
+-- (non-parallel ChunkAppend) which does not exercise the shared-memory
+-- subplan coordination path.
+SET max_parallel_workers_per_gather = 2;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET min_parallel_table_scan_size = 0;
+-- get the sequential result for comparison
+SET max_parallel_workers_per_gather = 0;
+SELECT count(*) AS expected FROM "test" WHERE j = (SELECT max(j) FROM "test") \gset
+-- parallel with leader participation
+SET max_parallel_workers_per_gather = 2;
+SELECT count(*) = :expected AS leader_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ leader_ok 
+-----------
+ t
+
+-- parallel with only workers (leader does not participate)
+SET parallel_leader_participation = off;
+SELECT count(*) = :expected AS workers_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ workers_ok 
+------------
+ t
+
+RESET parallel_leader_participation;
+RESET max_parallel_workers_per_gather;
+RESET parallel_tuple_cost;
+RESET parallel_setup_cost;
+RESET min_parallel_table_scan_size;

test/expected/parallel-18.out

@@ -578,6 +578,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=2);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_2_chunk
 
+SELECT count(*) FROM "test" WHERE i > 400000 AND length(version()) > 0;
+ count 
+-------
+ 59999
+
 ALTER TABLE :CHUNK1 SET (parallel_workers=2);
 ALTER TABLE :CHUNK2 SET (parallel_workers=0);
 :PREFIX SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
@@ -598,6 +603,11 @@ ALTER TABLE :CHUNK2 SET (parallel_workers=0);
                                  One-Time Filter: (length(version()) > 0)
                                  ->  Parallel Seq Scan on _hyper_1_1_chunk
 
+SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
+ count 
+-------
+ 60000
+
 -- Verify result correctness with mixed partial/non-partial subplans.
 -- chunk1 is partial (parallel_workers=2), chunk2 is non-partial (parallel_workers=0).
 SET max_parallel_workers_per_gather TO 0;
@@ -732,3 +742,36 @@ RESET max_parallel_workers_per_gather;
          ->  Seq Scan on _hyper_1_2_chunk
                Filter: (ts < now())
 
+-- test parallel ChunkAppend with InitPlan params (runtime exclusion)
+-- Filter on j (non-partitioning column) so that the planner produces a
+-- parallel-aware ChunkAppend with runtime_exclusion_parent but without
+-- startup_exclusion. Filtering on the partitioning column i would enable
+-- runtime_exclusion_children, making the planner pick Single Copy mode
+-- (non-parallel ChunkAppend) which does not exercise the shared-memory
+-- subplan coordination path.
+SET max_parallel_workers_per_gather = 2;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET min_parallel_table_scan_size = 0;
+-- get the sequential result for comparison
+SET max_parallel_workers_per_gather = 0;
+SELECT count(*) AS expected FROM "test" WHERE j = (SELECT max(j) FROM "test") \gset
+-- parallel with leader participation
+SET max_parallel_workers_per_gather = 2;
+SELECT count(*) = :expected AS leader_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ leader_ok 
+-----------
+ t
+
+-- parallel with only workers (leader does not participate)
+SET parallel_leader_participation = off;
+SELECT count(*) = :expected AS workers_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+ workers_ok 
+------------
+ t
+
+RESET parallel_leader_participation;
+RESET max_parallel_workers_per_gather;
+RESET parallel_tuple_cost;
+RESET parallel_setup_cost;
+RESET min_parallel_table_scan_size;

test/sql/parallel.sql.in

@@ -167,10 +167,12 @@ RESET max_parallel_workers_per_gather;
 ALTER TABLE :CHUNK1 SET (parallel_workers=0);
 ALTER TABLE :CHUNK2 SET (parallel_workers=2);
 :PREFIX SELECT count(*) FROM "test" WHERE i > 400000 AND length(version()) > 0;
+SELECT count(*) FROM "test" WHERE i > 400000 AND length(version()) > 0;
 
 ALTER TABLE :CHUNK1 SET (parallel_workers=2);
 ALTER TABLE :CHUNK2 SET (parallel_workers=0);
 :PREFIX SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
+SELECT count(*) FROM "test" WHERE i < 600000 AND length(version()) > 0;
 
 -- Verify result correctness with mixed partial/non-partial subplans.
 -- chunk1 is partial (parallel_workers=2), chunk2 is non-partial (parallel_workers=0).
@@ -240,3 +242,33 @@ RESET max_parallel_workers_per_gather;
 -- this won't be parallel query because now() is parallel restricted in PG < 12
 :PREFIX SELECT i FROM "test" WHERE ts < now();
 
+-- test parallel ChunkAppend with InitPlan params (runtime exclusion)
+-- Filter on j (non-partitioning column) so that the planner produces a
+-- parallel-aware ChunkAppend with runtime_exclusion_parent but without
+-- startup_exclusion. Filtering on the partitioning column i would enable
+-- runtime_exclusion_children, making the planner pick Single Copy mode
+-- (non-parallel ChunkAppend) which does not exercise the shared-memory
+-- subplan coordination path.
+SET max_parallel_workers_per_gather = 2;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET min_parallel_table_scan_size = 0;
+
+-- get the sequential result for comparison
+SET max_parallel_workers_per_gather = 0;
+SELECT count(*) AS expected FROM "test" WHERE j = (SELECT max(j) FROM "test") \gset
+
+-- parallel with leader participation
+SET max_parallel_workers_per_gather = 2;
+SELECT count(*) = :expected AS leader_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+
+-- parallel with only workers (leader does not participate)
+SET parallel_leader_participation = off;
+SELECT count(*) = :expected AS workers_ok FROM "test" WHERE j = (SELECT max(j) FROM "test");
+RESET parallel_leader_participation;
+
+RESET max_parallel_workers_per_gather;
+RESET parallel_tuple_cost;
+RESET parallel_setup_cost;
+RESET min_parallel_table_scan_size;
+

tsl/test/expected/chunk_append_space_runtime_exclusion.out

@@ -0,0 +1,104 @@
+-- This file and its contents are licensed under the Timescale License.
+-- Please see the included NOTICE for copyright information and
+-- LICENSE-TIMESCALE for a copy of the license.
+-- Test runtime chunk exclusion in an ordered ChunkAppend with space
+-- partitioning. When some time slices have multiple space partitions
+-- (producing MergeAppend children) and others have a single partition
+-- (producing direct scan children), the ChunkAppend must handle both
+-- child types during runtime exclusion.
+\set PREFIX 'EXPLAIN (COSTS OFF)'
+\set PREFIX_ANALYZE 'EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF, BUFFERS OFF, SUMMARY OFF)'
+SET max_parallel_workers_per_gather = 0;
+SET enable_material = off;
+SET enable_seqscan = off;
+SET enable_hashjoin = off;
+SET enable_mergejoin = off;
+CREATE TABLE space_mixed(time timestamptz NOT NULL, device_id int NOT NULL, value float);
+SELECT create_hypertable('space_mixed', 'time', 'device_id',
+    number_partitions => 2,
+    chunk_time_interval => '5 days'::interval);
+    create_hypertable     
+--------------------------
+ (1,public,space_mixed,t)
+
+-- Row counts and LIMIT below are chosen so EXPLAIN ANALYZE per-loop
+-- averages are integers; otherwise PG versions differ in how they
+-- format half-integer values.
+-- Time slice 1: both space partitions have data (produces MergeAppend).
+INSERT INTO space_mixed VALUES
+  ('2024-01-01 01:00', 1, 1.0), ('2024-01-01 02:00', 1, 2.0),
+  ('2024-01-02 01:00', 3, 3.0), ('2024-01-02 02:00', 3, 4.0);
+-- Time slice 2: only one space partition has data (produces direct scan).
+INSERT INTO space_mixed VALUES
+  ('2024-01-06 01:00', 1, 5.0), ('2024-01-06 02:00', 1, 6.0),
+  ('2024-01-07 01:00', 1, 7.0), ('2024-01-07 02:00', 1, 8.0);
+CREATE INDEX ON space_mixed(time);
+ANALYZE space_mixed;
+-- The driver table provides join parameters for runtime exclusion.
+CREATE TABLE driver_times(t timestamptz PRIMARY KEY);
+INSERT INTO driver_times VALUES ('2024-01-01'), ('2024-01-06');
+-- Plan should show ordered ChunkAppend with mixed children:
+-- MergeAppend for the first time slice, direct IndexScan for the second.
+:PREFIX
+SELECT d.t, m.*
+FROM driver_times d,
+LATERAL (SELECT * FROM space_mixed m WHERE m.time >= d.t ORDER BY m.time LIMIT 4) m;
+--- QUERY PLAN ---
+ Nested Loop
+   ->  Index Only Scan using driver_times_pkey on driver_times d
+   ->  Limit
+         ->  Custom Scan (ChunkAppend) on space_mixed m
+               Order: m."time"
+               ->  Merge Append
+                     Sort Key: m."time"
+                     ->  Index Scan using _hyper_1_1_chunk_space_mixed_time_idx1 on _hyper_1_1_chunk m_1
+                           Index Cond: ("time" >= d.t)
+                     ->  Index Scan using _hyper_1_2_chunk_space_mixed_time_idx1 on _hyper_1_2_chunk m_2
+                           Index Cond: ("time" >= d.t)
+               ->  Index Scan using _hyper_1_3_chunk_space_mixed_time_idx1 on _hyper_1_3_chunk m_3
+                     Index Cond: ("time" >= d.t)
+
+-- Execute to trigger runtime exclusion. The MergeAppend child is
+-- skipped by do_runtime_exclusion (scan == NULL path), while the
+-- direct scan child participates in constraint exclusion.
+:PREFIX_ANALYZE
+SELECT d.t, m.*
+FROM driver_times d,
+LATERAL (SELECT * FROM space_mixed m WHERE m.time >= d.t ORDER BY m.time LIMIT 4) m;
+--- QUERY PLAN ---
+ Nested Loop (actual rows=8.00 loops=1)
+   ->  Index Only Scan using driver_times_pkey on driver_times d (actual rows=2.00 loops=1)
+   ->  Limit (actual rows=4.00 loops=2)
+         ->  Custom Scan (ChunkAppend) on space_mixed m (actual rows=4.00 loops=2)
+               Order: m."time"
+               Chunks excluded during runtime: 0
+               ->  Merge Append (actual rows=2.00 loops=2)
+                     Sort Key: m."time"
+                     ->  Index Scan using _hyper_1_1_chunk_space_mixed_time_idx1 on _hyper_1_1_chunk m_1 (actual rows=1.00 loops=2)
+                           Index Cond: ("time" >= d.t)
+                     ->  Index Scan using _hyper_1_2_chunk_space_mixed_time_idx1 on _hyper_1_2_chunk m_2 (actual rows=1.00 loops=2)
+                           Index Cond: ("time" >= d.t)
+               ->  Index Scan using _hyper_1_3_chunk_space_mixed_time_idx1 on _hyper_1_3_chunk m_3 (actual rows=4.00 loops=1)
+                     Index Cond: ("time" >= d.t)
+
+SELECT d.t, m.*
+FROM driver_times d,
+LATERAL (SELECT * FROM space_mixed m WHERE m.time >= d.t ORDER BY m.time LIMIT 4) m;
+              t               |             time             | device_id | value 
+------------------------------+------------------------------+-----------+-------
+ Mon Jan 01 00:00:00 2024 PST | Mon Jan 01 01:00:00 2024 PST |         1 |     1
+ Mon Jan 01 00:00:00 2024 PST | Mon Jan 01 02:00:00 2024 PST |         1 |     2
+ Mon Jan 01 00:00:00 2024 PST | Tue Jan 02 01:00:00 2024 PST |         3 |     3
+ Mon Jan 01 00:00:00 2024 PST | Tue Jan 02 02:00:00 2024 PST |         3 |     4
+ Sat Jan 06 00:00:00 2024 PST | Sat Jan 06 01:00:00 2024 PST |         1 |     5
+ Sat Jan 06 00:00:00 2024 PST | Sat Jan 06 02:00:00 2024 PST |         1 |     6
+ Sat Jan 06 00:00:00 2024 PST | Sun Jan 07 01:00:00 2024 PST |         1 |     7
+ Sat Jan 06 00:00:00 2024 PST | Sun Jan 07 02:00:00 2024 PST |         1 |     8
+
+DROP TABLE space_mixed;
+DROP TABLE driver_times;
+RESET max_parallel_workers_per_gather;
+RESET enable_material;
+RESET enable_seqscan;
+RESET enable_hashjoin;
+RESET enable_mergejoin;

tsl/test/sql/CMakeLists.txt

@@ -10,6 +10,7 @@ set(TEST_FILES
     cagg_add_column.sql
     cagg_ddl.sql
     cagg_direct_compress.sql
+    chunk_append_space_runtime_exclusion.sql
     cagg_errors.sql
     cagg_granular_refresh_api.sql
     cagg_invalidation.sql