Use underlying type of domain type when defined

This fixes issues with primary keys being defined using domain types, and deselected by default.

E.g. for a domain type `email_domain` with base type `text`, the `data_type` column will now be `text` rather than `email_domain`.

Author: Limess

tap_postgres/discovery_utils.py

@@ -64,7 +64,6 @@ def produce_table_info(conn, filter_schemas=None, tables: Optional[List[str]] =
     with conn.cursor(cursor_factory=psycopg2.extras.DictCursor, name='stitch_cursor') as cur:
         cur.itersize = post_db.CURSOR_ITER_SIZE
         table_info = {}
-        # SELECT CASE WHEN $2.typtype = 'd' THEN $2.typbasetype ELSE $1.atttypid END
         sql = """
 SELECT
   pg_class.reltuples::BIGINT                            AS approximate_row_count,
@@ -73,7 +72,7 @@ def produce_table_info(conn, filter_schemas=None, tables: Optional[List[str]] =
   pg_class.relname                                      AS table_name,
   attname                                               AS column_name,
   i.indisprimary                                        AS primary_key,
-  format_type(a.atttypid, NULL::integer)                AS data_type,
+  format_type(CASE WHEN pgt.typtype = 'd' THEN pgt.typbasetype ELSE a.atttypid END, NULL::integer) AS data_type,
   information_schema._pg_char_max_length(CASE WHEN COALESCE(subpgt.typtype, pgt.typtype) = 'd'
                                               THEN COALESCE(subpgt.typbasetype, pgt.typbasetype) ELSE COALESCE(subpgt.oid, pgt.oid)
                                           END,

tests/test_discovery.py

@@ -452,6 +452,51 @@ def test_catalog(self):
                                   'definitions' : BASE_RECURSIVE_SCHEMAS},
                                  stream_dict.get('schema'))
 
+class TestDomainTable(unittest.TestCase):
+    maxDiff = None
+    table_name = 'CHICKEN TIMES'
+
+    def setUp(self):
+       table_spec = {"columns": [{"name" : 'our_test_domain_pk',          "type" : "test_domain", "primary_key" : True },
+                                 {"name" : 'our_test_domain',             "type" : "test_domain"},
+                                 {"name" : 'our_test_integer_domain',             "type" : "test_integer_domain" }],
+                     "name" : TestHStoreTable.table_name}
+       with get_test_connection() as conn:
+           cur = conn.cursor()
+           cur.execute("""     DROP DOMAIN IF EXISTS test_domain CASCADE """)
+           cur.execute("""     CREATE DOMAIN test_domain AS text; """)
+           cur.execute("""     DROP DOMAIN IF EXISTS test_integerdomain CASCADE """)
+           cur.execute("""     CREATE DOMAIN test_integerdomain AS text; """)
+
+       ensure_test_table(table_spec)
+
+    def test_catalog(self):
+        conn_config = get_test_connection_config()
+        streams = tap_postgres.do_discovery(conn_config)
+        chicken_streams = [s for s in streams if s['tap_stream_id'] == 'public-CHICKEN TIMES']
+        self.assertEqual(len(chicken_streams), 1)
+        stream_dict = chicken_streams[0]
+        stream_dict.get('metadata').sort(key=lambda md: md['breadcrumb'])
+
+        with get_test_connection() as conn:
+            with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+                cur.execute("""INSERT INTO "CHICKEN TIMES" (our_test_domain_pk, our_test_domain, our_test_integer_domain) VALUES ('sad', 'happy', 3)""")
+                cur.execute("""SELECT * FROM  "CHICKEN TIMES" """)
+
+                self.assertEqual(metadata.to_map(stream_dict.get('metadata')),
+                                 {() : {'table-key-properties': ['our_test_domain_pk'], 'database-name': 'postgres', 'schema-name': 'public', 'is-view': False, 'row-count': 0},
+                                  ('properties', 'our_test_domain_pk') : {'inclusion': 'automatic', 'sql-datatype' : 'text',  'selected-by-default' : True},
+                                  ('properties', 'our_test_domain') : {'inclusion': 'available', 'sql-datatype' : 'text',  'selected-by-default' : True},
+                                  ('properties', 'our_test_integer_domain') : {'inclusion': 'available', 'sql-datatype' : 'integer',  'selected-by-default' : True}})
+
+
+                self.assertEqual({'properties': {'our_test__integer_domain':                  {'type': ['null', 'integer']},
+                                                 'our_test_domain':                  {'type': ['null', 'string']},
+                                                 'our_test_domain_pk':               {'type': ['string']}},
+                                  'type': 'object',
+                                  'definitions' : BASE_RECURSIVE_SCHEMAS},
+                                 stream_dict.get('schema'))
+
 class TestArraysTable(unittest.TestCase):
     maxDiff = None
     table_name = 'CHICKEN TIMES'
@@ -536,7 +581,7 @@ class TestColumnGrants(unittest.TestCase):
     table_name = 'CHICKEN TIMES'
     user = 'tmp_user_for_grant_tests'
     password = 'password'
- 
+
     def setUp(self):
         table_spec = {"columns": [{"name" : "id",                "type" : "integer",  "serial" : True},
                                  {"name" : 'size integer',      "type" : "integer",  "quoted" : True},
@@ -560,8 +605,8 @@ def setUp(self):
            LOGGER.info("running sql: {}".format(sql))
            cur.execute(sql)
 
-           
-           
+
+
 
     def test_catalog(self):
         conn_config = get_test_connection_config()
@@ -587,7 +632,7 @@ def test_catalog(self):
                           ('properties', 'id'): {'inclusion': 'available',
                                                  'selected-by-default': True,
                                                  'sql-datatype': 'integer'}})
-        
+
         self.assertEqual({'definitions' : BASE_RECURSIVE_SCHEMAS,
                           'type': 'object',
                           'properties': {'id': {'type': ['null', 'integer'],