🎉 Airbyte CDK (File-based CDK): Stop the sync if the record could not be parsed

airbyte-cdk/python/.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.58.5
+current_version = 0.58.6
 commit = False
 
 [bumpversion:file:setup.py]

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.58.6
+File CDK: Added logic to emit logged `RecordParseError` errors and raise the single `AirbyteTracebackException` in the end of the sync, instead of silent skipping the parsing errors. PR: https://github.com/airbytehq/airbyte/pull/32589
+
 ## 0.58.5
 Handle private network exception as config error
 

airbyte-cdk/python/Dockerfile

@@ -32,5 +32,5 @@ ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
 # needs to be the same as CDK
-LABEL io.airbyte.version=0.58.5
+LABEL io.airbyte.version=0.58.6
 LABEL io.airbyte.name=airbyte/source-declarative-manifest

airbyte-cdk/python/airbyte_cdk/sources/file_based/config/avro_format.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+
 from airbyte_cdk.utils.oneof_option_config import OneOfOptionConfig
 from pydantic import BaseModel, Field
 

airbyte-cdk/python/airbyte_cdk/sources/file_based/config/parquet_format.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+
 from airbyte_cdk.utils.oneof_option_config import OneOfOptionConfig
 from pydantic import BaseModel, Field
 

airbyte-cdk/python/airbyte_cdk/sources/file_based/exceptions.py

@@ -3,8 +3,9 @@
 #
 
 from enum import Enum
-from typing import Union
+from typing import Any, List, Union
 
+from airbyte_cdk.models import AirbyteMessage, FailureType
 from airbyte_cdk.utils import AirbyteTracedException
 
 
@@ -40,6 +41,30 @@ class FileBasedSourceError(Enum):
     UNDEFINED_VALIDATION_POLICY = "The validation policy defined in the config does not exist for the source."
 
 
+class FileBasedErrorsCollector:
+    """
+    The placeholder for all errors collected.
+    """
+
+    errors: List[AirbyteMessage] = []
+
+    def yield_and_raise_collected(self) -> Any:
+        if self.errors:
+            # emit collected logged messages
+            yield from self.errors
+            # clean the collector
+            self.errors.clear()
+            # raising the single exception
+            raise AirbyteTracedException(
+                internal_message="Please check the logged errors for more information.",
+                message="Some errors occured while reading from the source.",
+                failure_type=FailureType.config_error,
+            )
+
+    def collect(self, logged_error: AirbyteMessage) -> None:
+        self.errors.append(logged_error)
+
+
 class BaseFileBasedSourceError(Exception):
     def __init__(self, error: Union[FileBasedSourceError, str], **kwargs):  # type: ignore # noqa
         if isinstance(error, FileBasedSourceError):

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_based_source.py

@@ -14,7 +14,7 @@
 from airbyte_cdk.sources.file_based.config.abstract_file_based_spec import AbstractFileBasedSpec
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig, ValidationPolicy
 from airbyte_cdk.sources.file_based.discovery_policy import AbstractDiscoveryPolicy, DefaultDiscoveryPolicy
