test(sink): add bigquery-sink compatible test (#16497)

integration_tests/big-query-sink/bq_prepare.sql

@@ -4,3 +4,35 @@ CREATE TABLE `rwctest.bqtest.bq_sink` (
     target_id STRING,
     event_timestamp TIMESTAMP
 );
+
+DROP TABLE IF EXISTS `rwctest.bqtest.bq_sink_data_types`;
+CREATE TABLE `rwctest.bqtest.bq_sink_data_types` (
+    types_id INT64,
+    c_boolean BOOL,
+    c_smallint INT64,
+    c_integer INT64,
+    c_bigint INT64,
+    c_decimal NUMERIC,
+    c_double_precision FLOAT64,
+    c_varchar STRING,
+    c_bytea BYTES,
+    c_date DATE,
+    c_timestamp DATETIME,
+    c_timestamptz TIMESTAMP,
+    c_interval INTERVAL,
+    c_jsonb JSON,
+    c_boolean_array ARRAY<BOOL>,
+    c_smallint_array ARRAY<INT64>,
+    c_integer_array ARRAY<INT64>,
+    c_bigint_array ARRAY<INT64>,
+    c_decimal_array ARRAY<NUMERIC>,
+    c_double_precision_array ARRAY<FLOAT64>,
+    c_varchar_array ARRAY<STRING>,
+    c_bytea_array ARRAY<BYTES>,
+    c_date_array ARRAY<DATE>,
+    c_timestamp_array ARRAY<DATETIME>,
+    c_timestamptz_array ARRAY<TIMESTAMP>,
+    c_interval_array ARRAY<INTERVAL>,
+    c_jsonb_array ARRAY<JSON>,
+    c_struct STRUCT<s_int INT64, s_bool BOOL>
+);

integration_tests/big-query-sink/create_sink.sql

@@ -27,3 +27,15 @@ FROM
     -- region = '${aws_region}',
     -- force_append_only='true',
 -- );
+
+CREATE SINK bq_sink_data_types_sink
+FROM
+    bq_sink_data_types WITH (
+    connector = 'bigquery',
+    type = 'append-only',
+    bigquery.local.path= '/gcp-rwctest.json',
+    bigquery.project= 'rwctest',
+    bigquery.dataset= 'bqtest',
+    bigquery.table= 'bq_sink_data_types',
+    force_append_only='true'
+);

integration_tests/big-query-sink/create_source.sql

@@ -14,3 +14,41 @@ CREATE table user_behaviors (
     fields.user_id.end = '100',
     datagen.rows.per.second = '10'
 ) FORMAT PLAIN ENCODE JSON;
+
+CREATE TABLE bq_sink_data_types (
+    types_id int,
+    c_boolean boolean,
+    c_smallint smallint,
+    c_integer integer,
+    c_bigint bigint,
+    c_decimal decimal,
+    c_double_precision double precision,
+    c_varchar varchar,
+    c_bytea bytea,
+    c_date date,
+    c_timestamp timestamp,
+    c_timestamptz timestamptz,
+    c_interval interval,
+    c_jsonb jsonb,
+    c_boolean_array boolean[],
+    c_smallint_array smallint[],
+    c_integer_array integer[],
+    c_bigint_array bigint[],
+    c_decimal_array decimal[],
+    c_double_precision_array double precision[],
+    c_varchar_array varchar[],
+    c_bytea_array bytea[],
+    c_date_array date[],
+    c_timestamp_array timestamp[],
+    c_timestamptz_array timestamptz[],
+    c_interval_array interval[],
+    c_jsonb_array jsonb[],
+    c_struct STRUCT<s_int INTEGER, s_bool boolean>,
+    PRIMARY KEY(types_id)
+);
+
+INSERT INTO bq_sink_data_types VALUES (1, False, 0, 0, 0, 0, 0, '', '\x00', '0001-01-01', '0001-01-01 00:00:00', '0001-01-01 00:00:00'::timestamptz, interval '0 second', '{}', array[]::boolean[], array[]::smallint[], array[]::integer[], array[]::bigint[], array[]::decimal[], array[]::double precision[], array[]::varchar[], array[]::bytea[], array[]::date[], array[]::timestamp[], array[]::timestamptz[], array[]::interval[], array[]::jsonb[], ROW(1, False));
+
+INSERT INTO bq_sink_data_types VALUES (2, False, -32767, -2147483647, -9223372036854775807, -10.0, -10000.0, 'aa', '\x00', '1970-01-01', '1970-01-01 00:00:00.123456', '1970-01-01 00:00:00.123456Z', interval '10 second', '{}', array[False::boolean]::boolean[], array[-32767::smallint]::smallint[], array[-2147483647::integer]::integer[], array[-9223372036854775807::bigint]::bigint[], array[-10.0::decimal]::decimal[], array[-10000.0::double precision]::double precision[], array[''::varchar]::varchar[], array['\x00'::bytea]::bytea[], array['1970-01-01'::date]::date[], array['1970-01-01 00:00:00'::timestamp]::timestamp[], array['1970-01-01 00:00:00Z'::timestamptz]::timestamptz[], array[interval '0 second'::interval]::interval[], array['{}'::jsonb]::jsonb[], ROW(2, True));
+
+INSERT INTO bq_sink_data_types VALUES (3, True, 32767, 2147483647, 9223372036854775807, -10.0, 10000.0, 'zzzzzz', '\xffffff', '9999-12-31', '9999-12-31 23:59:59', '9999-12-31 23:59:59Z', interval '9990 year', '{"whatever":"meaningless"}', array[True::boolean]::boolean[], array[32767::smallint]::smallint[], array[2147483647::integer]::integer[], array[9223372036854775807::bigint]::bigint[], array[-10.0::decimal]::decimal[], array[10000.0::double precision]::double precision[], array['zzzzzz'::varchar]::varchar[], array['\xffffff'::bytea]::bytea[], array['9999-12-31'::date]::date[], array['9999-12-31 23:59:59'::timestamp]::timestamp[], array['9999-12-31 23:59:59Z'::timestamptz]::timestamptz[], array[interval '9990 year'::interval]::interval[], array['{"whatever":"meaningless"}'::jsonb]::jsonb[], ROW(3, True));

integration_tests/big-query-sink/sink_check.py

@@ -2,7 +2,7 @@
 import subprocess
 import sys
 
-relations = ['rwctest.bqtest.bq_sink']
+relations = ['rwctest.bqtest.bq_sink', 'rwctest.bqtest.bq_sink_data_types']
 
 failed_cases = []
 for rel in relations: