Normalization: handle non-object top-level schemas; treat binary data as string (#22165)

* handle dumb top-level schemas

* version bump

* also definitions

* treat binary as string

* fallback case

* format

* new variable

Author: edgao

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -45,7 +45,7 @@
   icon: bigquery.svg
   normalizationConfig:
     normalizationRepository: airbyte/normalization
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: bigquery
   supportsDbt: true
   resourceRequirements:
@@ -91,7 +91,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization-clickhouse
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: clickhouse
   supportsDbt: true
 - name: Cloudflare R2
@@ -213,7 +213,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization-mssql
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: mssql
   supportsDbt: true
 - name: MeiliSearch
@@ -239,7 +239,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization-mysql
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: mysql
   supportsDbt: true
 - name: Oracle
@@ -251,7 +251,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization-oracle
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: oracle
   supportsDbt: true
 - name: Postgres
@@ -263,7 +263,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: postgres
   supportsDbt: true
 - name: Pulsar
@@ -295,7 +295,7 @@
   icon: redshift.svg
   normalizationConfig:
     normalizationRepository: airbyte/normalization-redshift
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: redshift
   supportsDbt: true
   resourceRequirements:
@@ -353,7 +353,7 @@
   icon: snowflake.svg
   normalizationConfig:
     normalizationRepository: airbyte/normalization-snowflake
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: snowflake
   supportsDbt: true
   resourceRequirements:
@@ -407,7 +407,7 @@
   releaseStage: alpha
   normalizationConfig:
     normalizationRepository: airbyte/normalization-tidb
-    normalizationTag: 0.3.1
+    normalizationTag: 0.3.2
     normalizationIntegrationType: tidb
   supportsDbt: true
 - name: Typesense

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -28,5 +28,5 @@ WORKDIR /airbyte
 ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.3.1
+LABEL io.airbyte.version=0.3.2
 LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/normalization/transform_catalog/catalog_processor.py

@@ -135,7 +135,17 @@ def build_stream_processor(
                 primary_key = get_field(configured_stream, "primary_key", f"Undefined primary key for stream {stream_name}")
 
             message = f"'json_schema'.'properties' are not defined for stream {stream_name}"
-            properties = get_field(get_field(stream_config, "json_schema", message), "properties", message)
+            stream_schema = get_field(stream_config, "json_schema", f"'json_schema' is not defined for stream {stream_name}")
+            if "properties" in stream_schema:
+                properties = get_field(stream_schema, "properties", message)
+            elif "oneOf" in stream_schema:
+                options = list(filter(lambda option: "properties" in option, stream_schema["oneOf"]))
+                if len(options) == 0:
+                    raise KeyError(f"Stream {stream_name} does not have any properties")
+                # If there are multiple oneOf options, just pick the first one - we don't really support oneOf to begin with
+                properties = options[0]["properties"]
+            else:
+                raise KeyError(f"Stream {stream_name} does not have any properties and no oneOf option with properties")
 
             from_table = dbt_macro.Source(schema_name, raw_table_name)
 

airbyte-integrations/bases/base-normalization/normalization/transform_catalog/utils.py

@@ -31,11 +31,14 @@ def is_reftype(definition: dict) -> bool:
 
 
 def is_string(definition: dict) -> bool:
-    return is_type_included(definition, get_reftype_function(data_type.STRING_TYPE))
+    return is_type_included(definition, get_reftype_function(data_type.STRING_TYPE)) or is_type_included(
+        definition, get_reftype_function(data_type.BINARY_DATA_TYPE)
+    )
 
 
 def is_binary_datatype(definition: dict) -> bool:
-    return is_type_included(definition, get_reftype_function(data_type.BINARY_DATA_TYPE))
+    return False
+    # return is_type_included(definition, get_reftype_function(data_type.BINARY_DATA_TYPE))
 
 
 def is_datetime(definition: dict) -> bool:

docs/understanding-airbyte/basic-normalization.md

@@ -353,6 +353,7 @@ Therefore, in order to "upgrade" to the desired normalization version, you need
 
 | Airbyte Version | Normalization Version | Date       | Pull Request                                                                                                                         | Subject                                                                                                              |
 |:----------------|:----------------------|:-----------|:-------------------------------------------------------------------------------------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------|
+|                 | 0.3.2                 | 2023-01-31 | [\#22165](https://github.com/airbytehq/airbyte/pull/22165)                                                                           | Fix support for non-object top-level schemas |
 |                 | 0.3.1                 | 2023-01-31 | [\#22161](https://github.com/airbytehq/airbyte/pull/22161)                                                                           | Fix handling for combined primitive types |
 |                 | 0.3.0                 | 2023-01-30 | [\#19721](https://github.com/airbytehq/airbyte/pull/19721)                                                                           | Update normalization to airbyte-protocol v1.0.0 |
 |                 | 0.2.25                | 2022-12-05 | [\#19573](https://github.com/airbytehq/airbyte/pull/19573)                                                                           | Update Clickhouse dbt version |