Timescaledb: handle schemas

Author: Alex Toker

storey/timescaledb_target.py

@@ -112,6 +112,9 @@ def __init__(
         self._dsn = dsn
         self._pool = None  # Connection pool will be created lazily during first use
         self._column_names = self._get_column_names()
+        self._schema = None
+        if "." in self._table:
+            self._schema, self._table = self._table.split(".", 1)
 
     def _init(self):
         """Initialize the target (called synchronously).
@@ -193,7 +196,9 @@ async def _emit(self, batch, batch_key, batch_time, batch_events, last_event_tim
         async with self._pool.acquire() as conn:
             # Use PostgreSQL's COPY protocol for optimal performance
             # This is significantly faster than individual INSERT statements
-            await conn.copy_records_to_table(self._table, records=records, columns=self._column_names)
+            await conn.copy_records_to_table(
+                self._table, schema_name=self._schema, records=records, columns=self._column_names
+            )
 
     async def _terminate(self):
         """Terminate and cleanup resources.