Basic support for logical decoding

The the limitations include: * TOASTed values are not supported, * Subtransactions are not supported, * Mixed transactions are not supported, * Replication origins are not supported. Logical decoding requires 32-bit transaction ids. This commit implements assignment of 32-bit "logical transaction ids". Those are assigned in unused range of 32-bit transaction ids. The number could repeat shortly, but same numbers can't co-exist in parallel. Also, this commit implements skipping of WAL-logging of empty transactions (with no data modifications) even if they contain subtransactions.
JoshInnis · May 23, 2024 · 6de6b11 · 6de6b11
1 parent 6544ef4
commit 6de6b11
Show file tree

Hide file tree

Showing 15 changed files with 616 additions and 90 deletions.
diff --git a/Makefile b/Makefile
@@ -45,6 +45,7 @@ OBJS = src/btree/btree.o \
 	   src/catalog/sys_trees.o \
 	   src/checkpoint/checkpoint.o \
 	   src/orioledb.o \
+	   src/recovery/logical.o \
 	   src/recovery/recovery.o \
 	   src/recovery/wal.o \
 	   src/recovery/worker.o \
@@ -151,7 +152,8 @@ TESTGRESCHECKS_PART_1 = t/checkpointer_test.py \
 						t/replication_test.py \
 						t/types_test.py \
 						t/undo_eviction_test.py \
-						t/toast_index_test.py
+						t/toast_index_test.py \
+						t/logical_test.py
 TESTGRESCHECKS_PART_2 = t/checkpoint_concurrent_test.py \
 						t/checkpoint_eviction_test.py \
 						t/checkpoint_same_trx_test.py \

diff --git a/ci/build.sh b/ci/build.sh
@@ -27,6 +27,8 @@ cd postgresql
 ./configure $CONFIG_ARGS
 make -sj `nproc`
 make -sj `nproc` install
+make -C contrib -sj `nproc`
+make -C contrib -sj `nproc` install
 cd ..
 
 if [ $CHECK_TYPE = "static" ] && [ $COMPILER = "clang" ]; then

diff --git a/include/recovery/logical.h b/include/recovery/logical.h
@@ -0,0 +1,25 @@
+/*-------------------------------------------------------------------------
+ *
+ * logical.h
+ *		External declarations for logical decoding of OrioleDB tables.
+ *
+ * Copyright (c) 2024, Oriole DB Inc.
+ *
+ * IDENTIFICATION
+ *	  contrib/orioledb/include/recovery/logical.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef __LOGICAL_H__
+#define __LOGICAL_H__
+
+#include "btree/btree.h"
+#include "recovery/internal.h"
+
+#include "replication/decode.h"
+#include "replication/logical.h"
+
+extern void orioledb_decode(LogicalDecodingContext *ctx,
+							XLogRecordBuffer *buf);
+
+#endif							/* __LOGICAL_H__ */
diff --git a/include/recovery/wal.h b/include/recovery/wal.h
@@ -44,6 +44,7 @@ typedef struct
 {
 	uint8		recType;
 	uint8		oxid[sizeof(OXid)];
+	uint8		logicalXid[sizeof(TransactionId)];
 } WALRecXid;
 
 typedef struct
@@ -116,15 +117,17 @@ extern void add_savepoint_wal_record(SubTransactionId parentSubid);
 extern void add_rollback_to_savepoint_wal_record(SubTransactionId parentSubid);
 extern bool local_wal_is_empty(void);
 extern XLogRecPtr flush_local_wal(bool commit);
-extern void wal_commit(OXid oxid);
-extern void wal_joint_commit(OXid oxid, TransactionId xid);
+extern void wal_commit(OXid oxid, TransactionId logicalXid);
+extern void wal_joint_commit(OXid oxid, TransactionId logicalXid, TransactionId xid);
 extern void wal_after_commit(void);
-extern void wal_rollback(OXid oxid);
+extern void wal_rollback(OXid oxid, TransactionId logicalXid);
 extern XLogRecPtr log_logical_wal_container(Pointer ptr, int length);
 extern void o_wal_insert(BTreeDescr *desc, OTuple tuple);
 extern void o_wal_update(BTreeDescr *desc, OTuple tuple);
 extern void o_wal_delete(BTreeDescr *desc, OTuple tuple);
 extern void o_wal_delete_key(BTreeDescr *desc, OTuple key);
 extern void add_truncate_wal_record(ORelOids oids);
+extern bool get_local_wal_has_material_changes(void);
+extern void set_local_wal_has_material_changes(bool value);
 
 #endif							/* __WAL_H__ */
diff --git a/include/transam/oxid.h b/include/transam/oxid.h
@@ -45,9 +45,11 @@ extern void advance_oxids(OXid new_xid);
 extern OXid get_current_oxid(void);
 extern void set_oxid_csn(OXid oxid, CommitSeqNo csn);
 extern void set_current_oxid(OXid oxid);
+extern void set_current_logical_xid(TransactionId xid);
 extern void parallel_worker_set_oxid(void);
 extern void reset_current_oxid(void);
 extern OXid get_current_oxid_if_any(void);
+extern TransactionId get_current_logical_xid(void);
 extern void current_oxid_precommit(void);
 extern void current_oxid_commit(CommitSeqNo csn);
 extern void current_oxid_abort(void);

diff --git a/include/transam/undo.h b/include/transam/undo.h
@@ -115,7 +115,9 @@ typedef struct
 {
 	bool		needs_wal_flush;
 	bool		has_retained_undo_location;
+	bool		local_wal_has_material_changes;
 	OXid		oxid;
+	TransactionId logicalXid;
 } OAutonomousTxState;
 
 /*

diff --git a/include/tuple/slot.h b/include/tuple/slot.h
@@ -32,6 +32,7 @@ typedef struct OTableSlot
 	bytea	   *rowid;
 	CommitSeqNo csn;
 	int			ixnum;
+	bool		leafTuple;
 	uint32		version;
 	OTupleReaderState state;
 	BTreeLocationHint hint;
@@ -44,6 +45,12 @@ extern void tts_orioledb_store_tuple(TupleTableSlot *slot, OTuple tuple,
 									 OTableDescr *descr, CommitSeqNo csn,
 									 int ixnum, bool shouldfree,
 									 BTreeLocationHint *hint);
+extern void tts_orioledb_store_non_leaf_tuple(TupleTableSlot *slot,
+											  OTuple tuple,
+											  OTableDescr *descr,
+											  CommitSeqNo csn,
+											  int ixnum, bool shouldfree,
+											  BTreeLocationHint *hint);
 extern OTuple tts_orioledb_make_secondary_tuple(TupleTableSlot *slot,
 												OIndexDescr *idx,
 												bool leaf);

diff --git a/src/orioledb.c b/src/orioledb.c
@@ -22,6 +22,7 @@
 #include "catalog/o_sys_cache.h"
 #include "catalog/sys_trees.h"
 #include "checkpoint/checkpoint.h"
+#include "recovery/logical.h"
 #include "recovery/recovery.h"
 #include "recovery/wal.h"
 #include "s3/headers.h"
@@ -235,7 +236,7 @@ static RmgrData rmgr =
 	.rm_desc = orioledb_rm_desc,
 	.rm_identify = orioledb_rm_identify,
 	.rm_mask = NULL,
-	.rm_decode = NULL
+	.rm_decode = orioledb_decode
 };
 
 void