[yugabyte#25431] YSQL: Perform read for deferred FK trigger at txn end

Summary:
FK constraint has the following optimization for performing batched read:
- FK constraint registers ybctids which need to be checked
- on checking particular ybctid for particular FK constraint all registered ybctds are read at once, result is cached
- checking ybctid for another FK constraint will check cached result instead of making real read request.

In Postgres constraint could be of 2 types:
- IMMEDIATE (default). Checked at statement end.
- DEFERRED. Checked at transaction end.

Nowadays FK optimization work incorrectly in case multiple constraint of different types is used in same transaction. And the reason is that YSQL registers ybctids for both types of constraint in single map.

Example:

```
1. CREATE TABLE pk_t(k INT PRIMARY KEY);
2. CREATE TABLE fk_t(k INT PRIMARY KEY,
                     pk_1 INT REFERENCES pk_t(k),
                     pk_2 INT REFERENCES pk_t(k) DEFERRABLE INITIALLY DEFERRED);
3. INSERT INTO pk_t VALUES (1);
4. BEGIN;
5. INSERT INTO fk_t VALUES(1, 1, 2);
6. INSERT INTO pk_t VALUES(2);
7. COMMIT;
```

- On step #5 YSQL inserts value `(1, 1, 2)` into table with 2 FK referenced columns. Where constraint for second column is `DEFERRED`.
- Both constraint registers ybctid for rows `k = 1` and `k = 2` in table `pk_t`.
- Because constraint for first column is non deferred is it executed immediately (at the end of the statement).
- Due to optimization both registered ybctids will be read at once. And result will be cached. And the result contains `k = 1` only, because `k = 2` is only inserted on step #6
- On step #7 YSQL will perform the check of constraint for second column and cached result will be used which doesn't have `k = 2` inserted on step #6

Solution is to store ybctids for deferred and non-deferred constraint in different structure. And read them independently. All the ybctids registered for deferred constraints will be read only on transaction commit step (step #7).
For this purpose the new `YBCNotifyDeferredTriggersProcessingStarted()` function is introduced. Which is called straight before deferred triggers firing at the beginning of `COMMIT` command processing.
Jira: DB-14665

Test Plan:
Jenkins

New unit test are introduced

```
./yb_build.sh --gtest_filter PgFKeyTest.DeferredConstraintReadAtTxnEnd
```

Reviewers: pjain, myang, kramanathan, patnaik.balivada

Reviewed By: pjain

Subscribers: yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D40896

Author: d-uspenskiy

src/postgres/src/backend/access/transam/xact.c

@@ -2367,6 +2367,9 @@ CommitTransaction(void)
 			 TransStateAsString(s->state));
 	Assert(s->parent == NULL);
 
+	if (IsYugaByteEnabled())
+		YBCNotifyDeferredTriggersProcessingStarted();
+
 	/*
 	 * Do pre-commit processing that involves calling user-defined code, such
 	 * as triggers.  SECURITY_RESTRICTED_OPERATION contexts must not queue an

src/postgres/src/backend/commands/tablecmds.c

@@ -12374,7 +12374,9 @@ YbGetNext(YbFKTriggerScanDesc desc, TupleTableSlot *slot)
 				ExecDropSingleTupleTableSlot(new_slot);
 				break;
 			}
-			YbAddTriggerFKReferenceIntent(desc->trigger, desc->fk_rel, new_slot, desc->estate);
+			YbAddTriggerFKReferenceIntent(desc->trigger,
+										  desc->fk_rel, new_slot, desc->estate,
+										  /* is_deferred= */ false);
 			desc->buffered_tuples[desc->buffered_tuples_size++] = new_slot;
 		}
 	}

src/postgres/src/backend/commands/trigger.c

@@ -6620,8 +6620,13 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
 			}
 		}
 
-		if (IsYBBackedRelation(rel) && RI_FKey_trigger_type(trigger->tgfoid) == RI_TRIGGER_FK)
-			YbAddTriggerFKReferenceIntent(trigger, rel, newslot, estate);
+		if (IsYBBackedRelation(rel) &&
+			RI_FKey_trigger_type(trigger->tgfoid) == RI_TRIGGER_FK)
+		{
+			const bool is_deferred = new_shared.ybc_txn_fdw_tuplestore != NULL;
+			YbAddTriggerFKReferenceIntent(trigger, rel, newslot, estate,
+										  is_deferred);
+		}
 
 		afterTriggerAddEvent(&afterTriggers.query_stack[afterTriggers.query_depth].events,
 							 &new_event, &new_shared);

