From caa613e1dcff8bf001e93a10118cc42465dd7942 Mon Sep 17 00:00:00 2001 From: Viet Vu Date: Thu, 20 Jun 2024 16:32:40 +0700 Subject: [PATCH 1/5] add check to disable validate_records --- target_clickhouse/sinks.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/target_clickhouse/sinks.py b/target_clickhouse/sinks.py index ce90a18..f7429b8 100644 --- a/target_clickhouse/sinks.py +++ b/target_clickhouse/sinks.py @@ -177,7 +177,8 @@ def _validate_and_parse(self, record: dict) -> dict: record = pre_validate_for_string_type(record, self.schema, self.logger) try: - self._validator.validate(record) + if self.validate_schema: + self._validator.validate(record) self._parse_timestamps_in_record( record=record, schema=self.schema, From 50f132d59c624d6412b07c2829c005b4f389a200 Mon Sep 17 00:00:00 2001 From: vietvudanh Date: Tue, 25 Jun 2024 20:05:57 +0700 Subject: [PATCH 2/5] add support for unsigned int based on minLength --- target_clickhouse/connectors.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/target_clickhouse/connectors.py b/target_clickhouse/connectors.py index db61806..60c75c8 100644 --- a/target_clickhouse/connectors.py +++ b/target_clickhouse/connectors.py @@ -91,6 +91,8 @@ def to_sql_type(self, jsonschema_type: dict) -> sqlalchemy.types.TypeEngine: """ sql_type = th.to_sql_type(jsonschema_type) + minLength = jsonschema_type['minLength'] + maxLength = jsonschema_type['maxLength'] # Clickhouse does not support the DECIMAL type without providing precision, # so we need to use the FLOAT type. @@ -99,9 +101,14 @@ def to_sql_type(self, jsonschema_type: dict) -> sqlalchemy.types.TypeEngine: sqlalchemy.types.TypeEngine, sqlalchemy.types.FLOAT(), ) elif type(sql_type) == sqlalchemy.types.INTEGER: - sql_type = typing.cast( - sqlalchemy.types.TypeEngine, clickhouse_sqlalchemy_types.Int64(), - ) + if minLength == 0: + sql_type = typing.cast( + sqlalchemy.types.TypeEngine, clickhouse_sqlalchemy_types.UInt64(), + ) + else: + sql_type = typing.cast( + sqlalchemy.types.TypeEngine, clickhouse_sqlalchemy_types.Int64(), + ) elif type(sql_type) == sqlalchemy.types.DATE: sql_type = typing.cast( sqlalchemy.types.TypeEngine, From 9deee84de1227829c0ac071350301f282502e020 Mon Sep 17 00:00:00 2001 From: vietvudanh Date: Tue, 25 Jun 2024 20:17:56 +0700 Subject: [PATCH 3/5] change block code to call parent's `super()._validate_and_parse(record)` --- target_clickhouse/sinks.py | 15 +-------------- 1 file changed, 1 insertion(+), 14 deletions(-) diff --git a/target_clickhouse/sinks.py b/target_clickhouse/sinks.py index f7429b8..d1fdcc6 100644 --- a/target_clickhouse/sinks.py +++ b/target_clickhouse/sinks.py @@ -176,20 +176,7 @@ def _validate_and_parse(self, record: dict) -> dict: # Pre-validate and correct string type mismatches. record = pre_validate_for_string_type(record, self.schema, self.logger) - try: - if self.validate_schema: - self._validator.validate(record) - self._parse_timestamps_in_record( - record=record, - schema=self.schema, - treatment=self.datetime_error_treatment, - ) - except jsonschema_exceptions.ValidationError as e: - if self.logger: - self.logger.exception(f"Record failed validation: {record}") - raise e # : RERAISES - - return record + return super()._validate_and_parse(record) def _parse_timestamps_in_record( self, From a8c66d9906d187b00f75df397684a34cf5a02cf7 Mon Sep 17 00:00:00 2001 From: vietvudanh Date: Tue, 25 Jun 2024 20:39:59 +0700 Subject: [PATCH 4/5] fix wrong key name --- target_clickhouse/connectors.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/target_clickhouse/connectors.py b/target_clickhouse/connectors.py index 60c75c8..a444163 100644 --- a/target_clickhouse/connectors.py +++ b/target_clickhouse/connectors.py @@ -91,8 +91,6 @@ def to_sql_type(self, jsonschema_type: dict) -> sqlalchemy.types.TypeEngine: """ sql_type = th.to_sql_type(jsonschema_type) - minLength = jsonschema_type['minLength'] - maxLength = jsonschema_type['maxLength'] # Clickhouse does not support the DECIMAL type without providing precision, # so we need to use the FLOAT type. @@ -101,7 +99,8 @@ def to_sql_type(self, jsonschema_type: dict) -> sqlalchemy.types.TypeEngine: sqlalchemy.types.TypeEngine, sqlalchemy.types.FLOAT(), ) elif type(sql_type) == sqlalchemy.types.INTEGER: - if minLength == 0: + minimum = jsonschema_type.get("minimum") + if minimum and minimum == 0: sql_type = typing.cast( sqlalchemy.types.TypeEngine, clickhouse_sqlalchemy_types.UInt64(), ) From 8125f677f226f009a10c409cb274a48d014f1145 Mon Sep 17 00:00:00 2001 From: Viet Vu Date: Wed, 26 Jun 2024 10:00:27 +0700 Subject: [PATCH 5/5] the value is 0, need to specify the check! --- target_clickhouse/connectors.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/target_clickhouse/connectors.py b/target_clickhouse/connectors.py index a444163..8252a60 100644 --- a/target_clickhouse/connectors.py +++ b/target_clickhouse/connectors.py @@ -100,7 +100,7 @@ def to_sql_type(self, jsonschema_type: dict) -> sqlalchemy.types.TypeEngine: ) elif type(sql_type) == sqlalchemy.types.INTEGER: minimum = jsonschema_type.get("minimum") - if minimum and minimum == 0: + if minimum is not None and minimum == 0: sql_type = typing.cast( sqlalchemy.types.TypeEngine, clickhouse_sqlalchemy_types.UInt64(), )