fix: resolve issue where some destinations will fail due to null/missing stream namespace keys in catalog

airbyte/_connector_base.py

@@ -469,6 +469,10 @@ def _execute(
             for line in self.executor.execute(args, stdin=stdin, suppress_stderr=suppress_stderr):
                 try:
                     message: AirbyteMessage = AirbyteMessage.model_validate_json(json_data=line)
+
+                    if message.type == Type.RECORD and message.record and message.record.namespace is None:
+                        message.record.namespace = ""
+
                     if progress_tracker and message.record:
                         stream_name = message.record.stream
                         progress_tracker.tally_bytes_read(

airbyte/_message_iterators.py

@@ -43,6 +43,7 @@ def _new_stream_success_message(stream_name: str) -> AirbyteMessage:
             stream_status=AirbyteStreamStatusTraceMessage(
                 stream_descriptor=StreamDescriptor(
                     name=stream_name,
+                    namespace="",  # Some destinations fail if null/missing
                 ),
                 status=AirbyteStreamStatus.COMPLETE,
                 reasons=None,
@@ -108,9 +109,8 @@ def generator() -> Generator[AirbyteMessage, None, None]:
                                     "datetime.datetime", record.get(AB_EXTRACTED_AT_COLUMN)
                                 ).timestamp()
                             ),
-                            # `meta` and `namespace` are not handled:
                             meta=None,
-                            namespace=None,
+                            namespace="",  # Some destinations fail if null/missing
                         ),
                     )
 
@@ -138,7 +138,11 @@ def generator() -> Generator[AirbyteMessage, None, None]:
                     break
                 try:
                     # Let Pydantic handle the JSON decoding from the raw string
-                    yield AirbyteMessage.model_validate_json(next_line)
+                    message = AirbyteMessage.model_validate_json(next_line)
+                    if message.type == Type.RECORD and message.record:
+                        if message.record.namespace is None:
+                            message.record.namespace = ""
+                    yield message
                 except pydantic.ValidationError:
                     # Handle JSON decoding errors (optional)
                     raise ValueError("Invalid JSON format")  # noqa: B904
@@ -153,7 +157,11 @@ def generator() -> Generator[AirbyteMessage, None, None]:
             for line in buffer:
                 try:
                     # Let Pydantic handle the JSON decoding from the raw string
-                    yield AirbyteMessage.model_validate_json(line)
+                    message = AirbyteMessage.model_validate_json(line)
+                    if message.type == Type.RECORD and message.record:
+                        if message.record.namespace is None:
+                            message.record.namespace = ""
+                    yield message
                 except pydantic.ValidationError:
                     # Handle JSON decoding errors (optional)
                     raise ValueError(f"Invalid JSON format in input string: {line}")  # noqa: B904

airbyte/shared/catalog_providers.py

@@ -55,6 +55,9 @@ def validate_catalog(catalog: ConfiguredAirbyteCatalog) -> None:
 
         This requires ensuring that `generationId` and `minGenerationId` are both set. If
         not, both values will be set to `1`.
+
+        Also replaces null namespaces with empty strings for compatibility with strict
+        destination validation.
         """
         for stream in catalog.streams:
             if stream.generation_id is None:
@@ -63,6 +66,13 @@ def validate_catalog(catalog: ConfiguredAirbyteCatalog) -> None:
                 stream.minimum_generation_id = 1
             if stream.sync_id is None:
                 stream.sync_id = 1  # This should ideally increment monotonically with each sync.
+
+            if stream.stream.namespace is None:
+                stream.stream.namespace = ""
+            if stream.stream.source_defined_cursor is None:
+                stream.stream.source_defined_cursor = False
+            if stream.stream.is_resumable is None:
+                stream.stream.is_resumable = False
 
         return catalog