fix: gate 'NA' string preservation behind opt-in preserve_na_values flag

ResponseToFileExtractor previously always passed keep_default_na=False,
which changed CSV parsing for every connector using it. Make the behavior
opt-in via a new preserve_na_values field (default False), preserving the
historical behavior (pandas NA tokens -> null) unless explicitly enabled.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tolik0

airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py

@@ -27,9 +27,14 @@ class ResponseToFileExtractor(RecordExtractor):
 
     Eventually, we want to support multiple file type by re-using the file based CDK parsers if possible. However, the lift is too high for
     a first iteration so we will only support CSV parsing using pandas as salesforce and sendgrid were doing.
+
+    By default, pandas interprets strings such as "NA", "N/A", "null" and "NaN" as missing values and converts them to null. When
+    ``preserve_na_values`` is True, these strings are kept as-is (only empty cells are treated as null). This is opt-in to avoid changing
+    behavior for existing connectors.
     """
 
     parameters: InitVar[Mapping[str, Any]]
+    preserve_na_values: bool = False
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.logger = logging.getLogger("airbyte")
@@ -136,16 +141,21 @@ def _read_with_chunks(
 
         try:
             with open(path, "r", encoding=file_encoding) as data:
+                # When preserving NA values, disable pandas' default NA parsing so strings like "NA"/"N/A" are kept as-is,
+                # and explicitly map empty cells to None. Otherwise, keep the historical behavior where pandas converts its
+                # default NA tokens (including empty cells) to NaN, which we then map to None.
+                read_kwargs = {"keep_default_na": False} if self.preserve_na_values else {}
+                na_replacements = {nan: None, "": None} if self.preserve_na_values else {nan: None}
                 chunks = pd.read_csv(
                     data,
                     chunksize=chunk_size,
                     iterator=True,
                     dialect="unix",
                     dtype=object,
-                    keep_default_na=False,
+                    **read_kwargs,
                 )
                 for chunk in chunks:
-                    chunk = chunk.replace({nan: None, "": None}).to_dict(orient="records")
+                    chunk = chunk.replace(na_replacements).to_dict(orient="records")
                     for row in chunk:
                         yield row
         except pd.errors.EmptyDataError as e:

unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py

@@ -15,6 +15,7 @@
 class ResponseToFileExtractorTest(TestCase):
     def setUp(self) -> None:
         self._extractor = ResponseToFileExtractor({})
+        self._preserving_extractor = ResponseToFileExtractor({}, preserve_na_values=True)
         self._http_mocker = requests_mock.Mocker()
         self._http_mocker.__enter__()
 
@@ -39,13 +40,13 @@ def test_text_response_with_null_bytes(self) -> None:
 
         assert extracted_records == [{"FIRST_NAME": "a first name", "LAST_NAME": "a last name"}]
 
-    def test_na_string_values_are_preserved(self) -> None:
+    def test_na_string_values_are_preserved_when_enabled(self) -> None:
         pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", "<NA>"]
         csv_rows = "\n".join(f'"{token}","{token} value"' for token in pandas_na_tokens)
         csv_with_na = f'"region","name"\n{csv_rows}\n'
         response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8")))
 
-        extracted_records = list(self._extractor.extract_records(response))
+        extracted_records = list(self._preserving_extractor.extract_records(response))
 
         assert len(extracted_records) == len(pandas_na_tokens)
         for record, token in zip(extracted_records, pandas_na_tokens):
@@ -54,11 +55,24 @@ def test_na_string_values_are_preserved(self) -> None:
             )
             assert record["name"] == f"{token} value"
 
-    def test_empty_fields_are_converted_to_none(self) -> None:
+    def test_na_string_values_are_converted_to_none_by_default(self) -> None:
+        pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", "<NA>"]
+        csv_rows = "\n".join(f'"{token}","preserved"' for token in pandas_na_tokens)
+        csv_with_na = f'"region","name"\n{csv_rows}\n'
+        response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8")))
+
+        extracted_records = list(self._extractor.extract_records(response))
+
+        assert len(extracted_records) == len(pandas_na_tokens)
+        for record in extracted_records:
+            assert record["region"] is None
+            assert record["name"] == "preserved"
+
+    def test_empty_fields_are_converted_to_none_when_preserving(self) -> None:
         csv_with_empty = '"region","name"\n"NA","North America"\n,""\n'
         response = self._mock_streamed_response(BytesIO(csv_with_empty.encode("utf-8")))
 
-        extracted_records = list(self._extractor.extract_records(response))
+        extracted_records = list(self._preserving_extractor.extract_records(response))
 
         assert extracted_records == [
             {"region": "NA", "name": "North America"},