Merge pull request #16 from Kaligo/fix/data-24245-columns-order

fix: columns order in fast-sync CSV file mismatches the target schema [DATA-24245]

Author: khoaanguyenn

README.md

@@ -291,3 +291,24 @@ Install python dependencies and run python linter
   make venv
   make pylint
 ```
+
+### To integrate and run this locally with the Meltano project:
+
+- Use `file://` as the `pip_url` to install the plugin from the local directory.
+- Run `meltano install` again to re-install it.
+
+```yaml
+# In extract/extractors.meltano.yml
+plugins:
+  extractors:
+  - name: tap-postgres
+    variant: transferwise
+    inherit_from: tap-postgres
+    pip_url: file:///home/dev/pipelinewise-tap-postgres
+```
+
+### To create a GH release:
+
+```
+gh release create v2.2.2 --generate-notes --repo Kaligo/pipelinewise-tap-postgres
+```

tap_postgres/sync_strategies/fast_sync_rds.py

@@ -11,7 +11,6 @@
 import datetime
 import time
 import uuid
-from functools import partial
 from typing import Dict, List, Optional
 
 import psycopg2
@@ -26,6 +25,9 @@
 
 LOGGER = singer.get_logger("tap_postgres")
 
+# It's important to use lowercase to match the default column order.
+METADATA_COLUMNS = {"_sdc_batched_at", "_sdc_deleted_at", "_sdc_extracted_at"}
+
 
 class FastSyncRdsStrategy:
     """
@@ -71,21 +73,54 @@ def _generate_s3_path(self, schema_name: str, table_name: str) -> str:
         # Remove leading slash if prefix is empty to avoid double slashes
         return path.lstrip("/")
 
-    def _prepend_metadata_columns(
-        self, columns: Optional[List[str]] = None
-    ) -> List[str]:
-        # Metadata columns need to go first in the same order with
-        # pipelinewise-target-redshift/target_redshift/__init__.py#add_metadata_columns_to_schema
-        if columns is None:
-            columns = []
+    def _get_metadata_column_names(self) -> List[str]:
         if self.conn_config.get("fast_sync_rds_add_metadata_columns", True):
-            columns[:0] = [
-                "NOW() AT TIME ZONE 'UTC' AS _SDC_BATCHED_AT",
-                "NULL AS _SDC_DELETED_AT",
-                "NOW() AT TIME ZONE 'UTC' AS _SDC_EXTRACTED_AT",
-            ]
+            return METADATA_COLUMNS
+        return []
+
+    def _get_metadata_column_sql(self, column_name: str) -> str:
+        """Get SQL expression for a metadata column."""
+        # Handle both lowercase and uppercase column names
+        column_name_lower = column_name.lower()
+
+        metadata_sql_map = {
+            "_sdc_batched_at": "current_timestamp at time zone 'UTC' as _sdc_batched_at",
+            "_sdc_deleted_at": "null as _sdc_deleted_at",
+            "_sdc_extracted_at": "current_timestamp at time zone 'UTC' as _sdc_extracted_at",
+        }
+
+        if column_name_lower in metadata_sql_map:
+            return metadata_sql_map[column_name_lower]
+
+        raise ValueError(f"Unknown metadata column: {column_name}")
+
+    def _build_sorted_column_expressions(
+        self, desired_columns: List[str], md_map: Dict
+    ) -> List[str]:
+        """
+        Build SQL expressions for all columns (metadata + desired) in sorted order.
 
