DBZ-8403 Fix PostgreSQL 17 compilation

When compiling against PG17 on Debian testing with gcc 14, building
fails because of incompatible-pointer-type error. This commit updates
the macros added in e1f8714 to handle the pointer types.

README.md

@@ -10,15 +10,17 @@
 A PostgreSQL logical decoder output plugin to deliver data as [Protocol Buffers](https://developers.google.com/protocol-buffers), adapted for Debezium
 
 ## Thanks to
+
 - The original [Decoderbufs Project](https://github.com/xstevens/decoderbufs) on which this is based
 - [The PostgreSQL Team](https://postgresql.org) for adding [logical decoding](http://www.postgresql.org/docs/9.4/static/logicaldecoding.html) support
 
 ## Dependencies
+
 This code depends on the following libraries and requires them for compilation:
 
-* [PostgreSQL](http://www.postgresql.org) 9.6+
-* [Protobuf-c](https://github.com/protobuf-c/protobuf-c) 1.2+ - used for data serialization
-* [PostGIS](http://www.postgis.net/) 2.1+ - used for Postgres geometric types support
+- [PostgreSQL](http://www.postgresql.org) 9.6+
+- [Protobuf-c](https://github.com/protobuf-c/protobuf-c) 1.2+ - used for data serialization
+- [PostGIS](http://www.postgis.net/) 2.1+ - used for Postgres geometric types support
 
 ## Building
 
@@ -26,8 +28,10 @@ This code depends on the following libraries and requires them for compilation:
 (for pg_config), PostgreSQL server development packages, protobuf-c for the Protocol Buffer support and some PostGIS development packages.
 
 ### Installing Dependencies
+
 #### Debian
 
+```bash
 # Core build utilities 
 apt-get update && apt-get install -f -y software-properties-common build-essential pkg-config git postgresql-server-dev-9.6 
 
@@ -37,10 +41,13 @@ This code depends on the following libraries and requires them for compilation:
 # Protobuf-c dependency (requires a non-stable Debian repo)
 add-apt-repository "deb http://ftp.debian.org/debian testing main contrib" && apt-get update
 apt-get install -y libprotobuf-c-dev=1.2.1-1+b1
+```
 
 When updating the ProtoBuf definition, also install the ProtoBuf C compiler:
 
+```bash
 apt-get install -y protobuf-c-compiler=1.2.*
+```
 
 The above are taken from the Debezium [container images](https://github.com/debezium/docker-images).
 
@@ -53,16 +60,20 @@ Note that the last step from the above sequence is only required for Debian to b
 
 If you have all of the above prerequisites installed, clone this git repo to build from source:
 
+```bash
 git clone https://github.com/debezium/postgres-decoderbufs.git
 cd postgres-decoderbufs
+```
 
 ### Optional: Re-generating ProtoBuf code
 
 This is only needed after changes to the ProtoBuf definition (_proto/pg_logicaldec.proto):
 
+```bash
 cd proto
 protoc-c --c_out=../src/proto pg_logicaldec.proto
 cd ..
+```
 
 Commit the generated files to git then.
 
@@ -71,13 +82,16 @@ Commit the generated files to git then.
 If you have multiple Postgres versions installed, you can select which version to install decoderbufs into by altering your `$PATH` to point to the right version.
 Then `make` and `make install` for each version.  Here is an example:
 
+```bash
 # Install for Postgres 9.6 if I have multiple local versions
 export PATH=/usr/lib/postgresql/9.6/bin:$PATH
 make
 make install
+```
 
 Once the extension has been installed you just need to enable it and logical replication in postgresql.conf:
 
+```bash
 # MODULES
 shared_preload_libraries = 'decoderbufs'
 
@@ -87,16 +101,21 @@ Once the extension has been installed you just need to enable it and logical rep
 wal_keep_segments = 4           # in logfile segments, 16MB each; 0 disables
 #wal_sender_timeout = 60s       # in milliseconds; 0 disables
 max_replication_slots = 4       # max number of replication slots (change requires restart)
+```
 
 In addition, permissions will have to be added for the user that connects to the DB to be able to replicate. This can be modified in _pg\_hba.conf_ like so:
 
+```make
 local   replication     <youruser>                          trust
 host    replication     <youruser>  127.0.0.1/32            trust
 host    replication     <youruser>  ::1/128                 trust
+```
 
 And restart PostgreSQL.
 
 ## Usage
+
+```sql
 -- can use SQL for demo purposes
 select * from pg_create_logical_replication_slot('decoderbufs_demo', 'decoderbufs');
 
@@ -109,6 +128,7 @@ And restart PostgreSQL.
 
 -- check the WAL position of logical replicators
 select * from pg_replication_slots where slot_type = 'logical';
+```
 
 If you're performing an UPDATE/DELETE on your table and you don't see results for those operations from logical decoding, make sure you have set [REPLICA IDENTITY](http://www.postgresql.org/docs/9.4/static/sql-altertable.html#SQL-CREATETABLE-REPLICA-IDENTITY) appropriately for your use case.
 

src/decoderbufs.c

@@ -60,6 +60,12 @@
 #error Expecting timestamps to be represented as integers, not as floating-point.
 #endif
 
+#if PG_VERSION_NUM >= 170000
+#define TUPLE_ACCESS(x) x
+#else
+#define TUPLE_ACCESS(x) &x->tuple
+#endif
+
 PG_MODULE_MAGIC;
 
 /* define a time macro to convert TimestampTz into something more sane,
@@ -269,10 +275,13 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT32;
       break;
     case INT8OID:
-    case OIDOID:
       datum_msg->datum_int64 = DatumGetInt64(datum);
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
       break;
+    case OIDOID:
+      datum_msg->datum_int64 = (Oid) DatumGetUInt64(datum);
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
+      break;
     case FLOAT4OID:
       datum_msg->datum_float = DatumGetFloat4(datum);
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_FLOAT;
@@ -304,23 +313,21 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
     case TIMESTAMPOID:
       ts = DatumGetTimestamp(datum);
       if (TIMESTAMP_NOT_FINITE(ts)) {
-           ereport(ERROR, (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
-           errmsg("timestamp \'%s\'out of range", ts ? strVal(ts) : "(null)")));
+           datum_msg->datum_int64 = ts;
       } else {
            datum_msg->datum_int64 = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(ts);
+      }
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
       break;
-      }
     case TIMESTAMPTZOID:
       ts = DatumGetTimestampTz(datum);
       if (TIMESTAMP_NOT_FINITE(ts)) {
-           ereport(ERROR, (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
-           errmsg("timestamp \'%s\'out of range", ts ? strVal(ts) : "(null)")));
+           datum_msg->datum_int64 = ts;
       } else {
            datum_msg->datum_int64 = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(ts);
+      }
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
       break;
-        }
     case DATEOID:
       /* simply get the number of days as the stored 32 bit value and convert to EPOCH */
       datum_msg->datum_int32 = DATE_TO_DAYS_SINCE_EPOCH(DatumGetDateADT(datum));
@@ -499,7 +506,11 @@ static void pg_decode_begin_txn(LogicalDecodingContext *ctx,
   rmsg.has_op = true;
   rmsg.transaction_id = txn->xid;
   rmsg.has_transaction_id = true;
+#if PG_VERSION_NUM >= 150000
+  rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->xact_time.commit_time);
+#else
   rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->commit_time);
+#endif
   rmsg.has_commit_time = true;
 
   /* write msg */
@@ -537,7 +548,11 @@ static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
   rmsg.has_op = true;
   rmsg.transaction_id = txn->xid;
   rmsg.has_transaction_id = true;
+#if PG_VERSION_NUM >= 150000
+  rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->xact_time.commit_time);
+#else
   rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->commit_time);
+#endif
   rmsg.has_commit_time = true;
 
   /* write msg */
@@ -589,7 +604,11 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
   /* set common fields */
   rmsg.transaction_id = txn->xid;
   rmsg.has_transaction_id = true;
+#if PG_VERSION_NUM >= 150000
+  rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->xact_time.commit_time);
+#else
   rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->commit_time);
+#endif
   rmsg.has_commit_time = true;
   rmsg.table = pstrdup(quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
     NameStr(class_form->relname)));
@@ -609,13 +628,13 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
         rmsg.new_tuple =
             palloc(sizeof(Decoderbufs__DatumMessage*) * rmsg.n_new_tuple);
         tuple_to_tuple_msg(rmsg.new_tuple, relation,
-                           &change->data.tp.newtuple->tuple, tupdesc);
+                           TUPLE_ACCESS(change->data.tp.newtuple), tupdesc);
 
         rmsg.n_new_typeinfo = rmsg.n_new_tuple;
         rmsg.new_typeinfo =
             palloc(sizeof(Decoderbufs__TypeInfo*) * rmsg.n_new_typeinfo);
         add_metadata_to_msg(rmsg.new_typeinfo, relation,
-                           &change->data.tp.newtuple->tuple, tupdesc);
+                           TUPLE_ACCESS(change->data.tp.newtuple), tupdesc);
       }
       break;
     case REORDER_BUFFER_CHANGE_UPDATE:
@@ -630,7 +649,7 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
           rmsg.old_tuple =
               palloc(sizeof(Decoderbufs__DatumMessage*) * rmsg.n_old_tuple);
           tuple_to_tuple_msg(rmsg.old_tuple, relation,
-                             &change->data.tp.oldtuple->tuple, tupdesc);
+                             TUPLE_ACCESS(change->data.tp.oldtuple), tupdesc);
         }
         if (change->data.tp.newtuple != NULL) {
           elog(DEBUG1, "decoding new tuple information");
@@ -640,13 +659,13 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
           rmsg.new_tuple =
               palloc(sizeof(Decoderbufs__DatumMessage*) * rmsg.n_new_tuple);
           tuple_to_tuple_msg(rmsg.new_tuple, relation,
-                             &change->data.tp.newtuple->tuple, tupdesc);
+                             TUPLE_ACCESS(change->data.tp.newtuple), tupdesc);
 
           rmsg.n_new_typeinfo = rmsg.n_new_tuple;
           rmsg.new_typeinfo =
               palloc(sizeof(Decoderbufs__TypeInfo*) * rmsg.n_new_typeinfo);
           add_metadata_to_msg(rmsg.new_typeinfo, relation,
-                             &change->data.tp.newtuple->tuple, tupdesc);
+                             TUPLE_ACCESS(change->data.tp.newtuple), tupdesc);
         }
       }
       break;
@@ -662,7 +681,7 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
         rmsg.old_tuple =
             palloc(sizeof(Decoderbufs__DatumMessage*) *  rmsg.n_old_tuple);
         tuple_to_tuple_msg(rmsg.old_tuple, relation,
-                           &change->data.tp.oldtuple->tuple, tupdesc);
+                           TUPLE_ACCESS(change->data.tp.oldtuple), tupdesc);
       } else {
         elog(DEBUG1, "no information to decode from DELETE because either no PK is present or REPLICA IDENTITY NOTHING or invalid ");
       }