[reshke]

DistributedLog truncate runs only on primaries and removes distributed xact log segments older than the cutoff page, emitting a TRUNCATE WAL record. Its redo branch in DistributedLog_redo() however only performed the SimpleLruTruncate() and never advanced DistributedLogShared->oldestXmin, so on a mirror/standby replaying the record the in-memory horizon stayed stale while the pre-truncation segments were already deleted.

without fix, standby query fails with

07:41:28.345963112 +0000
@@ -57,19 +57,8 @@
 -- above.  It must return the rows (or defer cleanly), never fail on the
 -- removed distributed log segment.
 -1S: select * from hs_dlog_trunc order by a;
- a  | msg
-----+--------
- 1  | row 1
- 2  | row 2
- 3  | row 3
- 4  | row 4
- 5  | row 5
- 6  | row 6
- 7  | row 7
- 8  | row 8
- 9  | row 9
- 10 | row 10
-(10 rows)
+DETAIL:  Could not open file "pg_distributedlog/0000": No such file or directory.
+ERROR:  could not access status of transaction 3
 -1Sq: ... <quitting>

 1: drop table hs_dlog_trunc;

A reader resolving a pre-truncation gxid then fell through the stale-horizon guard in DistributedLog_CommittedCheck() (and DistributedLog_GetDistributedXid(), which had no guard at all) into an SLRU read of a removed segment, producing a hard distributed-log read error instead of a soft not-found that takes the standby in-doubt path.

To fix, in the DISTRIBUTEDLOG_TRUNCATE redo branch, CAS-advance the local oldestXmin to the first xid of the cutoff page (never moving it backwards), and add a horizon guard to DistributedLog_GetDistributedXid() so trimmed xids return not-found cleanly.

Fixes #ISSUE_Number

What does this PR do?

Type of Change

• Bug fix (non-breaking change)
• New feature (non-breaking change)
• Breaking change (fix or feature with breaking changes)
• Documentation update

Breaking Changes

Test Plan

• Unit tests added/updated
• Integration tests added/updated
• Passed make installcheck
• Passed make -C src/test installcheck-cbdb-parallel

Impact

Performance:

User-facing changes:

Dependencies:

Checklist

• Followed contribution guide
• Added/updated documentation
• Reviewed code for security implications
• This PR contains AI-assisted code generation
• Requested review from cloudberry committers

Additional Context

CI Skip Instructions

---

[Alena0704]

Hi, thank you for working on this! I found a hot standby visibility regression with this change: a query can return rows inserted after its distributed snapshot was taken.

Steps to reproduce:

1. The standby QD takes a distributed snapshot and pauses before dispatching the query to the mirrors.
2. The primary inserts and commits 10 rows.
3. The mirrors replay DISTRIBUTEDLOG_TRUNCATE.
4. The query resumes with the original distributed snapshot, but the mirrors take their local snapshots after the insert.

The query should return 0 rows, but with this PR it returns 10 rows.

Output test with this PR (isolation2 test for this below)

-- Truncation must preserve visibility for an older standby snapshot.
-- Keep the INSERT's gxid entry in a retained segment and check that replay
-- does not make the rows visible to that snapshot.
-- DLOG layout: 4096 xids/page, 32 pages/segment (131072 xids).

create extension if not exists gp_inject_fault;
CREATE

1: create table hs_dlvis(a int, b int) distributed by (a);
CREATE
1: create table hs_dlvis_pin(a int) distributed by (a);
CREATE

-- Pin the primaries' oldestXmin with an old QD snapshot, without holding
-- back the mirrors' local xmin.
2: begin isolation level repeatable read;
BEGIN
2: select count(*) from hs_dlvis_pin;
 count 
-------
 0     
(1 row)
1: insert into hs_dlvis_pin select generate_series(1, 10);
INSERT 10

