From 74477ccd235df71ad4b37b12b325b54ae56c984c Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Mon, 24 Aug 2026 16:00:15 +0000 Subject: [PATCH 1/6] added oracle support to data validation --- .../teleport/v2/visitor/IUnifiedVisitor.java | 3 + .../v2/visitor/UnifiedHasherVisitor.java | 7 + .../v2/visitor/UnifiedStringVisitor.java | 5 + ...sonRecordMapperOracleAllDataTypesTest.java | 806 ++++++++++++++++++ .../GCSSpannerDVAvroSetupHelper.java | 27 +- .../templates/GCSSpannerDVOracleSmokeIT.java | 243 ++++++ .../v2/visitor/IUnifiedVisitorTest.java | 33 + .../v2/visitor/UnifiedHasherVisitorTest.java | 15 + .../oracle_all_datatypes.avsc | 86 ++ .../spanner-schema.sql | 34 + 10 files changed, 1256 insertions(+), 3 deletions(-) create mode 100644 v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java create mode 100644 v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java create mode 100644 v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc create mode 100644 v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java index d2ed372d0f..33675c1c25 100644 --- a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java @@ -37,6 +37,8 @@ public interface IUnifiedVisitor { void visitFloat64(double d); + void visitFloat32(float f); + void visitBool(boolean b); void visitBytes(byte[] b); @@ -64,6 +66,7 @@ static void dispatch(Value value, IUnifiedVisitor visitor) { case UUID -> visitor.visitUuid(value.getUuid()); case INT64 -> visitor.visitInt64(value.getInt64()); case FLOAT64 -> visitor.visitFloat64(value.getFloat64()); + case FLOAT32 -> visitor.visitFloat32(value.getFloat32()); case BOOL -> visitor.visitBool(value.getBool()); case BYTES -> visitor.visitBytes(value.getBytes().toByteArray()); case DATE -> visitor.visitDate(value.getDate()); diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitor.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitor.java index 9d5f4e6ea7..c67f4c6f0a 100644 --- a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitor.java +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitor.java @@ -66,6 +66,13 @@ public void visitFloat64(double d) { hasher.putDouble(d); } + @Override + public void visitFloat32(float f) { + // Float32 values are encoded with a sentinel byte 1 followed by the float value + markNonNull(); + hasher.putFloat(f); + } + @Override public void visitBool(boolean b) { // Bool values are encoded with a sentinel byte 1 followed by the boolean value diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitor.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitor.java index 8c9e860d6e..a943b20445 100644 --- a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitor.java +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitor.java @@ -49,6 +49,11 @@ public void visitFloat64(double d) { result = String.valueOf(d); } + @Override + public void visitFloat32(float f) { + result = String.valueOf(f); + } + @Override public void visitBool(boolean b) { result = String.valueOf(b); diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java new file mode 100644 index 0000000000..b8e3f6907c --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java @@ -0,0 +1,806 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ +package com.google.cloud.teleport.v2.mapper; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.cloud.ByteArray; +import com.google.cloud.Timestamp; +import com.google.cloud.spanner.Dialect; +import com.google.cloud.spanner.Struct; +import com.google.cloud.spanner.Value; +import com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants; +import com.google.cloud.teleport.v2.dto.ComparisonRecord; +import com.google.cloud.teleport.v2.spanner.ddl.Column; +import com.google.cloud.teleport.v2.spanner.ddl.Ddl; +import com.google.cloud.teleport.v2.spanner.ddl.IndexColumn; +import com.google.cloud.teleport.v2.spanner.ddl.Table; +import com.google.cloud.teleport.v2.spanner.migrations.avro.GenericRecordTypeConvertor; +import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper; +import com.google.cloud.teleport.v2.spanner.migrations.schema.IdentityMapper; +import com.google.cloud.teleport.v2.spanner.type.Type; +import com.google.cloud.teleport.v2.spanner.utils.ISpannerMigrationTransformer; +import com.google.cloud.teleport.v2.spanner.utils.MigrationTransformationResponse; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVAvroSetupHelper; +import com.google.cloud.teleport.v2.visitor.IUnifiedVisitor; +import com.google.cloud.teleport.v2.visitor.UnifiedHasherVisitor; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.avro.LogicalTypes; +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * Unit test validating that all Oracle data types specified in the Oracle Datatype Mapping Matrix + * correctly convert from Avro GenericRecord and Spanner Struct into identical ComparisonRecord + * hashes. + */ +@RunWith(JUnit4.class) +public class ComparisonRecordMapperOracleAllDataTypesTest { + + private ISchemaMapper mockSchemaMapper; + private ISpannerMigrationTransformer mockTransformer; + private Ddl mockDdl; + private ComparisonRecordMapper mapper; + + @Before + public void setUp() { + mockSchemaMapper = mock(ISchemaMapper.class); + mockTransformer = mock(ISpannerMigrationTransformer.class); + mockDdl = mock(Ddl.class); + mapper = new ComparisonRecordMapper(mockSchemaMapper, mockTransformer, mockDdl); + } + + @Test + public void testAllOracleDataTypesHashParity() throws Exception { + String tableName = "AllDatatypes"; + + // 1. Define Avro Schemas for Oracle Datatypes + Schema decimalSchema = + LogicalTypes.decimal(10, 2).addToSchema(Schema.create(Schema.Type.BYTES)); + Schema timestampMicrosSchema = + LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)); + + Schema payloadSchema = + SchemaBuilder.record("Payload") + .fields() + .name("id") + .type(Schema.create(Schema.Type.LONG)) + .noDefault() + .name("varchar2_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("varchar_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("char_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("character_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("nvarchar2_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("nchar_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("number_col") + .type(decimalSchema) + .noDefault() + .name("numeric_col") + .type(decimalSchema) + .noDefault() + .name("decimal_col") + .type(decimalSchema) + .noDefault() + .name("dec_col") + .type(decimalSchema) + .noDefault() + .name("float_col") + .type(Schema.create(Schema.Type.DOUBLE)) + .noDefault() + .name("double_precision_col") + .type(Schema.create(Schema.Type.DOUBLE)) + .noDefault() + .name("real_col") + .type(Schema.create(Schema.Type.DOUBLE)) + .noDefault() + .name("binary_float_col") + .type(Schema.create(Schema.Type.FLOAT)) + .noDefault() + .name("binary_double_col") + .type(Schema.create(Schema.Type.DOUBLE)) + .noDefault() + .name("integer_col") + .type(Schema.create(Schema.Type.LONG)) + .noDefault() + .name("int_col") + .type(Schema.create(Schema.Type.LONG)) + .noDefault() + .name("smallint_col") + .type(Schema.create(Schema.Type.LONG)) + .noDefault() + .name("date_col") + .type(timestampMicrosSchema) + .noDefault() + .name("timestamp_col") + .type(timestampMicrosSchema) + .noDefault() + .name("timestamp_tz_col") + .type(timestampMicrosSchema) + .noDefault() + .name("timestamp_ltz_col") + .type(timestampMicrosSchema) + .noDefault() + .name("interval_ym_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("interval_ds_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("raw_col") + .type(Schema.create(Schema.Type.BYTES)) + .noDefault() + .name("blob_col") + .type(Schema.create(Schema.Type.BYTES)) + .noDefault() + .name("clob_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("nclob_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("rowid_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("json_col") + .type(SchemaBuilder.builder().stringBuilder().prop("logicalType", "json").endString()) + .noDefault() + .name("xmltype_col") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .endRecord(); + + Schema avroSchema = + SchemaBuilder.record("SourceRow") + .fields() + .name("tableName") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("shardId") + .type(Schema.create(Schema.Type.STRING)) + .noDefault() + .name("payload") + .type(payloadSchema) + .noDefault() + .endRecord(); + + // 2. Populate Avro payload + GenericRecord payload = new GenericData.Record(payloadSchema); + payload.put("id", 1L); + payload.put("varchar2_col", "test_varchar2"); + payload.put("varchar_col", "test_varchar"); + payload.put("char_col", "test_char "); + payload.put("character_col", "test_char "); + payload.put("nvarchar2_col", "test_nvarchar2"); + payload.put("nchar_col", "test_nchar"); + payload.put( + "number_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); + payload.put( + "numeric_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); + payload.put( + "decimal_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); + payload.put( + "dec_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); + payload.put("float_col", 123.456d); + payload.put("double_precision_col", 123.456d); + payload.put("real_col", 123.456d); + payload.put("binary_float_col", 123.0f); + payload.put("binary_double_col", 123.0d); + payload.put("integer_col", 12345L); + payload.put("int_col", 12345L); + payload.put("smallint_col", 123L); + long timestampMicros = 1704103200000000L; + payload.put("date_col", timestampMicros); + payload.put("timestamp_col", timestampMicros); + payload.put("timestamp_tz_col", timestampMicros); + payload.put("timestamp_ltz_col", timestampMicros); + payload.put("interval_ym_col", "P1Y2M"); + payload.put("interval_ds_col", "PT3H4M5S"); + payload.put("raw_col", ByteBuffer.wrap(new byte[] {0x41, 0x42, 0x43})); + payload.put("blob_col", ByteBuffer.wrap(new byte[] {0x41, 0x42, 0x43, 0x44})); + payload.put("clob_col", "test_clob_content"); + payload.put("nclob_col", "test_nclob_content"); + payload.put("rowid_col", "AAAB12AADAAAAwPAAA"); + payload.put("json_col", "{\"k1\":\"v1\"}"); + payload.put("xmltype_col", "test"); + + GenericRecord avroRecord = new GenericData.Record(avroSchema); + avroRecord.put("tableName", tableName); + avroRecord.put("shardId", "shard1"); + avroRecord.put("payload", payload); + + // 3. Configure Schema Mapper mocks + List columnNames = + Arrays.asList( + "id", + "varchar2_col", + "varchar_col", + "char_col", + "character_col", + "nvarchar2_col", + "nchar_col", + "number_col", + "numeric_col", + "decimal_col", + "dec_col", + "float_col", + "double_precision_col", + "real_col", + "binary_float_col", + "binary_double_col", + "integer_col", + "int_col", + "smallint_col", + "date_col", + "timestamp_col", + "timestamp_tz_col", + "timestamp_ltz_col", + "interval_ym_col", + "interval_ds_col", + "raw_col", + "blob_col", + "clob_col", + "nclob_col", + "rowid_col", + "json_col", + "xmltype_col"); + + when(mockSchemaMapper.getSpannerTableName(anyString(), anyString())).thenReturn(tableName); + when(mockSchemaMapper.getSpannerColumnName(anyString(), anyString(), anyString())) + .thenAnswer(invocation -> invocation.getArgument(2)); + when(mockSchemaMapper.getSourceColumnName(anyString(), anyString(), anyString())) + .thenAnswer(invocation -> invocation.getArgument(2)); + when(mockSchemaMapper.getDialect()).thenReturn(Dialect.GOOGLE_STANDARD_SQL); + when(mockSchemaMapper.getSpannerColumns(anyString(), anyString())).thenReturn(columnNames); + when(mockSchemaMapper.colExistsAtSource(anyString(), anyString(), anyString())) + .thenReturn(true); + + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("id"))) + .thenReturn(Type.int64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("varchar2_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("varchar_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("char_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("character_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nvarchar2_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nchar_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("number_col"))) + .thenReturn(Type.numeric()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("numeric_col"))) + .thenReturn(Type.numeric()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("decimal_col"))) + .thenReturn(Type.numeric()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("dec_col"))) + .thenReturn(Type.numeric()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("float_col"))) + .thenReturn(Type.float64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("double_precision_col"))) + .thenReturn(Type.float64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("real_col"))) + .thenReturn(Type.float64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("binary_float_col"))) + .thenReturn(Type.float32()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("binary_double_col"))) + .thenReturn(Type.float64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("integer_col"))) + .thenReturn(Type.int64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("int_col"))) + .thenReturn(Type.int64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("smallint_col"))) + .thenReturn(Type.int64()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("date_col"))) + .thenReturn(Type.timestamp()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_col"))) + .thenReturn(Type.timestamp()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_tz_col"))) + .thenReturn(Type.timestamp()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_ltz_col"))) + .thenReturn(Type.timestamp()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("interval_ym_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("interval_ds_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("raw_col"))) + .thenReturn(Type.bytes()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("blob_col"))) + .thenReturn(Type.bytes()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("clob_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nclob_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("rowid_col"))) + .thenReturn(Type.string()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("json_col"))) + .thenReturn(Type.json()); + when(mockSchemaMapper.getSpannerColumnType( + anyString(), anyString(), org.mockito.ArgumentMatchers.eq("xmltype_col"))) + .thenReturn(Type.string()); + + Table mockTable = mock(Table.class); + when(mockDdl.table(tableName)).thenReturn(mockTable); + IndexColumn pkCol = IndexColumn.create("id", IndexColumn.Order.ASC); + when(mockTable.primaryKeys()).thenReturn(com.google.common.collect.ImmutableList.of(pkCol)); + + MigrationTransformationResponse mockResponse = mock(MigrationTransformationResponse.class); + when(mockResponse.isEventFiltered()).thenReturn(false); + when(mockResponse.getResponseRow()).thenReturn(Collections.emptyMap()); + when(mockTransformer.toSpannerRow(org.mockito.ArgumentMatchers.any())).thenReturn(mockResponse); + + // 4. Map Avro Record to ComparisonRecord + ComparisonRecord avroRecordResult = mapper.mapFrom(avroRecord); + assertNotNull(avroRecordResult); + assertEquals(tableName, avroRecordResult.getTableName()); + assertEquals("shard1", avroRecordResult.getShardId()); + + // 5. Build identical Spanner Struct + Timestamp spannerTimestamp = Timestamp.ofTimeMicroseconds(timestampMicros); + Struct spannerStruct = + Struct.newBuilder() + .set(GCSSpannerDVConstants.TABLE_NAME_COLUMN) + .to(tableName) + .set("id") + .to(1L) + .set("varchar2_col") + .to("test_varchar2") + .set("varchar_col") + .to("test_varchar") + .set("char_col") + .to("test_char ") + .set("character_col") + .to("test_char ") + .set("nvarchar2_col") + .to("test_nvarchar2") + .set("nchar_col") + .to("test_nchar") + .set("number_col") + .to(new BigDecimal("1234.560000000")) + .set("numeric_col") + .to(new BigDecimal("1234.560000000")) + .set("decimal_col") + .to(new BigDecimal("1234.560000000")) + .set("dec_col") + .to(new BigDecimal("1234.560000000")) + .set("float_col") + .to(123.456d) + .set("double_precision_col") + .to(123.456d) + .set("real_col") + .to(123.456d) + .set("binary_float_col") + .to(123.0f) + .set("binary_double_col") + .to(123.0d) + .set("integer_col") + .to(12345L) + .set("int_col") + .to(12345L) + .set("smallint_col") + .to(123L) + .set("date_col") + .to(spannerTimestamp) + .set("timestamp_col") + .to(spannerTimestamp) + .set("timestamp_tz_col") + .to(spannerTimestamp) + .set("timestamp_ltz_col") + .to(spannerTimestamp) + .set("interval_ym_col") + .to("P1Y2M") + .set("interval_ds_col") + .to("PT3H4M5S") + .set("raw_col") + .to(ByteArray.copyFrom(new byte[] {0x41, 0x42, 0x43})) + .set("blob_col") + .to(ByteArray.copyFrom(new byte[] {0x41, 0x42, 0x43, 0x44})) + .set("clob_col") + .to("test_clob_content") + .set("nclob_col") + .to("test_nclob_content") + .set("rowid_col") + .to("AAAB12AADAAAAwPAAA") + .set("json_col") + .to(Value.json("{\"k1\":\"v1\"}")) + .set("xmltype_col") + .to("test") + .build(); + + // 6. Map Spanner Struct to ComparisonRecord + ComparisonRecord spannerRecordResult = mapper.mapFrom(spannerStruct); + assertNotNull(spannerRecordResult); + + GenericRecordTypeConvertor convertor = + new GenericRecordTypeConvertor(mockSchemaMapper, "", "shard1", mockTransformer); + java.util.Map avroValues = convertor.transformChangeEvent(payload, tableName); + + for (String col : columnNames) { + Value avroVal = avroValues.get(col); + Value spannerVal = spannerStruct.getValue(col); + + com.google.common.hash.Hasher h1 = com.google.common.hash.Hashing.murmur3_128().newHasher(); + UnifiedHasherVisitor v1 = new UnifiedHasherVisitor(h1); + IUnifiedVisitor.dispatch(avroVal, v1); + + com.google.common.hash.Hasher h2 = com.google.common.hash.Hashing.murmur3_128().newHasher(); + UnifiedHasherVisitor v2 = new UnifiedHasherVisitor(h2); + IUnifiedVisitor.dispatch(spannerVal, v2); + + org.junit.Assert.assertEquals( + "Hash mismatch for column " + + col + + " (avroVal: " + + avroVal + + " vs spannerVal: " + + spannerVal + + ")", + h2.hash().toString(), + h1.hash().toString()); + } + + // 7. Verify Hashes Match Exactly! + assertEquals( + "Avro hash and Spanner Struct hash must match for all Oracle datatypes", + spannerRecordResult.getHash(), + avroRecordResult.getHash()); + } + + @Test + public void testSmokeSchemaAndRecordsDirectly() throws Exception { + Ddl ddl = + Ddl.builder(Dialect.GOOGLE_STANDARD_SQL) + .createTable("OracleAllDatatypes") + .column("id") + .int64() + .notNull() + .endColumn() + .column("varchar2_col") + .string() + .max() + .endColumn() + .column("varchar_col") + .string() + .max() + .endColumn() + .column("char_col") + .string() + .max() + .endColumn() + .column("character_col") + .string() + .max() + .endColumn() + .column("nvarchar2_col") + .string() + .max() + .endColumn() + .column("nchar_col") + .string() + .max() + .endColumn() + .column("number_col") + .numeric() + .endColumn() + .column("numeric_col") + .numeric() + .endColumn() + .column("decimal_col") + .numeric() + .endColumn() + .column("dec_col") + .numeric() + .endColumn() + .column("float_col") + .float64() + .endColumn() + .column("double_precision_col") + .float64() + .endColumn() + .column("real_col") + .float64() + .endColumn() + .column("binary_float_col") + .float32() + .endColumn() + .column("binary_double_col") + .float64() + .endColumn() + .column("integer_col") + .int64() + .endColumn() + .column("int_col") + .int64() + .endColumn() + .column("smallint_col") + .int64() + .endColumn() + .column("date_col") + .timestamp() + .endColumn() + .column("timestamp_col") + .timestamp() + .endColumn() + .column("timestamp_tz_col") + .timestamp() + .endColumn() + .column("timestamp_ltz_col") + .timestamp() + .endColumn() + .column("interval_ym_col") + .string() + .max() + .endColumn() + .column("interval_ds_col") + .string() + .max() + .endColumn() + .column("raw_col") + .bytes() + .max() + .endColumn() + .column("blob_col") + .bytes() + .max() + .endColumn() + .column("clob_col") + .string() + .max() + .endColumn() + .column("nclob_col") + .string() + .max() + .endColumn() + .column("rowid_col") + .string() + .max() + .endColumn() + .column("json_col") + .json() + .endColumn() + .column("xmltype_col") + .string() + .max() + .endColumn() + .primaryKey() + .asc("id") + .end() + .endTable() + .build(); + + IdentityMapper identityMapper = new IdentityMapper(ddl); + ComparisonRecordMapper smokeMapper = new ComparisonRecordMapper(identityMapper, null, ddl); + + GCSSpannerDVAvroSetupHelper.TableDef tableDef = + new GCSSpannerDVAvroSetupHelper.TableDef( + new Schema.Parser() + .parse( + ComparisonRecordMapperOracleAllDataTypesTest.class + .getClassLoader() + .getResourceAsStream( + "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc")), + "OracleAllDatatypes", + Arrays.asList("id")); + + java.time.Instant testTimestamp = java.time.Instant.parse("2024-01-01T10:00:00Z"); + BigDecimal testNumeric = new BigDecimal("1234.560000000"); + byte[] testBytes = new byte[] {0x41, 0x42, 0x43, 0x44}; + + GenericRecord avroRecord = + new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null) + .set("id", 1L) + .set("varchar2_col", "test_varchar2") + .set("varchar_col", "test_varchar") + .set("char_col", "test_char ") + .set("character_col", "test_char ") + .set("nvarchar2_col", "test_nvarchar2") + .set("nchar_col", "test_nchar ") + .set("number_col", testNumeric) + .set("numeric_col", testNumeric) + .set("decimal_col", testNumeric) + .set("dec_col", testNumeric) + .set("float_col", 123.456) + .set("double_precision_col", 123.456) + .set("real_col", 123.456) + .set("binary_float_col", 123.0f) + .set("binary_double_col", 123.0) + .set("integer_col", 12345L) + .set("int_col", 12345L) + .set("smallint_col", 123L) + .set("date_col", testTimestamp) + .set("timestamp_col", testTimestamp) + .set("timestamp_tz_col", testTimestamp) + .set("timestamp_ltz_col", testTimestamp) + .set("interval_ym_col", "P1Y2M") + .set("interval_ds_col", "PT3H4M5S") + .set("raw_col", testBytes) + .set("blob_col", testBytes) + .set("clob_col", "test_clob_content") + .set("nclob_col", "test_nclob_content") + .set("rowid_col", "AAAB12AADAAAAwPAAA") + .set("json_col", "{}") + .set("xmltype_col", "test") + .build(); + + Struct spannerStruct = + Struct.newBuilder() + .set(GCSSpannerDVConstants.TABLE_NAME_COLUMN) + .to("OracleAllDatatypes") + .set("id") + .to(1L) + .set("varchar2_col") + .to("test_varchar2") + .set("varchar_col") + .to("test_varchar") + .set("char_col") + .to("test_char ") + .set("character_col") + .to("test_char ") + .set("nvarchar2_col") + .to("test_nvarchar2") + .set("nchar_col") + .to("test_nchar ") + .set("number_col") + .to(testNumeric) + .set("numeric_col") + .to(testNumeric) + .set("decimal_col") + .to(testNumeric) + .set("dec_col") + .to(testNumeric) + .set("float_col") + .to(123.456) + .set("double_precision_col") + .to(123.456) + .set("real_col") + .to(123.456) + .set("binary_float_col") + .to(123.0f) + .set("binary_double_col") + .to(123.0) + .set("integer_col") + .to(12345L) + .set("int_col") + .to(12345L) + .set("smallint_col") + .to(123L) + .set("date_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_tz_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_ltz_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("interval_ym_col") + .to("P1Y2M") + .set("interval_ds_col") + .to("PT3H4M5S") + .set("raw_col") + .to(ByteArray.copyFrom(testBytes)) + .set("blob_col") + .to(ByteArray.copyFrom(testBytes)) + .set("clob_col") + .to("test_clob_content") + .set("nclob_col") + .to("test_nclob_content") + .set("rowid_col") + .to("AAAB12AADAAAAwPAAA") + .set("json_col") + .to(Value.json("{}")) + .set("xmltype_col") + .to("test") + .build(); + + ComparisonRecord avroResult = smokeMapper.mapFrom(avroRecord); + ComparisonRecord spannerResult = smokeMapper.mapFrom(spannerStruct); + + assertNotNull(avroResult); + assertNotNull(spannerResult); + + GenericRecord payload = (GenericRecord) avroRecord.get("payload"); + GenericRecordTypeConvertor convertor = + new GenericRecordTypeConvertor(identityMapper, "", null, null); + java.util.Map avroValues = + convertor.transformChangeEvent(payload, "OracleAllDatatypes"); + + for (String col : + ddl.table("OracleAllDatatypes").columns().stream().map(Column::name).toList()) { + Value avroVal = avroValues.get(col); + Value spannerVal = spannerStruct.getValue(col); + + com.google.common.hash.Hasher h1 = com.google.common.hash.Hashing.murmur3_128().newHasher(); + UnifiedHasherVisitor v1 = new UnifiedHasherVisitor(h1); + IUnifiedVisitor.dispatch(avroVal, v1); + + com.google.common.hash.Hasher h2 = com.google.common.hash.Hashing.murmur3_128().newHasher(); + UnifiedHasherVisitor v2 = new UnifiedHasherVisitor(h2); + IUnifiedVisitor.dispatch(spannerVal, v2); + + org.junit.Assert.assertEquals( + "Hash mismatch for column " + + col + + " (avroVal: " + + avroVal + + " vs spannerVal: " + + spannerVal + + ")", + h2.hash().toString(), + h1.hash().toString()); + } + + System.out.println("UNIT TEST COMPUTED HASH: " + avroResult.getHash()); + assertEquals(spannerResult.getHash(), avroResult.getHash()); + } +} diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java index 7d81b3ee2e..3fb6b3e1c8 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java @@ -29,6 +29,18 @@ */ public class GCSSpannerDVAvroSetupHelper { + public static Schema parseAvroSchema(String resourceName) { + try (java.io.InputStream is = + GCSSpannerDVAvroSetupHelper.class.getClassLoader().getResourceAsStream(resourceName)) { + if (is == null) { + throw new IllegalArgumentException("Resource not found: " + resourceName); + } + return new Schema.Parser().parse(is); + } catch (Exception e) { + throw new RuntimeException("Failed to parse Avro schema from resource: " + resourceName, e); + } + } + /** * Defines standard table schemas that are universally used across multiple integration tests. * Centralizing these definitions prevents schema drift across tests and minimizes setup code. @@ -36,13 +48,12 @@ public class GCSSpannerDVAvroSetupHelper { public static class TableDef { public static final TableDef USERS = new TableDef( - GCSSpannerDVITBase.getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/users.avsc"), + parseAvroSchema("GCSSpannerDVAvroSetupHelper/users.avsc"), "Users", Arrays.asList("user_id", "event_id")); public static final TableDef ACCOUNT_ROLES = new TableDef( - GCSSpannerDVITBase.getSchemaFromAvscFile( - "GCSSpannerDVAvroSetupHelper/account_roles.avsc"), + parseAvroSchema("GCSSpannerDVAvroSetupHelper/account_roles.avsc"), "AccountRoles", Arrays.asList("role_id")); @@ -122,6 +133,16 @@ private static Object convertToAvroFormat(Object value) { return (t.getEpochSecond() * 1_000_000L) + (t.getNano() / 1000L); } + if (value instanceof java.math.BigDecimal) { + java.math.BigDecimal bd = (java.math.BigDecimal) value; + return java.nio.ByteBuffer.wrap( + bd.setScale(9, java.math.RoundingMode.HALF_UP).unscaledValue().toByteArray()); + } + + if (value instanceof byte[]) { + return java.nio.ByteBuffer.wrap((byte[]) value); + } + // Default fallback (String, Integer, Long, Double, Float) return value; } diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java new file mode 100644 index 0000000000..e465e6de25 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java @@ -0,0 +1,243 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ +package com.google.cloud.teleport.v2.templates; + +import com.google.cloud.ByteArray; +import com.google.cloud.Timestamp; +import com.google.cloud.spanner.Mutation; +import com.google.cloud.spanner.Value; +import com.google.cloud.teleport.metadata.DirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.TableValidationStatsDto; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.ValidationSummaryDto; +import java.io.IOException; +import java.math.BigDecimal; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; +import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Integration smoke test for GCSSpannerDV validating all Oracle data types. */ +@Category({TemplateIntegrationTest.class, DirectRunnerTest.class}) +@RunWith(JUnit4.class) +@TemplateIntegrationTest(GCSSpannerDV.class) +public class GCSSpannerDVOracleSmokeIT extends GCSSpannerDVITBase { + + private static final String SPANNER_DDL_RESOURCE = "GCSSpannerDVOracleSmokeIT/spanner-schema.sql"; + private static final String AVRO_SCHEMA_RESOURCE = + "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc"; + + @Before + public void setUp() throws IOException { + spannerResourceManager = setUpSpannerResourceManager(); + bigQueryResourceManager = setUpBigQueryResourceManager(); + bigQueryResourceManager.createDataset(REGION); + createSpannerDDL(spannerResourceManager, SPANNER_DDL_RESOURCE); + } + + @Test + public void testOracleAllDataTypesValidationSmoke() throws Exception { + GCSSpannerDVAvroSetupHelper.TableDef tableDef = + new GCSSpannerDVAvroSetupHelper.TableDef( + getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), "OracleAllDatatypes", Arrays.asList("id")); + + java.time.Instant testTimestamp = java.time.Instant.parse("2024-01-01T10:00:00Z"); + BigDecimal testNumeric = new BigDecimal("1234.567890123"); + byte[] testBytes = new byte[] {0x41, 0x42, 0x43, 0x44}; + + // 1. Generate Avro Source Records + List records = + Arrays.asList( + // Row 1: Standard Row with all types populated + new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null) + .set("id", 1L) + .set("varchar2_col", "test_varchar2") + .set("varchar_col", "test_varchar") + .set("char_col", "test_char ") + .set("character_col", "test_char ") + .set("nvarchar2_col", "test_nvarchar2") + .set("nchar_col", "test_nchar ") + .set("number_col", testNumeric) + .set("numeric_col", testNumeric) + .set("decimal_col", testNumeric) + .set("dec_col", testNumeric) + .set("float_col", 123.456) + .set("double_precision_col", 123.456) + .set("real_col", 123.456) + .set("binary_float_col", 123.0f) + .set("binary_double_col", 123.0) + .set("integer_col", 12345L) + .set("int_col", 12345L) + .set("smallint_col", 123L) + .set("date_col", testTimestamp) + .set("timestamp_col", testTimestamp) + .set("timestamp_tz_col", testTimestamp) + .set("timestamp_ltz_col", testTimestamp) + .set("interval_ym_col", "P1Y2M") + .set("interval_ds_col", "PT3H4M5S") + .set("raw_col", testBytes) + .set("blob_col", testBytes) + .set("clob_col", "test_clob_content") + .set("nclob_col", "test_nclob_content") + .set("rowid_col", "AAAB12AADAAAAwPAAA") + .set("json_col", "{}") + .set("xmltype_col", "test") + .build(), + // Row 2: Null Row + new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null).set("id", 2L).build()); + + String gcsInputDirectory = getGcsPath("input"); + uploadAvroFileToGcs("input/oracle_all_datatypes.avro", tableDef.schema, records); + + // 2. Insert Matching Records in Destination (Spanner) + spannerResourceManager.write( + Arrays.asList( + // Row 1 + Mutation.newInsertOrUpdateBuilder("OracleAllDatatypes") + .set("id") + .to(1L) + .set("varchar2_col") + .to("test_varchar2") + .set("varchar_col") + .to("test_varchar") + .set("char_col") + .to("test_char ") + .set("character_col") + .to("test_char ") + .set("nvarchar2_col") + .to("test_nvarchar2") + .set("nchar_col") + .to("test_nchar ") + .set("number_col") + .to(testNumeric) + .set("numeric_col") + .to(testNumeric) + .set("decimal_col") + .to(testNumeric) + .set("dec_col") + .to(testNumeric) + .set("float_col") + .to(123.456) + .set("double_precision_col") + .to(123.456) + .set("real_col") + .to(123.456) + .set("binary_float_col") + .to(123.0f) + .set("binary_double_col") + .to(123.0) + .set("integer_col") + .to(12345L) + .set("int_col") + .to(12345L) + .set("smallint_col") + .to(123L) + .set("date_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_tz_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("timestamp_ltz_col") + .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) + .set("interval_ym_col") + .to("P1Y2M") + .set("interval_ds_col") + .to("PT3H4M5S") + .set("raw_col") + .to(ByteArray.copyFrom(testBytes)) + .set("blob_col") + .to(ByteArray.copyFrom(testBytes)) + .set("clob_col") + .to("test_clob_content") + .set("nclob_col") + .to("test_nclob_content") + .set("rowid_col") + .to("AAAB12AADAAAAwPAAA") + .set("json_col") + .to(Value.json("{}")) + .set("xmltype_col") + .to("test") + .build(), + // Row 2 + Mutation.newInsertOrUpdateBuilder("OracleAllDatatypes").set("id").to(2L).build())); + + // Wait for Spanner's 20-second exact staleness read bound in SpannerReaderTransform + Thread.sleep(20000); + + // 3. Launch Pipeline + LaunchConfig.Builder options = LaunchConfig.builder(testName, specPath); + LaunchInfo jobInfo = + launchDataflowJob( + options, + testName, + PROJECT, + spannerResourceManager, + bigQueryResourceManager.getDatasetId(), + gcsInputDirectory, + null, + null, + null, + null, + null, + null); + + pipelineOperator().waitUntilDone(createConfig(jobInfo)); + + List> mismatches = + org.apache.beam.it.gcp.bigquery.matchers.BigQueryAsserts.tableResultToRecords( + bigQueryResourceManager.readTable("MismatchedRecords")); + System.out.println("DEBUG MISMATCHED RECORDS COUNT: " + mismatches.size()); + for (java.util.Map m : mismatches) { + System.out.println("DEBUG MISMATCH: " + m); + } + + // 4. Assert Validation Results in BigQuery + GCSSpannerDVTestAsserts.assertValidationSummary( + bigQueryResourceManager, + Collections.singletonList( + new ValidationSummaryDto( + "MATCH", + 1L, // Total tables validated + 2L, // Total rows matched + 0L, // Total rows mismatched + "" // Tables with mismatches + ))); + + GCSSpannerDVTestAsserts.assertTableValidationStats( + bigQueryResourceManager, + Collections.singletonList( + new TableValidationStatsDto( + null, // Schema name + "OracleAllDatatypes", // Table name + "MATCH", // Status + 2L, // Source row count + 2L, // Destination row count + 2L, // Matched row count + 0L // Mismatch row count + ))); + + GCSSpannerDVTestAsserts.assertMismatchedRecords( + bigQueryResourceManager, Collections.emptyList()); + } +} diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitorTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitorTest.java index 7f4bf66e1d..7002375dde 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitorTest.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitorTest.java @@ -64,6 +64,39 @@ public void testDispatchMatchesFloat64() { verify(visitor).visitFloat64(input); } + @Test + public void testDispatchMatchesFloat32() { + IUnifiedVisitor visitor = mock(IUnifiedVisitor.class); + float input = 123.456f; + Value value = Value.float32(input); + + IUnifiedVisitor.dispatch(value, visitor); + + verify(visitor).visitFloat32(input); + } + + @Test + public void testDispatchMatchesPgNumeric() { + IUnifiedVisitor visitor = mock(IUnifiedVisitor.class); + BigDecimal input = new BigDecimal("123.456"); + Value value = Value.pgNumeric(input.toString()); + + IUnifiedVisitor.dispatch(value, visitor); + + verify(visitor).visitNumeric(input); + } + + @Test + public void testDispatchMatchesPgJsonb() { + IUnifiedVisitor visitor = mock(IUnifiedVisitor.class); + String input = "{\"key\": \"value\"}"; + Value value = Value.pgJsonb(input); + + IUnifiedVisitor.dispatch(value, visitor); + + verify(visitor).visitJson(input); + } + @Test public void testDispatchMatchesBool() { IUnifiedVisitor visitor = mock(IUnifiedVisitor.class); diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitorTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitorTest.java index a5ac4fd48a..0e4ff4abe0 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitorTest.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedHasherVisitorTest.java @@ -103,6 +103,21 @@ public void testVisitFloat64() { assertEquals(expectedHash, actualHash); } + @Test + public void testVisitFloat32() { + Hasher hasher = Hashing.murmur3_128().newHasher(); + UnifiedHasherVisitor visitor = new UnifiedHasherVisitor(hasher); + float input = 123.456f; + + visitor.visitFloat32(input); + HashCode actualHash = hasher.hash(); + + HashCode expectedHash = + Hashing.murmur3_128().newHasher().putByte((byte) 1).putFloat(input).hash(); + + assertEquals(expectedHash, actualHash); + } + @Test public void testVisitBool() { Hasher hasher = Hashing.murmur3_128().newHasher(); diff --git a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc new file mode 100644 index 0000000000..cfcf3a1ac4 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc @@ -0,0 +1,86 @@ +{ + "type": "record", + "name": "SourceRowWithMetadata", + "fields": [ + { "name": "tableName", "type": "string" }, + { "name": "shardId", "type": [ "null", "string" ], "default": null }, + { "name": "primaryKeys", "type": { "type": "array", "items": "string" } }, + { + "name": "payload", + "type": { + "type": "record", + "name": "OracleAllDatatypesPayload", + "fields": [ + { "name": "id", "type": "long" }, + { "name": "varchar2_col", "type": [ "null", "string" ], "default": null }, + { "name": "varchar_col", "type": [ "null", "string" ], "default": null }, + { "name": "char_col", "type": [ "null", "string" ], "default": null }, + { "name": "character_col", "type": [ "null", "string" ], "default": null }, + { "name": "nvarchar2_col", "type": [ "null", "string" ], "default": null }, + { "name": "nchar_col", "type": [ "null", "string" ], "default": null }, + { + "name": "number_col", + "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], + "default": null + }, + { + "name": "numeric_col", + "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], + "default": null + }, + { + "name": "decimal_col", + "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], + "default": null + }, + { + "name": "dec_col", + "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], + "default": null + }, + { "name": "float_col", "type": [ "null", "double" ], "default": null }, + { "name": "double_precision_col", "type": [ "null", "double" ], "default": null }, + { "name": "real_col", "type": [ "null", "double" ], "default": null }, + { "name": "binary_float_col", "type": [ "null", "float" ], "default": null }, + { "name": "binary_double_col", "type": [ "null", "double" ], "default": null }, + { "name": "integer_col", "type": [ "null", "long" ], "default": null }, + { "name": "int_col", "type": [ "null", "long" ], "default": null }, + { "name": "smallint_col", "type": [ "null", "long" ], "default": null }, + { + "name": "date_col", + "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], + "default": null + }, + { + "name": "timestamp_col", + "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], + "default": null + }, + { + "name": "timestamp_tz_col", + "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], + "default": null + }, + { + "name": "timestamp_ltz_col", + "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], + "default": null + }, + { "name": "interval_ym_col", "type": [ "null", "string" ], "default": null }, + { "name": "interval_ds_col", "type": [ "null", "string" ], "default": null }, + { "name": "raw_col", "type": [ "null", "bytes" ], "default": null }, + { "name": "blob_col", "type": [ "null", "bytes" ], "default": null }, + { "name": "clob_col", "type": [ "null", "string" ], "default": null }, + { "name": "nclob_col", "type": [ "null", "string" ], "default": null }, + { "name": "rowid_col", "type": [ "null", "string" ], "default": null }, + { + "name": "json_col", + "type": [ "null", { "type": "string", "logicalType": "json" } ], + "default": null + }, + { "name": "xmltype_col", "type": [ "null", "string" ], "default": null } + ] + } + } + ] +} diff --git a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql new file mode 100644 index 0000000000..0bf2551f5c --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql @@ -0,0 +1,34 @@ +CREATE TABLE OracleAllDatatypes ( + id INT64 NOT NULL, + varchar2_col STRING(MAX), + varchar_col STRING(MAX), + char_col STRING(MAX), + character_col STRING(MAX), + nvarchar2_col STRING(MAX), + nchar_col STRING(MAX), + number_col NUMERIC, + numeric_col NUMERIC, + decimal_col NUMERIC, + dec_col NUMERIC, + float_col FLOAT64, + double_precision_col FLOAT64, + real_col FLOAT64, + binary_float_col FLOAT32, + binary_double_col FLOAT64, + integer_col INT64, + int_col INT64, + smallint_col INT64, + date_col TIMESTAMP, + timestamp_col TIMESTAMP, + timestamp_tz_col TIMESTAMP, + timestamp_ltz_col TIMESTAMP, + interval_ym_col STRING(MAX), + interval_ds_col STRING(MAX), + raw_col BYTES(MAX), + blob_col BYTES(MAX), + clob_col STRING(MAX), + nclob_col STRING(MAX), + rowid_col STRING(MAX), + json_col JSON, + xmltype_col STRING(MAX) +) PRIMARY KEY(id); From 6ae4981106a2e22a3aec153d781c028c3ad9e810 Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Fri, 28 Aug 2026 05:53:41 +0000 Subject: [PATCH 2/6] fixed pr issues --- .../cloud/teleport/v2/visitor/IUnifiedVisitor.java | 3 ++- .../ComparisonRecordMapperOracleAllDataTypesTest.java | 9 ++------- .../v2/templates/GCSSpannerDVAvroSetupHelper.java | 11 ++++++----- .../v2/templates/GCSSpannerDVOracleSmokeIT.java | 8 -------- 4 files changed, 10 insertions(+), 21 deletions(-) diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java index 33675c1c25..6165d08937 100644 --- a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/visitor/IUnifiedVisitor.java @@ -72,7 +72,8 @@ static void dispatch(Value value, IUnifiedVisitor visitor) { case DATE -> visitor.visitDate(value.getDate()); case NUMERIC, PG_NUMERIC -> visitor.visitNumeric(value.getNumeric()); case TIMESTAMP -> visitor.visitTimestamp(value.getTimestamp()); - case JSON, PG_JSONB -> visitor.visitJson(value.getJson()); + case JSON -> visitor.visitJson(value.getJson()); + case PG_JSONB -> visitor.visitJson(value.getPgJsonb()); default -> visitor.visitDefault(value); } } diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java index b8e3f6907c..e09abd4434 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java @@ -644,12 +644,8 @@ public void testSmokeSchemaAndRecordsDirectly() throws Exception { GCSSpannerDVAvroSetupHelper.TableDef tableDef = new GCSSpannerDVAvroSetupHelper.TableDef( - new Schema.Parser() - .parse( - ComparisonRecordMapperOracleAllDataTypesTest.class - .getClassLoader() - .getResourceAsStream( - "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc")), + GCSSpannerDVAvroSetupHelper.parseAvroSchema( + "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc"), "OracleAllDatatypes", Arrays.asList("id")); @@ -800,7 +796,6 @@ public void testSmokeSchemaAndRecordsDirectly() throws Exception { h1.hash().toString()); } - System.out.println("UNIT TEST COMPUTED HASH: " + avroResult.getHash()); assertEquals(spannerResult.getHash(), avroResult.getHash()); } } diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java index 3fb6b3e1c8..1312c26bb7 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java @@ -113,12 +113,13 @@ public GenericRecord build() { * Converts standard Java types into the primitive formats required by Avro logical types. * *

Because this helper uses a generic {@code Map} to dynamically build records, - * standard Java objects (like {@link Instant}) must be manually translated into Avro's expected - * underlying primitives (like {@code Long} for timestamp-micros) before serialization. + * standard Java objects (like {@link Instant}, {@link java.math.BigDecimal}, {@code byte[]}) must + * be manually translated into Avro's expected underlying primitives (like {@code Long} for + * timestamp-micros, {@link java.nio.ByteBuffer} for decimals with scale 9 or raw bytes) before + * serialization. * - *

IMPORTANT: This method currently only supports {@link Instant}. If new test tables - * are introduced that use other complex Avro mappings (e.g., Dates, Decimals, UUIDs, or custom - * Datastream composites like Datetime), this method MUST be updated to coerce those types. + *

NOTE: {@link java.math.BigDecimal} is currently normalized to scale 9 for Avro + * decimal serialization. * * @param value The standard Java object provided in the test map. * @return The Avro-compatible primitive value ready for serialization. diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java index e465e6de25..7016c5ac8e 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java @@ -204,14 +204,6 @@ public void testOracleAllDataTypesValidationSmoke() throws Exception { pipelineOperator().waitUntilDone(createConfig(jobInfo)); - List> mismatches = - org.apache.beam.it.gcp.bigquery.matchers.BigQueryAsserts.tableResultToRecords( - bigQueryResourceManager.readTable("MismatchedRecords")); - System.out.println("DEBUG MISMATCHED RECORDS COUNT: " + mismatches.size()); - for (java.util.Map m : mismatches) { - System.out.println("DEBUG MISMATCH: " + m); - } - // 4. Assert Validation Results in BigQuery GCSSpannerDVTestAsserts.assertValidationSummary( bigQueryResourceManager, From d0a17eecc06d122f3cd240155ec19418dd9d5af5 Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Fri, 28 Aug 2026 06:44:21 +0000 Subject: [PATCH 3/6] added check for json handling --- .../spanner/migrations/avro/GenericRecordTypeConvertor.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertor.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertor.java index 3b82c11394..88d2cf1d92 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertor.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertor.java @@ -647,7 +647,9 @@ static String handleLogicalFieldType( return timestamp.atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_OFFSET_DATE_TIME); } else if (fieldSchema.getProp(LOGICAL_TYPE) != null && fieldSchema.getProp(LOGICAL_TYPE).equals(CustomAvroTypes.JSON)) { - if (cassandraAnnotations.cassandraType().getKind().equals(Kind.MAP)) { + if (cassandraAnnotations != null + && cassandraAnnotations.cassandraType() != null + && cassandraAnnotations.cassandraType().getKind().equals(Kind.MAP)) { return AvroJsonToCassandraMapConvertor.handleJsonToMap( recordValue, cassandraAnnotations, fieldName, fieldSchema); } else { From 4cddf1864ea98606e0a29fd3928c66e5b16963fd Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Fri, 28 Aug 2026 08:14:08 +0000 Subject: [PATCH 4/6] standardized the flow for oracle tests similar to other sources --- ...sonRecordMapperOracleAllDataTypesTest.java | 801 ------------------ .../GCSSpannerDVAvroSetupHelper.java | 22 +- .../v2/templates/GCSSpannerDVITBase.java | 15 - .../templates/GCSSpannerDVOracleSmokeIT.java | 235 ----- .../v2/templates/GCSSpannerDVSchemaIT.java | 3 +- .../GCSSpannerDVWideRowMax10MibCellIT.java | 4 +- .../GCSSpannerDVWideRowMax16KeyTableIT.java | 2 +- .../GCSSpannerDVWideRowMaxColumnNameIT.java | 4 +- ...nAndValidationOracleAllDataTypesE2EIT.java | 165 ++++ .../endtoend/EndToEndTestingITBase.java | 5 + .../oracle-schema.sql | 111 +++ .../spanner-schema.sql | 30 + .../oracle_all_datatypes.avsc | 86 -- .../spanner-schema.sql | 34 - 14 files changed, 333 insertions(+), 1184 deletions(-) delete mode 100644 v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java delete mode 100644 v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java create mode 100644 v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/BulkMigrationAndValidationOracleAllDataTypesE2EIT.java create mode 100644 v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/oracle-schema.sql create mode 100644 v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/spanner-schema.sql delete mode 100644 v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc delete mode 100644 v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java deleted file mode 100644 index e09abd4434..0000000000 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/mapper/ComparisonRecordMapperOracleAllDataTypesTest.java +++ /dev/null @@ -1,801 +0,0 @@ -/* - * Copyright (C) 2026 Google LLC - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not - * use this file except in compliance with the License. You may obtain a copy of - * the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT - * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the - * License for the specific language governing permissions and limitations under - * the License. - */ -package com.google.cloud.teleport.v2.mapper; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import com.google.cloud.ByteArray; -import com.google.cloud.Timestamp; -import com.google.cloud.spanner.Dialect; -import com.google.cloud.spanner.Struct; -import com.google.cloud.spanner.Value; -import com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants; -import com.google.cloud.teleport.v2.dto.ComparisonRecord; -import com.google.cloud.teleport.v2.spanner.ddl.Column; -import com.google.cloud.teleport.v2.spanner.ddl.Ddl; -import com.google.cloud.teleport.v2.spanner.ddl.IndexColumn; -import com.google.cloud.teleport.v2.spanner.ddl.Table; -import com.google.cloud.teleport.v2.spanner.migrations.avro.GenericRecordTypeConvertor; -import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper; -import com.google.cloud.teleport.v2.spanner.migrations.schema.IdentityMapper; -import com.google.cloud.teleport.v2.spanner.type.Type; -import com.google.cloud.teleport.v2.spanner.utils.ISpannerMigrationTransformer; -import com.google.cloud.teleport.v2.spanner.utils.MigrationTransformationResponse; -import com.google.cloud.teleport.v2.templates.GCSSpannerDVAvroSetupHelper; -import com.google.cloud.teleport.v2.visitor.IUnifiedVisitor; -import com.google.cloud.teleport.v2.visitor.UnifiedHasherVisitor; -import java.math.BigDecimal; -import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; -import org.apache.avro.LogicalTypes; -import org.apache.avro.Schema; -import org.apache.avro.SchemaBuilder; -import org.apache.avro.generic.GenericData; -import org.apache.avro.generic.GenericRecord; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; - -/** - * Unit test validating that all Oracle data types specified in the Oracle Datatype Mapping Matrix - * correctly convert from Avro GenericRecord and Spanner Struct into identical ComparisonRecord - * hashes. - */ -@RunWith(JUnit4.class) -public class ComparisonRecordMapperOracleAllDataTypesTest { - - private ISchemaMapper mockSchemaMapper; - private ISpannerMigrationTransformer mockTransformer; - private Ddl mockDdl; - private ComparisonRecordMapper mapper; - - @Before - public void setUp() { - mockSchemaMapper = mock(ISchemaMapper.class); - mockTransformer = mock(ISpannerMigrationTransformer.class); - mockDdl = mock(Ddl.class); - mapper = new ComparisonRecordMapper(mockSchemaMapper, mockTransformer, mockDdl); - } - - @Test - public void testAllOracleDataTypesHashParity() throws Exception { - String tableName = "AllDatatypes"; - - // 1. Define Avro Schemas for Oracle Datatypes - Schema decimalSchema = - LogicalTypes.decimal(10, 2).addToSchema(Schema.create(Schema.Type.BYTES)); - Schema timestampMicrosSchema = - LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)); - - Schema payloadSchema = - SchemaBuilder.record("Payload") - .fields() - .name("id") - .type(Schema.create(Schema.Type.LONG)) - .noDefault() - .name("varchar2_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("varchar_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("char_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("character_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("nvarchar2_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("nchar_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("number_col") - .type(decimalSchema) - .noDefault() - .name("numeric_col") - .type(decimalSchema) - .noDefault() - .name("decimal_col") - .type(decimalSchema) - .noDefault() - .name("dec_col") - .type(decimalSchema) - .noDefault() - .name("float_col") - .type(Schema.create(Schema.Type.DOUBLE)) - .noDefault() - .name("double_precision_col") - .type(Schema.create(Schema.Type.DOUBLE)) - .noDefault() - .name("real_col") - .type(Schema.create(Schema.Type.DOUBLE)) - .noDefault() - .name("binary_float_col") - .type(Schema.create(Schema.Type.FLOAT)) - .noDefault() - .name("binary_double_col") - .type(Schema.create(Schema.Type.DOUBLE)) - .noDefault() - .name("integer_col") - .type(Schema.create(Schema.Type.LONG)) - .noDefault() - .name("int_col") - .type(Schema.create(Schema.Type.LONG)) - .noDefault() - .name("smallint_col") - .type(Schema.create(Schema.Type.LONG)) - .noDefault() - .name("date_col") - .type(timestampMicrosSchema) - .noDefault() - .name("timestamp_col") - .type(timestampMicrosSchema) - .noDefault() - .name("timestamp_tz_col") - .type(timestampMicrosSchema) - .noDefault() - .name("timestamp_ltz_col") - .type(timestampMicrosSchema) - .noDefault() - .name("interval_ym_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("interval_ds_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("raw_col") - .type(Schema.create(Schema.Type.BYTES)) - .noDefault() - .name("blob_col") - .type(Schema.create(Schema.Type.BYTES)) - .noDefault() - .name("clob_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("nclob_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("rowid_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("json_col") - .type(SchemaBuilder.builder().stringBuilder().prop("logicalType", "json").endString()) - .noDefault() - .name("xmltype_col") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .endRecord(); - - Schema avroSchema = - SchemaBuilder.record("SourceRow") - .fields() - .name("tableName") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("shardId") - .type(Schema.create(Schema.Type.STRING)) - .noDefault() - .name("payload") - .type(payloadSchema) - .noDefault() - .endRecord(); - - // 2. Populate Avro payload - GenericRecord payload = new GenericData.Record(payloadSchema); - payload.put("id", 1L); - payload.put("varchar2_col", "test_varchar2"); - payload.put("varchar_col", "test_varchar"); - payload.put("char_col", "test_char "); - payload.put("character_col", "test_char "); - payload.put("nvarchar2_col", "test_nvarchar2"); - payload.put("nchar_col", "test_nchar"); - payload.put( - "number_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); - payload.put( - "numeric_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); - payload.put( - "decimal_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); - payload.put( - "dec_col", ByteBuffer.wrap(new BigDecimal("1234.56").unscaledValue().toByteArray())); - payload.put("float_col", 123.456d); - payload.put("double_precision_col", 123.456d); - payload.put("real_col", 123.456d); - payload.put("binary_float_col", 123.0f); - payload.put("binary_double_col", 123.0d); - payload.put("integer_col", 12345L); - payload.put("int_col", 12345L); - payload.put("smallint_col", 123L); - long timestampMicros = 1704103200000000L; - payload.put("date_col", timestampMicros); - payload.put("timestamp_col", timestampMicros); - payload.put("timestamp_tz_col", timestampMicros); - payload.put("timestamp_ltz_col", timestampMicros); - payload.put("interval_ym_col", "P1Y2M"); - payload.put("interval_ds_col", "PT3H4M5S"); - payload.put("raw_col", ByteBuffer.wrap(new byte[] {0x41, 0x42, 0x43})); - payload.put("blob_col", ByteBuffer.wrap(new byte[] {0x41, 0x42, 0x43, 0x44})); - payload.put("clob_col", "test_clob_content"); - payload.put("nclob_col", "test_nclob_content"); - payload.put("rowid_col", "AAAB12AADAAAAwPAAA"); - payload.put("json_col", "{\"k1\":\"v1\"}"); - payload.put("xmltype_col", "test"); - - GenericRecord avroRecord = new GenericData.Record(avroSchema); - avroRecord.put("tableName", tableName); - avroRecord.put("shardId", "shard1"); - avroRecord.put("payload", payload); - - // 3. Configure Schema Mapper mocks - List columnNames = - Arrays.asList( - "id", - "varchar2_col", - "varchar_col", - "char_col", - "character_col", - "nvarchar2_col", - "nchar_col", - "number_col", - "numeric_col", - "decimal_col", - "dec_col", - "float_col", - "double_precision_col", - "real_col", - "binary_float_col", - "binary_double_col", - "integer_col", - "int_col", - "smallint_col", - "date_col", - "timestamp_col", - "timestamp_tz_col", - "timestamp_ltz_col", - "interval_ym_col", - "interval_ds_col", - "raw_col", - "blob_col", - "clob_col", - "nclob_col", - "rowid_col", - "json_col", - "xmltype_col"); - - when(mockSchemaMapper.getSpannerTableName(anyString(), anyString())).thenReturn(tableName); - when(mockSchemaMapper.getSpannerColumnName(anyString(), anyString(), anyString())) - .thenAnswer(invocation -> invocation.getArgument(2)); - when(mockSchemaMapper.getSourceColumnName(anyString(), anyString(), anyString())) - .thenAnswer(invocation -> invocation.getArgument(2)); - when(mockSchemaMapper.getDialect()).thenReturn(Dialect.GOOGLE_STANDARD_SQL); - when(mockSchemaMapper.getSpannerColumns(anyString(), anyString())).thenReturn(columnNames); - when(mockSchemaMapper.colExistsAtSource(anyString(), anyString(), anyString())) - .thenReturn(true); - - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("id"))) - .thenReturn(Type.int64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("varchar2_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("varchar_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("char_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("character_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nvarchar2_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nchar_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("number_col"))) - .thenReturn(Type.numeric()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("numeric_col"))) - .thenReturn(Type.numeric()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("decimal_col"))) - .thenReturn(Type.numeric()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("dec_col"))) - .thenReturn(Type.numeric()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("float_col"))) - .thenReturn(Type.float64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("double_precision_col"))) - .thenReturn(Type.float64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("real_col"))) - .thenReturn(Type.float64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("binary_float_col"))) - .thenReturn(Type.float32()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("binary_double_col"))) - .thenReturn(Type.float64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("integer_col"))) - .thenReturn(Type.int64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("int_col"))) - .thenReturn(Type.int64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("smallint_col"))) - .thenReturn(Type.int64()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("date_col"))) - .thenReturn(Type.timestamp()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_col"))) - .thenReturn(Type.timestamp()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_tz_col"))) - .thenReturn(Type.timestamp()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("timestamp_ltz_col"))) - .thenReturn(Type.timestamp()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("interval_ym_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("interval_ds_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("raw_col"))) - .thenReturn(Type.bytes()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("blob_col"))) - .thenReturn(Type.bytes()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("clob_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("nclob_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("rowid_col"))) - .thenReturn(Type.string()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("json_col"))) - .thenReturn(Type.json()); - when(mockSchemaMapper.getSpannerColumnType( - anyString(), anyString(), org.mockito.ArgumentMatchers.eq("xmltype_col"))) - .thenReturn(Type.string()); - - Table mockTable = mock(Table.class); - when(mockDdl.table(tableName)).thenReturn(mockTable); - IndexColumn pkCol = IndexColumn.create("id", IndexColumn.Order.ASC); - when(mockTable.primaryKeys()).thenReturn(com.google.common.collect.ImmutableList.of(pkCol)); - - MigrationTransformationResponse mockResponse = mock(MigrationTransformationResponse.class); - when(mockResponse.isEventFiltered()).thenReturn(false); - when(mockResponse.getResponseRow()).thenReturn(Collections.emptyMap()); - when(mockTransformer.toSpannerRow(org.mockito.ArgumentMatchers.any())).thenReturn(mockResponse); - - // 4. Map Avro Record to ComparisonRecord - ComparisonRecord avroRecordResult = mapper.mapFrom(avroRecord); - assertNotNull(avroRecordResult); - assertEquals(tableName, avroRecordResult.getTableName()); - assertEquals("shard1", avroRecordResult.getShardId()); - - // 5. Build identical Spanner Struct - Timestamp spannerTimestamp = Timestamp.ofTimeMicroseconds(timestampMicros); - Struct spannerStruct = - Struct.newBuilder() - .set(GCSSpannerDVConstants.TABLE_NAME_COLUMN) - .to(tableName) - .set("id") - .to(1L) - .set("varchar2_col") - .to("test_varchar2") - .set("varchar_col") - .to("test_varchar") - .set("char_col") - .to("test_char ") - .set("character_col") - .to("test_char ") - .set("nvarchar2_col") - .to("test_nvarchar2") - .set("nchar_col") - .to("test_nchar") - .set("number_col") - .to(new BigDecimal("1234.560000000")) - .set("numeric_col") - .to(new BigDecimal("1234.560000000")) - .set("decimal_col") - .to(new BigDecimal("1234.560000000")) - .set("dec_col") - .to(new BigDecimal("1234.560000000")) - .set("float_col") - .to(123.456d) - .set("double_precision_col") - .to(123.456d) - .set("real_col") - .to(123.456d) - .set("binary_float_col") - .to(123.0f) - .set("binary_double_col") - .to(123.0d) - .set("integer_col") - .to(12345L) - .set("int_col") - .to(12345L) - .set("smallint_col") - .to(123L) - .set("date_col") - .to(spannerTimestamp) - .set("timestamp_col") - .to(spannerTimestamp) - .set("timestamp_tz_col") - .to(spannerTimestamp) - .set("timestamp_ltz_col") - .to(spannerTimestamp) - .set("interval_ym_col") - .to("P1Y2M") - .set("interval_ds_col") - .to("PT3H4M5S") - .set("raw_col") - .to(ByteArray.copyFrom(new byte[] {0x41, 0x42, 0x43})) - .set("blob_col") - .to(ByteArray.copyFrom(new byte[] {0x41, 0x42, 0x43, 0x44})) - .set("clob_col") - .to("test_clob_content") - .set("nclob_col") - .to("test_nclob_content") - .set("rowid_col") - .to("AAAB12AADAAAAwPAAA") - .set("json_col") - .to(Value.json("{\"k1\":\"v1\"}")) - .set("xmltype_col") - .to("test") - .build(); - - // 6. Map Spanner Struct to ComparisonRecord - ComparisonRecord spannerRecordResult = mapper.mapFrom(spannerStruct); - assertNotNull(spannerRecordResult); - - GenericRecordTypeConvertor convertor = - new GenericRecordTypeConvertor(mockSchemaMapper, "", "shard1", mockTransformer); - java.util.Map avroValues = convertor.transformChangeEvent(payload, tableName); - - for (String col : columnNames) { - Value avroVal = avroValues.get(col); - Value spannerVal = spannerStruct.getValue(col); - - com.google.common.hash.Hasher h1 = com.google.common.hash.Hashing.murmur3_128().newHasher(); - UnifiedHasherVisitor v1 = new UnifiedHasherVisitor(h1); - IUnifiedVisitor.dispatch(avroVal, v1); - - com.google.common.hash.Hasher h2 = com.google.common.hash.Hashing.murmur3_128().newHasher(); - UnifiedHasherVisitor v2 = new UnifiedHasherVisitor(h2); - IUnifiedVisitor.dispatch(spannerVal, v2); - - org.junit.Assert.assertEquals( - "Hash mismatch for column " - + col - + " (avroVal: " - + avroVal - + " vs spannerVal: " - + spannerVal - + ")", - h2.hash().toString(), - h1.hash().toString()); - } - - // 7. Verify Hashes Match Exactly! - assertEquals( - "Avro hash and Spanner Struct hash must match for all Oracle datatypes", - spannerRecordResult.getHash(), - avroRecordResult.getHash()); - } - - @Test - public void testSmokeSchemaAndRecordsDirectly() throws Exception { - Ddl ddl = - Ddl.builder(Dialect.GOOGLE_STANDARD_SQL) - .createTable("OracleAllDatatypes") - .column("id") - .int64() - .notNull() - .endColumn() - .column("varchar2_col") - .string() - .max() - .endColumn() - .column("varchar_col") - .string() - .max() - .endColumn() - .column("char_col") - .string() - .max() - .endColumn() - .column("character_col") - .string() - .max() - .endColumn() - .column("nvarchar2_col") - .string() - .max() - .endColumn() - .column("nchar_col") - .string() - .max() - .endColumn() - .column("number_col") - .numeric() - .endColumn() - .column("numeric_col") - .numeric() - .endColumn() - .column("decimal_col") - .numeric() - .endColumn() - .column("dec_col") - .numeric() - .endColumn() - .column("float_col") - .float64() - .endColumn() - .column("double_precision_col") - .float64() - .endColumn() - .column("real_col") - .float64() - .endColumn() - .column("binary_float_col") - .float32() - .endColumn() - .column("binary_double_col") - .float64() - .endColumn() - .column("integer_col") - .int64() - .endColumn() - .column("int_col") - .int64() - .endColumn() - .column("smallint_col") - .int64() - .endColumn() - .column("date_col") - .timestamp() - .endColumn() - .column("timestamp_col") - .timestamp() - .endColumn() - .column("timestamp_tz_col") - .timestamp() - .endColumn() - .column("timestamp_ltz_col") - .timestamp() - .endColumn() - .column("interval_ym_col") - .string() - .max() - .endColumn() - .column("interval_ds_col") - .string() - .max() - .endColumn() - .column("raw_col") - .bytes() - .max() - .endColumn() - .column("blob_col") - .bytes() - .max() - .endColumn() - .column("clob_col") - .string() - .max() - .endColumn() - .column("nclob_col") - .string() - .max() - .endColumn() - .column("rowid_col") - .string() - .max() - .endColumn() - .column("json_col") - .json() - .endColumn() - .column("xmltype_col") - .string() - .max() - .endColumn() - .primaryKey() - .asc("id") - .end() - .endTable() - .build(); - - IdentityMapper identityMapper = new IdentityMapper(ddl); - ComparisonRecordMapper smokeMapper = new ComparisonRecordMapper(identityMapper, null, ddl); - - GCSSpannerDVAvroSetupHelper.TableDef tableDef = - new GCSSpannerDVAvroSetupHelper.TableDef( - GCSSpannerDVAvroSetupHelper.parseAvroSchema( - "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc"), - "OracleAllDatatypes", - Arrays.asList("id")); - - java.time.Instant testTimestamp = java.time.Instant.parse("2024-01-01T10:00:00Z"); - BigDecimal testNumeric = new BigDecimal("1234.560000000"); - byte[] testBytes = new byte[] {0x41, 0x42, 0x43, 0x44}; - - GenericRecord avroRecord = - new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null) - .set("id", 1L) - .set("varchar2_col", "test_varchar2") - .set("varchar_col", "test_varchar") - .set("char_col", "test_char ") - .set("character_col", "test_char ") - .set("nvarchar2_col", "test_nvarchar2") - .set("nchar_col", "test_nchar ") - .set("number_col", testNumeric) - .set("numeric_col", testNumeric) - .set("decimal_col", testNumeric) - .set("dec_col", testNumeric) - .set("float_col", 123.456) - .set("double_precision_col", 123.456) - .set("real_col", 123.456) - .set("binary_float_col", 123.0f) - .set("binary_double_col", 123.0) - .set("integer_col", 12345L) - .set("int_col", 12345L) - .set("smallint_col", 123L) - .set("date_col", testTimestamp) - .set("timestamp_col", testTimestamp) - .set("timestamp_tz_col", testTimestamp) - .set("timestamp_ltz_col", testTimestamp) - .set("interval_ym_col", "P1Y2M") - .set("interval_ds_col", "PT3H4M5S") - .set("raw_col", testBytes) - .set("blob_col", testBytes) - .set("clob_col", "test_clob_content") - .set("nclob_col", "test_nclob_content") - .set("rowid_col", "AAAB12AADAAAAwPAAA") - .set("json_col", "{}") - .set("xmltype_col", "test") - .build(); - - Struct spannerStruct = - Struct.newBuilder() - .set(GCSSpannerDVConstants.TABLE_NAME_COLUMN) - .to("OracleAllDatatypes") - .set("id") - .to(1L) - .set("varchar2_col") - .to("test_varchar2") - .set("varchar_col") - .to("test_varchar") - .set("char_col") - .to("test_char ") - .set("character_col") - .to("test_char ") - .set("nvarchar2_col") - .to("test_nvarchar2") - .set("nchar_col") - .to("test_nchar ") - .set("number_col") - .to(testNumeric) - .set("numeric_col") - .to(testNumeric) - .set("decimal_col") - .to(testNumeric) - .set("dec_col") - .to(testNumeric) - .set("float_col") - .to(123.456) - .set("double_precision_col") - .to(123.456) - .set("real_col") - .to(123.456) - .set("binary_float_col") - .to(123.0f) - .set("binary_double_col") - .to(123.0) - .set("integer_col") - .to(12345L) - .set("int_col") - .to(12345L) - .set("smallint_col") - .to(123L) - .set("date_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_tz_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_ltz_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("interval_ym_col") - .to("P1Y2M") - .set("interval_ds_col") - .to("PT3H4M5S") - .set("raw_col") - .to(ByteArray.copyFrom(testBytes)) - .set("blob_col") - .to(ByteArray.copyFrom(testBytes)) - .set("clob_col") - .to("test_clob_content") - .set("nclob_col") - .to("test_nclob_content") - .set("rowid_col") - .to("AAAB12AADAAAAwPAAA") - .set("json_col") - .to(Value.json("{}")) - .set("xmltype_col") - .to("test") - .build(); - - ComparisonRecord avroResult = smokeMapper.mapFrom(avroRecord); - ComparisonRecord spannerResult = smokeMapper.mapFrom(spannerStruct); - - assertNotNull(avroResult); - assertNotNull(spannerResult); - - GenericRecord payload = (GenericRecord) avroRecord.get("payload"); - GenericRecordTypeConvertor convertor = - new GenericRecordTypeConvertor(identityMapper, "", null, null); - java.util.Map avroValues = - convertor.transformChangeEvent(payload, "OracleAllDatatypes"); - - for (String col : - ddl.table("OracleAllDatatypes").columns().stream().map(Column::name).toList()) { - Value avroVal = avroValues.get(col); - Value spannerVal = spannerStruct.getValue(col); - - com.google.common.hash.Hasher h1 = com.google.common.hash.Hashing.murmur3_128().newHasher(); - UnifiedHasherVisitor v1 = new UnifiedHasherVisitor(h1); - IUnifiedVisitor.dispatch(avroVal, v1); - - com.google.common.hash.Hasher h2 = com.google.common.hash.Hashing.murmur3_128().newHasher(); - UnifiedHasherVisitor v2 = new UnifiedHasherVisitor(h2); - IUnifiedVisitor.dispatch(spannerVal, v2); - - org.junit.Assert.assertEquals( - "Hash mismatch for column " - + col - + " (avroVal: " - + avroVal - + " vs spannerVal: " - + spannerVal - + ")", - h2.hash().toString(), - h1.hash().toString()); - } - - assertEquals(spannerResult.getHash(), avroResult.getHash()); - } -} diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java index 1312c26bb7..b740009796 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVAvroSetupHelper.java @@ -15,6 +15,8 @@ */ package com.google.cloud.teleport.v2.templates; +import com.google.common.io.Resources; +import java.io.InputStream; import java.time.Instant; import java.util.Arrays; import java.util.List; @@ -29,15 +31,17 @@ */ public class GCSSpannerDVAvroSetupHelper { - public static Schema parseAvroSchema(String resourceName) { - try (java.io.InputStream is = - GCSSpannerDVAvroSetupHelper.class.getClassLoader().getResourceAsStream(resourceName)) { - if (is == null) { - throw new IllegalArgumentException("Resource not found: " + resourceName); - } + /** + * Helper function to load an Avro Schema from a resource file. + * + * @param resourceName The path to the Avro schema file relative to the resources directory + * @return The parsed Avro Schema + */ + public static Schema getSchemaFromAvscFile(String resourceName) { + try (InputStream is = Resources.getResource(resourceName).openStream()) { return new Schema.Parser().parse(is); } catch (Exception e) { - throw new RuntimeException("Failed to parse Avro schema from resource: " + resourceName, e); + throw new RuntimeException("Failed to load Avro schema from resource: " + resourceName, e); } } @@ -48,12 +52,12 @@ public static Schema parseAvroSchema(String resourceName) { public static class TableDef { public static final TableDef USERS = new TableDef( - parseAvroSchema("GCSSpannerDVAvroSetupHelper/users.avsc"), + getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/users.avsc"), "Users", Arrays.asList("user_id", "event_id")); public static final TableDef ACCOUNT_ROLES = new TableDef( - parseAvroSchema("GCSSpannerDVAvroSetupHelper/account_roles.avsc"), + getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/account_roles.avsc"), "AccountRoles", Arrays.asList("role_id")); diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVITBase.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVITBase.java index 51f0ec1bc3..75112f5c85 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVITBase.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVITBase.java @@ -21,7 +21,6 @@ import com.google.common.io.Resources; import java.io.File; import java.io.IOException; -import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.HashMap; @@ -206,20 +205,6 @@ public void uploadCustomShardJarToGcs(String gcsPathPrefix) throws IOException { "../spanner-custom-shard/target/spanner-custom-shard-1.0-SNAPSHOT.jar"); } - /** - * Helper function to load an Avro Schema from a resource file. - * - * @param resourceName The path to the Avro schema file relative to the resources directory - * @return The parsed Avro Schema - */ - public static Schema getSchemaFromAvscFile(String resourceName) { - try (InputStream is = Resources.getResource(resourceName).openStream()) { - return new Schema.Parser().parse(is); - } catch (Exception e) { - throw new RuntimeException("Failed to load Avro schema from resource: " + resourceName, e); - } - } - /** * Serializes a list of GenericRecords into a binary .avro file and uploads it to GCS. * diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java deleted file mode 100644 index 7016c5ac8e..0000000000 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVOracleSmokeIT.java +++ /dev/null @@ -1,235 +0,0 @@ -/* - * Copyright (C) 2026 Google LLC - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not - * use this file except in compliance with the License. You may obtain a copy of - * the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT - * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the - * License for the specific language governing permissions and limitations under - * the License. - */ -package com.google.cloud.teleport.v2.templates; - -import com.google.cloud.ByteArray; -import com.google.cloud.Timestamp; -import com.google.cloud.spanner.Mutation; -import com.google.cloud.spanner.Value; -import com.google.cloud.teleport.metadata.DirectRunnerTest; -import com.google.cloud.teleport.metadata.TemplateIntegrationTest; -import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.TableValidationStatsDto; -import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.ValidationSummaryDto; -import java.io.IOException; -import java.math.BigDecimal; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; -import org.apache.avro.generic.GenericRecord; -import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; -import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; -import org.junit.Before; -import org.junit.Test; -import org.junit.experimental.categories.Category; -import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; - -/** Integration smoke test for GCSSpannerDV validating all Oracle data types. */ -@Category({TemplateIntegrationTest.class, DirectRunnerTest.class}) -@RunWith(JUnit4.class) -@TemplateIntegrationTest(GCSSpannerDV.class) -public class GCSSpannerDVOracleSmokeIT extends GCSSpannerDVITBase { - - private static final String SPANNER_DDL_RESOURCE = "GCSSpannerDVOracleSmokeIT/spanner-schema.sql"; - private static final String AVRO_SCHEMA_RESOURCE = - "GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc"; - - @Before - public void setUp() throws IOException { - spannerResourceManager = setUpSpannerResourceManager(); - bigQueryResourceManager = setUpBigQueryResourceManager(); - bigQueryResourceManager.createDataset(REGION); - createSpannerDDL(spannerResourceManager, SPANNER_DDL_RESOURCE); - } - - @Test - public void testOracleAllDataTypesValidationSmoke() throws Exception { - GCSSpannerDVAvroSetupHelper.TableDef tableDef = - new GCSSpannerDVAvroSetupHelper.TableDef( - getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), "OracleAllDatatypes", Arrays.asList("id")); - - java.time.Instant testTimestamp = java.time.Instant.parse("2024-01-01T10:00:00Z"); - BigDecimal testNumeric = new BigDecimal("1234.567890123"); - byte[] testBytes = new byte[] {0x41, 0x42, 0x43, 0x44}; - - // 1. Generate Avro Source Records - List records = - Arrays.asList( - // Row 1: Standard Row with all types populated - new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null) - .set("id", 1L) - .set("varchar2_col", "test_varchar2") - .set("varchar_col", "test_varchar") - .set("char_col", "test_char ") - .set("character_col", "test_char ") - .set("nvarchar2_col", "test_nvarchar2") - .set("nchar_col", "test_nchar ") - .set("number_col", testNumeric) - .set("numeric_col", testNumeric) - .set("decimal_col", testNumeric) - .set("dec_col", testNumeric) - .set("float_col", 123.456) - .set("double_precision_col", 123.456) - .set("real_col", 123.456) - .set("binary_float_col", 123.0f) - .set("binary_double_col", 123.0) - .set("integer_col", 12345L) - .set("int_col", 12345L) - .set("smallint_col", 123L) - .set("date_col", testTimestamp) - .set("timestamp_col", testTimestamp) - .set("timestamp_tz_col", testTimestamp) - .set("timestamp_ltz_col", testTimestamp) - .set("interval_ym_col", "P1Y2M") - .set("interval_ds_col", "PT3H4M5S") - .set("raw_col", testBytes) - .set("blob_col", testBytes) - .set("clob_col", "test_clob_content") - .set("nclob_col", "test_nclob_content") - .set("rowid_col", "AAAB12AADAAAAwPAAA") - .set("json_col", "{}") - .set("xmltype_col", "test") - .build(), - // Row 2: Null Row - new GCSSpannerDVAvroSetupHelper.RecordBuilder(tableDef, null).set("id", 2L).build()); - - String gcsInputDirectory = getGcsPath("input"); - uploadAvroFileToGcs("input/oracle_all_datatypes.avro", tableDef.schema, records); - - // 2. Insert Matching Records in Destination (Spanner) - spannerResourceManager.write( - Arrays.asList( - // Row 1 - Mutation.newInsertOrUpdateBuilder("OracleAllDatatypes") - .set("id") - .to(1L) - .set("varchar2_col") - .to("test_varchar2") - .set("varchar_col") - .to("test_varchar") - .set("char_col") - .to("test_char ") - .set("character_col") - .to("test_char ") - .set("nvarchar2_col") - .to("test_nvarchar2") - .set("nchar_col") - .to("test_nchar ") - .set("number_col") - .to(testNumeric) - .set("numeric_col") - .to(testNumeric) - .set("decimal_col") - .to(testNumeric) - .set("dec_col") - .to(testNumeric) - .set("float_col") - .to(123.456) - .set("double_precision_col") - .to(123.456) - .set("real_col") - .to(123.456) - .set("binary_float_col") - .to(123.0f) - .set("binary_double_col") - .to(123.0) - .set("integer_col") - .to(12345L) - .set("int_col") - .to(12345L) - .set("smallint_col") - .to(123L) - .set("date_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_tz_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("timestamp_ltz_col") - .to(Timestamp.parseTimestamp("2024-01-01T10:00:00Z")) - .set("interval_ym_col") - .to("P1Y2M") - .set("interval_ds_col") - .to("PT3H4M5S") - .set("raw_col") - .to(ByteArray.copyFrom(testBytes)) - .set("blob_col") - .to(ByteArray.copyFrom(testBytes)) - .set("clob_col") - .to("test_clob_content") - .set("nclob_col") - .to("test_nclob_content") - .set("rowid_col") - .to("AAAB12AADAAAAwPAAA") - .set("json_col") - .to(Value.json("{}")) - .set("xmltype_col") - .to("test") - .build(), - // Row 2 - Mutation.newInsertOrUpdateBuilder("OracleAllDatatypes").set("id").to(2L).build())); - - // Wait for Spanner's 20-second exact staleness read bound in SpannerReaderTransform - Thread.sleep(20000); - - // 3. Launch Pipeline - LaunchConfig.Builder options = LaunchConfig.builder(testName, specPath); - LaunchInfo jobInfo = - launchDataflowJob( - options, - testName, - PROJECT, - spannerResourceManager, - bigQueryResourceManager.getDatasetId(), - gcsInputDirectory, - null, - null, - null, - null, - null, - null); - - pipelineOperator().waitUntilDone(createConfig(jobInfo)); - - // 4. Assert Validation Results in BigQuery - GCSSpannerDVTestAsserts.assertValidationSummary( - bigQueryResourceManager, - Collections.singletonList( - new ValidationSummaryDto( - "MATCH", - 1L, // Total tables validated - 2L, // Total rows matched - 0L, // Total rows mismatched - "" // Tables with mismatches - ))); - - GCSSpannerDVTestAsserts.assertTableValidationStats( - bigQueryResourceManager, - Collections.singletonList( - new TableValidationStatsDto( - null, // Schema name - "OracleAllDatatypes", // Table name - "MATCH", // Status - 2L, // Source row count - 2L, // Destination row count - 2L, // Matched row count - 0L // Mismatch row count - ))); - - GCSSpannerDVTestAsserts.assertMismatchedRecords( - bigQueryResourceManager, Collections.emptyList()); - } -} diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVSchemaIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVSchemaIT.java index ea0f6d5ac5..bc1906be57 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVSchemaIT.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVSchemaIT.java @@ -104,7 +104,8 @@ public void validationTestWithEmptyTables() throws Exception { public void validationTestWithReservedKeywords() throws Exception { Schema reservedKeywordsSchema = - getSchemaFromAvscFile("GCSSpannerDVSchemaIT/reserved_keywords.avsc"); + GCSSpannerDVAvroSetupHelper.getSchemaFromAvscFile( + "GCSSpannerDVSchemaIT/reserved_keywords.avsc"); GCSSpannerDVAvroSetupHelper.TableDef reservedTableDef = new GCSSpannerDVAvroSetupHelper.TableDef( reservedKeywordsSchema, "ORDER", Arrays.asList("SELECT")); diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax10MibCellIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax10MibCellIT.java index a92f545981..21e9e524ff 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax10MibCellIT.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax10MibCellIT.java @@ -57,7 +57,9 @@ public void setUp() throws IOException { public void test10MibCell() throws Exception { GCSSpannerDVAvroSetupHelper.TableDef tableDef = new GCSSpannerDVAvroSetupHelper.TableDef( - getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), "Max10MibCellTable", Arrays.asList("id")); + GCSSpannerDVAvroSetupHelper.getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), + "Max10MibCellTable", + Arrays.asList("id")); final int safeBlobSize = (10 * 1024 * 1024) - 1024; // 9.9MB to avoid limit issues byte[] matchBytes = new byte[safeBlobSize]; diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax16KeyTableIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax16KeyTableIT.java index 000b4721ce..a5682d042b 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax16KeyTableIT.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMax16KeyTableIT.java @@ -55,7 +55,7 @@ public void setUp() throws IOException { public void test16KeyColumns() throws Exception { GCSSpannerDVAvroSetupHelper.TableDef tableDef = new GCSSpannerDVAvroSetupHelper.TableDef( - getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), + GCSSpannerDVAvroSetupHelper.getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), "Max16KeyTable", Arrays.asList( "col_1", "col_2", "col_3", "col_4", "col_5", "col_6", "col_7", "col_8", "col_9", diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMaxColumnNameIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMaxColumnNameIT.java index b5abc76602..e6a7f65f58 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMaxColumnNameIT.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVWideRowMaxColumnNameIT.java @@ -62,7 +62,9 @@ public void setUp() throws IOException { public void testMaxColumnNameLength() throws Exception { GCSSpannerDVAvroSetupHelper.TableDef maxColTableDef = new GCSSpannerDVAvroSetupHelper.TableDef( - getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), "MaxColumnNameTable", Arrays.asList("id")); + GCSSpannerDVAvroSetupHelper.getSchemaFromAvscFile(AVRO_SCHEMA_RESOURCE), + "MaxColumnNameTable", + Arrays.asList("id")); List records = Arrays.asList( diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/BulkMigrationAndValidationOracleAllDataTypesE2EIT.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/BulkMigrationAndValidationOracleAllDataTypesE2EIT.java new file mode 100644 index 0000000000..d637ae2189 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/BulkMigrationAndValidationOracleAllDataTypesE2EIT.java @@ -0,0 +1,165 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ +package com.google.cloud.teleport.v2.templates.endtoend; + +import static com.google.common.truth.Truth.assertThat; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; + +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import com.google.cloud.teleport.v2.templates.GCSSpannerDV; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.TableValidationStatsDto; +import com.google.cloud.teleport.v2.templates.GCSSpannerDVTestAsserts.ValidationSummaryDto; +import java.io.IOException; +import java.util.Collections; +import java.util.TimeZone; +import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; +import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * End-to-End Integration test validating the migration and validation of all supported data types + * from Oracle to Spanner. + * + *

This test verifies the entire lifecycle of data types across two pipelines (Bulk Migration and + * Data Validation). Specifically, it evaluates how the bulk migration pipeline maps each Oracle + * data type to Avro and Spanner, and subsequently, how the validation pipeline uses those Avro + * files to perform end-to-end data validation. + * + *

The test is driven by schemas that reflect real-world mappings: + * + *

    + *
  • The Oracle schema contains all supported Oracle data types. + *
  • The Spanner schema utilizes the default data type mapping provided by Spanner Migration + * Tool (SMT). + *
+ * + *

To ensure comprehensive boundary coverage, the test injects and validates four distinct rows + * of data: + * + *

    + *
  • Standard Row: Typical, everyday values. + *
  • Null Row: Tests NULL value handling across all nullable columns. + *
  • Minimum Row: Tests lower bounds, negative limits, and minimum string lengths. + *
  • Maximum Row: Tests upper bounds, large text/blob limits, and maximum string sizes. + *
+ */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@RunWith(JUnit4.class) +@TemplateIntegrationTest(GCSSpannerDV.class) +public class BulkMigrationAndValidationOracleAllDataTypesE2EIT extends EndToEndTestingITBase { + + private static final String SPANNER_DDL_RESOURCE = + "BulkMigrationAndValidationOracleAllDataTypesE2EIT/spanner-schema.sql"; + private static final String ORACLE_DDL_RESOURCE = + "BulkMigrationAndValidationOracleAllDataTypesE2EIT/oracle-schema.sql"; + + private CloudOracleResourceManager oracleResourceManager; + private TimeZone originalTimeZone; + + @Before + public void setUp() throws IOException { + originalTimeZone = TimeZone.getDefault(); + TimeZone.setDefault(TimeZone.getTimeZone("UTC")); + oracleResourceManager = CloudOracleResourceManager.builder(testName).build(); + spannerResourceManager = setUpSpannerResourceManager(); + bigQueryResourceManager = setUpBigQueryResourceManager(); + bigQueryResourceManager.createDataset(REGION); + } + + @After + public void tearDown() { + if (originalTimeZone != null) { + TimeZone.setDefault(originalTimeZone); + } + ResourceManagerUtils.cleanResources( + oracleResourceManager, flexTemplateDataflowJobResourceManager); + // Spanner and BigQuery are automatically cleaned up in tearDownBase() + } + + @Test + public void allDataTypesE2E() throws Exception { + /* + * Creates a table and inserts 4 boundary testing rows (Standard, NULL, Minimum, Maximum). + */ + executeSqlScript(oracleResourceManager, ORACLE_DDL_RESOURCE); + createSpannerDDL(spannerResourceManager, SPANNER_DDL_RESOURCE); + + // 2. Launch Bulk Pipeline (SourceDbToSpanner) + String gcsOutputDirectory = "gs://" + artifactBucketName + "/" + testId; + + // Launch Bulk Pipeline (SourceDbToSpanner) + LaunchInfo bulkJobInfo = + launchBulkDataflowJob( + testName, spannerResourceManager, gcsClient, oracleResourceManager, null, false); + assertThatPipeline(bulkJobInfo).isRunning(); + pipelineOperator().waitUntilDone(createConfig(bulkJobInfo)); + + // 3. Assert on spanner rows to verify the bulk job was actually successful + assertThat(spannerResourceManager.getRowCount("AllDatatypes")).isEqualTo(4L); + + // 4. Launch Validation Pipeline (GCSSpannerDV) + LaunchConfig.Builder dvOptions = LaunchConfig.builder(testName, specPath); + LaunchInfo validationJobInfo = + launchDataflowJob( + dvOptions, + testName, + PROJECT, + spannerResourceManager, + bigQueryResourceManager.getDatasetId(), + gcsOutputDirectory, + null, + null, + null, + null, + null, + null); + + assertThatPipeline(validationJobInfo).isRunning(); + pipelineOperator().waitUntilDone(createConfig(validationJobInfo)); + + // 5. Assert BigQuery Validation Results (Expect PERFECT MATCH) + GCSSpannerDVTestAsserts.assertValidationSummary( + bigQueryResourceManager, + Collections.singletonList( + new ValidationSummaryDto( + /* status= */ "MATCH", + /* totalTablesValidated= */ 1L, + /* totalRowsMatched= */ 4L, + /* totalRowsMismatched= */ 0L, + /* tablesWithMismatches= */ ""))); + + GCSSpannerDVTestAsserts.assertTableValidationStats( + bigQueryResourceManager, + Collections.singletonList( + new TableValidationStatsDto( + /* schemaName= */ null, + /* tableName= */ "AllDatatypes", + /* status= */ "MATCH", + /* sourceRowCount= */ 4L, + /* destinationRowCount= */ 4L, + /* matchedRowCount= */ 4L, + /* mismatchRowCount= */ 0L))); + } +} diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/EndToEndTestingITBase.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/EndToEndTestingITBase.java index eec24eaa25..fe5c8b5c97 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/EndToEndTestingITBase.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/endtoend/EndToEndTestingITBase.java @@ -24,6 +24,7 @@ import java.util.Collections; import java.util.List; import org.apache.beam.it.common.PipelineLauncher; +import org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager; import org.apache.beam.it.gcp.cloudsql.CloudPostgresResourceManager; import org.apache.beam.it.gcp.cloudsql.CloudSqlResourceManager; import org.apache.beam.it.gcp.dataflow.FlexTemplateDataflowJobResourceManager; @@ -114,6 +115,10 @@ protected PipelineLauncher.LaunchInfo launchBulkDataflowJob( connectionProps = null; jdbcDriver = "org.postgresql.Driver"; dialect = "POSTGRESQL"; + } else if (cloudSqlResourceManager instanceof CloudOracleResourceManager) { + connectionProps = null; + jdbcDriver = "oracle.jdbc.OracleDriver"; + dialect = "ORACLE"; } if (!multiSharded) { diff --git a/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/oracle-schema.sql b/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/oracle-schema.sql new file mode 100644 index 0000000000..af473cc667 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/oracle-schema.sql @@ -0,0 +1,111 @@ +CREATE TABLE AllDatatypes ( + id NUMBER(38, 0) PRIMARY KEY, + varchar_col VARCHAR2(2000), + varchar2_col VARCHAR2(2000), + char_col CHAR(255), + nvarchar2_col NVARCHAR2(2000), + nchar_col NCHAR(255), + number_col NUMBER(38, 0), + numeric_col NUMBER(38, 0), + decimal_col NUMBER(38, 10), + float_col FLOAT, + double_precision_col DOUBLE PRECISION, + binary_float_col BINARY_FLOAT, + binary_double_col BINARY_DOUBLE, + integer_col INTEGER, + int_col INT, + smallint_col SMALLINT, + date_col DATE, + timestamp_col TIMESTAMP(6), + timestamp_tz_col TIMESTAMP(6) WITH TIME ZONE, + timestamp_ltz_col TIMESTAMP(6) WITH LOCAL TIME ZONE, + interval_ym_col INTERVAL YEAR TO MONTH, + interval_ds_col INTERVAL DAY TO SECOND, + raw_col RAW(2000), + clob_col CLOB, + nclob_col NCLOB, + blob_col BLOB, + rowid_col ROWID, + json_col VARCHAR2(4000) +); + +-- Row 1 (Standard Values) +INSERT INTO AllDatatypes ( + id, varchar_col, varchar2_col, char_col, nvarchar2_col, nchar_col, + number_col, numeric_col, decimal_col, float_col, double_precision_col, + binary_float_col, binary_double_col, integer_col, int_col, smallint_col, + date_col, timestamp_col, timestamp_tz_col, timestamp_ltz_col, + interval_ym_col, interval_ds_col, raw_col, clob_col, nclob_col, blob_col, + rowid_col, json_col +) VALUES ( + 1, 'varchar', 'varchar2', 'char', 'nvarchar2', 'nchar', + 12345, 12345, 123.456, 123.45, 123.45, + 123.45, 123.45, 12345, 12345, 123, + TO_DATE('2024-01-01 10:00:00', 'YYYY-MM-DD HH24:MI:SS'), + TO_TIMESTAMP('2024-01-01 10:00:00.000000', 'YYYY-MM-DD HH24:MI:SS.FF6'), + TO_TIMESTAMP_TZ('2024-01-01 10:00:00 +00:00', 'YYYY-MM-DD HH24:MI:SS TZH:TZM'), + TO_TIMESTAMP('2024-01-01 10:00:00.000000', 'YYYY-MM-DD HH24:MI:SS.FF6'), + INTERVAL '1-2' YEAR TO MONTH, + INTERVAL '1 02:03:04.555555' DAY TO SECOND, + HEXTORAW('414243'), + 'clob text', + 'nclob text', + HEXTORAW('414243'), + 'AAAB12AADAAAA12AAA', + '{}' +); + +-- Row 2 (All NULL values) +INSERT INTO AllDatatypes (id) VALUES (2); + +-- Row 3 (Minimum Values) +INSERT INTO AllDatatypes ( + id, varchar_col, varchar2_col, char_col, nvarchar2_col, nchar_col, + number_col, numeric_col, decimal_col, float_col, double_precision_col, + binary_float_col, binary_double_col, integer_col, int_col, smallint_col, + date_col, timestamp_col, timestamp_tz_col, timestamp_ltz_col, + interval_ym_col, interval_ds_col, raw_col, clob_col, nclob_col, blob_col, + rowid_col, json_col +) VALUES ( + 3, 'a', 'a', 'a', 'a', 'a', + -999999999999999999, -999999999999999999, -999999999999999.9999999999, -1.0E38, -1.0E308, + -3.402823E+38, -1.7976931348623157E+308, -2147483648, -2147483648, -32768, + TO_DATE('1970-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS'), + TO_TIMESTAMP('1970-01-01 00:00:00.000000', 'YYYY-MM-DD HH24:MI:SS.FF6'), + TO_TIMESTAMP_TZ('1970-01-01 00:00:00 +00:00', 'YYYY-MM-DD HH24:MI:SS TZH:TZM'), + TO_TIMESTAMP('1970-01-01 00:00:00.000000', 'YYYY-MM-DD HH24:MI:SS.FF6'), + INTERVAL '-99-11' YEAR TO MONTH, + INTERVAL '-99 23:59:59.999999' DAY TO SECOND, + HEXTORAW('00'), + 'min clob', + 'min nclob', + HEXTORAW('00'), + 'AAAAAAAAAAAAAAAAAA', + '{}' +); + +-- Row 4 (Maximum Values) +INSERT INTO AllDatatypes ( + id, varchar_col, varchar2_col, char_col, nvarchar2_col, nchar_col, + number_col, numeric_col, decimal_col, float_col, double_precision_col, + binary_float_col, binary_double_col, integer_col, int_col, smallint_col, + date_col, timestamp_col, timestamp_tz_col, timestamp_ltz_col, + interval_ym_col, interval_ds_col, raw_col, clob_col, nclob_col, blob_col, + rowid_col, json_col +) VALUES ( + 4, RPAD('Z', 2000, 'Z'), RPAD('Z', 2000, 'Z'), RPAD('Z', 255, 'Z'), RPAD('Z', 2000, 'Z'), RPAD('Z', 255, 'Z'), + 999999999999999999, 999999999999999999, 999999999999999.9999999999, 1.0E38, 1.0E308, + 3.402823E+38, 1.7976931348623157E+308, 2147483647, 2147483647, 32767, + TO_DATE('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS'), + TO_TIMESTAMP('9999-12-31 23:59:59.999999', 'YYYY-MM-DD HH24:MI:SS.FF6'), + TO_TIMESTAMP_TZ('9999-12-31 23:59:59 +00:00', 'YYYY-MM-DD HH24:MI:SS TZH:TZM'), + TO_TIMESTAMP('9999-12-31 23:59:59.999999', 'YYYY-MM-DD HH24:MI:SS.FF6'), + INTERVAL '99-11' YEAR TO MONTH, + INTERVAL '99 23:59:59.999999' DAY TO SECOND, + HEXTORAW('FFFF'), + RPAD('Z', 4000, 'Z'), + RPAD('Z', 4000, 'Z'), + HEXTORAW('FFFF'), + 'ZZZZZZZZZZZZZZZZZZ', + '{}' +); diff --git a/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/spanner-schema.sql b/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/spanner-schema.sql new file mode 100644 index 0000000000..4ab82bdcc5 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/resources/BulkMigrationAndValidationOracleAllDataTypesE2EIT/spanner-schema.sql @@ -0,0 +1,30 @@ +CREATE TABLE AllDatatypes ( + id INT64, + varchar_col STRING(2000), + varchar2_col STRING(2000), + char_col STRING(255), + nvarchar2_col STRING(2000), + nchar_col STRING(255), + number_col NUMERIC, + numeric_col NUMERIC, + decimal_col NUMERIC, + float_col FLOAT64, + double_precision_col FLOAT64, + binary_float_col FLOAT32, + binary_double_col FLOAT64, + integer_col INT64, + int_col INT64, + smallint_col INT64, + date_col TIMESTAMP, + timestamp_col TIMESTAMP, + timestamp_tz_col TIMESTAMP, + timestamp_ltz_col TIMESTAMP, + interval_ym_col STRING(MAX), + interval_ds_col STRING(MAX), + raw_col BYTES(MAX), + clob_col STRING(MAX), + nclob_col STRING(MAX), + blob_col BYTES(MAX), + rowid_col STRING(MAX), + json_col JSON +) PRIMARY KEY (id); diff --git a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc deleted file mode 100644 index cfcf3a1ac4..0000000000 --- a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/oracle_all_datatypes.avsc +++ /dev/null @@ -1,86 +0,0 @@ -{ - "type": "record", - "name": "SourceRowWithMetadata", - "fields": [ - { "name": "tableName", "type": "string" }, - { "name": "shardId", "type": [ "null", "string" ], "default": null }, - { "name": "primaryKeys", "type": { "type": "array", "items": "string" } }, - { - "name": "payload", - "type": { - "type": "record", - "name": "OracleAllDatatypesPayload", - "fields": [ - { "name": "id", "type": "long" }, - { "name": "varchar2_col", "type": [ "null", "string" ], "default": null }, - { "name": "varchar_col", "type": [ "null", "string" ], "default": null }, - { "name": "char_col", "type": [ "null", "string" ], "default": null }, - { "name": "character_col", "type": [ "null", "string" ], "default": null }, - { "name": "nvarchar2_col", "type": [ "null", "string" ], "default": null }, - { "name": "nchar_col", "type": [ "null", "string" ], "default": null }, - { - "name": "number_col", - "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], - "default": null - }, - { - "name": "numeric_col", - "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], - "default": null - }, - { - "name": "decimal_col", - "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], - "default": null - }, - { - "name": "dec_col", - "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 38, "scale": 9 } ], - "default": null - }, - { "name": "float_col", "type": [ "null", "double" ], "default": null }, - { "name": "double_precision_col", "type": [ "null", "double" ], "default": null }, - { "name": "real_col", "type": [ "null", "double" ], "default": null }, - { "name": "binary_float_col", "type": [ "null", "float" ], "default": null }, - { "name": "binary_double_col", "type": [ "null", "double" ], "default": null }, - { "name": "integer_col", "type": [ "null", "long" ], "default": null }, - { "name": "int_col", "type": [ "null", "long" ], "default": null }, - { "name": "smallint_col", "type": [ "null", "long" ], "default": null }, - { - "name": "date_col", - "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], - "default": null - }, - { - "name": "timestamp_col", - "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], - "default": null - }, - { - "name": "timestamp_tz_col", - "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], - "default": null - }, - { - "name": "timestamp_ltz_col", - "type": [ "null", { "type": "long", "logicalType": "timestamp-micros" } ], - "default": null - }, - { "name": "interval_ym_col", "type": [ "null", "string" ], "default": null }, - { "name": "interval_ds_col", "type": [ "null", "string" ], "default": null }, - { "name": "raw_col", "type": [ "null", "bytes" ], "default": null }, - { "name": "blob_col", "type": [ "null", "bytes" ], "default": null }, - { "name": "clob_col", "type": [ "null", "string" ], "default": null }, - { "name": "nclob_col", "type": [ "null", "string" ], "default": null }, - { "name": "rowid_col", "type": [ "null", "string" ], "default": null }, - { - "name": "json_col", - "type": [ "null", { "type": "string", "logicalType": "json" } ], - "default": null - }, - { "name": "xmltype_col", "type": [ "null", "string" ], "default": null } - ] - } - } - ] -} diff --git a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql b/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql deleted file mode 100644 index 0bf2551f5c..0000000000 --- a/v2/gcs-spanner-dv/src/test/resources/GCSSpannerDVOracleSmokeIT/spanner-schema.sql +++ /dev/null @@ -1,34 +0,0 @@ -CREATE TABLE OracleAllDatatypes ( - id INT64 NOT NULL, - varchar2_col STRING(MAX), - varchar_col STRING(MAX), - char_col STRING(MAX), - character_col STRING(MAX), - nvarchar2_col STRING(MAX), - nchar_col STRING(MAX), - number_col NUMERIC, - numeric_col NUMERIC, - decimal_col NUMERIC, - dec_col NUMERIC, - float_col FLOAT64, - double_precision_col FLOAT64, - real_col FLOAT64, - binary_float_col FLOAT32, - binary_double_col FLOAT64, - integer_col INT64, - int_col INT64, - smallint_col INT64, - date_col TIMESTAMP, - timestamp_col TIMESTAMP, - timestamp_tz_col TIMESTAMP, - timestamp_ltz_col TIMESTAMP, - interval_ym_col STRING(MAX), - interval_ds_col STRING(MAX), - raw_col BYTES(MAX), - blob_col BYTES(MAX), - clob_col STRING(MAX), - nclob_col STRING(MAX), - rowid_col STRING(MAX), - json_col JSON, - xmltype_col STRING(MAX) -) PRIMARY KEY(id); From b39c0641f4f8ab627edd98e3aadff7e54560f13f Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Fri, 28 Aug 2026 08:41:18 +0000 Subject: [PATCH 5/6] improved test coverage --- .../teleport/v2/visitor/UnifiedStringVisitorTest.java | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitorTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitorTest.java index 5070759405..8542be344a 100644 --- a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitorTest.java +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/visitor/UnifiedStringVisitorTest.java @@ -70,6 +70,16 @@ public void testVisitFloat64() { assertEquals(String.valueOf(input), visitor.getResult()); } + @Test + public void testVisitFloat32() { + UnifiedStringVisitor visitor = new UnifiedStringVisitor(); + float input = 12.3f; + + visitor.visitFloat32(input); + + assertEquals(String.valueOf(input), visitor.getResult()); + } + @Test public void testVisitBool() { UnifiedStringVisitor visitor = new UnifiedStringVisitor(); From 33dd3978e69847e87aba8344bfc97a01ec1de218 Mon Sep 17 00:00:00 2001 From: Aditya Bharadwaj Date: Fri, 28 Aug 2026 09:00:34 +0000 Subject: [PATCH 6/6] improved test coverage --- .../migrations/avro/GenericRecordTypeConvertorTest.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java index d055a7e4a9..3c326c220a 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java @@ -300,6 +300,15 @@ public void testHandleLogicalFieldType() { getTestCassandraAnnotationNone()); assertEquals("Test json_col conversion: ", "{\"k1\":\"476F6F676C65\"}", result); + col = "json_col"; + result = + GenericRecordTypeConvertor.handleLogicalFieldType( + col, + genericRecord.get(col), + genericRecord.getSchema().getField(col).schema(), + null); + assertEquals("Test json_col conversion: ", "{\"k1\":\"476F6F676C65\"}", result); + col = "json_col"; result = GenericRecordTypeConvertor.handleLogicalFieldType(