src/postgres/src/backend/utils/adt/ri_triggers.c

@@ -3375,7 +3375,8 @@ RI_FKey_trigger_type(Oid tgfoid)
 
 void
 YbAddTriggerFKReferenceIntent(Trigger *trigger, Relation fk_rel,
-							  TupleTableSlot *new_slot, EState *estate)
+							  TupleTableSlot *new_slot, EState *estate,
+							  bool is_deferred)
 {
 	YbcPgYBTupleIdDescriptor *descr;
 
@@ -3402,7 +3403,9 @@ YbAddTriggerFKReferenceIntent(Trigger *trigger, Relation fk_rel,
 			null_found = attr->is_null && (attr->attr_num > 0);
 
 		if (!null_found)
-			HandleYBStatus(YBCAddForeignKeyReferenceIntent(descr, YBCIsRegionLocal(fk_rel)));
+			HandleYBStatus(YBCAddForeignKeyReferenceIntent(descr,
+														   YBCIsRegionLocal(fk_rel),
+														   is_deferred));
 		pfree(descr);
 	}
 }

src/postgres/src/include/commands/trigger.h

@@ -283,7 +283,7 @@ extern void RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel,
 									 Relation pk_rel);
 extern void YbAddTriggerFKReferenceIntent(Trigger *trigger, Relation fk_rel,
 										  TupleTableSlot *new_slot,
-										  EState *estate);
+										  EState *estate, bool is_deferred);
 
 /* result values for RI_FKey_trigger_type: */
 #define RI_TRIGGER_PK	1		/* is a trigger on the PK relation */

src/postgres/src/test/regress/expected/yb.orig.foreign_key.out

@@ -618,7 +618,7 @@ INSERT INTO c VALUES(1, 1, 1, 1, 1);
  Trigger for constraint c_p2_fk_fkey: calls=1
  Trigger for constraint c_p4_fk_fkey: calls=1
  Storage Read Requests: 1
- Storage Rows Scanned: 4
+ Storage Rows Scanned: 2
  Storage Write Requests: 1
  Storage Flush Requests: 1
 (9 rows)
@@ -699,7 +699,7 @@ INSERT INTO fk VALUES (500, 1); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_id_name_fkey"
 DETAIL:  Key (id, name)=(500, 1) is not present in table "pk".
 SELECT * from fk;
- id | name 
+ id | name
 ----+------
   1 |  500
 (1 row)
@@ -721,7 +721,7 @@ INSERT INTO fk VALUES (200); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_id_fkey"
 DETAIL:  Key (id)=(200) is not present in table "pk".
 SELECT * FROM fk;
- id  
+ id
 -----
    1
  105
@@ -743,7 +743,7 @@ INSERT INTO fk VALUES (200); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_b_fkey"
 DETAIL:  Key (b)=(200) is not present in table "pk".
 SELECT * FROM fk;
-  b  
+  b
 -----
    1
  105
@@ -764,7 +764,7 @@ INSERT INTO fk VALUES (150, 20); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_a_c_fkey"
 DETAIL:  Key (a, c)=(150, 20) is not present in table "pk".
 SELECT * FROM fk;
- a | c  
+ a | c
 ---+----
  1 | 20
 (1 row)
@@ -787,7 +787,7 @@ INSERT INTO fk(a, c) VALUES (150, 20); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_a_c_fkey"
 DETAIL:  Key (a, c)=(150, 20) is not present in table "pk".
 SELECT * FROM fk;
- b | d | a | c  
+ b | d | a | c
 ---+---+---+----
    |   | 1 | 20
 (1 row)
@@ -810,7 +810,7 @@ DETAIL:  Partition key of the failing row contains (a) = (20).
 INSERT into pk values (5);
 INSERT into fk values (5);
 SELECT * FROM fk;
- a 
+ a
 ---
  5
 (1 row)
@@ -834,7 +834,7 @@ INSERT INTO fk VALUES (1, 10); -- should fail
 ERROR:  insert or update on table "fk" violates foreign key constraint "fk_b2_fkey"
 DETAIL:  Key (b2)=(10) is not present in table "pk2".
 SELECT * from fk;
- b | b2 
+ b | b2
 ---+----
  1 |  1
 (1 row)
@@ -851,7 +851,7 @@ DETAIL:  Key (a, b)=(2, 1) is not present in table "pk".
 INSERT INTO pk VALUES (5, 1);
 UPDATE fk set a = a + 4; -- should pass
 SELECT * from fk;