-- Advance local xids to the first page of the next segment.
0U: set synchronous_commit = off;
SET
0U: do $$ declare t bigint := (txid_current() / 131072 + 1) * 131072 + 8; begin loop exit when txid_current() >= t; commit; end loop; end $$;
DO
1U: set synchronous_commit = off;
SET
1U: do $$ declare t bigint := (txid_current() / 131072 + 1) * 131072 + 8; begin loop exit when txid_current() >= t; commit; end loop; end $$;
DO

-- Pause after taking the standby distributed snapshot, before dispatch.
-1S: select gp_inject_fault('select_after_qd_create_snapshot', 'suspend', dbid) from gp_segment_configuration where content = -1 and role = 'm';
 gp_inject_fault 
-----------------
 Success:        
(1 row)
-1S&: select * from hs_dlvis order by a;  <waiting ...>
1: select gp_wait_until_triggered_fault('select_after_qd_create_snapshot', 1, dbid) from gp_segment_configuration where content = -1 and role = 'm';
 gp_wait_until_triggered_fault 
-------------------------------
 Success:                      
(1 row)

1: insert into hs_dlvis select g, g from generate_series(1, 10) g;
INSERT 10
-- Check that the INSERT landed on the first page of each segment.
1: select gp_segment_id, xmin::text::bigint / 4096 % 32 as page_in_segment from hs_dlvis group by 1, 2 order by 1;
 gp_segment_id | page_in_segment 
---------------+-----------------
 0             | 0               
 1             | 0               
(2 rows)

-- Advance the mirrors' oldestXmin to the INSERT's entry.
-- Retry the recovery WAL error at segment boundaries: oldestXmin is
-- already updated before the query fails.
!\retcode sh -c 'p=$(psql -XAtq -d isolation2-hot-standby -c "select port from gp_segment_configuration where content = -1 and role = chr(109)"); for i in 1 2 3 4 5; do psql -X -At -d isolation2-hot-standby -p $p -c "select count(*) from hs_dlvis" && exit 0; done; exit 1';
-- start_ignore
ERROR:  cannot make new WAL entries during recovery (xloginsert.c:163)  (seg0 slice1 172.17.0.2:7324 pid=3166720) (xloginsert.c:163)
10

-- end_ignore
(exited with code 0)

-- Move every primary to the second page of the segment.
0U: do $$ declare t bigint := (txid_current() / 131072) * 131072 + 4096 + 100; begin loop exit when txid_current() >= t; commit; end loop; end $$;
DO
1U: do $$ declare t bigint := (txid_current() / 131072) * 131072 + 4096 + 100; begin loop exit when txid_current() >= t; commit; end loop; end $$;
DO
0U: select txid_current() / 4096 % 32 as page_in_segment;
 page_in_segment 
-----------------
 1               
(1 row)
1U: select txid_current() / 4096 % 32 as page_in_segment;
 page_in_segment 
-----------------
 1               
(1 row)
0Uq: ... <quitting>
1Uq: ... <quitting>

-- Release the pin and trigger TRUNCATE with the second page as cutoff.
2: commit;
COMMIT
1: select count(*) from hs_dlvis_pin;
 count 
-------
 10    
(1 row)

-- Wait until only the INSERT's segment remains on each mirror.
0M: do $$ begin for i in 1..120 loop exit when (select count(*) from pg_ls_dir('pg_distributedlog')) = 1; perform pg_sleep(1); end loop; end $$;
DO
1M: do $$ begin for i in 1..120 loop exit when (select count(*) from pg_ls_dir('pg_distributedlog')) = 1; perform pg_sleep(1); end loop; end $$;
DO

-- Flush DLOG to disk; remote_apply makes the DDL wait for checkpoint replay.
1: checkpoint;
CHECKPOINT
1: create table hs_dlvis_sync(a int);
CREATE
1: drop table hs_dlvis_sync;
DROP
0M: checkpoint;
CHECKPOINT
1M: checkpoint;
CHECKPOINT

