From e46e859897ade025754aa1c9a049e5fcd061fda0 Mon Sep 17 00:00:00 2001
From: Ben Theunissen <ben@shaped.ai>
Date: Mon, 31 Jul 2023 14:26:36 -0600
Subject: [PATCH] Allow override for target table name (#17)

* Fix

* Fix version
---
 pyproject.toml             |  2 +-
 target_clickhouse/sinks.py | 22 ++++++++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index a34ba10..100f03f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "shaped-target-clickhouse"
-version = "0.0.3"
+version = "0.0.4"
 description = "`target-clickhouse` is a Singer target for clickhouse, built with the Meltano Singer SDK."
 readme = "README.md"
 authors = ["Ben Theunissen"]
diff --git a/target_clickhouse/sinks.py b/target_clickhouse/sinks.py
index 2d6d426..5bd358e 100644
--- a/target_clickhouse/sinks.py
+++ b/target_clickhouse/sinks.py
@@ -95,6 +95,11 @@ def create_empty_table(
         _ = partition_keys  # Not supported in generic implementation.
 
         _, _, table_name = self.parse_full_table_name(full_table_name)
+
+        # If config table name is set, then use it instead of the table name.
+        if self.config.get("table_name"):
+            table_name = self.config.get("table_name")
+
         # Do not set schema, as it is not supported by Clickhouse.
         meta = MetaData(schema=None, bind=self._engine)
         columns: list[Column] = []
@@ -132,3 +137,20 @@ class ClickhouseSink(SQLSink):
     """clickhouse target sink class."""
 
     connector_class = ClickhouseConnector
+
+    @property
+    def full_table_name(self) -> str:
+        """Return the fully qualified table name.
+
+        Returns
+            The fully qualified table name.
+        """
+        # Use the config table name if set.
+        if self.config.get("table_name"):
+            return self.config.get("table_name")
+
+        return self.connector.get_fully_qualified_name(
+            table_name=self.table_name,
+            schema_name=self.schema_name,
+            db_name=self.database_name,
+        )