Use VARIANT type for properties without a type

target_snowflake/db_sync.py

@@ -72,10 +72,10 @@ def validate_config(config):
 
 def column_type(schema_property):
     """Take a specific schema property and return the snowflake equivalent column type"""
-    property_type = schema_property['type']
-    property_format = schema_property['format'] if 'format' in schema_property else None
+    property_type = schema_property.get('type')
+    property_format = schema_property.get('format')
     col_type = 'text'
-    if 'object' in property_type or 'array' in property_type:
+    if property_type is None or 'object' in property_type or 'array' in property_type:
         col_type = 'variant'
 
     # Every date-time JSON value is currently mapped to TIMESTAMP_NTZ
@@ -101,9 +101,11 @@ def column_type(schema_property):
 
 def column_trans(schema_property):
     """Generate SQL transformed columns syntax"""
-    property_type = schema_property['type']
+    property_type = schema_property.get('type')
     col_trans = ''
-    if 'object' in property_type or 'array' in property_type:
+    if property_type is None:
+        col_trans = 'to_variant'
+    elif 'object' in property_type or 'array' in property_type:
         col_trans = 'parse_json'
     elif schema_property.get('format') == 'binary':
         col_trans = 'to_binary'

target_snowflake/stream_utils.py

@@ -66,8 +66,10 @@ def reset_new_value(record: Dict, key: str, _format: str):
                         reset_new_value(record, key, type_dict['format'])
                         break
             else:
-                if 'string' in schema['properties'][key]['type'] and \
-                        schema['properties'][key].get('format', None) in {'date-time', 'time', 'date'}:
+                property_type = schema['properties'][key].get('type')
+                property_format = schema['properties'][key].get('format')
+                if property_type is not None and 'string' in property_type and \
+                        property_format in {'date-time', 'time', 'date'}:
                     reset_new_value(record, key, schema['properties'][key]['format'])