added oracle support to data validation

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,12 +66,14 @@ 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());
       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);
     }
   }

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

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);

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,20 +31,33 @@
  */
 public class GCSSpannerDVAvroSetupHelper {
 
+  /**
+   * 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);
+    }
+  }
+
   /**
    * Defines standard table schemas that are universally used across multiple integration tests.
    * Centralizing these definitions prevents schema drift across tests and minimizes setup code.
    */
   public static class TableDef {
     public static final TableDef USERS =
         new TableDef(
-            GCSSpannerDVITBase.getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/users.avsc"),
+            getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/users.avsc"),
             "Users",
             Arrays.asList("user_id", "event_id"));
     public static final TableDef ACCOUNT_ROLES =
         new TableDef(
-            GCSSpannerDVITBase.getSchemaFromAvscFile(
-                "GCSSpannerDVAvroSetupHelper/account_roles.avsc"),
+            getSchemaFromAvscFile("GCSSpannerDVAvroSetupHelper/account_roles.avsc"),
             "AccountRoles",
             Arrays.asList("role_id"));
 
@@ -102,12 +117,13 @@ public GenericRecord build() {
    * Converts standard Java types into the primitive formats required by Avro logical types.
    *
    * <p>Because this helper uses a generic {@code Map<String, Object>} 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.
    *
-   * <p><b>IMPORTANT:</b> 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.
+   * <p><b>NOTE:</b> {@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.
@@ -122,6 +138,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;
   }

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.
    *

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"));

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];

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",

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<GenericRecord> records =
         Arrays.asList(

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.
+ *
+ * <p>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.
+ *
+ * <p>The test is driven by schemas that reflect real-world mappings:
+ *
+ * <ul>
+ *   <li>The Oracle schema contains all supported Oracle data types.
+ *   <li>The Spanner schema utilizes the default data type mapping provided by Spanner Migration
+ *       Tool (SMT).
+ * </ul>
+ *
+ * <p>To ensure comprehensive boundary coverage, the test injects and validates four distinct rows
+ * of data:
+ *
+ * <ul>
+ *   <li><b>Standard Row:</b> Typical, everyday values.
+ *   <li><b>Null Row:</b> Tests NULL value handling across all nullable columns.
+ *   <li><b>Minimum Row:</b> Tests lower bounds, negative limits, and minimum string lengths.
+ *   <li><b>Maximum Row:</b> Tests upper bounds, large text/blob limits, and maximum string sizes.
+ * </ul>
+ */
+@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)));
+  }
+}

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) {