kafka_fdw_get_watermarks: use bigint (2nd take) (#44)

* kafka_fdw_get_watermarks: the C symbol returns int64_t (and so does rdkafka), use bigint

* kafka_fdw_get_watermarks: move function argument type changes into a new update script, bump the
extension's version.

* Remove PARALLEL SAFE label kafka_get_watermarks() function just to be on
the safe side; it hardly makes sense to use it in parallel queries
anyway.

Co-authored-by: Felix Wischke (65278) <[email protected]>

kafka_fdw--0.0.2--0.0.3.sql

@@ -0,0 +1,12 @@
+DROP FUNCTION kafka_get_watermarks(IN regclass, OUT int, OUT int, OUT int);
+
+CREATE FUNCTION kafka_get_watermarks(
+    IN rel regclass,
+	OUT partition int,
+	OUT offset_low bigint,
+	OUT offset_high bigint)
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'kafka_get_watermarks'
+LANGUAGE C STRICT;
+
+ALTER TABLE kafka_fdw_offset_dump ALTER COLUMN "offset" TYPE bigint;

kafka_fdw.control

@@ -1,5 +1,5 @@
 # kafka FDW
 comment = 'kafka Foreign Data Wrapper for CSV formated messages'
-default_version = '0.0.2'
+default_version = '0.0.3'
 module_pathname = '$libdir/kafka_fdw'
 relocatable = true

sql/kafka_fdw.sql

@@ -1,7 +1,7 @@
 CREATE TABLE kafka_fdw_offset_dump(
     tbloid oid,
     partition int,
-    "offset" int,
+    "offset" bigint,
     last_fetch timestamp DEFAULT statement_timestamp(),
     PRIMARY KEY(tbloid, partition)
 );
@@ -23,18 +23,8 @@ CREATE FOREIGN DATA WRAPPER kafka_fdw
 
 CREATE FUNCTION kafka_get_watermarks(IN rel regclass,
 	OUT partition int,
-	OUT offset_low int,
-	OUT offset_high int)
+	OUT offset_low bigint,
+	OUT offset_high bigint)
 RETURNS SETOF record
 AS 'MODULE_PATHNAME', 'kafka_get_watermarks'
 LANGUAGE C STRICT;
-
-DO $$
-DECLARE version_num INTEGER;
-BEGIN
-    SELECT current_setting('server_version_num') INTO STRICT version_num;
-    IF version_num > 90600 THEN
-        EXECUTE 'ALTER FUNCTION kafka_get_watermarks(regclass) PARALLEL SAFE';
-    END IF;
-END
-$$;