-        return columns
+        Columns are sorted alphabetically to match target's schema order.
+        This ensures the exported data column order matches the table column
+        order exactly.
+
+        Args:
+            desired_columns: List of desired column names from the source table
+            md_map: Metadata map for column transformations
+
+        Returns:
+            List of SQL expressions for columns in sorted order
+        """
+        metadata_column_names = self._get_metadata_column_names()
+        all_column_names = [*metadata_column_names, *desired_columns]
+        # Sort columns to ensure the output CSV headers match target's schema order.
+        all_column_names.sort()
+
+        return [
+            self._get_metadata_column_sql(name) if name in metadata_column_names
+            else post_db.prepare_columns_for_select_sql(name, md_map=md_map)
+            for name in all_column_names
+        ]
 
     def _build_select_query(  # pylint: disable=too-many-arguments,too-many-positional-arguments
         self,
@@ -97,14 +132,7 @@ def _build_select_query(  # pylint: disable=too-many-arguments,too-many-position
         replication_key_value: Optional[str] = None,
         replication_key_sql_datatype: Optional[str] = None,
     ) -> str:
-        columns = self._prepend_metadata_columns([])
-        escaped_columns = list(
-            map(
-                partial(post_db.prepare_columns_for_select_sql, md_map=md_map),
-                desired_columns,
-            )
-        )
-        columns.extend(escaped_columns)
+        columns = self._build_sorted_column_expressions(desired_columns, md_map)
 
         return sync_common.get_query_for_replication_data(
             {

tests/unit/test_fast_sync_rds.py

@@ -144,14 +144,14 @@ def test_sync_table_full(self, mock_open_conn):
         # Verify metadata columns are in the query (tests _prepend_metadata_columns indirectly)
         export_query = self._extract_export_query(mock_cursor)
         self.assertIsNotNone(export_query)
-        self.assertIn("_SDC_BATCHED_AT", export_query)
-        self.assertIn("_SDC_DELETED_AT", export_query)
-        self.assertIn("_SDC_EXTRACTED_AT", export_query)
-
-        # Verify metadata columns order
-        batched_pos = export_query.find("_SDC_BATCHED_AT")
-        deleted_pos = export_query.find("_SDC_DELETED_AT")
-        extracted_pos = export_query.find("_SDC_EXTRACTED_AT")
+        self.assertIn("_sdc_batched_at", export_query)
+        self.assertIn("_sdc_deleted_at", export_query)
+        self.assertIn("_sdc_extracted_at", export_query)
+
+        # Verify metadata columns are present and in correct order
+        batched_pos = export_query.find("_sdc_batched_at")
+        deleted_pos = export_query.find("_sdc_deleted_at")
+        extracted_pos = export_query.find("_sdc_extracted_at")
         id_pos = export_query.find('"id"')
 
         self.assertGreater(batched_pos, 0)
@@ -160,6 +160,65 @@ def test_sync_table_full(self, mock_open_conn):
         if id_pos > 0:
             self.assertGreater(id_pos, extracted_pos)
 
+    @patch("tap_postgres.sync_strategies.fast_sync_rds.post_db.open_connection")
+    def test_sync_table_full_column_ordering(self, mock_open_conn):
+        """Test sync_table_full orders columns alphabetically including metadata columns"""
+        desired_columns = ["zebra", "_id", "active"]
+        md_map_with_columns = {
+            (): {"schema-name": "test_schema"},
+            ("properties", "_id"): {"sql-datatype": "integer"},
+            ("properties", "active"): {"sql-datatype": "boolean"},
+            ("properties", "zebra"): {"sql-datatype": "varchar"},
+        }
+
+        mock_conn, mock_cursor = self._setup_mock_connection()
+        mock_open_conn.return_value.__enter__.return_value = mock_conn
+
+        self.fast_sync_rds_strategy.sync_table_full(
+            stream=self.stream,
+            state=self.state,
+            desired_columns=desired_columns,
+            md_map=md_map_with_columns,
+        )
+
+        # Verify export query contains all columns
+        export_query = self._extract_export_query(mock_cursor)
+        self.assertIsNotNone(export_query)
+        self.assertIn("_sdc_batched_at", export_query)
+        self.assertIn("_sdc_deleted_at", export_query)
+        self.assertIn("_sdc_extracted_at", export_query)
+        self.assertIn('"_id"', export_query)
+        self.assertIn('"active"', export_query)
+        self.assertIn('"zebra"', export_query)
+
+        # Verify complete column ordering: all columns should be sorted alphabetically
+        # Expected order: _id, _sdc_batched_at, _sdc_deleted_at, _sdc_extracted_at, active, zebra
+        expected_columns = [
+            '"_id"',
+            "_sdc_batched_at",
+            "_sdc_deleted_at",
+            "_sdc_extracted_at",
+            '"active"',
+            '"zebra"',
+        ]
+
+        # Find positions of all columns
+        column_positions = [export_query.find(col) for col in expected_columns]
+
+        self.assertEqual(len(column_positions), len(expected_columns))
+
+        # Verify all columns are found
+        for i, (col, pos) in enumerate(zip(expected_columns, column_positions)):
+            self.assertGreater(pos, 0, f"{col} should be found in query")
+
+        # Verify columns are in correct order (positions should be ascending)
+        self.assertEqual(
+            column_positions,
+            sorted(column_positions),
+            f"Columns should be in alphabetical order. Found positions: {dict(zip(expected_columns, column_positions))}",
+        )
+
+
     @patch("tap_postgres.sync_strategies.fast_sync_rds.post_db.open_connection")
     def test_sync_table_full_no_prefix(self, mock_open_conn):
         """Test sync_table_full with empty prefix - verifies S3 path generation"""
@@ -205,9 +264,9 @@ def test_sync_table_full_metadata_columns_disabled(self, mock_open_conn):
         # Verify metadata columns are NOT in the query
         export_query = self._extract_export_query(mock_cursor)
         self.assertIsNotNone(export_query)
-        self.assertNotIn("_SDC_BATCHED_AT", export_query)
-        self.assertNotIn("_SDC_DELETED_AT", export_query)
-        self.assertNotIn("_SDC_EXTRACTED_AT", export_query)
+        self.assertNotIn("_sdc_batched_at", export_query)
+        self.assertNotIn("_sdc_deleted_at", export_query)
+        self.assertNotIn("_sdc_extracted_at", export_query)
 
     @patch("tap_postgres.sync_strategies.fast_sync_rds.post_db.open_connection")
     @patch("tap_postgres.sync_strategies.fast_sync_rds.singer.get_bookmark")