[CBRD-26463] [POC] support replication for OOS

src/connection/server_support.c

@@ -2683,6 +2683,7 @@ css_set_thread_info (THREAD_ENTRY * thread_p, int client_id, int rid, int tran_i
   thread_p->lockwait_state = -1;
   thread_p->query_entry = NULL;
   thread_p->tran_next_wait = NULL;
+  thread_p->oos_oids.clear ();
 
   thread_p->end_resource_tracks ();
   thread_clear_recursion_depth (thread_p);

src/storage/heap_file.c

@@ -938,7 +938,6 @@ static int heap_update_and_log_header (THREAD_ENTRY * thread_p, const HFID * hfi
 				       const PGBUF_WATCHER heap_header_watcher, HEAP_HDR_STATS * heap_hdr,
 				       const VPID new_next_vpid, const VPID new_last_vpid, const int new_num_pages);
 
-static bool heap_recdes_contains_oos (const RECDES * record);
 static SCAN_CODE heap_record_replace_oos_oids_with_values_if_exists (THREAD_ENTRY * thread_p,
 								     HEAP_GET_CONTEXT * context);
 
@@ -12264,6 +12263,8 @@ heap_attrinfo_insert_to_oos (THREAD_ENTRY * thread_p, HEAP_CACHE_ATTRINFO * attr
   VFID oos_vfid;
   OID oos_oid;
   RECDES recdes;
+  LOG_TDES *tdes;
+  int tran_index;
   int i;
 
   recdes.area_size = IO_MAX_PAGE_SIZE;
@@ -12280,6 +12281,19 @@ heap_attrinfo_insert_to_oos (THREAD_ENTRY * thread_p, HEAP_CACHE_ATTRINFO * attr
       return S_ERROR;
     }
 
+  /* Find transaction descriptor for current logging transaction */
+  tran_index = LOG_FIND_THREAD_TRAN_INDEX (thread_p);
+  tdes = LOG_FIND_TDES (tran_index);
+  if (tdes == NULL)
+    {
+      er_set (ER_FATAL_ERROR_SEVERITY, ARG_FILE_LINE, ER_LOG_UNKNOWN_TRANINDEX, 1, tran_index);
+      return S_ERROR;
+    }
+
+  /* init oos tracking info */
+  tdes->oos_insert_lsa_queue.clear ();
+  thread_p->oos_oids.clear ();
+
   for (i = 0; i < attr_info->num_values; i++)
     {
       /* is this oos column ? */
@@ -12297,6 +12311,8 @@ heap_attrinfo_insert_to_oos (THREAD_ENTRY * thread_p, HEAP_CACHE_ATTRINFO * attr
 	    {
 	      return S_ERROR;
 	    }
+
+	  thread_p->oos_oids.push_back (oos_oid);	/* for replication log */
 	  (*oos_oids)[i] = oos_oid;
 	  if (recdes.data != PTR_ALIGN (recbuf, MAX_ALIGNMENT))
 	    {
@@ -27596,7 +27612,7 @@ heap_log_postpone_heap_append_pages (THREAD_ENTRY * thread_p, const HFID * hfid,
 
 // *INDENT-ON*
 
-static bool
+bool
 heap_recdes_contains_oos (const RECDES * record)
 {
   int flag = (INT32) OR_GET_MVCC_FLAG (record->data);

src/storage/heap_file.h

@@ -716,6 +716,10 @@ extern int heap_nonheader_page_capacity ();
 
 extern int heap_rv_postpone_append_pages_to_heap (THREAD_ENTRY * thread_p, LOG_RCV * recv);
 extern void heap_rv_dump_append_pages_to_heap (FILE * fp, int length, void *data);
+
+extern bool heap_oos_find_vfid (THREAD_ENTRY * thread_p, const HFID * hfid, VFID * oos_vfid, bool docreate);
+extern bool heap_recdes_contains_oos (const RECDES * record);
+
 // *INDENT-OFF*
 extern void heap_log_postpone_heap_append_pages (THREAD_ENTRY * thread_p, const HFID * hfid, const OID * class_oid,
 						 const std::vector<VPID> &heap_pages_array);

src/storage/oos_file.cpp

@@ -31,14 +31,6 @@
 // XXX: SHOULD BE THE LAST INCLUDE HEADER
 #include "memory_wrapper.hpp"
 
-struct oos_record_header
-{
-  int total_size;
-  int chunk_index;
-  OID next_chunk_oid;
-};
-using OOS_RECORD_HEADER = struct oos_record_header;
-
 // ****************************************************************************
 // static functions
 // ****************************************************************************
@@ -674,6 +666,12 @@ oos_rv_redo_insert (THREAD_ENTRY *thread_p, LOG_RCV *rcv)
   return NO_ERROR;
 }
 
+void
+oos_push_oos_oid (THREAD_ENTRY *thread_p, const OID *oid)
+{
+  thread_p->oos_oids.push_back (*oid);
+}
+
 #if defined(CUBRID_UNIT_TEST_ENABLED)
 int
 bridge_oos_get_max_chunk_size_within_page ()

src/storage/oos_file.hpp

@@ -21,10 +21,32 @@
 #include "storage_common.h"
 #include "thread_compat.hpp"
 
+struct oos_record_header
+{
+  int total_size;
+  int chunk_index;
+  OID next_chunk_oid;
+};
+using OOS_RECORD_HEADER = struct oos_record_header;
+
+#define OOS_RECORD_HEADER_SIZE ((int) sizeof (OOS_RECORD_HEADER))
+
 extern int oos_file_create (THREAD_ENTRY *thread_p, VFID &oos_vfid);
 extern int oos_file_destroy (THREAD_ENTRY *thread_p, const VFID &oos_vfid);
 extern int oos_insert (THREAD_ENTRY *thread_p, const VFID &oos_vfid, RECDES &recdes, OID &oid);
 extern int oos_read (THREAD_ENTRY *thread_p, const OID &oid, RECDES &recdes);
 
 extern int oos_rv_redo_delete (THREAD_ENTRY *thread_p, LOG_RCV *rcv);
-extern int oos_rv_redo_insert (THREAD_ENTRY *thread_p, LOG_RCV *rcv);
+extern int oos_rv_redo_insert (THREAD_ENTRY *thread_p, LOG_RCV *rcv);
+
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern void oos_push_oos_oid (THREAD_ENTRY *thread_p, const OID *oid);
+
+#ifdef __cplusplus
+}
+#endif

src/storage/tde.h

@@ -136,7 +136,9 @@ typedef struct tde_mk_file_item
    || (rcvindex) == RVBT_ONLINE_INDEX_UNDO_TRAN_INSERT \
    || (rcvindex) == RVBT_ONLINE_INDEX_UNDO_TRAN_DELETE \
    || (rcvindex) == RVOOS_INSERT \
-   || (rcvindex) == RVOOS_DELETE)
+   || (rcvindex) == RVOOS_DELETE \
+   || (rcvindex) == RVREPL_OOS_INSERT \
+   || (rcvindex) == RVREPL_OOS_DELETE)
 
 #define tde_er_log(...) if (prm_get_bool_value (PRM_ID_ER_LOG_TDE)) _er_log_debug (ARG_FILE_LINE, "TDE: " __VA_ARGS__)
 

src/thread/thread_entry.hpp

@@ -37,6 +37,8 @@
 
 #include <cassert>
 
+#include <vector>
+#include "oid.h"
 // forward definitions
 
 // from connection_defs.h
@@ -308,6 +310,8 @@ namespace cubthread
 
       bool m_skip_end_resource_tracks_in_recycle;
 
+      std::vector<OID> oos_oids;
+
       thread_id_t get_id ();
       pthread_t get_posix_id ();
       void register_id ();

src/transaction/locator.h

@@ -111,16 +111,21 @@ typedef enum
   LC_FLUSH_INSERT,
   LC_FLUSH_INSERT_PRUNE,
   LC_FLUSH_INSERT_PRUNE_VERIFY,
+  LC_FLUSH_INSERT_OOS,
   LC_FLUSH_DELETE,
   LC_FLUSH_UPDATE,
   LC_FLUSH_UPDATE_PRUNE,
   LC_FLUSH_UPDATE_PRUNE_VERIFY,
   LC_FETCH_VERIFY_CHN
 } LC_COPYAREA_OPERATION;
 
+/* TODO:
+ *    Handle LC_FLUSH_INSERT_OOS in LC_IS_FLUSH_INSERT usage to prevent malfunctions.
+ *    Post-POC, refactor by separating macros or implementing exception handling for OOS call sites.
+ */
 #define LC_IS_FLUSH_INSERT(operation) \
   (operation == LC_FLUSH_INSERT || operation == LC_FLUSH_INSERT_PRUNE \
-   || operation == LC_FLUSH_INSERT_PRUNE_VERIFY)
+   || operation == LC_FLUSH_INSERT_PRUNE_VERIFY || operation == LC_FLUSH_INSERT_OOS)
 
 #define LC_IS_FLUSH_UPDATE(operation) \
   (operation == LC_FLUSH_UPDATE || operation == LC_FLUSH_UPDATE_PRUNE \