DBZ-3 Adds support for the CASH, BIT, JSONB, VARBIT, DATE,TIME,TIMETZ and INTERVAL oids

It also changes the previous behavior for TIMESTAMP and TIMESTAMPTZ to return UTC style LONGs as opposed to STRINGS.

src/decoderbufs.c

@@ -51,8 +51,15 @@
 #include "utils/syscache.h"
 #include "utils/typcache.h"
 #include "utils/uuid.h"
+#include "utils/timestamp.h"
+#include "utils/date.h"
+#include "utils/cash.h"
 #include "proto/pg_logicaldec.pb-c.h"
 
+#ifndef HAVE_INT64_TIMESTAMP
+#error Expecting timestamps to be represented as integers, not as floating-point.
+#endif
+
 /* POSTGIS version define so it doesn't redef macros */
 #define POSTGIS_PGSQL_VERSION 94
 #include "liblwgeom.h"
@@ -60,15 +67,11 @@
 PG_MODULE_MAGIC;
 
 /* define a time macro to convert TimestampTz into something more sane,
- * which in this case is microseconds since epoch
+ * which in this case is microseconds since epoch. This is because PG stores internally the timestamps relative to
+ * 2000-01-01T00:00:00Z and not the Unix epoch.
  */
-#ifdef HAVE_INT64_TIMESTAMP
+#define TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(t)    t + (POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * USECS_PER_DAY;
-#define TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(t)                                     \
+#define DATE_TO_DAYS_SINCE_EPOCH(t)    t + (POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE);
-  t + ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY *USECS_PER_SEC);
-#else
-#define TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(t)                                     \
-  (t + ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY)) * 1000.0;
-#endif
 
 typedef struct {
   MemoryContext context;
@@ -395,6 +398,11 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
   bytea *valptr = NULL;
   const char *output = NULL;
   Point *p = NULL;
+  Timestamp ts = NULL;
+  double duration;
+  TimeTzADT *timetz = NULL;
+  Interval   *interval = NULL;
+     
   int size = 0;
   switch (typid) {
     case BOOLOID:
@@ -429,25 +437,67 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
         datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_DOUBLE;
       }
       break;
+    case CASHOID: 
+      datum_msg->datum_int64 = DatumGetCash(datum);
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
+      break;
     case CHAROID:
     case VARCHAROID:
     case BPCHAROID:
     case TEXTOID:
     case JSONOID:
+    case JSONBOID: 
     case XMLOID:
+    case BITOID:
+    case VARBITOID:
     case UUIDOID:
       output = OidOutputFunctionCall(typoutput, datum);
       datum_msg->datum_string = pnstrdup(output, strlen(output));
       datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_STRING;
       break;
     case TIMESTAMPOID:
-    /*
+      ts = DatumGetTimestamp(datum);
-     * THIS FALLTHROUGH IS MAKING THE ASSUMPTION WE ARE ON UTC
+      if (TIMESTAMP_NOT_FINITE(ts)) {
-     */
+           ereport(ERROR, (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
+           errmsg("timestamp \'%s\'out of range", ts ? strVal(ts) : "(null)")));
+      } else {
+           datum_msg->datum_int64 = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(ts);
+           datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
+           break;       
+      }
     case TIMESTAMPTZOID:
-      output = timestamptz_to_str(DatumGetTimestampTz(datum));
+      ts = DatumGetTimestampTz(datum);
-      datum_msg->datum_string = pnstrdup(output, strlen(output));
+      if (TIMESTAMP_NOT_FINITE(ts)) {
-      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_STRING;
+           ereport(ERROR, (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
+           errmsg("timestamp \'%s\'out of range", ts ? strVal(ts) : "(null)")));
+       } 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));
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT32;
+      break;
+    case TIMEOID:
+      datum_msg->datum_int64 = DatumGetTimeADT(datum);
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_INT64;
+      break;                                                 
+    case TIMETZOID:
+      timetz = DatumGetTimeTzADTP(datum);
+      /* use GMT-equivalent time */     
+      datum_msg->datum_double = (double) (timetz->time + (timetz->zone * 1000000.0));
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_DOUBLE;
+      break;     
+    case INTERVALOID:
+      interval = DatumGetIntervalP(datum);
+      /* 
+        Convert the month part of Interval to days using assumed average month length of 365.25/12.0 days. 
+      */  
+      duration = interval->time + interval->day * (double) USECS_PER_DAY + interval->month * ((DAYS_PER_YEAR / (double) MONTHS_PER_YEAR) * USECS_PER_DAY);
+      datum_msg->datum_double = duration;
+      datum_msg->datum_case = DECODERBUFS__DATUM_MESSAGE__DATUM_DATUM_DOUBLE;
       break;
     case BYTEAOID:
       valptr = DatumGetByteaPCopy(datum);