external
/
postgres-decoderbufs
mirror of https://github.com/debezium/postgres-decoderbufs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixing styling using clang-format

pull/1/head
Xavier Stevens 2015-01-20 15:23:45 -08:00
parent b211211fdf
commit 5594ad04ef
2 changed files with 45 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/decoderbufs.c
View File

@ -24,11 +24,11 @@
*/ */
#if defined(__linux__) #if defined(__linux__)
#include <endian.h> #include <endian.h>
#elif defined(__APPLE__) #elif defined(__APPLE__)
#include <machine/endian.h> #include <machine/endian.h>
#include <libkern/OSByteOrder.h> #include <libkern/OSByteOrder.h>
#define htobe64(x) OSSwapHostToBigInt64(x) #define htobe64(x) OSSwapHostToBigInt64(x)
#endif #endif
#include <inttypes.h> #include <inttypes.h>
@ -91,8 +91,7 @@ static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Relation rel, ReorderBufferChange *change); Relation rel, ReorderBufferChange *change);
void _PG_init(void) { void _PG_init(void) {}
}
/* specify output plugin callbacks */ /* specify output plugin callbacks */
void _PG_output_plugin_init(OutputPluginCallbacks *cb) { void _PG_output_plugin_init(OutputPluginCallbacks *cb) {
@ -148,21 +147,34 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
char *geom_oid_str = NULL; char *geom_oid_str = NULL;
char *geog_oid_str = NULL; char *geog_oid_str = NULL;
if (SPI_connect() == SPI_ERROR_CONNECT) { if (SPI_connect() == SPI_ERROR_CONNECT) {
elog(NOTICE, "Could not connect to SPI manager to scan for PostGIS types"); elog(NOTICE,
"Could not connect to SPI manager to scan for PostGIS types");
SPI_finish(); SPI_finish();
return; return;
} }
if (SPI_OK_SELECT == SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geometry'", true, 1) && SPI_processed > 0) { if (SPI_OK_SELECT ==
geom_oid_str = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1); SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geometry'",
true, 1) &&
SPI_processed > 0) {
geom_oid_str =
SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
if (geom_oid_str != NULL) { if (geom_oid_str != NULL) {
fprintf(stderr, "Decoderbufs detected PostGIS geometry type with oid: %s\n", geom_oid_str); fprintf(stderr,
"Decoderbufs detected PostGIS geometry type with oid: %s\n",
geom_oid_str);
geometry_oid = atoi(geom_oid_str); geometry_oid = atoi(geom_oid_str);
} }
} }
if (SPI_OK_SELECT == SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geography'", true, 1) && SPI_processed > 0) { if (SPI_OK_SELECT ==
geog_oid_str = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1); SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geography'",
true, 1) &&
SPI_processed > 0) {
geog_oid_str =
SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
if (geog_oid_str != NULL) { if (geog_oid_str != NULL) {
fprintf(stderr, "Decoderbufs detected PostGIS geography type with oid: %s\n", geog_oid_str); fprintf(stderr,
"Decoderbufs detected PostGIS geography type with oid: %s\n",
geog_oid_str);
geography_oid = atoi(geog_oid_str); geography_oid = atoi(geog_oid_str);
} }
} }
@ -180,8 +192,7 @@ static void pg_decode_shutdown(LogicalDecodingContext *ctx) {
/* BEGIN callback */ /* BEGIN callback */
static void pg_decode_begin_txn(LogicalDecodingContext *ctx, static void pg_decode_begin_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn) { ReorderBufferTXN *txn) {}
}
/* COMMIT callback */ /* COMMIT callback */
static void pg_decode_commit_txn(LogicalDecodingContext *ctx, static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
@ -268,13 +279,18 @@ static void print_tuple_msg(StringInfo out, Decoderbufs__DatumMessage **tup,
appendStringInfo(out, ", datum[%s]", dmsg->datum_string); appendStringInfo(out, ", datum[%s]", dmsg->datum_string);
break; break;
case POINTOID: case POINTOID:
appendStringInfo(out, ", datum[POINT(%f, %f)]", dmsg->datum_point->x, dmsg->datum_point->y); appendStringInfo(out, ", datum[POINT(%f, %f)]",
dmsg->datum_point->x, dmsg->datum_point->y);
break; break;
default: default:
if (dmsg->column_type == geometry_oid && dmsg->datum_point != NULL) { if (dmsg->column_type == geometry_oid &&
appendStringInfo(out, ", datum[GEOMETRY(POINT(%f,%f))]", dmsg->datum_point->x, dmsg->datum_point->y); dmsg->datum_point != NULL) {
} else if (dmsg->column_type == geography_oid && dmsg->datum_point != NULL) { appendStringInfo(out, ", datum[GEOMETRY(POINT(%f,%f))]",
appendStringInfo(out, ", datum[GEOGRAPHY(POINT(%f,%f))]", dmsg->datum_point->x, dmsg->datum_point->y); dmsg->datum_point->x, dmsg->datum_point->y);
} else if (dmsg->column_type == geography_oid &&
dmsg->datum_point != NULL) {
appendStringInfo(out, ", datum[GEOGRAPHY(POINT(%f,%f))]",
dmsg->datum_point->x, dmsg->datum_point->y);
} }
break; break;
} }
@ -307,13 +323,14 @@ static double numeric_to_double_no_overflow(Numeric num) {
return val; return val;
} }
static bool geography_point_as_decoderbufs_point(Datum datum, Decoderbufs__Point *p) { static bool geography_point_as_decoderbufs_point(Datum datum,
Decoderbufs__Point *p) {
GSERIALIZED *geom; GSERIALIZED *geom;
LWGEOM *lwgeom; LWGEOM *lwgeom;
LWPOINT *point = NULL; LWPOINT *point = NULL;
POINT2D p2d; POINT2D p2d;
geom = (GSERIALIZED*)PG_DETOAST_DATUM(datum); geom = (GSERIALIZED *)PG_DETOAST_DATUM(datum);
if (gserialized_get_type(geom) != POINTTYPE) { if (gserialized_get_type(geom) != POINTTYPE) {
return false; return false;
} }
@ -585,7 +602,7 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
size_t ssize = decoderbufs__row_message__pack(&rmsg, packed); size_t ssize = decoderbufs__row_message__pack(&rmsg, packed);
uint64_t flen = htobe64(ssize); uint64_t flen = htobe64(ssize);
/* frame encoding size */ /* frame encoding size */
appendBinaryStringInfo(ctx->out, (char *) &flen, sizeof(flen)); appendBinaryStringInfo(ctx->out, (char *)&flen, sizeof(flen));
/* frame encoding payload */ /* frame encoding payload */
appendBinaryStringInfo(ctx->out, packed, ssize); appendBinaryStringInfo(ctx->out, packed, ssize);
OutputPluginWrite(ctx, true); OutputPluginWrite(ctx, true);