diff --git a/v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java b/v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java index d9bfd97d44..106781afc3 100644 --- a/v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java +++ b/v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java @@ -25,11 +25,14 @@ import java.time.Duration; import java.time.Instant; import java.time.LocalDate; +import java.time.LocalTime; +import java.time.OffsetTime; import java.time.Period; import java.time.ZoneId; import java.time.ZoneOffset; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; +import java.time.temporal.ChronoUnit; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -650,6 +653,54 @@ static void handleDatastreamRecordType( .withZoneSameInstant(ZoneId.of("UTC")) .format(DEFAULT_TIMESTAMP_WITH_TZ_FORMATTER)); break; + case "timeTz": + long timeTzNanos = + ((Number) getOrDefault(element, "time", 0L)).longValue() + * TimeUnit.MICROSECONDS.toNanos(1); + int offsetSeconds = ((Number) getOrDefault(element, "offset", 0)).intValue() / 1000; + + ZoneOffset timeTzOffset = ZoneOffset.ofTotalSeconds(offsetSeconds); + + if (timeTzNanos == 86400000000000L) { + jsonObject.put(fieldName, "24:00:00" + timeTzOffset.toString()); + break; + } + + LocalTime localTime = LocalTime.ofNanoOfDay(timeTzNanos); + OffsetTime offsetTime = OffsetTime.of(localTime, timeTzOffset); + jsonObject.put(fieldName, offsetTime.format(DateTimeFormatter.ISO_OFFSET_TIME)); + break; + case "interval": + int months = ((Number) getOrDefault(element, "months", 0)).intValue(); + int hours = ((Number) getOrDefault(element, "hours", 0)).intValue(); + long micros = ((Number) getOrDefault(element, "micros", 0L)).longValue(); + + int days = hours / 24; + int remainingHours = hours % 24; + + Period intervalPeriod = Period.ZERO.plusMonths(months).plusDays(days).normalized(); + Duration intervalDuration = + Duration.ZERO.plusHours(remainingHours).plus(micros, ChronoUnit.MICROS); + + if (intervalPeriod.isZero() && intervalDuration.isZero()) { + jsonObject.put(fieldName, "PT0S"); + break; + } + + StringBuilder result = new StringBuilder(); + if (!intervalPeriod.isZero()) { + result.append(intervalPeriod.toString()); + } + if (!intervalDuration.isZero()) { + if (result.length() == 0) { + result.append(intervalDuration.toString()); + } else { + // Remove the 'P' from Duration + result.append(intervalDuration.toString().substring(1)); + } + } + jsonObject.put(fieldName, result.toString()); + break; /* * The `intervalNano` maps to nano second precision interval type used by Cassandra Interval. * On spanner this will map to `string` or `Interval` type. diff --git a/v2/datastream-common/src/test/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJsonTest.java b/v2/datastream-common/src/test/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJsonTest.java index 9b06541856..d3d8a806be 100644 --- a/v2/datastream-common/src/test/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJsonTest.java +++ b/v2/datastream-common/src/test/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJsonTest.java @@ -197,7 +197,7 @@ public void testPostgresByteArray() throws IOException, URISyntaxException { ObjectMapper mapper = new ObjectMapper(); JsonNode changeEvent = mapper.readTree(jsonData); // The avro file contains binary_content: b'\xde\xad\xbe\xef', which is converted to - // base64 encoded string by Jackson library. + // base64 encoded string by Jackson. assertEquals("3q2+7w==", changeEvent.get("binary_content").textValue()); } @@ -316,6 +316,76 @@ public void testIntervalNano() throws JsonProcessingException { assertEquals(expected, new ObjectMapper().writeValueAsString(objectNode)); } + @Test + public void testInterval() throws JsonProcessingException { + ObjectNode objectNode = new ObjectNode(new JsonNodeFactory(true)); + + /* Basic Test: 1 month, 26 hours (1 day + 2 hours), 3000000 micros (3 seconds) */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "basic", generateIntervalSchema(), generateIntervalRecord(1, 26, 3000000L), objectNode); + + /* Zero interval */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "zero_interval", generateIntervalSchema(), generateIntervalRecord(0, 0, 0L), objectNode); + + /* Only months */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "only_months", generateIntervalSchema(), generateIntervalRecord(12, 0, 0L), objectNode); + + /* Only time (5 hours + 123456 micros = 5H0.123456S) */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "only_time", generateIntervalSchema(), generateIntervalRecord(0, 5, 123456L), objectNode); + + /* Negative interval (-1 month, -26 hours = -1 day - 2 hours, -3000000 micros = -3 seconds) */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "neg_basic", + generateIntervalSchema(), + generateIntervalRecord(-1, -26, -3000000L), + objectNode); + + String expected = + "{\"basic\":\"P1M1DT2H3S\"," + + "\"zero_interval\":\"PT0S\"," + + "\"only_months\":\"P1Y\"," + + "\"only_time\":\"PT5H0.123456S\"," + + "\"neg_basic\":\"P-1M-1DT-2H-3S\"}"; + assertEquals(expected, new ObjectMapper().writeValueAsString(objectNode)); + } + + @Test + public void testTimeTz() throws JsonProcessingException { + ObjectNode objectNode = new ObjectNode(new JsonNodeFactory(true)); + + /* Basic Test: 23:59:59 + 10 hours offset */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "basic", generateTimeTzSchema(), generateTimeTzRecord(86399000000L, 36000000), objectNode); + + /* Negative offset: 12:30:00 - 5 hours offset */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "neg_offset", + generateTimeTzSchema(), + generateTimeTzRecord(45000000000L, -18000000), + objectNode); + + /* Zero offset (UTC): 08:00:00Z */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "utc", generateTimeTzSchema(), generateTimeTzRecord(28800000000L, 0), objectNode); + + /* 24:00:00 special case */ + UnifiedTypesFormatter.handleDatastreamRecordType( + "max_time", + generateTimeTzSchema(), + generateTimeTzRecord(86400000000L, 36000000), + objectNode); + + String expected = + "{\"basic\":\"23:59:59+10:00\"," + + "\"neg_offset\":\"12:30:00-05:00\"," + + "\"utc\":\"08:00:00Z\"," + + "\"max_time\":\"24:00:00+10:00\"}"; + assertEquals(expected, new ObjectMapper().writeValueAsString(objectNode)); + } + @Test public void testGetPrimaryKeys_primaryKeysField() throws IOException { Schema arraySchema = Schema.createArray(Schema.create(Schema.Type.STRING)); @@ -570,4 +640,48 @@ private GenericRecord buildOuterRecord(GenericRecord sourceMetadata, String read record.put("payload", payload); return record; } + + private GenericRecord generateIntervalRecord(Integer months, Integer hours, Long micros) { + GenericRecord genericRecord = new GenericData.Record(generateIntervalSchema()); + genericRecord.put("months", months); + genericRecord.put("hours", hours); + genericRecord.put("micros", micros); + return genericRecord; + } + + private Schema generateIntervalSchema() { + return SchemaBuilder.builder() + .record("interval") + .fields() + .name("months") + .type(SchemaBuilder.builder().intType()) + .withDefault(0) + .name("hours") + .type(SchemaBuilder.builder().intType()) + .withDefault(0) + .name("micros") + .type(SchemaBuilder.builder().longType()) + .withDefault(0L) + .endRecord(); + } + + private GenericRecord generateTimeTzRecord(Long time, Integer offset) { + GenericRecord genericRecord = new GenericData.Record(generateTimeTzSchema()); + genericRecord.put("time", time); + genericRecord.put("offset", offset); + return genericRecord; + } + + private Schema generateTimeTzSchema() { + return SchemaBuilder.builder() + .record("timeTz") + .fields() + .name("time") + .type(SchemaBuilder.builder().longType()) + .withDefault(0L) + .name("offset") + .type(SchemaBuilder.builder().intType()) + .withDefault(0) + .endRecord(); + } } diff --git a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PostgreSQLDatastreamToSpannerDataTypesIT.java b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PostgreSQLDatastreamToSpannerDataTypesIT.java index 657134d421..cc39760ff0 100644 --- a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PostgreSQLDatastreamToSpannerDataTypesIT.java +++ b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PostgreSQLDatastreamToSpannerDataTypesIT.java @@ -19,6 +19,7 @@ import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; +import com.google.cloud.ByteArray; import com.google.cloud.spanner.Struct; import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; import com.google.cloud.teleport.metadata.TemplateIntegrationTest; @@ -84,7 +85,6 @@ public class PostgreSQLDatastreamToSpannerDataTypesIT extends DataStreamToSpanne "t_circle_to_float64_array", "t_datemultirange", "t_daterange", - "t_enum", "t_float_array_to_float64_array", "t_float_array_to_string", "t_int_array_to_int64_array", @@ -93,7 +93,6 @@ public class PostgreSQLDatastreamToSpannerDataTypesIT extends DataStreamToSpanne "t_int4range", "t_int8multirange", "t_int8range", - "t_interval", "t_interval_to_int64", "t_line_to_float64_array", "t_lseg_to_float64_array", @@ -110,17 +109,10 @@ public class PostgreSQLDatastreamToSpannerDataTypesIT extends DataStreamToSpanne "t_real_array_to_string", "t_smallint_array_to_int64_array", "t_smallint_array_to_string", - "t_time", - "t_time_with_time_zone", - "t_time_without_time_zone", - "t_timetz", "t_tsmultirange", - "t_tsquery", "t_tsrange", "t_tstzmultirange", "t_tstzrange", - "t_tsvector", - "t_txid_snapshot", "t_varbit_to_bool_array"); private static CloudPostgresResourceManager.ReplicationInfo replicationInfo; private static CloudPostgresResourceManager.ReplicationInfo pgDialectReplicationInfo; @@ -297,18 +289,11 @@ private void validateResult( // These types are not mapped as expected, ignore them to avoid failing the test. Set ignoredTypeMappings = Set.of( - "bit", - "bit_to_string", - "bit_varying", - "bit_varying_to_string", - "bytea", - "json", - "json_to_string", - "macaddr", - "macaddr8", + "time", + "time_with_time_zone", + "time_without_time_zone", + "timetz", "uuid_to_bytes", - "varbit", - "varbit_to_string", "t_bigint_array_to_int64_array", "t_bigint_array_to_string", "t_bit_to_bool_array", @@ -322,6 +307,7 @@ private void validateResult( "t_float_array_to_string", "t_int_array_to_int64_array", "t_int_array_to_string", + "t_interval_to_int64", "t_line_to_float64_array", "t_lseg_to_float64_array", "t_money_to_int64", @@ -411,6 +397,7 @@ private ConditionCheck buildConditionCheck( // for the timeout. Set ignoredTables = Set.of( + "uuid_to_bytes", "t_bigint_array_to_int64_array", "t_bigint_array_to_string", "t_bit_to_bool_array", @@ -473,26 +460,32 @@ private Map>> getExpectedData() { result.put( "bigint_to_string", createRows("-9223372036854775808", "9223372036854775807", "42", "NULL")); + result.put( + "bigint_to_numeric", + createRows("-9223372036854775808", "9223372036854775807", "42", "NULL")); result.put("bigserial", createRows("-9223372036854775808", "9223372036854775807", "42")); result.put( "bigserial_to_string", createRows("-9223372036854775808", "9223372036854775807", "42")); - result.put("bit", createRows("AA==", "gA==", "NULL")); - result.put("bit_to_string", createRows("AA==", "gA==", "NULL")); - result.put("bit_varying", createRows("UA==", "NULL")); - result.put("bit_varying_to_string", createRows("UA==", "NULL")); + result.put( + "bigserial_to_numeric", createRows("-9223372036854775808", "9223372036854775807", "42")); + result.put( + "bit", + createRows(ByteArray.copyFrom("0").toBase64(), ByteArray.copyFrom("1").toBase64(), "NULL")); + result.put("bit_varying", createRows(ByteArray.copyFrom("0101").toBase64(), "NULL")); result.put("bool", createRows("false", "true", "NULL")); result.put("bool_to_string", createRows("false", "true", "NULL")); result.put("boolean", createRows("false", "true", "NULL")); result.put("boolean_to_string", createRows("false", "true", "NULL")); - result.put("bytea", createRows("YWJj", "NULL")); - result.put("bytea_to_string", createRows("YWJj", "NULL")); + result.put("bytea", createRows(ByteArray.copyFrom("abc").toBase64(), "NULL")); result.put("char", createRows("a", "Θ", "NULL")); + result.put("char_n", createRows("a ", "test ", "NULL")); result.put("character", createRows("a", "Ξ", "NULL")); + result.put("character_n", createRows("a ", "test ", "NULL")); result.put("character_varying", createRows("testing character varying", "NULL")); + result.put("character_varying_n", createRows("testing", "NULL")); result.put("cidr", createRows("192.168.100.128/25", "NULL")); result.put("date", createRows("0001-01-01", "9999-12-31", "NULL")); result.put("date_to_string", createRows("0001-01-01", "9999-12-31", "NULL")); - result.put("decimal", createRows("0.12", "NULL")); result.put("decimal_to_string", createRows("0.12", "NULL")); result.put( "double_precision", @@ -502,6 +495,7 @@ private Map>> getExpectedData() { "double_precision_to_string", createRows( "-1.9876542E+307", "1.9876542E+307", "NaN", "-Infinity", "Infinity", "1.23", "NULL")); + result.put("enum", createRows("enum1", "NULL")); result.put( "float_to_float64", createRows( @@ -533,26 +527,35 @@ private Map>> getExpectedData() { result.put("inet", createRows("192.168.1.0/24", "NULL")); result.put("int", createRows("-2147483648", "2147483647", "1", "NULL")); result.put("int_to_string", createRows("-2147483648", "2147483647", "1", "NULL")); + result.put("int_to_numeric", createRows("-2147483648", "2147483647", "1", "NULL")); + result.put("int_to_float64", createRows("-2.147483648E9", "2.147483647E9", "1.0", "NULL")); result.put("integer", createRows("-2147483648", "2147483647", "2", "NULL")); result.put("integer_to_string", createRows("-2147483648", "2147483647", "2", "NULL")); + result.put("integer_to_numeric", createRows("-2147483648", "2147483647", "2", "NULL")); + result.put("integer_to_float64", createRows("-2.147483648E9", "2.147483647E9", "2.0", "NULL")); result.put("int2", createRows("-32768", "32767", "3", "NULL")); result.put("int2_to_string", createRows("-32768", "32767", "3", "NULL")); + result.put("int2_to_numeric", createRows("-32768", "32767", "3", "NULL")); + result.put("int2_to_float32", createRows("-32768.0", "32767.0", "3.0", "NULL")); + result.put("int2_to_float64", createRows("-32768.0", "32767.0", "3.0", "NULL")); result.put("int4", createRows("-2147483648", "2147483647", "4", "NULL")); result.put("int4_to_string", createRows("-2147483648", "2147483647", "4", "NULL")); + result.put("int4_to_numeric", createRows("-2147483648", "2147483647", "4", "NULL")); + result.put("int4_to_float64", createRows("-2.147483648E9", "2.147483647E9", "4.0", "NULL")); result.put("int8", createRows("-9223372036854775808", "9223372036854775807", "5", "NULL")); result.put( "int8_to_string", createRows("-9223372036854775808", "9223372036854775807", "5", "NULL")); - result.put("json", createRows("{\"duplicate_key\":2}", "{\"null_key\":null}", "NULL")); result.put( - "json_to_string", createRows("{\"duplicate_key\": 2}", "{\"null_key\": null}", "NULL")); + "int8_to_numeric", createRows("-9223372036854775808", "9223372036854775807", "5", "NULL")); + result.put( + "interval", createRows("P1Y2M3DT4H5M6.789S", "PT0S", "P3M-2DT-2H-16M-13.210988S", "NULL")); + result.put("json", createRows("{\"duplicate_key\":1}", "{\"null_key\":null}", "NULL")); + result.put( + "json_to_string", + createRows("{\"duplicate_key\": 1, \"duplicate_k...", "{\"null_key\": null}", "NULL")); result.put("jsonb", createRows("{\"duplicate_key\":2}", "{\"null_key\":null}", "NULL")); result.put( "jsonb_to_string", createRows("{\"duplicate_key\": 2}", "{\"null_key\": null}", "NULL")); - result.put( - "large_decimal_to_numeric", - createRows( - // Decimals with scale larger than supported in Spanner are rounded - "0.12", "100000000000000000000000", "12345678901234567890.123456789", "NULL")); result.put( "large_decimal_to_string", createRows( @@ -560,11 +563,6 @@ private Map>> getExpectedData() { "99999999999999999999999.9999999999", "123456789012345678901234567890.12...", "NULL")); - result.put( - "large_numeric_to_numeric", - createRows( - // Decimals with scale larger than supported in Spanner are rounded - "0.12", "100000000000000000000000", "12345678901234567890.123456789", "NULL")); result.put( "large_numeric_to_string", createRows( @@ -575,9 +573,12 @@ private Map>> getExpectedData() { result.put("macaddr", createRows("08:00:2b:01:02:03", "NULL")); result.put("macaddr8", createRows("08:00:2b:01:02:03:04:05", "NULL")); result.put("money", createRows("123.45", "NULL")); - result.put("numeric", createRows("4.56", "NULL")); + result.put("money_to_numeric", createRows("123.45", "NULL")); result.put("numeric_to_string", createRows("4.56", "NULL")); result.put("oid", createRows("1000", "NULL")); + result.put("oid_to_string", createRows("1000", "NULL")); + result.put("oid_to_numeric", createRows("1000", "NULL")); + result.put("oid_to_float64", createRows("1000.0", "NULL")); result.put( "real", createRows( @@ -592,17 +593,40 @@ private Map>> getExpectedData() { "-1.9876542E+38", "1.9876542E+38", "NaN", "-Infinity", "Infinity", "5.67", "NULL")); result.put("serial", createRows("-2147483648", "2147483647", "6")); result.put("serial_to_string", createRows("-2147483648", "2147483647", "6")); + result.put("serial_to_numeric", createRows("-2147483648", "2147483647", "6")); + result.put("serial_to_float64", createRows("-2.147483648E9", "2.147483647E9", "6.0")); result.put("serial2", createRows("-32768", "32767", "7")); result.put("serial2_to_string", createRows("-32768", "32767", "7")); + result.put("serial2_to_numeric", createRows("-32768", "32767", "7")); + result.put("serial2_to_float32", createRows("-32768.0", "32767.0", "7.0")); + result.put("serial2_to_float64", createRows("-32768.0", "32767.0", "7.0")); result.put("serial4", createRows("-2147483648", "2147483647", "8")); result.put("serial4_to_string", createRows("-2147483648", "2147483647", "8")); + result.put("serial4_to_numeric", createRows("-2147483648", "2147483647", "8")); + result.put("serial4_to_float64", createRows("-2.147483648E9", "2.147483647E9", "8.0")); result.put("serial8", createRows("-9223372036854775808", "9223372036854775807", "9")); result.put("serial8_to_string", createRows("-9223372036854775808", "9223372036854775807", "9")); + result.put( + "serial8_to_numeric", createRows("-9223372036854775808", "9223372036854775807", "9")); result.put("smallint", createRows("-32768", "32767", "10", "NULL")); result.put("smallint_to_string", createRows("-32768", "32767", "10", "NULL")); + result.put("smallint_to_numeric", createRows("-32768", "32767", "10", "NULL")); + result.put("smallint_to_float32", createRows("-32768.0", "32767.0", "10.0", "NULL")); + result.put("smallint_to_float64", createRows("-32768.0", "32767.0", "10.0", "NULL")); result.put("smallserial", createRows("-32768", "32767", "11")); result.put("smallserial_to_string", createRows("-32768", "32767", "11")); + result.put("smallserial_to_numeric", createRows("-32768", "32767", "11")); + result.put("smallserial_to_float32", createRows("-32768.0", "32767.0", "11.0")); + result.put("smallserial_to_float64", createRows("-32768.0", "32767.0", "11.0")); result.put("text", createRows("testing text", "NULL")); + // Datastream incorrectly wraps 24:00:00 to 0 microseconds during extraction. + // This causes 24:00:00 to be silently rewritten to Spanner as 'PT0S' instead of 'PT24H', + // and '24:00:00+10:00' as '00:00:00+10:00' instead of '24:00:00+10:00'. + // Ignored in ignoredTypeMappings until the Datastream bug is resolved. + result.put("time", createRows("PT24H", "NULL")); + result.put("time_without_time_zone", createRows("PT24H", "NULL")); + result.put("time_with_time_zone", createRows("23:59:59+10:00", "24:00:00+10:00", "NULL")); + result.put("timetz", createRows("23:59:59+10:00", "24:00:00+10:00", "NULL")); result.put("timestamp", createRows("1970-01-02T03:04:05.123456Z", "NULL")); result.put("timestamp_to_timestamp", createRows("1970-01-02T03:04:05.123456000Z", "NULL")); result.put( @@ -618,15 +642,22 @@ private Map>> getExpectedData() { "timestamp_with_timezone_to_string", createRows("1970-02-02T18:05:06.123456Z", "1970-02-03T05:05:06.123456Z", "NULL")); result.put("timestamp_without_time_zone", createRows("1970-01-02T03:04:05.123456Z", "NULL")); + result.put( + "timestamp_without_time_zone_to_timestamp", + createRows("1970-01-02T03:04:05.123456000Z", "NULL")); + result.put("tsquery", createRows("'fat' & 'rat'", "NULL")); + result.put("tsvector", createRows("'a' 'cat' 'fat' 'mat' 'on' 'sat'", "NULL")); + result.put("txid_snapshot", createRows("10:20:10,14,15", "NULL")); result.put("uuid", createRows("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11", "NULL")); result.put( "uuid_pk", createUuidPkRows( "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11", "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a12")); result.put("uuid_to_bytes", createRows("oO68mZwLTvi7bWu5vTgKEQ==", "NULL")); - result.put("varbit", createRows("wA==", "NULL")); - result.put("varbit_to_string", createRows("wA==", "NULL")); + result.put("uuid_to_string", createRows("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11", "NULL")); + result.put("varbit", createRows(ByteArray.copyFrom("1100").toBase64(), "NULL")); result.put("varchar", createRows("testing varchar", "NULL")); + result.put("varchar_n", createRows("testing", "NULL")); result.put("xml", createRows("123", "NULL")); return result; } @@ -636,26 +667,58 @@ private Map>> getExpectedDataPGDialect() { // differences. Notably, some data types like numeric have slightly different behaviour. Map>> expectedData = getExpectedData(); - expectedData.put("decimal", createRows("0.120000000", "NULL")); expectedData.put("json", createRows("{\"duplicate_key\": 2}", "{\"null_key\": null}", "NULL")); expectedData.put("jsonb", createRows("{\"duplicate_key\": 2}", "{\"null_key\": null}", "NULL")); expectedData.put( - "large_decimal_to_numeric", + "bigint_to_numeric", createRows( - // Decimals with scale larger than supported in Spanner are rounded - "0.120000000", - "100000000000000000000000.000000000", - "12345678901234567890.123456789", + "-9223372036854775808.000000000", + "9223372036854775807.000000000", + "42.000000000", "NULL")); expectedData.put( - "large_numeric_to_numeric", + "bigserial_to_numeric", createRows( - // Decimals with scale larger than supported in Spanner are rounded - "0.120000000", - "100000000000000000000000.000000000", - "12345678901234567890.123456789", + "-9223372036854775808.000000000", "9223372036854775807.000000000", "42.000000000")); + expectedData.put( + "int2_to_numeric", + createRows("-32768.000000000", "32767.000000000", "3.000000000", "NULL")); + expectedData.put( + "int4_to_numeric", + createRows("-2147483648.000000000", "2147483647.000000000", "4.000000000", "NULL")); + expectedData.put( + "int8_to_numeric", + createRows( + "-9223372036854775808.000000000", + "9223372036854775807.000000000", + "5.000000000", "NULL")); - expectedData.put("numeric", createRows("4.560000000", "NULL")); + expectedData.put( + "int_to_numeric", + createRows("-2147483648.000000000", "2147483647.000000000", "1.000000000", "NULL")); + expectedData.put( + "integer_to_numeric", + createRows("-2147483648.000000000", "2147483647.000000000", "2.000000000", "NULL")); + expectedData.put("money_to_numeric", createRows("123.450000000", "NULL")); + expectedData.put("oid_to_numeric", createRows("1000.000000000", "NULL")); + expectedData.put( + "serial2_to_numeric", createRows("-32768.000000000", "32767.000000000", "7.000000000")); + expectedData.put( + "serial4_to_numeric", + createRows("-2147483648.000000000", "2147483647.000000000", "8.000000000")); + expectedData.put( + "serial8_to_numeric", + createRows( + "-9223372036854775808.000000000", "9223372036854775807.000000000", "9.000000000")); + expectedData.put( + "serial_to_numeric", + createRows("-2147483648.000000000", "2147483647.000000000", "6.000000000")); + expectedData.put( + "smallint_to_numeric", + createRows("-32768.000000000", "32767.000000000", "10.000000000", "NULL")); + expectedData.put( + "smallserial_to_numeric", + createRows("-32768.000000000", "32767.000000000", "11.000000000")); return expectedData; } diff --git a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/source/postgresql/PostgresChangeEventContextTest.java b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/source/postgresql/PostgresChangeEventContextTest.java index e18e731c2d..686b0ab070 100644 --- a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/source/postgresql/PostgresChangeEventContextTest.java +++ b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/source/postgresql/PostgresChangeEventContextTest.java @@ -31,6 +31,7 @@ import com.google.cloud.teleport.v2.templates.datastream.ChangeEventConvertorTest; import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants; import java.io.IOException; +import java.util.Base64; import java.util.Map; import org.json.JSONObject; import org.junit.Test; @@ -61,6 +62,8 @@ public void canGenerateShadowTableMutation() throws Exception { changeEvent.put(PostgresqlDsToSpSourceConnector.POSTGRES_LSN_KEY, "1/867"); changeEvent.put( DatastreamConstants.EVENT_SOURCE_TYPE_KEY, SourceConstants.POSTGRES_SOURCE_TYPE); + changeEvent.put( + "bytes_field", Base64.getEncoder().encodeToString(new byte[] {120, 53, 56, 48, 48})); ChangeEventContext changeEventContext = new PostgresqlDsToSpSourceConnector() @@ -97,6 +100,8 @@ public void canGenerateShadowTableMutationForBackfillEvent() throws Exception { changeEvent.put(PostgresqlDsToSpSourceConnector.POSTGRES_LSN_KEY, JSONObject.NULL); changeEvent.put( DatastreamConstants.EVENT_SOURCE_TYPE_KEY, SourceConstants.POSTGRES_SOURCE_TYPE); + changeEvent.put( + "bytes_field", Base64.getEncoder().encodeToString(new byte[] {120, 53, 56, 48, 48})); ChangeEventContext changeEventContext = new PostgresqlDsToSpSourceConnector() @@ -132,6 +137,8 @@ public void canGenerateShadowTableMutationForBackfillEventWithMissingKeys() thro changeEvent.put(PostgresqlDsToSpSourceConnector.POSTGRES_TIMESTAMP_KEY, eventTimestamp); changeEvent.put( DatastreamConstants.EVENT_SOURCE_TYPE_KEY, SourceConstants.POSTGRES_SOURCE_TYPE); + changeEvent.put( + "bytes_field", Base64.getEncoder().encodeToString(new byte[] {120, 53, 56, 48, 48})); ChangeEventContext changeEventContext = new PostgresqlDsToSpSourceConnector() diff --git a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/pg-dialect-spanner-schema.sql b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/pg-dialect-spanner-schema.sql index 3d8e8d81a8..94a8a37680 100644 --- a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/pg-dialect-spanner-schema.sql +++ b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/pg-dialect-spanner-schema.sql @@ -1,15 +1,15 @@ CREATE TABLE IF NOT EXISTS t_bigint (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bigint_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_bigint_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bigint_array_to_int64_array (id INT8, col INT8[], PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bigint_array_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bigserial (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bigserial_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_bigserial_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bit (id INT8, col BYTEA, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bit_to_bool_array (id INT8, col BOOL[], PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_bit_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bit_varying (id INT8, col BYTEA, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bit_varying_to_bool_array (id INT8, col BOOL[], PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_bit_varying_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bool (id INT8, col BOOL, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bool_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bool_array_to_bool_array (id INT8, col BOOL[], PRIMARY KEY (id)); @@ -19,10 +19,12 @@ CREATE TABLE IF NOT EXISTS t_boolean_to_string (id INT8, col VARCHAR, PRIMARY KE CREATE TABLE IF NOT EXISTS t_box (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_box_to_float64_array (id INT8, col FLOAT8[], PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_bytea (id INT8, col BYTEA, PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_bytea_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_char (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_char_n (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_character (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_character_n (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_character_varying (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_character_varying_n (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_cidr (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_circle (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_circle_to_float64_array (id INT8, col FLOAT8[], PRIMARY KEY (id)); @@ -30,7 +32,6 @@ CREATE TABLE IF NOT EXISTS t_date (id INT8, col DATE, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_date_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_datemultirange (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_daterange (id INT8, col VARCHAR, PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_decimal (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_decimal_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_double_precision (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_double_precision_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); @@ -47,29 +48,37 @@ CREATE TABLE IF NOT EXISTS t_float8_to_string (id INT8, col VARCHAR, PRIMARY KEY CREATE TABLE IF NOT EXISTS t_inet (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int_array_to_int64_array (id INT8, col INT8[], PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int_array_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int2 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int2_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int2_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int2_to_float32 (id INT8, col FLOAT4, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int2_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int4 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int4_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int4_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int4_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int4multirange (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int4range (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int8 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int8_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_int8_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int8multirange (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_int8range (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_integer (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_integer_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_integer_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_integer_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_interval (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_interval_to_int64 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_json (id INT8, col JSONB, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_json_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_jsonb (id INT8, col JSONB, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_jsonb_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_large_decimal_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_large_decimal_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_large_numeric_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_large_numeric_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_line (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_line_to_float64_array (id INT8, col FLOAT8[], PRIMARY KEY (id)); @@ -79,11 +88,14 @@ CREATE TABLE IF NOT EXISTS t_macaddr (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_macaddr8 (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_money (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_money_to_int64 (id INT8, col INT8, PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_money_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_numeric_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_nummultirange (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_numrange (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_oid (id INT8, col INT8, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_oid_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_oid_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_oid_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_path (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_path_to_float64_array (id INT8, col FLOAT8[], PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_pg_lsn (id INT8, col VARCHAR, PRIMARY KEY (id)); @@ -99,18 +111,32 @@ CREATE TABLE IF NOT EXISTS t_real_array_to_float32_array (id INT8, col FLOAT4[], CREATE TABLE IF NOT EXISTS t_real_array_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial2 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial2_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial2_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial2_to_float32 (id INT8, col FLOAT4, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial2_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial4 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial4_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial4_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial4_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial8 (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_serial8_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_serial8_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallint (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallint_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallint_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallint_to_float32 (id INT8, col FLOAT4, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallint_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallint_array_to_int64_array (id INT8, col INT8[], PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallint_array_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallserial (id INT8, col INT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_smallserial_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallserial_to_numeric (id INT8, col NUMERIC, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallserial_to_float32 (id INT8, col FLOAT4, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_smallserial_to_float64 (id INT8, col FLOAT8, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_text (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_time (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_time_with_time_zone (id INT8, col VARCHAR, PRIMARY KEY (id)); @@ -120,6 +146,7 @@ CREATE TABLE IF NOT EXISTS t_timestamp_to_timestamp (id INT8, col TIMESTAMPTZ, P CREATE TABLE IF NOT EXISTS t_timestamp_with_time_zone (id INT8, col TIMESTAMPTZ, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_timestamp_with_timezone_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_timestamp_without_time_zone (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_timestamp_without_time_zone_to_timestamp (id INT8, col TIMESTAMPTZ, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_timestamptz (id INT8, col TIMESTAMPTZ, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_timestamptz_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_timetz (id INT8, col VARCHAR, PRIMARY KEY (id)); @@ -132,9 +159,10 @@ CREATE TABLE IF NOT EXISTS t_tsvector (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_txid_snapshot (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_uuid (id INT8, col UUID, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_uuid_to_bytes (id INT8, col BYTEA, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_uuid_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_varbit (id INT8, col BYTEA, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_varbit_to_bool_array (id INT8, col BOOL[], PRIMARY KEY (id)); -CREATE TABLE IF NOT EXISTS t_varbit_to_string (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_varchar (id INT8, col VARCHAR, PRIMARY KEY (id)); +CREATE TABLE IF NOT EXISTS t_varchar_n (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_xml (id INT8, col VARCHAR, PRIMARY KEY (id)); CREATE TABLE IF NOT EXISTS t_uuid_pk (id UUID NOT NULL, col UUID, PRIMARY KEY (id)); diff --git a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/postgresql-data-types.sql b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/postgresql-data-types.sql index f62281dd4e..b2c24e9330 100644 --- a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/postgresql-data-types.sql +++ b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/postgresql-data-types.sql @@ -2,16 +2,16 @@ CREATE TYPE myenum AS ENUM ('enum1', 'enum2', 'enum3'); CREATE TABLE t_bigint (id serial primary key, col bigint); CREATE TABLE t_bigint_to_string (id serial primary key, col bigint); +CREATE TABLE t_bigint_to_numeric (id serial primary key, col bigint); CREATE TABLE t_bigint_array_to_int64_array (id serial primary key, col bigint[]); CREATE TABLE t_bigint_array_to_string (id serial primary key, col bigint[]); CREATE TABLE t_bigserial (id serial primary key, col bigserial); CREATE TABLE t_bigserial_to_string (id serial primary key, col bigserial); +CREATE TABLE t_bigserial_to_numeric (id serial primary key, col bigserial); CREATE TABLE t_bit (id serial primary key, col bit); -CREATE TABLE t_bit_to_string (id serial primary key, col bit(32)); CREATE TABLE t_bit_to_bool_array (id serial primary key, col bit(32)); CREATE TABLE t_bit_varying (id serial primary key, col bit varying); CREATE TABLE t_bit_varying_to_bool_array (id serial primary key, col bit varying(32)); -CREATE TABLE t_bit_varying_to_string (id serial primary key, col bit varying(32)); CREATE TABLE t_bool (id serial primary key, col bool); CREATE TABLE t_bool_to_string (id serial primary key, col bool); CREATE TABLE t_bool_array_to_bool_array (id serial primary key, col bool[]); @@ -21,10 +21,12 @@ CREATE TABLE t_boolean_to_string (id serial primary key, col boolean); CREATE TABLE t_box (id serial primary key, col box); CREATE TABLE t_box_to_float64_array (id serial primary key, col box); CREATE TABLE t_bytea (id serial primary key, col bytea); -CREATE TABLE t_bytea_to_string (id serial primary key, col bytea); CREATE TABLE t_char (id serial primary key, col char); +CREATE TABLE t_char_n (id serial primary key, col char(10)); CREATE TABLE t_character (id serial primary key, col character); +CREATE TABLE t_character_n (id serial primary key, col character(10)); CREATE TABLE t_character_varying (id serial primary key, col character varying); +CREATE TABLE t_character_varying_n (id serial primary key, col character varying(10)); CREATE TABLE t_cidr (id serial primary key, col cidr); CREATE TABLE t_circle (id serial primary key, col circle); CREATE TABLE t_circle_to_float64_array (id serial primary key, col circle); @@ -32,7 +34,6 @@ CREATE TABLE t_date (id serial primary key, col date); CREATE TABLE t_date_to_string (id serial primary key, col date); CREATE TABLE t_datemultirange (id serial primary key, col datemultirange); CREATE TABLE t_daterange (id serial primary key, col daterange); -CREATE TABLE t_decimal (id serial primary key, col numeric(10,2)); CREATE TABLE t_decimal_to_string (id serial primary key, col decimal(10,2)); CREATE TABLE t_double_precision (id serial primary key, col double precision); CREATE TABLE t_double_precision_to_string (id serial primary key, col double precision); @@ -49,29 +50,37 @@ CREATE TABLE t_float8_to_string (id serial primary key, col float8); CREATE TABLE t_inet (id serial primary key, col inet); CREATE TABLE t_int (id serial primary key, col int); CREATE TABLE t_int_to_string (id serial primary key, col int); +CREATE TABLE t_int_to_numeric (id serial primary key, col int); +CREATE TABLE t_int_to_float64 (id serial primary key, col int); CREATE TABLE t_int_array_to_int64_array (id serial primary key, col int[]); CREATE TABLE t_int_array_to_string (id serial primary key, col int[]); CREATE TABLE t_int2 (id serial primary key, col int2); CREATE TABLE t_int2_to_string (id serial primary key, col int2); +CREATE TABLE t_int2_to_numeric (id serial primary key, col int2); +CREATE TABLE t_int2_to_float32 (id serial primary key, col int2); +CREATE TABLE t_int2_to_float64 (id serial primary key, col int2); CREATE TABLE t_int4 (id serial primary key, col int4); CREATE TABLE t_int4_to_string (id serial primary key, col int4); +CREATE TABLE t_int4_to_numeric (id serial primary key, col int4); +CREATE TABLE t_int4_to_float64 (id serial primary key, col int4); CREATE TABLE t_int4multirange (id serial primary key, col int4multirange); CREATE TABLE t_int4range (id serial primary key, col int4range); CREATE TABLE t_int8 (id serial primary key, col int8); CREATE TABLE t_int8_to_string (id serial primary key, col int8); +CREATE TABLE t_int8_to_numeric (id serial primary key, col int8); CREATE TABLE t_int8multirange (id serial primary key, col int8multirange); CREATE TABLE t_int8range (id serial primary key, col int8range); CREATE TABLE t_integer (id serial primary key, col integer); CREATE TABLE t_integer_to_string (id serial primary key, col integer); +CREATE TABLE t_integer_to_numeric (id serial primary key, col integer); +CREATE TABLE t_integer_to_float64 (id serial primary key, col integer); CREATE TABLE t_interval (id serial primary key, col interval); CREATE TABLE t_interval_to_int64 (id serial primary key, col interval); CREATE TABLE t_json (id serial primary key, col json); CREATE TABLE t_json_to_string (id serial primary key, col json); CREATE TABLE t_jsonb (id serial primary key, col jsonb); CREATE TABLE t_jsonb_to_string (id serial primary key, col jsonb); -CREATE TABLE t_large_decimal_to_numeric (id serial primary key, col decimal(40,10)); CREATE TABLE t_large_decimal_to_string (id serial primary key, col decimal(40,10)); -CREATE TABLE t_large_numeric_to_numeric (id serial primary key, col numeric(40,10)); CREATE TABLE t_large_numeric_to_string (id serial primary key, col numeric(40,10)); CREATE TABLE t_line (id serial primary key, col line); CREATE TABLE t_line_to_float64_array (id serial primary key, col line); @@ -81,11 +90,14 @@ CREATE TABLE t_macaddr (id serial primary key, col macaddr); CREATE TABLE t_macaddr8 (id serial primary key, col macaddr8); CREATE TABLE t_money (id serial primary key, col money); CREATE TABLE t_money_to_int64 (id serial primary key, col money); -CREATE TABLE t_numeric (id serial primary key, col numeric(10,2)); +CREATE TABLE t_money_to_numeric (id serial primary key, col money); CREATE TABLE t_numeric_to_string (id serial primary key, col numeric(10,2)); CREATE TABLE t_nummultirange (id serial primary key, col nummultirange); CREATE TABLE t_numrange (id serial primary key, col numrange); CREATE TABLE t_oid (id serial primary key, col oid); +CREATE TABLE t_oid_to_string (id serial primary key, col oid); +CREATE TABLE t_oid_to_numeric (id serial primary key, col oid); +CREATE TABLE t_oid_to_float64 (id serial primary key, col oid); CREATE TABLE t_path (id serial primary key, col path); CREATE TABLE t_path_to_float64_array (id serial primary key, col path); CREATE TABLE t_pg_lsn (id serial primary key, col pg_lsn); @@ -101,18 +113,32 @@ CREATE TABLE t_real_array_to_float32_array (id serial primary key, col real[]); CREATE TABLE t_real_array_to_string (id serial primary key, col real[]); CREATE TABLE t_serial (id serial primary key, col serial); CREATE TABLE t_serial_to_string (id serial primary key, col serial); +CREATE TABLE t_serial_to_numeric (id serial primary key, col serial); +CREATE TABLE t_serial_to_float64 (id serial primary key, col serial); CREATE TABLE t_serial2 (id serial primary key, col serial2); CREATE TABLE t_serial2_to_string (id serial primary key, col serial2); +CREATE TABLE t_serial2_to_numeric (id serial primary key, col serial2); +CREATE TABLE t_serial2_to_float32 (id serial primary key, col serial2); +CREATE TABLE t_serial2_to_float64 (id serial primary key, col serial2); CREATE TABLE t_serial4 (id serial primary key, col serial4); CREATE TABLE t_serial4_to_string (id serial primary key, col serial4); +CREATE TABLE t_serial4_to_numeric (id serial primary key, col serial4); +CREATE TABLE t_serial4_to_float64 (id serial primary key, col serial4); CREATE TABLE t_serial8 (id serial primary key, col serial8); CREATE TABLE t_serial8_to_string (id serial primary key, col serial8); +CREATE TABLE t_serial8_to_numeric (id serial primary key, col serial8); CREATE TABLE t_smallint (id serial primary key, col smallint); CREATE TABLE t_smallint_to_string (id serial primary key, col smallint); +CREATE TABLE t_smallint_to_numeric (id serial primary key, col smallint); +CREATE TABLE t_smallint_to_float32 (id serial primary key, col smallint); +CREATE TABLE t_smallint_to_float64 (id serial primary key, col smallint); CREATE TABLE t_smallint_array_to_int64_array (id serial primary key, col smallint[]); CREATE TABLE t_smallint_array_to_string (id serial primary key, col smallint[]); CREATE TABLE t_smallserial (id serial primary key, col smallserial); CREATE TABLE t_smallserial_to_string (id serial primary key, col smallserial); +CREATE TABLE t_smallserial_to_numeric (id serial primary key, col smallserial); +CREATE TABLE t_smallserial_to_float32 (id serial primary key, col smallserial); +CREATE TABLE t_smallserial_to_float64 (id serial primary key, col smallserial); CREATE TABLE t_text (id serial primary key, col text); CREATE TABLE t_time (id serial primary key, col time); CREATE TABLE t_time_with_time_zone (id serial primary key, col time with time zone); @@ -122,6 +148,7 @@ CREATE TABLE t_timestamp_to_timestamp (id serial primary key, col timestamp); CREATE TABLE t_timestamp_with_time_zone (id serial primary key, col timestamp with time zone); CREATE TABLE t_timestamp_with_timezone_to_string (id serial primary key, col timestamp with time zone); CREATE TABLE t_timestamp_without_time_zone (id serial primary key, col timestamp without time zone); +CREATE TABLE t_timestamp_without_time_zone_to_timestamp (id serial primary key, col timestamp without time zone); CREATE TABLE t_timestamptz (id serial primary key, col timestamptz); CREATE TABLE t_timestamptz_to_string (id serial primary key, col timestamptz); CREATE TABLE t_timetz (id serial primary key, col timetz); @@ -134,23 +161,24 @@ CREATE TABLE t_tsvector (id serial primary key, col tsvector); CREATE TABLE t_txid_snapshot (id serial primary key, col txid_snapshot); CREATE TABLE t_uuid (id serial primary key, col uuid); CREATE TABLE t_uuid_to_bytes (id serial primary key, col uuid); +CREATE TABLE t_uuid_to_string (id serial primary key, col uuid); CREATE TABLE t_varbit (id serial primary key, col varbit); -CREATE TABLE t_varbit_to_string (id serial primary key, col varbit(32)); CREATE TABLE t_varbit_to_bool_array (id serial primary key, col varbit(32)); CREATE TABLE t_varchar (id serial primary key, col varchar); +CREATE TABLE t_varchar_n (id serial primary key, col varchar(10)); CREATE TABLE t_xml (id serial primary key, col xml); INSERT INTO t_bigint (col) VALUES (-9223372036854775808), (9223372036854775807), (42), (NULL); INSERT INTO t_bigint_to_string (col) VALUES (-9223372036854775808), (9223372036854775807), (42), (NULL); +INSERT INTO t_bigint_to_numeric (col) VALUES (-9223372036854775808), (9223372036854775807), (42), (NULL); INSERT INTO t_bigint_array_to_int64_array (col) VALUES ('{-9223372036854775808, 9223372036854775807}'), (NULL); INSERT INTO t_bigint_array_to_string (col) VALUES ('{-9223372036854775808, 9223372036854775807}'), (NULL); INSERT INTO t_bigserial (col) VALUES (-9223372036854775808), (9223372036854775807), (42); INSERT INTO t_bigserial_to_string (col) VALUES (-9223372036854775808), (9223372036854775807), (42); +INSERT INTO t_bigserial_to_numeric (col) VALUES (-9223372036854775808), (9223372036854775807), (42); INSERT INTO t_bit (col) VALUES (0::bit), (1::bit), (NULL); -INSERT INTO t_bit_to_string (col) VALUES (0::bit(32)), (1::bit(32)), (NULL); INSERT INTO t_bit_to_bool_array (col) VALUES (0::bit(32)), (NULL); INSERT INTO t_bit_varying (col) VALUES ('0101'::bit varying), (NULL); -INSERT INTO t_bit_varying_to_string (col) VALUES ('0101'::bit varying(32)), (NULL); INSERT INTO t_bit_varying_to_bool_array (col) VALUES ('0101'::bit varying(32)), (NULL); INSERT INTO t_bool (col) VALUES (false), (true), (NULL); INSERT INTO t_bool_to_string (col) VALUES (false), (true), (NULL); @@ -161,10 +189,12 @@ INSERT INTO t_boolean_to_string (col) VALUES (false), (true), (NULL); INSERT INTO t_box (col) VALUES ('((1, 2), (3, 4))'), (NULL); INSERT INTO t_box_to_float64_array (col) VALUES ('((1, 2), (3, 4))'), (NULL); INSERT INTO t_bytea (col) VALUES ('abc'::bytea), (NULL); -INSERT INTO t_bytea_to_string (col) VALUES ('abc'::bytea), (NULL); INSERT INTO t_char (col) VALUES ('a'), ('Θ'), (NULL); +INSERT INTO t_char_n (col) VALUES ('a'), ('test'), (NULL); INSERT INTO t_character (col) VALUES ('a'), ('Ξ'), (NULL); +INSERT INTO t_character_n (col) VALUES ('a'), ('test'), (NULL); INSERT INTO t_character_varying (col) VALUES ('testing character varying'), (NULL); +INSERT INTO t_character_varying_n (col) VALUES ('testing'), (NULL); INSERT INTO t_cidr (col) VALUES ('192.168.100.128/25'), (NULL); INSERT INTO t_circle (col) VALUES ('((1, 2), 3)'), (NULL); INSERT INTO t_circle_to_float64_array (col) VALUES ('((1, 2), 3)'), (NULL); @@ -172,7 +202,6 @@ INSERT INTO t_date (col) VALUES ('0001-01-01'::date), ('9999-12-31'::date), (NUL INSERT INTO t_date_to_string (col) VALUES ('0001-01-01'::date), ('9999-12-31'::date), (NULL); INSERT INTO t_datemultirange (col) VALUES ('{[0001-01-01, 9999-12-31]}'), (NULL); INSERT INTO t_daterange (col) VALUES ('[0001-01-01, 9999-12-31]'), (NULL); -INSERT INTO t_decimal (col) VALUES (0.12), (NULL); INSERT INTO t_decimal_to_string (col) VALUES (0.12), (NULL); INSERT INTO t_double_precision (col) VALUES ('-1.9876542e307'), ('1.9876542e307'), ('NaN'), ('-Infinity'), ('Infinity'), (1.23), (NULL); INSERT INTO t_double_precision_to_string (col) VALUES ('-1.9876542e307'), ('1.9876542e307'), ('NaN'), ('-Infinity'), ('Infinity'), (1.23), (NULL); @@ -189,29 +218,37 @@ INSERT INTO t_float8_to_string (col) VALUES ('-1.9876542e307'), ('1.9876542e307' INSERT INTO t_inet (col) VALUES ('192.168.1.0/24'), (NULL); INSERT INTO t_int (col) VALUES (-2147483648), (2147483647), (1), (NULL); INSERT INTO t_int_to_string (col) VALUES (-2147483648), (2147483647), (1), (NULL); +INSERT INTO t_int_to_numeric (col) VALUES (-2147483648), (2147483647), (1), (NULL); +INSERT INTO t_int_to_float64 (col) VALUES (-2147483648), (2147483647), (1), (NULL); INSERT INTO t_int_array_to_int64_array (col) VALUES ('{-2147483648, 2147483647}'), (NULL); INSERT INTO t_int_array_to_string (col) VALUES ('{-2147483648, 2147483647}'), (NULL); INSERT INTO t_int2 (col) VALUES (-32768), (32767), (3), (NULL); INSERT INTO t_int2_to_string (col) VALUES (-32768), (32767), (3), (NULL); +INSERT INTO t_int2_to_numeric (col) VALUES (-32768), (32767), (3), (NULL); +INSERT INTO t_int2_to_float32 (col) VALUES (-32768), (32767), (3), (NULL); +INSERT INTO t_int2_to_float64 (col) VALUES (-32768), (32767), (3), (NULL); INSERT INTO t_int4 (col) VALUES (-2147483648), (2147483647), (4), (NULL); INSERT INTO t_int4_to_string (col) VALUES (-2147483648), (2147483647), (4), (NULL); +INSERT INTO t_int4_to_numeric (col) VALUES (-2147483648), (2147483647), (4), (NULL); +INSERT INTO t_int4_to_float64 (col) VALUES (-2147483648), (2147483647), (4), (NULL); INSERT INTO t_int4multirange (col) VALUES ('{[10, 20]}'), (NULL); INSERT INTO t_int4range (col) VALUES ('[10, 20]'), (NULL); INSERT INTO t_int8 (col) VALUES (-9223372036854775808), (9223372036854775807), (5), (NULL); INSERT INTO t_int8_to_string (col) VALUES (-9223372036854775808), (9223372036854775807), (5), (NULL); +INSERT INTO t_int8_to_numeric (col) VALUES (-9223372036854775808), (9223372036854775807), (5), (NULL); INSERT INTO t_int8multirange (col) VALUES ('{[30, 40]}'), (NULL); INSERT INTO t_int8range (col) VALUES ('[30, 40]'), (NULL); INSERT INTO t_integer (col) VALUES (-2147483648), (2147483647), (2), (NULL); INSERT INTO t_integer_to_string (col) VALUES (-2147483648), (2147483647), (2), (NULL); -INSERT INTO t_interval (col) VALUES ('1 hour'), (NULL); +INSERT INTO t_integer_to_numeric (col) VALUES (-2147483648), (2147483647), (2), (NULL); +INSERT INTO t_integer_to_float64 (col) VALUES (-2147483648), (2147483647), (2), (NULL); +INSERT INTO t_interval (col) VALUES ('1 year 2 mons 3 days 04:05:06.789'::interval), ('00:00:00'::interval), ('-1 year 15 months -3 days 23 hours -78 minutes 106.789012 seconds'::interval), (NULL); INSERT INTO t_interval_to_int64 (col) VALUES ('1 hour'), (NULL); INSERT INTO t_json (col) VALUES ('{"duplicate_key": 1, "duplicate_key": 2}'), ('{"null_key": null}'), (NULL); INSERT INTO t_json_to_string (col) VALUES ('{"duplicate_key": 1, "duplicate_key": 2}'), ('{"null_key": null}'), (NULL); INSERT INTO t_jsonb (col) VALUES ('{"duplicate_key": 1, "duplicate_key": 2}'), ('{"null_key": null}'), (NULL); INSERT INTO t_jsonb_to_string (col) VALUES ('{"duplicate_key": 1, "duplicate_key": 2}'), ('{"null_key": null}'), (NULL); -INSERT INTO t_large_decimal_to_numeric (col) VALUES (0.12), (99999999999999999999999.9999999999), (12345678901234567890.1234567890), (NULL); INSERT INTO t_large_decimal_to_string (col) VALUES (0.12), (99999999999999999999999.9999999999), (123456789012345678901234567890.1234567890), (NULL); -INSERT INTO t_large_numeric_to_numeric (col) VALUES (0.12), (99999999999999999999999.9999999999), (12345678901234567890.1234567890), (NULL); INSERT INTO t_large_numeric_to_string (col) VALUES (0.12), (99999999999999999999999.9999999999), (123456789012345678901234567890.1234567890), (NULL); INSERT INTO t_line (col) VALUES ('{ 1, 2, 3 }'), (NULL); INSERT INTO t_line_to_float64_array (col) VALUES ('{ 1, 2, 3 }'), (NULL); @@ -221,11 +258,14 @@ INSERT INTO t_macaddr (col) VALUES ('08:00:2b:01:02:03'), (NULL); INSERT INTO t_macaddr8 (col) VALUES ('08:00:2b:01:02:03:04:05'), (NULL); INSERT INTO t_money (col) VALUES ('123.45'::money), (NULL); INSERT INTO t_money_to_int64 (col) VALUES ('123.45'::money), (NULL); -INSERT INTO t_numeric (col) VALUES (4.56), (NULL); +INSERT INTO t_money_to_numeric (col) VALUES ('123.45'::money), (NULL); INSERT INTO t_numeric_to_string (col) VALUES (4.56), (NULL); INSERT INTO t_nummultirange (col) VALUES ('{[50, 60]}'), (NULL); INSERT INTO t_numrange (col) VALUES ('[50, 60]'), (NULL); INSERT INTO t_oid (col) VALUES (1000::oid), (NULL); +INSERT INTO t_oid_to_string (col) VALUES (1000::oid), (NULL); +INSERT INTO t_oid_to_numeric (col) VALUES (1000::oid), (NULL); +INSERT INTO t_oid_to_float64 (col) VALUES (1000::oid), (NULL); INSERT INTO t_path (col) VALUES ('[ (1, 2), (3, 4), (5, 6) ]'), (NULL); INSERT INTO t_path_to_float64_array (col) VALUES ('[ (1, 2), (3, 4), (5, 6) ]'), (NULL); INSERT INTO t_pg_lsn (col) VALUES ('123/0'::pg_lsn), (NULL); @@ -241,30 +281,45 @@ INSERT INTO t_real_array_to_float32_array (col) VALUES ('{-1.9876542e38, 1.98765 INSERT INTO t_real_array_to_string (col) VALUES ('{-1.9876542e38, 1.9876542e38}'), (NULL); INSERT INTO t_serial (col) VALUES (-2147483648), (2147483647), (6); INSERT INTO t_serial_to_string (col) VALUES (-2147483648), (2147483647), (6); +INSERT INTO t_serial_to_numeric (col) VALUES (-2147483648), (2147483647), (6); +INSERT INTO t_serial_to_float64 (col) VALUES (-2147483648), (2147483647), (6); INSERT INTO t_serial2 (col) VALUES (-32768), (32767), (7); INSERT INTO t_serial2_to_string (col) VALUES (-32768), (32767), (7); +INSERT INTO t_serial2_to_numeric (col) VALUES (-32768), (32767), (7); +INSERT INTO t_serial2_to_float32 (col) VALUES (-32768), (32767), (7); +INSERT INTO t_serial2_to_float64 (col) VALUES (-32768), (32767), (7); INSERT INTO t_serial4 (col) VALUES (-2147483648), (2147483647), (8); INSERT INTO t_serial4_to_string (col) VALUES (-2147483648), (2147483647), (8); +INSERT INTO t_serial4_to_numeric (col) VALUES (-2147483648), (2147483647), (8); +INSERT INTO t_serial4_to_float64 (col) VALUES (-2147483648), (2147483647), (8); INSERT INTO t_serial8 (col) VALUES (-9223372036854775808), (9223372036854775807), (9); INSERT INTO t_serial8_to_string (col) VALUES (-9223372036854775808), (9223372036854775807), (9); +INSERT INTO t_serial8_to_numeric (col) VALUES (-9223372036854775808), (9223372036854775807), (9); INSERT INTO t_smallint (col) VALUES (-32768), (32767), (10), (NULL); INSERT INTO t_smallint_to_string (col) VALUES (-32768), (32767), (10), (NULL); +INSERT INTO t_smallint_to_numeric (col) VALUES (-32768), (32767), (10), (NULL); +INSERT INTO t_smallint_to_float32 (col) VALUES (-32768), (32767), (10), (NULL); +INSERT INTO t_smallint_to_float64 (col) VALUES (-32768), (32767), (10), (NULL); INSERT INTO t_smallint_array_to_int64_array (col) VALUES ('{-32768, 32767}'), (NULL); INSERT INTO t_smallint_array_to_string (col) VALUES ('{-32768, 32767}'), (NULL); INSERT INTO t_smallserial (col) VALUES (-32768), (32767), (11); INSERT INTO t_smallserial_to_string (col) VALUES (-32768), (32767), (11); +INSERT INTO t_smallserial_to_numeric (col) VALUES (-32768), (32767), (11); +INSERT INTO t_smallserial_to_float32 (col) VALUES (-32768), (32767), (11); +INSERT INTO t_smallserial_to_float64 (col) VALUES (-32768), (32767), (11); INSERT INTO t_text (col) VALUES ('testing text'), (NULL); INSERT INTO t_time (col) VALUES ('24:00:00'::time), (NULL); -INSERT INTO t_time_with_time_zone (col) VALUES ('23:59:59+10:00'), (NULL); +INSERT INTO t_time_with_time_zone (col) VALUES ('23:59:59+10:00'), ('24:00:00+10:00'), (NULL); INSERT INTO t_time_without_time_zone (col) VALUES ('24:00:00'::time), (NULL); INSERT INTO t_timestamp (col) VALUES ('1970-01-02 03:04:05.123456'::timestamp), (NULL); INSERT INTO t_timestamp_to_timestamp (col) VALUES ('1970-01-02 03:04:05.123456'::timestamp), (NULL); INSERT INTO t_timestamp_with_time_zone (col) VALUES ('1970-02-03 04:05:06.123456+10:00'::timestamptz), ('1970-02-03 04:05:06.123456-01'::timestamptz), (NULL); INSERT INTO t_timestamp_with_timezone_to_string (col) VALUES ('1970-02-03 04:05:06.123456+10:00'::timestamptz), ('1970-02-03 04:05:06.123456-01'::timestamptz), (NULL); INSERT INTO t_timestamp_without_time_zone (col) VALUES ('1970-01-02 03:04:05.123456'::timestamp), (NULL); +INSERT INTO t_timestamp_without_time_zone_to_timestamp (col) VALUES ('1970-01-02 03:04:05.123456'::timestamp), (NULL); INSERT INTO t_timestamptz (col) VALUES ('1970-02-03 04:05:06.123456+10:00'::timestamptz), ('1970-02-03 04:05:06.123456-01'::timestamptz), (NULL); INSERT INTO t_timestamptz_to_string (col) VALUES ('1970-02-03 04:05:06.123456+10:00'::timestamptz), ('1970-02-03 04:05:06.123456-01'::timestamptz), (NULL); -INSERT INTO t_timetz (col) VALUES ('23:59:59+10:00'), (NULL); +INSERT INTO t_timetz (col) VALUES ('23:59:59+10:00'), ('24:00:00+10:00'), (NULL); INSERT INTO t_tsmultirange (col) VALUES ('{[1970-01-01 01:00, 1970-01-01 02:00]}'), (NULL); INSERT INTO t_tsquery (col) VALUES ('fat & rat'::tsquery), (NULL); INSERT INTO t_tsrange (col) VALUES ('[1970-01-01 01:00, 1970-01-01 02:00]'), (NULL); @@ -274,10 +329,11 @@ INSERT INTO t_tsvector (col) VALUES ('a fat cat sat on a mat'::tsvector), (NULL) INSERT INTO t_txid_snapshot (col) VALUES ('10:20:10,14,15'::txid_snapshot), (NULL); INSERT INTO t_uuid (col) VALUES ('a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid), (NULL); INSERT INTO t_uuid_to_bytes (col) VALUES ('a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid), (NULL); +INSERT INTO t_uuid_to_string (col) VALUES ('a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid), (NULL); INSERT INTO t_varbit (col) VALUES ('1100'::varbit), (NULL); -INSERT INTO t_varbit_to_string (col) VALUES ('1100'::varbit(32)), (NULL); INSERT INTO t_varbit_to_bool_array (col) VALUES ('1100'::varbit(32)), (NULL); INSERT INTO t_varchar (col) VALUES ('testing varchar'), (NULL); +INSERT INTO t_varchar_n (col) VALUES ('testing'), (NULL); INSERT INTO t_xml (col) VALUES ('123'::xml), (NULL); CREATE TABLE t_uuid_pk (id uuid primary key, col uuid); diff --git a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/spanner-schema.sql b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/spanner-schema.sql index c0099bd470..43fd9ae3c8 100644 --- a/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/spanner-schema.sql +++ b/v2/datastream-to-spanner/src/test/resources/PostgreSQLDataTypesIT/spanner-schema.sql @@ -1,15 +1,15 @@ CREATE TABLE IF NOT EXISTS t_bigint (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bigint_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_bigint_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bigint_array_to_int64_array (id INT64, col ARRAY) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bigint_array_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bigserial (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bigserial_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_bigserial_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bit (id INT64, col BYTES(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bit_to_bool_array (id INT64, col ARRAY) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_bit_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bit_varying (id INT64, col BYTES(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bit_varying_to_bool_array (id INT64, col ARRAY) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_bit_varying_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bool (id INT64, col BOOL) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bool_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bool_array_to_bool_array (id INT64, col ARRAY) PRIMARY KEY (id); @@ -19,10 +19,12 @@ CREATE TABLE IF NOT EXISTS t_boolean_to_string (id INT64, col STRING(MAX)) PRIMA CREATE TABLE IF NOT EXISTS t_box (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_box_to_float64_array (id INT64, col ARRAY) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_bytea (id INT64, col BYTES(MAX)) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_bytea_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_char (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_char_n (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_character (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_character_n (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_character_varying (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_character_varying_n (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_cidr (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_circle (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_circle_to_float64_array (id INT64, col ARRAY) PRIMARY KEY (id); @@ -30,7 +32,6 @@ CREATE TABLE IF NOT EXISTS t_date (id INT64, col DATE) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_date_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_datemultirange (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_daterange (id INT64, col STRING(MAX)) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_decimal (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_decimal_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_double_precision (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_double_precision_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); @@ -47,29 +48,37 @@ CREATE TABLE IF NOT EXISTS t_float8_to_string (id INT64, col STRING(MAX)) PRIMAR CREATE TABLE IF NOT EXISTS t_inet (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int_array_to_int64_array (id INT64, col ARRAY) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int_array_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int2 (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int2_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int2_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int2_to_float32 (id INT64, col FLOAT32) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int2_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int4 (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int4_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int4_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int4_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int4multirange (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int4range (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int8 (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int8_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_int8_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int8multirange (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_int8range (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_integer (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_integer_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_integer_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_integer_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_interval (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_interval_to_int64 (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_json (id INT64, col JSON) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_json_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_jsonb (id INT64, col JSON) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_jsonb_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_large_decimal_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_large_decimal_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_large_numeric_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_large_numeric_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_line (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_line_to_float64_array (id INT64, col ARRAY) PRIMARY KEY (id); @@ -79,11 +88,14 @@ CREATE TABLE IF NOT EXISTS t_macaddr (id INT64, col STRING(MAX)) PRIMARY KEY (id CREATE TABLE IF NOT EXISTS t_macaddr8 (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_money (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_money_to_int64 (id INT64, col INT64) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_money_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_numeric_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_nummultirange (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_numrange (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_oid (id INT64, col INT64) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_oid_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_oid_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_oid_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_path (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_path_to_float64_array (id INT64, col ARRAY) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_pg_lsn (id INT64, col STRING(MAX)) PRIMARY KEY (id); @@ -99,18 +111,32 @@ CREATE TABLE IF NOT EXISTS t_real_array_to_float32_array (id INT64, col ARRAY) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_smallint_array_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_smallserial (id INT64, col INT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_smallserial_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_smallserial_to_numeric (id INT64, col NUMERIC) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_smallserial_to_float32 (id INT64, col FLOAT32) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_smallserial_to_float64 (id INT64, col FLOAT64) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_text (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_time (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_time_with_time_zone (id INT64, col STRING(MAX)) PRIMARY KEY (id); @@ -120,6 +146,7 @@ CREATE TABLE IF NOT EXISTS t_timestamp_to_timestamp (id INT64, col TIMESTAMP) PR CREATE TABLE IF NOT EXISTS t_timestamp_with_time_zone (id INT64, col TIMESTAMP) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_timestamp_with_timezone_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_timestamp_without_time_zone (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_timestamp_without_time_zone_to_timestamp (id INT64, col TIMESTAMP) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_timestamptz (id INT64, col TIMESTAMP) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_timestamptz_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_timetz (id INT64, col STRING(MAX)) PRIMARY KEY (id); @@ -132,10 +159,11 @@ CREATE TABLE IF NOT EXISTS t_tsvector (id INT64, col STRING(MAX)) PRIMARY KEY (i CREATE TABLE IF NOT EXISTS t_txid_snapshot (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_uuid (id INT64, col UUID) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_uuid_to_bytes (id INT64, col BYTES(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_uuid_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_varbit (id INT64, col BYTES(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_varbit_to_bool_array (id INT64, col ARRAY) PRIMARY KEY (id); -CREATE TABLE IF NOT EXISTS t_varbit_to_string (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_varchar (id INT64, col STRING(MAX)) PRIMARY KEY (id); +CREATE TABLE IF NOT EXISTS t_varchar_n (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_xml (id INT64, col STRING(MAX)) PRIMARY KEY (id); CREATE TABLE IF NOT EXISTS t_uuid_pk (id UUID NOT NULL, col UUID) PRIMARY KEY (id); diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertor.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertor.java index 0b483a9e96..c59216967b 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertor.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertor.java @@ -31,6 +31,7 @@ import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; +import java.util.Base64; import java.util.regex.Pattern; import org.apache.commons.lang3.BooleanUtils; import org.apache.commons.lang3.math.NumberUtils; @@ -215,12 +216,28 @@ public static ByteArray toByteArray(JsonNode changeEvent, String key, boolean re return ByteArray.copyFrom(bigIntValue.toByteArray()); } - // For data with Spanner type as BYTES, Datastream returns a hex encoded string. We need to - // decode it before returning to ensure data correctness. + // For all databases except PostgreSQL, for data with Spanner type as BYTES, + // Datastream returns a hex encoded string. We need to decode it before + // returning to ensure data correctness. String s = node.asText(); if (s.equalsIgnoreCase("NULL")) { return null; } + + // For PostgreSQL, we rely on `_metadata_source_type` here because during the Avro-to-JSON + // conversion (in FormatDatastreamRecordToJson), all structural Avro schema information is + // lost. + // Jackson automatically serializes Avro `bytes` fields (used natively by PostgreSQL + // Datastream for binary columns) into Base64 JSON strings. Since this method only + // receives the resulting JSON tree, it cannot check the original Avro schema type, + // so it checks if the source database type is PostgreSQL instead to safely decode + // the Base64. + JsonNode sourceTypeNode = changeEvent.get("_metadata_source_type"); + if (sourceTypeNode != null && "postgresql".equalsIgnoreCase(sourceTypeNode.asText())) { + byte[] decodedBytes = Base64.getDecoder().decode(s); + return ByteArray.copyFrom(decodedBytes); + } + // Make an odd length hex string even by appending a 0 in the beginning. if (s.length() % 2 == 1) { s = "0" + s; diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertorTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertorTest.java index a4488aaa14..beffe9872e 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertorTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertorTest.java @@ -575,6 +575,20 @@ public void canConvertToByteArray() throws Exception { ByteArray.copyFrom(new byte[] {-1, 0})); } + @Test + public void canConvertToByteArrayPostgresqlBase64() throws Exception { + JSONObject changeEvent = new JSONObject(); + changeEvent.put("_metadata_source_type", "postgresql"); + // Base64 encoding of byte array: { (byte) 0xde, (byte) 0xad, (byte) 0xbe, (byte) 0xef } + changeEvent.put("field1", "3q2+7w=="); + + JsonNode ce = getJsonNode(changeEvent.toString()); + + assertEquals( + ChangeEventTypeConvertor.toByteArray(ce, "field1", /* requiredField= */ true), + ByteArray.copyFrom(new byte[] {(byte) 0xde, (byte) 0xad, (byte) 0xbe, (byte) 0xef})); + } + @Test(expected = ChangeEventConvertorException.class) public void cannotConvertNonExistentRequiredFieldToByteArray() throws Exception { JSONObject changeEvent = new JSONObject();