Add support for year as a primary key

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/jdbc/dialectadapter/mysql/MysqlDialectAdapter.java

@@ -376,6 +376,7 @@ private ImmutableMap<String, SourceColumnType> getTableCols(
           .put("TINYTEXT", IndexType.STRING)
           .put("DATETIME", IndexType.TIME_STAMP)
           .put("TIMESTAMP", IndexType.TIME_STAMP)
+          .put("YEAR", IndexType.NUMERIC)
           .build();
 
   /**

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/reader/io/jdbc/dialectadapter/mysql/MysqlDialectAdapterTest.java

@@ -280,7 +280,7 @@ public void testDiscoverTablesRsException() throws SQLException {
   public void testDiscoverIndexesBasic() throws SQLException, RetriableSchemaDiscoveryException {
     ImmutableList<String> testTables = ImmutableList.of("testTable1");
     ImmutableList<String> colTypes =
-        ImmutableList.of("float", "integer", "bit", "char", "varbinary", "binary");
+        ImmutableList.of("float", "integer", "bit", "char", "varbinary", "binary", "year");
     ImmutableList<SourceColumnIndexInfo> expectedSourceColumnIndexInfos =
         ImmutableList.of(
             SourceColumnIndexInfo.builder()
@@ -344,6 +344,15 @@ public void testDiscoverIndexesBasic() throws SQLException, RetriableSchemaDisco
                 .setIndexType(IndexType.BINARY)
                 .setOrdinalPosition(4)
                 .build());
+    SourceColumnIndexInfo.builder()
+        .setColumnName("testColYear")
+        .setIndexName("primary")
+        .setIsUnique(true)
+        .setIsPrimary(true)
+        .setCardinality(100L)
+        .setIndexType(IndexType.NUMERIC)
+        .setOrdinalPosition(6)
+        .build();
 
     final JdbcSchemaReference sourceSchemaReference =
         JdbcSchemaReference.builder().setDbName("testDB").build();

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

@@ -277,6 +277,7 @@ private Map<String, List<Map<String, Object>>> getExpectedData() {
             "2005-01-01T00:01:54.123456000Z",
             "2037-12-30T23:59:59Z",
             "2038-01-18T23:59:59Z"));
+    expectedData.put("year_pk", createRows("year_pk", "1901", "2000"));
     return expectedData;
   }
 

v2/sourcedb-to-spanner/src/test/resources/DataTypesIT/mysql-data-types.sql

@@ -291,6 +291,10 @@ CREATE TABLE timestamp_pk_table (
                                     timestamp_pk_col TIMESTAMP(6) NOT NULL,
                                     CONSTRAINT PRIMARY KEY (id)
 );
+CREATE TABLE year_pk_table (
+    id YEAR PRIMARY KEY,
+    year_pk_col YEAR NOT NULL
+);
 
 ALTER TABLE `bigint_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `bigint_unsigned_table` MODIFY `id` INT AUTO_INCREMENT;
@@ -448,6 +452,7 @@ INSERT INTO `timestamp_pk_table` (`id`, `timestamp_pk_col`) VALUES ('1970-01-01
 SET time_zone = 'Asia/Kolkata';
 INSERT INTO `timestamp_pk_table` (`id`, `timestamp_pk_col`) VALUES ('2005-01-01 05:31:54.123456', '2005-01-01 05:31:54.123456');
 SET time_zone = SYSTEM;
+INSERT INTO `year_pk_table` (`id`, `year_pk_col`) VALUES (1901, 1901), (2000, 2000);
 
 INSERT INTO `bigint_table` (`bigint_col`) VALUES (NULL);
 INSERT INTO `bigint_unsigned_table` (`bigint_unsigned_col`) VALUES (NULL);

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

@@ -322,3 +322,8 @@ CREATE TABLE IF NOT EXISTS timestamp_pk_table (
                                                   id TIMESTAMP NOT NULL,
                                                   timestamp_pk_col TIMESTAMP NOT NULL,
 ) PRIMARY KEY(id);
+
+CREATE TABLE IF NOT EXISTS year_pk_table (
+  id INT64 NOT NULL,
+  year_pk_col INT64 NOT NULL,
+) PRIMARY KEY(id);