feat: add insert rows json function in redshift

Author: vishnurk6247

wavefront/server/packages/flo_cloud/flo_cloud/aws/redshift.py

@@ -1,4 +1,5 @@
 import os
+import json
 import string
 import logging
 from typing import List, Dict, Any, Optional, Tuple
@@ -328,6 +329,52 @@ def execute_many(self, command: str, params_list: List[Dict[str, Any]]) -> int:
                 logger.error(f'Batch execution error: {e}')
                 raise
 
+    def insert_rows_json(self, table_name: str, data: List[Dict[str, Any]]) -> int:
+        """
+        Insert rows provided as a list of dictionaries into a table.
+
+        Dict/list values are serialized to JSON strings so they can be stored
+        in VARCHAR/SUPER columns. Columns are derived from the first row, so all
+        rows are expected to share the same keys.
+
+        Args:
+            table_name: Fully qualified target table name (e.g. 'db.schema.table')
+            data: List of dictionaries, each representing a row
+
+        Returns:
+            Number of rows inserted
+        """
+        if not data:
+            return 0
+
+        columns = list(data[0].keys())
+        serialized = [
+            {
+                col: json.dumps(row.get(col))
+                if isinstance(row.get(col), (dict, list))
+                else row.get(col)
+                for col in columns
+            }
+            for row in data
+        ]
+
+        column_list = ', '.join(f'"{col}"' for col in columns)
+        placeholders = ', '.join(f':{col}' for col in columns)
+        command = f'INSERT INTO {table_name} ({column_list}) VALUES ({placeholders})'
+
+        with self.get_connection() as connection:
+            cursor = connection.cursor()
+            try:
+                cursor.executemany(command, serialized)
+                connection.commit()
+                return cursor.rowcount
+            except RedshiftError as e:
+                connection.rollback()
+                logger.error(f'Insert error: {e}')
+                raise
+            finally:
+                cursor.close()
+
     def execute_transaction(
         self, commands: List[Tuple[str, Optional[Dict[str, Any]]]]
     ) -> bool:

wavefront/server/plugins/datasource/datasource/redshift/__init__.py

@@ -61,8 +61,8 @@ def fetch_data(
             group_by=group_by,
         )
 
-    def insert_rows_json(self, table_name: str, data):
-        pass
+    def insert_rows_json(self, table_name: str, data: List[Dict[str, Any]]) -> None:
+        self.client.insert_rows_json(f'{self.db_name}.{table_name}', data)
 
     async def execute_query(
         self, query: str, use_legacy_sql: bool = False, dry_run: bool = False, **kwargs