- a | b 
+ a | b
 ---+---
  5 | 1
 (1 row)

src/yb/yql/pggate/pg_fk_reference_cache.cc

@@ -47,8 +47,10 @@ class PgFKReferenceCache::Impl {
 
   void Clear() {
     references_.clear();
-    intents_.clear();
+    regular_intents_.clear();
+    deferred_intents_.clear();
     region_local_tables_.clear();
+    intents_ = &regular_intents_;
   }
 
   void DeleteReference(const LightweightTableYbctid& key) {
@@ -66,26 +68,31 @@ class PgFKReferenceCache::Impl {
       return true;
     }
 
+    auto reader = reader_provider_();
+    auto residual_capacity = std::min<size_t>(intents_->size(), buffering_settings_.max_batch_size);
+    reader.Reserve(residual_capacity);
+
     // Check existence of required FK intent.
-    // Absence means the key was checked by previous batched request and was not found.
-    // We don't need to call the reader in this case.
-    auto it = intents_.find(key);
-    if (it == intents_.end()) {
-      return false;
+    const auto intents_end = intents_->end();
+    auto it = intents_->find(key);
+    if (it == intents_end) {
+      if (!IsDeferredTriggersProcessingStarted()) {
+        // In case of processing non deferred intents absence means the key was checked by previous
+        // batched request and was not found. We don't need to call the reader in this case.
+        return false;
+      }
+      // In case of processing deferred intents absence of intent could be caused by
+      // subtxnransaction rollback. In this case we have to make a read attempt.
+      reader.Add(TableYbctid{key.table_id, std::string{key.ybctid}});
+    } else {
+      // If the reader fails to get the result, we fail the whole operation (and transaction).
+      // Hence it's ok to extract (erase) the keys from intent before calling reader.
+      reader.Add(std::move(intents_->extract(it).value()));
     }
+    --residual_capacity;
 
-    auto reader = reader_provider_();
-    auto available_capacity = std::min<size_t>(
-        intents_.size(), buffering_settings_.max_batch_size);
-    reader.Reserve(available_capacity);
-    // If the reader fails to get the result, we fail the whole operation (and transaction).
-    // Hence it's ok to extract (erase) the keys from intent before calling reader.
-    reader.Add(std::move(intents_.extract(it).value()));
-    --available_capacity;
-
-    for (auto it = intents_.begin();
-         it != intents_.end() && available_capacity; --available_capacity) {
-      reader.Add(std::move(intents_.extract(it++).value()));
+    for (auto it = intents_->begin(); it != intents_end && residual_capacity; --residual_capacity) {
+      reader.Add(std::move(intents_->extract(it++).value()));
     }
 
     // Add the keys found in docdb to the FK cache.
@@ -98,23 +105,42 @@ class PgFKReferenceCache::Impl {
     return references_.contains(key);
   }
 
-  void AddIntent(const LightweightTableYbctid& key, bool is_region_local) {
+  void AddIntent(const LightweightTableYbctid& key, const IntentOptions& options) {
+    LOG_IF(DFATAL, IsDeferredTriggersProcessingStarted())
+        << "AddIntent is not expected after deferred trigger processing start";
+
     if (references_.contains(key)) {
         return;
     }
 
-    if (is_region_local) {
+    if (options.is_region_local) {
       region_local_tables_.insert(key.table_id);
     }
-    DCHECK(is_region_local || !region_local_tables_.contains(key.table_id));
-    intents_.emplace(key.table_id, std::string(key.ybctid));
+    LOG_IF(DFATAL, !options.is_region_local && region_local_tables_.contains(key.table_id))
+        << "The " << key.table_id << " table was previously reported as region local";
+    (options.is_deferred ? &deferred_intents_ : &regular_intents_)->emplace(
+        key.table_id, std::string(key.ybctid));
+  }
+
+  void OnDeferredTriggersProcessingStarted() {
+    LOG_IF(DFATAL, IsDeferredTriggersProcessingStarted())
+        << "Multiple call of OnDeferredTriggersProcessingStarted is not expected";
+    LOG_IF(DFATAL, !regular_intents_.empty())
+        << "OnDeferredTriggersProcessingStarted implies all non deferred intents were processed";
+    intents_ = &deferred_intents_;
   }
 
  private:
+  [[nodiscard]] bool IsDeferredTriggersProcessingStarted() const {
+    return intents_ == &deferred_intents_;
+  }
+
   YbctidReaderProvider& reader_provider_;
   const BufferingSettings& buffering_settings_;
   MemoryOptimizedTableYbctidSet references_;
-  TableYbctidSet intents_;
+  TableYbctidSet regular_intents_;
+  TableYbctidSet deferred_intents_;
+  TableYbctidSet* intents_ = &regular_intents_;
   OidSet region_local_tables_;
 };
 
@@ -142,8 +168,13 @@ Result<bool> PgFKReferenceCache::IsReferenceExists(
   return impl_->IsReferenceExists(database_id, key);
 }
 
-void PgFKReferenceCache::AddIntent(const LightweightTableYbctid& key, bool is_region_local) {
-  impl_->AddIntent(key, is_region_local);
+void PgFKReferenceCache::AddIntent(
+    const LightweightTableYbctid& key, const IntentOptions& options) {
+  impl_->AddIntent(key, options);
+}
+
+void PgFKReferenceCache::OnDeferredTriggersProcessingStarted() {
+  impl_->OnDeferredTriggersProcessingStarted();
 }
 
 } // namespace yb::pggate

src/yb/yql/pggate/pg_fk_reference_cache.h

@@ -28,6 +28,11 @@ struct BufferingSettings;
 
 class PgFKReferenceCache {
  public:
+  struct IntentOptions {
+    bool is_region_local;
+    bool is_deferred;
+  };
+
   PgFKReferenceCache(YbctidReaderProvider& reader_provider,
                      std::reference_wrapper<const BufferingSettings> buffering_settings);
   ~PgFKReferenceCache();
@@ -36,7 +41,8 @@ class PgFKReferenceCache {
   void DeleteReference(const LightweightTableYbctid& key);
   void AddReference(const LightweightTableYbctid& key);
   Result<bool> IsReferenceExists(PgOid database_id, const LightweightTableYbctid& key);
-  void AddIntent(const LightweightTableYbctid& key, bool is_region_local);
+  void AddIntent(const LightweightTableYbctid& key, const IntentOptions& options);
+  void OnDeferredTriggersProcessingStarted();
 
  private:
   class Impl;

src/yb/yql/pggate/pggate.cc

@@ -1968,8 +1968,8 @@ Result<bool> PgApiImpl::ForeignKeyReferenceExists(
 }
 
 void PgApiImpl::AddForeignKeyReferenceIntent(
-    PgOid table_id, bool is_region_local, const Slice& ybctid) {
-  fk_reference_cache_.AddIntent(LightweightTableYbctid{table_id, ybctid}, is_region_local);
+    PgOid table_id, const Slice& ybctid, const PgFKReferenceCache::IntentOptions& options) {
+  fk_reference_cache_.AddIntent(LightweightTableYbctid{table_id, ybctid}, options);
 }
 
 void PgApiImpl::DeleteForeignKeyReference(PgOid table_id, const Slice& ybctid) {
@@ -1980,6 +1980,10 @@ void PgApiImpl::AddForeignKeyReference(PgOid table_id, const Slice& ybctid) {
   fk_reference_cache_.AddReference(LightweightTableYbctid{table_id, ybctid});
 }
 
+void PgApiImpl::NotifyDeferredTriggersProcessingStarted() {
+  fk_reference_cache_.OnDeferredTriggersProcessingStarted();
+}
+
 Status PgApiImpl::AddExplicitRowLockIntent(
     const PgObjectId& table_id, const Slice& ybctid, const YbcPgExplicitRowLockParams& params,
     bool is_region_local, YbcPgExplicitRowLockErrorInfo& error_info) {

src/yb/yql/pggate/pggate.h

@@ -704,7 +704,9 @@ class PgApiImpl {
   void DeleteForeignKeyReference(PgOid table_id, const Slice& ybctid);
   void AddForeignKeyReference(PgOid table_id, const Slice& ybctid);
   Result<bool> ForeignKeyReferenceExists(PgOid table_id, const Slice& ybctid, PgOid database_id);
-  void AddForeignKeyReferenceIntent(PgOid table_id, bool is_region_local, const Slice& ybctid);
+  void AddForeignKeyReferenceIntent(
+        PgOid table_id, const Slice& ybctid, const PgFKReferenceCache::IntentOptions& options);
+  void NotifyDeferredTriggersProcessingStarted();
 
   Status AddExplicitRowLockIntent(
       const PgObjectId& table_id, const Slice& ybctid,