-- Check that the INSERT's segment and nonzero gxid entry are still on disk.
0M: select f = upper(lpad(to_hex(x / 131072), 4, '0')) as entry_segment_kept, pg_read_binary_file('pg_distributedlog/' || f, (x % 131072) * 8, 8) <> '\x0000000000000000'::bytea as entry_has_gxid from (select distinct xmin::text::bigint as x from hs_dlvis) t, pg_ls_dir('pg_distributedlog') f;
 entry_segment_kept | entry_has_gxid 
--------------------+----------------
 t                  | t              
(1 row)
1M: select f = upper(lpad(to_hex(x / 131072), 4, '0')) as entry_segment_kept, pg_read_binary_file('pg_distributedlog/' || f, (x % 131072) * 8, 8) <> '\x0000000000000000'::bytea as entry_has_gxid from (select distinct xmin::text::bigint as x from hs_dlvis) t, pg_ls_dir('pg_distributedlog') f;
 entry_segment_kept | entry_has_gxid 
--------------------+----------------
 t                  | t              
(1 row)
0Mq: ... <quitting>
1Mq: ... <quitting>

-- Resume the old standby snapshot: expect 0 rows.
1: select gp_inject_fault('select_after_qd_create_snapshot', 'reset', dbid) from gp_segment_configuration where content = -1 and role = 'm';
 gp_inject_fault 
-----------------
 Success:        
(1 row)
-1S<:  <... completed>
 a  | b  
----+----
 1  | 1  
 2  | 2  
 3  | 3  
 4  | 4  
 5  | 5  
 6  | 6  
 7  | 7  
 8  | 8  
 9  | 9  
 10 | 10 
(10 rows)

-- A new standby snapshot must see all 10 rows.
-1S: select count(*) from hs_dlvis;
 count 
-------
 10    
(1 row)
-1Sq: ... <quitting>

1: drop table hs_dlvis;
DROP
1: drop table hs_dlvis_pin;
DROP

Expected result for the old snapshot:

-- Resume the old standby snapshot: expect 0 rows.
1: select gp_inject_fault('select_after_qd_create_snapshot', 'reset', dbid) from gp_segment_configuration where content = -1 and role = 'm';
 gp_inject_fault 
-----------------
 Success:        
(1 row)
-1S<:  <... completed>
 a | b 
---+---
(0 rows)

While testing, I also ran into an issue unrelated to this PR: when a mirror query advances oldestXmin across a segment boundary, it can call DistributedLog_Truncate() and fail with cannot make new WAL entries during recovery.
This happens without the PR, so it may be worth opening a separate issue, unless you’d prefer to fix it here.

I've included the isolation2 test below (just in case).

Isolation2 test

-- Truncation must preserve visibility for an older standby snapshot.
-- Keep the INSERT's gxid entry in a retained segment and check that replay
-- does not make the rows visible to that snapshot.
-- DLOG layout: 4096 xids/page, 32 pages/segment (131072 xids).

create extension if not exists gp_inject_fault;

1: create table hs_dlvis(a int, b int) distributed by (a);
1: create table hs_dlvis_pin(a int) distributed by (a);

-- Pin the primaries' oldestXmin with an old QD snapshot, without holding
-- back the mirrors' local xmin.
2: begin isolation level repeatable read;
2: select count(*) from hs_dlvis_pin;
1: insert into hs_dlvis_pin select generate_series(1, 10);

-- Advance local xids to the first page of the next segment.
0U: set synchronous_commit = off;
0U: do $$ declare t bigint := (txid_current() / 131072 + 1) * 131072 + 8; begin loop exit when txid_current() >= t; commit; end loop; end $$;
1U: set synchronous_commit = off;
1U: do $$ declare t bigint := (txid_current() / 131072 + 1) * 131072 + 8; begin loop exit when txid_current() >= t; commit; end loop; end $$;

