feat: add skip_rows_before_header and skip_rows_after_header to CsvDecoder

Add skip_rows_before_header and skip_rows_after_header optional integer
properties (default: 0) to the declarative CsvDecoder schema and parser.
This enables Connector Builder users to parse CSV/TSV responses that
contain metadata lines before or after the header row.

The implementation follows the same pattern already used by file-based
connectors (CsvFormat in csv_format.py) and mirrors the approach used
when set_values_to_none was added to CsvDecoder.

Co-Authored-By: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3886,6 +3886,16 @@ definitions:
         type: array
         items:
           type: string
+      skip_rows_before_header:
+        title: Skip Rows Before Header
+        description: The number of rows to skip before the header row. For example, if the header row is on the 3rd row, enter 2 in this field.
+        type: integer
+        default: 0
+      skip_rows_after_header:
+        title: Skip Rows After Header
+        description: The number of rows to skip after the header row.
+        type: integer
+        default: 0
   AsyncJobStatusMap:
     description: Matches the api job status to Async Job Status.
     type: object

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -104,24 +104,35 @@ class CsvParser(Parser):
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
     set_values_to_none: Optional[List[str]] = None
+    skip_rows_before_header: int = 0
+    skip_rows_after_header: int = 0
 
     def _get_delimiter(self) -> Optional[str]:
-        """
-        Get delimiter from the configuration. Check for the escape character and decode it.
-        """
+        """Get delimiter from the configuration. Check for the escape character and decode it."""
         if self.delimiter is not None:
             if self.delimiter.startswith("\\"):
                 self.delimiter = self.delimiter.encode("utf-8").decode("unicode_escape")
 
         return self.delimiter
 
+    @staticmethod
+    def _skip_rows(text_data: TextIOWrapper, rows_to_skip: int) -> None:
+        """Skip a specified number of rows from the current position in the text stream."""
+        for _ in range(rows_to_skip):
+            text_data.readline()
+
     def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
-        """
-        Parse CSV data from decompressed bytes.
-        """
+        """Parse CSV data from decompressed bytes."""
         text_data = TextIOWrapper(data, encoding=self.encoding)  # type: ignore
+        self._skip_rows(text_data, self.skip_rows_before_header)
         reader = csv.DictReader(text_data, delimiter=self._get_delimiter() or ",")
+        # After DictReader reads the header (first row it encounters), skip additional rows
+        # We need to handle skip_rows_after_header by consuming rows from the reader
+        skipped_after_header = 0
         for row in reader:
+            if skipped_after_header < self.skip_rows_after_header:
+                skipped_after_header += 1
+                continue
             if self.set_values_to_none:
                 row = {k: (None if v in self.set_values_to_none else v) for k, v in row.items()}
             yield row