[SpannerToSourceDb] Add support for PostgreSQL native enums

v2/sourcedb-to-spanner/README.md

@@ -208,3 +208,6 @@ However, because the continuous reader watches the `retry/` directory indefinite
   * **`JSON` / `JSONB`:**
     * *Spanner GoogleSQL:* `JSON`, `STRING`
     * *Spanner PostgreSQL:* `JSONB`, `VARCHAR` / `TEXT`
+  * **`ENUM`:**
+    * *Spanner GoogleSQL:* `STRING`
+    * *Spanner PostgreSQL:* `VARCHAR` / `TEXT`

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/postgres/PostgresSrcToSpSourceConnector.java

@@ -64,6 +64,7 @@ public class PostgresSrcToSpSourceConnector extends AbstractJdbcSrcToSpSourceCon
           // precision and scale are >= 0, map to DECIMAL)
           .put("DECIMAL", UnifiedMappingProvider.Type.NUMBER)
           .put("DOUBLE PRECISION", UnifiedMappingProvider.Type.DOUBLE)
+          .put("ENUM", UnifiedMappingProvider.Type.STRING)
           .put("FLOAT4", UnifiedMappingProvider.Type.FLOAT)
           .put("FLOAT8", UnifiedMappingProvider.Type.DOUBLE)
           .put("INET", UnifiedMappingProvider.Type.STRING)

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/postgres/reader/io/jdbc/dialectadapter/postgresql/PostgreSQLDialectAdapter.java

@@ -256,18 +256,23 @@ public ImmutableMap<String, ImmutableMap<String, SourceColumnType>> discoverTabl
         tables);
 
     final String query =
-        "SELECT table_name,"
-            + "  column_name,"
-            + "  data_type,"
-            + "  character_maximum_length,"
-            + "  numeric_precision,"
-            + "  numeric_scale"
-            + " FROM information_schema.columns"
-            + " WHERE table_catalog = ?"
-            + "  AND table_schema = ?"
-            + "  AND table_name IN "
+        "SELECT c.table_name,"
+            + "  c.column_name,"
+            // Enum columns are reported as the placeholder `USER-DEFINED` by
+            // information_schema; follow udt_name into pg_type to report them as `ENUM`.
+            + "  CASE WHEN t.typtype = 'e' THEN 'ENUM' ELSE c.data_type END AS data_type,"
+            + "  c.character_maximum_length,"
+            + "  c.numeric_precision,"
+            + "  c.numeric_scale"
+            + " FROM information_schema.columns c"
+            + "  LEFT OUTER JOIN pg_catalog.pg_namespace n ON n.nspname = c.udt_schema"
+            + "  LEFT OUTER JOIN pg_catalog.pg_type t ON t.typname = c.udt_name"
+            + "   AND t.typnamespace = n.oid"
+            + " WHERE c.table_catalog = ?"
+            + "  AND c.table_schema = ?"
+            + "  AND c.table_name IN "
             + DialectAdapter.generateInClause(tables.size())
-            + " ORDER BY table_name, ordinal_position";
+            + " ORDER BY c.table_name, c.ordinal_position";
 
     Map<String, ImmutableMap.Builder<String, SourceColumnType>> builders = new HashMap<>();
     tables.forEach(

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/postgres/reader/io/jdbc/rowmapper/provider/PostgreSQLJdbcValueMappings.java

@@ -276,6 +276,12 @@ private static long toMicros(Instant instant) {
                 return (int) (n / 2 + 8);
               })
           .put("DOUBLE PRECISION", ResultSet::getDouble, valuePassThrough, 8)
+          .put(
+              "ENUM",
+              ResultSet::getString,
+              valuePassThrough,
+              63) // Enum labels are limited to NAMEDATALEN-1 = 63 bytes.
+          // https://www.postgresql.org/docs/current/datatype-enum.html
           .put("FLOAT4", ResultSet::getFloat, valuePassThrough, 4)
           .put("FLOAT8", ResultSet::getDouble, valuePassThrough, 8)
           .put("INET", ResultSet::getString, valuePassThrough, 196)

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/reader/io/jdbc/rowmapper/JdbcSourceRowMapperTest.java

@@ -630,7 +630,7 @@ private ImmutableList<Column> postgreSQLColumns() {
                 .derbyColumnType("VARCHAR(100)")
                 .sourceColumnType("ENUM")
                 .inputValue("ENUM VALUE")
-                .mappedValue(null) // Unsupported
+                .mappedValue("ENUM VALUE")
                 .build())
         .add(
             Column.builder()

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/postgres/reader/io/schema/typemapping/provider/PostgreSQLMappingProviderTest.java

@@ -60,6 +60,7 @@ private ImmutableMap<String, String> expectedMapping() {
         .put("DATE", "{\"type\":\"int\",\"logicalType\":\"date\"}")
         .put("DECIMAL", "{\"type\":\"string\",\"logicalType\":\"number\"}")
         .put("DOUBLE PRECISION", "\"double\"")
+        .put("ENUM", "\"string\"")
         .put("FLOAT4", "\"float\"")
         .put("FLOAT8", "\"double\"")
         .put("INET", "\"string\"")

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

@@ -121,7 +121,6 @@ public void allTypesTest() throws Exception {
             // "t_circle_to_float64_array",
             "t_datemultirange",
             "t_daterange",
-            "t_enum",
             // "t_float_array_to_float64_array",
             "t_float_array_to_string",
             // "t_int_array_to_int64_array",
@@ -206,6 +205,7 @@ private Map<String, List<Map<String, Object>>> getExpectedData() {
     result.put("date", createRows("0001-01-01", "9999-12-31", "NULL"));
     result.put("date_to_string", createRows("0001-01-01", "9999-12-31", "NULL"));
     result.put("decimal_to_string", createRows("0.12", "NULL"));
+    result.put("enum", createRows("enum1", "NULL"));
     result.put(
         "double_precision",
         createRows(

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

@@ -122,7 +122,6 @@ public void allTypesTest() throws Exception {
             // "t_circle_to_float64_array",
             "t_datemultirange",
             "t_daterange",
-            "t_enum",
             // "t_float_array_to_float64_array",
             "t_float_array_to_string",
             // "t_int_array_to_int64_array",
@@ -213,6 +212,7 @@ private Map<String, List<Map<String, Object>>> getExpectedData() {
     result.put("date", createRows("0001-01-01", "9999-12-31", "NULL"));
     result.put("date_to_string", createRows("0001-01-01", "9999-12-31", "NULL"));
     result.put("decimal_to_string", createRows("0.12", "NULL"));
+    result.put("enum", createRows("enum1", "NULL"));
     result.put(
         "double_precision",
         createRows(