Add support for Postgres' hstore type to fast sync

pipelinewise/fastsync/commons/tap_postgres.py

@@ -415,9 +415,13 @@ def get_table_columns(self, table_name, max_num=None, date_type='date'):
                     ,character_maximum_length
                 FROM (SELECT
                 column_name,
-                data_type,
+                CASE 
+                  WHEN data_type = 'USER-DEFINED' AND udt_name = 'hstore' THEN 'hstore'
+                  ELSE data_type
+                END as data_type,
                 CASE
                     WHEN data_type = 'ARRAY' THEN 'array_to_json("' || column_name || '") AS ' || column_name
+                    WHEN data_type = 'USER-DEFINED' AND udt_name = 'hstore' THEN column_name|| '::json AS ' || column_name
                     WHEN data_type = 'date' THEN
                        'CASE WHEN "' ||column_name|| E'" < \\'0001-01-01\\' '
                             'OR "' ||column_name|| E'" > \\'9999-12-31\\' THEN \\'9999-12-31\\' '

pipelinewise/fastsync/postgres_to_snowflake.py

@@ -41,8 +41,13 @@
 LOCK = multiprocessing.Lock()
 
 
-def tap_type_to_target_type(pg_type, *_):
+def tap_type_to_target_type(pg_type, extra=None):
     """Data type mapping from Postgres to Snowflake"""
+
+    # Needed for call from assertion test helper
+    if pg_type == 'user-defined' and extra == 'hstore':
+        pg_type = 'hstore'
+
     return {
         'char': 'VARCHAR',
         'character': 'VARCHAR',
@@ -75,6 +80,7 @@ def tap_type_to_target_type(pg_type, *_):
         'ARRAY': 'VARIANT',
         'json': 'VARIANT',
         'jsonb': 'VARIANT',
+        'hstore': 'VARIANT'
     }.get(pg_type, 'VARCHAR')
 
 

tests/db/tap_postgres_data.sql

@@ -58,6 +58,26 @@ Maatjies', DATE '2021-01-30'),
 
 COMMIT;
 
+BEGIN;
+SET client_encoding = 'UTF8';
+
+CREATE EXTENSION IF NOT EXISTS hstore;
+
+DROP TABLE IF EXISTS public.special_data_types;
+
+CREATE TABLE public.special_data_types(
+    id serial PRIMARY KEY,
+    cHstore hstore
+);
+
+INSERt INTO public.special_data_types
+    (cHstore)
+VALUES
+    ('"key" => "some value"'),
+    ('"spaced key" => "another value"');
+COMMIT;
+
+
 BEGIN;
 SET client_encoding = 'LATIN1';
 

tests/end_to_end/helpers/db.py

@@ -133,7 +133,7 @@ def sql_get_columns_postgres(schemas: list) -> str:
     sql_schemas = ', '.join(f"'{schema}'" for schema in schemas)
 
     return f"""
-    SELECT table_name, STRING_AGG(CONCAT(column_name, ':', data_type, ':'), ';' ORDER BY column_name)
+    SELECT table_name, STRING_AGG(CONCAT(column_name, ':', data_type, ':', udt_name), ';' ORDER BY column_name)
      FROM information_schema.columns
     WHERE table_schema IN ({sql_schemas})
     GROUP BY table_name

tests/end_to_end/test-project/tap_postgres_to_sf.yml.template

@@ -67,6 +67,10 @@ schemas:
         replication_method: "INCREMENTAL"
         replication_key: "id"
 
+      ### Table with special data types
+      - table_name: "special_data_types"
+        replication_method: "LOG_BASED"
+
       ### Table with space and mixed upper and lowercase characters
       - table_name: "table_with_space and UPPERCase"
         replication_method: "LOG_BASED"