Merge pull request #19 from Kaligo/DATA-27839/define-on-fly-mapper-cols

feat: featuring adding new on-fly col if configured [DATA-27839]

Author: quocnguyendinh

setup.py

@@ -7,7 +7,7 @@
 
 setup(
     name="pipelinewise-tap-postgres",
-    version="2.2.1",
+    version="2.2.6",
     description="Singer.io tap for extracting data from PostgresSQL - PipelineWise compatible",
     long_description=long_description,
     long_description_content_type="text/markdown",

tap_postgres/__init__.py

@@ -150,8 +150,14 @@ def do_fast_sync_rds(conn_config, stream, state, desired_columns, md_map, sync_m
     state = singer.write_bookmark(
         state, stream["tap_stream_id"], "replication_key", replication_key
     )
+
+    desired_columns = fast_sync_rds_strategy.prepare_stream_with_extra_columns(
+        desired_columns, stream
+    )
+
     sync_common.send_schema_message(
-        stream, [replication_key] if replication_key else []
+        stream,
+        [replication_key] if replication_key else [],
     )
 
     LOGGER.info("Performing %s", sync_method)
@@ -598,15 +604,15 @@ def main_impl():
         ),
         "use_secondary": args.config.get("use_secondary", False),
         "limit": int(limit) if limit else None,
-        "skip_last_n_seconds": int(skip_last_n_seconds)
-        if skip_last_n_seconds
-        else None,
-        "look_back_n_seconds": int(look_back_n_seconds)
-        if look_back_n_seconds
-        else None,
-        "recover_mappings": ast.literal_eval(recover_mappings)
-        if recover_mappings
-        else {},
+        "skip_last_n_seconds": (
+            int(skip_last_n_seconds) if skip_last_n_seconds else None
+        ),
+        "look_back_n_seconds": (
+            int(look_back_n_seconds) if look_back_n_seconds else None
+        ),
+        "recover_mappings": (
+            ast.literal_eval(recover_mappings) if recover_mappings else {}
+        ),
         # Fast sync RDS config
         "fast_sync_rds": args.config.get("fast_sync_rds", False),
         "fast_sync_rds_s3_bucket": args.config.get("fast_sync_rds_s3_bucket"),

tap_postgres/sync_strategies/fast_sync_rds.py

@@ -77,6 +77,25 @@ def _get_metadata_column_names(self) -> List[str]:
         if self.conn_config.get("fast_sync_rds_add_metadata_columns", True):
             return METADATA_COLUMNS
         return []
+    
+    def _get_fast_sync_rds_transformation(self, tap_stream_id: str) -> Dict[str, str]:
+        return self.conn_config.get("fast_sync_rds_transformations", {}).get(
+            tap_stream_id, {}
+        )
+
+    def _get_extra_column_properties(
+        self, desired_columns: List[str], stream: Dict
+    ) -> List[str]:
+        transformation_columns = self._get_fast_sync_rds_transformation(
+            stream["tap_stream_id"]
+        )
+        all_columns = {*self._get_metadata_column_names(), *desired_columns}
+        extra_columns = [
+            col for col in transformation_columns if col not in all_columns
+        ]
+
+        # On-the-fly columns use string type; other types could be added if needed.
+        return {col: {"type": ["string", "null"]} for col in extra_columns}
 
     def _get_metadata_column_sql(self, column_name: str) -> str:
         """Get SQL expression for a metadata column."""
@@ -155,12 +174,7 @@ def _build_sorted_column_expressions(
         all_column_names.sort()
 
         # Get transformations for this stream if available
-        transformations = {}
-        if tap_stream_id:
-            all_transformations = self.conn_config.get(
-                "fast_sync_rds_transformations", {}
-            )
-            transformations = all_transformations.get(tap_stream_id, {})
+        transformations = self._get_fast_sync_rds_transformation(tap_stream_id)
 
         column_expressions = []
         for name in all_column_names:
@@ -380,6 +394,30 @@ def _track_latest_replication_key_value(  # pylint: disable=invalid-name,too-man
 
         return state
 
+    def prepare_stream_with_extra_columns(
+        self, desired_columns: List[str], stream: Dict
+    ):
+        """
+        Prepare stream with extra columns for fast sync RDS strategy.
+        Extra columns are columns that are created on the fly (not in orginal catalog)
+        provided by fast_sync_rds_transformations configuration.
+        To make it simple, their type is always ["string", "null"].
+
+        Args:
+            desired_columns: List of desired column names from the source table
+            stream: Stream dictionary
+
+        Returns:
+            List of desired column names with extra columns
+        """
+        extra_columns_properties = self._get_extra_column_properties(
+            desired_columns, stream
+        )
+        stream["schema"]["properties"].update(extra_columns_properties)
+        desired_columns.extend(extra_columns_properties.keys())
+        desired_columns.sort()
+        return desired_columns
+
     def sync_table(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
         self,
         stream: Dict,