-- Pause after taking the standby distributed snapshot, before dispatch.
-1S: select gp_inject_fault('select_after_qd_create_snapshot', 'suspend', dbid) from gp_segment_configuration where content = -1 and role = 'm';
-1S&: select * from hs_dlvis order by a;
1: select gp_wait_until_triggered_fault('select_after_qd_create_snapshot', 1, dbid) from gp_segment_configuration where content = -1 and role = 'm';

1: insert into hs_dlvis select g, g from generate_series(1, 10) g;
-- Check that the INSERT landed on the first page of each segment.
1: select gp_segment_id, xmin::text::bigint / 4096 % 32 as page_in_segment from hs_dlvis group by 1, 2 order by 1;

-- Advance the mirrors' oldestXmin to the INSERT's entry.
-- Retry the recovery WAL error at segment boundaries: oldestXmin is
-- already updated before the query fails.
!\retcode sh -c 'p=$(psql -XAtq -d isolation2-hot-standby -c "select port from gp_segment_configuration where content = -1 and role = chr(109)"); for i in 1 2 3 4 5; do psql -X -At -d isolation2-hot-standby -p $p -c "select count(*) from hs_dlvis" && exit 0; done; exit 1';

-- Move every primary to the second page of the segment.
0U: do $$ declare t bigint := (txid_current() / 131072) * 131072 + 4096 + 100; begin loop exit when txid_current() >= t; commit; end loop; end $$;
1U: do $$ declare t bigint := (txid_current() / 131072) * 131072 + 4096 + 100; begin loop exit when txid_current() >= t; commit; end loop; end $$;
0U: select txid_current() / 4096 % 32 as page_in_segment;
1U: select txid_current() / 4096 % 32 as page_in_segment;
0Uq:
1Uq:

-- Release the pin and trigger TRUNCATE with the second page as cutoff.
2: commit;
1: select count(*) from hs_dlvis_pin;

-- Wait until only the INSERT's segment remains on each mirror.
0M: do $$ begin for i in 1..120 loop exit when (select count(*) from pg_ls_dir('pg_distributedlog')) = 1; perform pg_sleep(1); end loop; end $$;
1M: do $$ begin for i in 1..120 loop exit when (select count(*) from pg_ls_dir('pg_distributedlog')) = 1; perform pg_sleep(1); end loop; end $$;

-- Flush DLOG to disk; remote_apply makes the DDL wait for checkpoint replay.
1: checkpoint;
1: create table hs_dlvis_sync(a int);
1: drop table hs_dlvis_sync;
0M: checkpoint;
1M: checkpoint;

-- Check that the INSERT's segment and nonzero gxid entry are still on disk.
0M: select f = upper(lpad(to_hex(x / 131072), 4, '0')) as entry_segment_kept, pg_read_binary_file('pg_distributedlog/' || f, (x % 131072) * 8, 8) <> '\x0000000000000000'::bytea as entry_has_gxid from (select distinct xmin::text::bigint as x from hs_dlvis) t, pg_ls_dir('pg_distributedlog') f;
1M: select f = upper(lpad(to_hex(x / 131072), 4, '0')) as entry_segment_kept, pg_read_binary_file('pg_distributedlog/' || f, (x % 131072) * 8, 8) <> '\x0000000000000000'::bytea as entry_has_gxid from (select distinct xmin::text::bigint as x from hs_dlvis) t, pg_ls_dir('pg_distributedlog') f;
0Mq:
1Mq:

-- Resume the old standby snapshot: expect 0 rows.
1: select gp_inject_fault('select_after_qd_create_snapshot', 'reset', dbid) from gp_segment_configuration where content = -1 and role = 'm';
-1S<:

-- A new standby snapshot must see all 10 rows.
-1S: select count(*) from hs_dlvis;
-1Sq:

1: drop table hs_dlvis;
1: drop table hs_dlvis_pin;