-from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, FileBasedSourceError
+from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, FileBasedErrorsCollector, FileBasedSourceError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.file_types import default_parsers
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
@@ -49,6 +49,7 @@ def __init__(
         self.stream_schemas = {s.stream.name: s.stream.json_schema for s in catalog.streams} if catalog else {}
         self.cursor_cls = cursor_cls
         self.logger = logging.getLogger(f"airbyte.{self.name}")
+        self.errors_collector: FileBasedErrorsCollector = FileBasedErrorsCollector()
 
     def check_connection(self, logger: logging.Logger, config: Mapping[str, Any]) -> Tuple[bool, Optional[Any]]:
         """
@@ -106,6 +107,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
                         parsers=self.parsers,
                         validation_policy=self._validate_and_get_validation_policy(stream_config),
                         cursor=self.cursor_cls(stream_config),
+                        errors_collector=self.errors_collector,
                     )
                 )
             return streams
@@ -121,6 +123,8 @@ def read(
         state: Optional[Union[List[AirbyteStateMessage], MutableMapping[str, Any]]] = None,
     ) -> Iterator[AirbyteMessage]:
         yield from super().read(logger, config, catalog, state)
+        # emit all the errors collected
+        yield from self.errors_collector.yield_and_raise_collected()
         # count streams using a certain parser
         parsed_config = self._get_parsed_config(config)
         for parser, count in Counter(stream.format.filetype for stream in parsed_config.streams).items():

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_types/avro_parser.py

@@ -8,6 +8,7 @@
 import fastavro
 from airbyte_cdk.sources.file_based.config.avro_format import AvroFormat
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig
+from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader, FileReadMode
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
@@ -144,15 +145,20 @@ def parse_records(
         if not isinstance(avro_format, AvroFormat):
             raise ValueError(f"Expected ParquetFormat, got {avro_format}")
 
-        with stream_reader.open_file(file, self.file_read_mode, self.ENCODING, logger) as fp:
-            avro_reader = fastavro.reader(fp)
-            schema = avro_reader.writer_schema
-            schema_field_name_to_type = {field["name"]: field["type"] for field in schema["fields"]}
-            for record in avro_reader:
-                yield {
-                    record_field: self._to_output_value(avro_format, schema_field_name_to_type[record_field], record[record_field])
-                    for record_field, record_value in schema_field_name_to_type.items()
-                }
+        line_no = 0
+        try:
+            with stream_reader.open_file(file, self.file_read_mode, self.ENCODING, logger) as fp:
+                avro_reader = fastavro.reader(fp)
+                schema = avro_reader.writer_schema
+                schema_field_name_to_type = {field["name"]: field["type"] for field in schema["fields"]}
+                for record in avro_reader:
+                    line_no += 1
+                    yield {
+                        record_field: self._to_output_value(avro_format, schema_field_name_to_type[record_field], record[record_field])
+                        for record_field, record_value in schema_field_name_to_type.items()
+                    }
+        except Exception as exc:
+            raise RecordParseError(FileBasedSourceError.ERROR_PARSING_RECORD, filename=file.uri, lineno=line_no) from exc
 
     @property
     def file_read_mode(self) -> FileReadMode:

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_types/csv_parser.py

@@ -178,17 +178,25 @@ def parse_records(
         logger: logging.Logger,
         discovered_schema: Optional[Mapping[str, SchemaType]],
     ) -> Iterable[Dict[str, Any]]:
-        config_format = _extract_format(config)
-        if discovered_schema:
-            property_types = {col: prop["type"] for col, prop in discovered_schema["properties"].items()}  # type: ignore # discovered_schema["properties"] is known to be a mapping
-            deduped_property_types = CsvParser._pre_propcess_property_types(property_types)
-        else:
-            deduped_property_types = {}
-        cast_fn = CsvParser._get_cast_function(deduped_property_types, config_format, logger, config.schemaless)
-        data_generator = self._csv_reader.read_data(config, file, stream_reader, logger, self.file_read_mode)
-        for row in data_generator:
-            yield CsvParser._to_nullable(cast_fn(row), deduped_property_types, config_format.null_values, config_format.strings_can_be_null)
-        data_generator.close()
+        line_no = 0
+        try:
+            config_format = _extract_format(config)
+            if discovered_schema:
+                property_types = {col: prop["type"] for col, prop in discovered_schema["properties"].items()}  # type: ignore # discovered_schema["properties"] is known to be a mapping
+                deduped_property_types = CsvParser._pre_propcess_property_types(property_types)
+            else:
+                deduped_property_types = {}
+            cast_fn = CsvParser._get_cast_function(deduped_property_types, config_format, logger, config.schemaless)
+            data_generator = self._csv_reader.read_data(config, file, stream_reader, logger, self.file_read_mode)
+            for row in data_generator:
+                line_no += 1
+                yield CsvParser._to_nullable(
+                    cast_fn(row), deduped_property_types, config_format.null_values, config_format.strings_can_be_null
+                )
+        except RecordParseError as parse_err:
+            raise RecordParseError(FileBasedSourceError.ERROR_PARSING_RECORD, filename=file.uri, lineno=line_no) from parse_err
+        finally:
+            data_generator.close()
 
     @property
     def file_read_mode(self) -> FileReadMode:

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_types/jsonl_parser.py

@@ -119,7 +119,7 @@ def _parse_jsonl_entries(
                     break
 
             if had_json_parsing_error and not yielded_at_least_once:
-                raise RecordParseError(FileBasedSourceError.ERROR_PARSING_RECORD)
+                raise RecordParseError(FileBasedSourceError.ERROR_PARSING_RECORD, filename=file.uri, lineno=line)
 
     @staticmethod
     def _instantiate_accumulator(line: Union[bytes, str]) -> Union[bytes, str]:

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_types/parquet_parser.py

@@ -11,7 +11,7 @@
 import pyarrow as pa
 import pyarrow.parquet as pq
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig, ParquetFormat
-from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, FileBasedSourceError
+from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, FileBasedSourceError, RecordParseError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader, FileReadMode
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
@@ -64,19 +64,27 @@ def parse_records(
         if not isinstance(parquet_format, ParquetFormat):
             logger.info(f"Expected ParquetFormat, got {parquet_format}")
             raise ConfigValidationError(FileBasedSourceError.CONFIG_VALIDATION_ERROR)
-        with stream_reader.open_file(file, self.file_read_mode, self.ENCODING, logger) as fp:
-            reader = pq.ParquetFile(fp)
-            partition_columns = {x.split("=")[0]: x.split("=")[1] for x in self._extract_partitions(file.uri)}
-            for row_group in range(reader.num_row_groups):
-                batch = reader.read_row_group(row_group)
-                for row in range(batch.num_rows):
-                    yield {
-                        **{
-                            column: ParquetParser._to_output_value(batch.column(column)[row], parquet_format)
-                            for column in batch.column_names
-                        },
-                        **partition_columns,
-                    }
+
+        line_no = 0
+        try:
+            with stream_reader.open_file(file, self.file_read_mode, self.ENCODING, logger) as fp:
+                reader = pq.ParquetFile(fp)
+                partition_columns = {x.split("=")[0]: x.split("=")[1] for x in self._extract_partitions(file.uri)}
+                for row_group in range(reader.num_row_groups):
+                    batch = reader.read_row_group(row_group)
+                    for row in range(batch.num_rows):
+                        line_no += 1
+                        yield {
+                            **{
+                                column: ParquetParser._to_output_value(batch.column(column)[row], parquet_format)
+                                for column in batch.column_names
+                            },
+                            **partition_columns,
+                        }
+        except Exception as exc:
+            raise RecordParseError(
+                FileBasedSourceError.ERROR_PARSING_RECORD, filename=file.uri, lineno=f"{row_group=}, {line_no=}"
+            ) from exc
 
     @staticmethod
     def _extract_partitions(filepath: str) -> List[str]:

airbyte-cdk/python/airbyte_cdk/sources/file_based/stream/abstract_file_based_stream.py

@@ -10,7 +10,7 @@
 from airbyte_cdk.sources.file_based.availability_strategy import AbstractFileBasedAvailabilityStrategy
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig, PrimaryKeyType
 from airbyte_cdk.sources.file_based.discovery_policy import AbstractDiscoveryPolicy
-from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError, UndefinedParserError
+from airbyte_cdk.sources.file_based.exceptions import FileBasedErrorsCollector, FileBasedSourceError, RecordParseError, UndefinedParserError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
@@ -44,6 +44,7 @@ def __init__(
         discovery_policy: AbstractDiscoveryPolicy,
         parsers: Dict[Type[Any], FileTypeParser],
         validation_policy: AbstractSchemaValidationPolicy,
+        errors_collector: FileBasedErrorsCollector,
     ):
         super().__init__()
         self.config = config
@@ -53,6 +54,7 @@ def __init__(
         self._discovery_policy = discovery_policy
         self._availability_strategy = availability_strategy
         self._parsers = parsers
+        self.errors_collector = errors_collector
 
     @property
     @abstractmethod

airbyte-cdk/python/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py

@@ -112,12 +112,14 @@ def read_records_from_slice(self, stream_slice: StreamSlice) -> Iterable[Airbyte
             except RecordParseError:
                 # Increment line_no because the exception was raised before we could increment it
                 line_no += 1
-                yield AirbyteMessage(
-                    type=MessageType.LOG,
-                    log=AirbyteLogMessage(
-                        level=Level.ERROR,
-                        message=f"{FileBasedSourceError.ERROR_PARSING_RECORD.value} stream={self.name} file={file.uri} line_no={line_no} n_skipped={n_skipped}",
-                        stack_trace=traceback.format_exc(),
+                self.errors_collector.collect(
+                    AirbyteMessage(
+                        type=MessageType.LOG,
+                        log=AirbyteLogMessage(
+                            level=Level.ERROR,
+                            message=f"{FileBasedSourceError.ERROR_PARSING_RECORD.value} stream={self.name} file={file.uri} line_no={line_no} n_skipped={n_skipped}",
+                            stack_trace=traceback.format_exc(),
+                        ),
                     ),
                 )
 

airbyte-cdk/python/setup.py

@@ -36,7 +36,7 @@
     name="airbyte-cdk",
     # The version of the airbyte-cdk package is used at runtime to validate manifests. That validation must be
     # updated if our semver format changes such as using release candidate versions.
-    version="0.58.5",
+    version="0.58.6",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",