[airbyte-cdk] Increase the maximum parseable field size for CSV files (…

…#36320)

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

@@ -131,7 +131,11 @@ def _skip_rows(fp: IOBase, rows_to_skip: int) -> None:
 class CsvParser(FileTypeParser):
     _MAX_BYTES_PER_FILE_FOR_SCHEMA_INFERENCE = 1_000_000
 
-    def __init__(self, csv_reader: Optional[_CsvReader] = None):
+    def __init__(self, csv_reader: Optional[_CsvReader] = None, csv_field_max_bytes: int = 2**31):
+        # Increase the maximum length of data that can be parsed in a single CSV field. The default is 128k, which is typically sufficient
+        # but given the use of Airbyte in loading a large variety of data it is best to allow for a larger maximum field size to avoid
+        # skipping data on load. https://stackoverflow.com/questions/15063936/csv-error-field-larger-than-field-limit-131072
+        csv.field_size_limit(csv_field_max_bytes)
         self._csv_reader = csv_reader if csv_reader else _CsvReader()
 
     def check_config(self, config: FileBasedStreamConfig) -> Tuple[bool, Optional[str]]:

airbyte-cdk/python/unit_tests/sources/file_based/file_types/test_csv_parser.py

@@ -501,6 +501,25 @@ def test_given_too_few_values_for_columns_when_read_data_then_raise_exception_an
             next(data_generator)
         assert new_dialect not in csv.list_dialects()
 
+    def test_parse_field_size_larger_than_default_python_maximum(self) -> None:
+        # The field size for the csv module will be set as a side-effect of initializing the CsvParser class.
+        assert csv.field_size_limit() == 2**31
+        long_string = 130 * 1024 * "a"
+        assert len(long_string.encode("utf-8")) > (128 * 1024)
+        self._stream_reader.open_file.return_value = (
+            CsvFileBuilder()
+            .with_data(
+                [
+                    "header1,header2",
+                    f'1,"{long_string}"',
+                ]
+            )
+            .build()
+        )
+
+        data_generator = self._read_data()
+        assert list(data_generator) == [{"header1": "1", "header2": long_string}]
+
     def _read_data(self) -> Generator[Dict[str, str], None, None]:
         data_generator = self._csv_reader.read_data(
             self._config,