tests: Bulk IT Tests

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/MySQLDDLIT.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright (C) 2024 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.spanner.Struct;
+import com.google.cloud.teleport.metadata.SkipDirectRunnerTest;
+import com.google.cloud.teleport.metadata.TemplateIntegrationTest;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.it.common.PipelineLauncher;
+import org.apache.beam.it.common.PipelineOperator;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import org.apache.beam.it.gcp.spanner.SpannerResourceManager;
+import org.apache.beam.it.gcp.spanner.matchers.SpannerAsserts;
+import org.apache.beam.it.jdbc.MySQLResourceManager;
+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;
+
+/**
+ * An integration test for {@link SourceDbToSpanner} Flex template which tests a migration with DDL
+ * changes to schema. Changes include Index changes, Primary key transformations and Generated
+ * columns migration.
+ */
+@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class})
+@TemplateIntegrationTest(SourceDbToSpanner.class)
+@RunWith(JUnit4.class)
+public class MySQLDDLIT extends SourceDbToSpannerITBase {
+  private static PipelineLauncher.LaunchInfo jobInfo;
+
+  public static MySQLResourceManager mySQLResourceManager;
+  public static SpannerResourceManager spannerResourceManager;
+
+  private static final String SESSION_FILE_RESOURCE = "DDLIT/company-session.json";
+  private static final String MYSQL_DDL_RESOURCE = "DDLIT/company-mysql-schema.sql";
+  private static final String SPANNER_DDL_RESOURCE = "DDLIT/company-spanner-schema.sql";
+
+  /**
+   * Setup resource managers and Launch dataflow job once during the execution of this test class. \
+   */
+  @Before
+  public void setUp() {
+    mySQLResourceManager = setUpMySQLResourceManager();
+    spannerResourceManager = setUpSpannerResourceManager();
+  }
+
+  /** Cleanup dataflow job and all the resources and resource managers. */
+  @After
+  public void cleanUp() {
+    ResourceManagerUtils.cleanResources(spannerResourceManager, mySQLResourceManager);
+  }
+
+  @Test
+  public void ddlModificationTest() throws Exception {
+    loadSQLFileResource(mySQLResourceManager, MYSQL_DDL_RESOURCE);
+    createSpannerDDL(spannerResourceManager, SPANNER_DDL_RESOURCE);
+    jobInfo =
+        launchDataflowJob(
+            getClass().getSimpleName(),
+            SESSION_FILE_RESOURCE,
+            "mapper",
+            mySQLResourceManager,
+            spannerResourceManager,
+            null,
+            null);
+    PipelineOperator.Result result = pipelineOperator().waitUntilDone(createConfig(jobInfo));
+
+    List<Map<String, Object>> companyMySQL =
+        mySQLResourceManager.runSQLQuery("SELECT company_id, company_name FROM company");
+    ImmutableList<Struct> companySpanner =
+        spannerResourceManager.readTableRecords("company", "company_id", "company_name");
+
+    SpannerAsserts.assertThatStructs(companySpanner)
+        .hasRecordsUnorderedCaseInsensitiveColumns(companyMySQL);
+
+    List<Map<String, Object>> employeeMySQL =
+        mySQLResourceManager.runSQLQuery(
+            "SELECT employee_id, company_id, employee_name, employee_address FROM employee");
+    ImmutableList<Struct> employeeSpanner =
+        spannerResourceManager.readTableRecords(
+            "employee", "employee_id", "company_id", "employee_name", "employee_address");
+
+    SpannerAsserts.assertThatStructs(employeeSpanner)
+        .hasRecordsUnorderedCaseInsensitiveColumns(employeeMySQL);
+
+    ImmutableList<Struct> employeeAttribute =
+        spannerResourceManager.readTableRecords(
+            "employee_attribute", "employee_id", "attribute_name", "value");
+
+    SpannerAsserts.assertThatStructs(employeeAttribute).hasRows(4); // Supports composite keys
+
+    ImmutableList<Struct> vendor =
+        spannerResourceManager.readTableRecords("vendor", "vendor_id", "full_name");
+
+    SpannerAsserts.assertThatStructs(vendor).hasRows(3);
+  }
+}

v2/sourcedb-to-spanner/src/test/resources/DDLIT/company-mysql-schema.sql

@@ -0,0 +1,59 @@
+CREATE TABLE `company` (
+                           `company_id` int(11) PRIMARY KEY NOT NULL,
+                           `company_name` varchar(100) DEFAULT NULL,
+                           `created_on` date
+);
+
+INSERT INTO `company` VALUES
+                          (1,'gog','1998-09-04'),
+                          (2,'app','1976-04-01'),
+                          (3,'ama','1994-07-05');
+
+CREATE TABLE `employee` (
+                            `employee_id` int(11) PRIMARY KEY NOT NULL,
+                            `company_id` int(11) DEFAULT NULL,
+                            `employee_name` varchar(100) DEFAULT NULL,
+                            `employee_address` varchar(100) DEFAULT NULL,
+                            `created_on` date
+);
+
+INSERT INTO `employee` VALUES
+                           (100,1,'emp1','add1','1996-01-01'),
+                           (101,1,'emp2','add2','1999-01-01'),
+                           (102,1,'emp3','add3','2012-01-01'),
+                           (300,3,'emp300','add300','1996-01-01');
+
+CREATE TABLE `employee_attribute` (
+                                      `employee_id` int(11) NOT NULL,
+                                      `attribute_name` varchar(100) NOT NULL,
+                                      `value` varchar(100) DEFAULT NULL,
+                                      `updated_on` date,
+                                      PRIMARY KEY (`employee_id`,`attribute_name`)
+);
+
+INSERT INTO `employee_attribute` VALUES
+                                     (100,'iq','150','2024-06-10'),
+                                     (101,'iq','120','2024-06-10'),
+                                     (102,'iq','20','2024-06-10'),
+                                     (300,'endurance','20','2024-06-10');
+
+CREATE TABLE `vendor` (
+                      `vendor_id` INT AUTO_INCREMENT PRIMARY KEY,
+                      `first_name` VARCHAR(255) NOT NULL,
+                      `last_name` VARCHAR(255) NOT NULL,
+                      `email` VARCHAR(255) UNIQUE NOT NULL,
+                      `full_name` VARCHAR(512) GENERATED ALWAYS AS (CONCAT(first_name, ' ', last_name)),
+                      INDEX full_name_idx (full_name)
+);
+
+INSERT INTO vendor (vendor_id, first_name, last_name, email) VALUES
+                                     (1, 'David', 'Lee', 'david.lee@example.com'),
+                                     (2, 'Sarah', 'Jones', 'sarah.jones@example.com'),
+                                     (3, 'Michael', 'Brown', 'michael.brown@example.com');
+
+CREATE TABLE `mysql_extra` (
+                           `test_id` int(11) PRIMARY KEY NOT NULL,
+                           `test_name` varchar(100) DEFAULT NULL
+);
+
+CREATE VIEW company_view AS  SELECT company_id FROM company;