🎉 New Destination: ClickHouse #7620

burmecia · 2021-11-04T05:15:55Z

What

Add new destination ClickHouse
Participating in hacktoberfest-2021 Destination: ClickHouse connector-contest#31

How

Implemented this destination connector using ClickHouse official JDBC driver.
It supports normalization and dbt, but schema change and nested stream are not support yet.
Upstream dbt clickhouse adapter has some bugs, for example "IndexError: no such group" for FixString column ClickHouse/dbt-clickhouse#20, so I made my own fork and the dbt test is based on my fork.

Test Runs

Integration test

Note: Some of the normalization and dbt test cases are disabled because,

old version of dbt generate incompatible test SQL for ClickHouse, accepted_values test doesn't work in all database engines dbt-labs/dbt-core#3905
the normalization container needs native port, while destination container needs HTTP port, we can't inject the port switch statement into DestinationAcceptanceTest.runSync() method. But I think that should be fine because there are more test cases in the following normalization test are covered.

Command used:
airbyte$ ./gradlew :airbyte-integrations:connectors:destination-clickhouse:integrationTest

Normalization test

Command used:
airbyte/airbyte-integrations/bases/base-normalization$ NORMALIZATION_TEST_TARGET=clickhouse pytest ./integration_tests/test_normalization.py

Pre-merge Checklist

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
Credentials added to Github CI. Instructions.
/test connector=connectors/<name> command is passing.
New Connector version released on Dockerhub by running the /publish command described here

Updating a connector

Community member or Airbyter

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
Credentials added to Github CI. Instructions.
/test connector=connectors/<name> command is passing.
New Connector version released on Dockerhub by running the /publish command described here

Connector Generator

Issue acceptance criteria met
PR name follows PR naming conventions
If adding a new generator, add it to the list of scaffold modules being tested
The generator test modules (all connectors with -scaffold in their name) have been updated with the latest scaffold by running ./gradlew :airbyte-integrations:connector-templates:generator:testScaffoldTemplates then checking in your changes
Documentation which references the generator is updated as needed.

marcosmarxm · 2021-11-04T16:29:35Z

This is amazing @burmecia !!! I'll ask the team to review this contribution in mid-time can you run ./gradlew format?

burmecia · 2021-11-04T23:02:20Z

This is amazing @burmecia !!! I'll ask the team to review this contribution in mid-time can you run ./gradlew format?

Sorry I forgot it, it is done now.

ChristopheDuong

Thanks for the contribution!!

Few suggestions to keep the code cleaner

ChristopheDuong · 2021-11-04T16:50:57Z

airbyte-workers/src/main/java/io/airbyte/workers/normalization/NormalizationRunnerFactory.java

@@ -27,6 +27,7 @@
          .put("airbyte/destination-postgres-strict-encrypt", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.POSTGRES))
          .put("airbyte/destination-redshift", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.REDSHIFT))
          .put("airbyte/destination-snowflake", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.SNOWFLAKE))
+          .put("airbyte/destination-clickhouse", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.CLICKHOUSE))


Suggested change

.put("airbyte/destination-clickhouse", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.CLICKHOUSE))

.put("airbyte/destination-clickhouse", ImmutablePair.of("airbyte/normalization-clickhouse", DestinationType.CLICKHOUSE))

which means this PR is also missing some changes in: https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/bases/base-normalization/docker-compose.build.yaml to publish the normalization docker image for clickhouse

thanks, done.

ChristopheDuong · 2021-11-04T16:57:40Z

...te-integrations/bases/base-normalization/normalization/transform_catalog/stream_processor.py

+    select *,
+      {{ case_begin }} {{ airbyte_end_at }} is null {{ cdc_active_row }} {{ case_then }} 1 {{ case_else }} 0 {{ case_end }} as {{ active_row }}
+    from (


instead of doing nested select (Subquery), it would be best to keep using the CTE syntax (already heavily used in this step)

thanks for the suggestion, I've put it in a new CTE

ChristopheDuong · 2021-11-04T17:02:14Z

...te-integrations/bases/base-normalization/normalization/transform_catalog/stream_processor.py

+        lag_begin = "lag"
+        lag_end = ""
+        if self.destination_type == DestinationType.CLICKHOUSE:
+            # ClickHouse doesn't support lag() yet, this is a workaround solution
+            # Ref: https://clickhouse.com/docs/en/sql-reference/window-functions/
+            lag_begin = "anyOrNull"
+            lag_end = "ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING"
+
+        case_begin = "case when"
+        case_then = "then"
+        case_else = "else"
+        case_end = "end"
+        if self.destination_type == DestinationType.CLICKHOUSE:
+            # ClickHouse doesn't have CASE WHEN, use multiIf instead
+            # Ref: https://clickhouse.com/docs/en/sql-reference/functions/conditional-functions/#multiif


This might be a lot of "hacks" because clickhouse doesn't support some SQL syntax...

Maybe it'd be worth breaking the whole generate_scd_type_2_model method into a special clickhouse__generate_scd_type_2_model instead? (with its own jinja template)

It'll make it easier to refactor normalization code when splitting the code base per destination type and break this part of the code into its own class probably where the function is overriden

but put it in a separate clickhouse__generate_scd_type_2_model function will create a large code duplication because most of them are still in common. There are actually only 4 ClickHouse specific parts:

lag() function

case when

cast

left outer join null

All the other parts are same as other database. Maybe need some code refactoring on generate_scd_type_2_model in the future.

alexandertsukanov · 2021-11-05T13:00:54Z

.../src/test/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationTest.java

+    db.close();
+  }
+
+  // @Test


Do not clear, why do we have this line commented? Whether add a comment or uncomment it, please.

sorry forgot removing it, done.

alexandertsukanov · 2021-11-05T13:58:55Z

...ouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java

+    if (config.has("password")) {
+      configBuilder.put("password", config.get("password").asText());


Looks like "password" good candidate for constant.

...ouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java

alexandertsukanov · 2021-11-05T14:02:02Z

...se/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseSqlOperations.java

+
+public class ClickhouseSqlOperations extends JdbcSqlOperations {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseSqlOperations.class);


Looks like LOGGER is unused in the class. So please remove or add logging.

I will use it for logging info.

alexandertsukanov · 2021-11-05T14:09:34Z

...se/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseSqlOperations.java

+      } finally {
+        try {
+          if (tmpFile != null) {
+            Files.delete(tmpFile.toPath());
+          }
+        } catch (final IOException e) {
+          throw new RuntimeException(e);


[Optional] Throwing an exception from within a finally block will mask any exception which was previously thrown in the try or catch block, and the masked’s exception message and stack trace will be lost.

thanks, I copied that code from Postgres connector:

airbyte/airbyte-integrations/connectors/destination-postgres/src/main/java/io/airbyte/integrations/destination/postgres/PostgresSqlOperations.java

Line 54 in b5af3f1

} catch (final IOException e) {

If you have better resolution, I'd happy to change.

How about this: (I'm implementing similar pattern for Destination MariaDB ColumnStore now)

Exception primaryException = null; try { // your code } catch (final Exception e) { primaryException = e; throw new RuntimeException(primaryException); } finally { try { // your code } catch (final IOException e) { if (primaryException != null) e.addSuppressed(primaryException); throw new RuntimeException(e); } }

thanks @koji-m, that's great.

.../src/test/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationTest.java

alexandertsukanov · 2021-11-05T14:13:07Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+
+  private static final String DB_NAME = "default";
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseDestinationAcceptanceTest.class);


Do we use LOGGER here?

alexandertsukanov · 2021-11-05T14:13:54Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+  private static final String DB_NAME = "default";
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseDestinationAcceptanceTest.class);
+  private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(RecordFormat.OBJECT);


Do we use JSON_FORMAT in the class?

alexandertsukanov · 2021-11-05T14:14:27Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+  private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
+
+  private ClickHouseContainer db;
+  private Network network;


Looks like not used field

network removed.

alexandertsukanov · 2021-11-05T14:15:18Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+  private ProcessFactory processFactory;
+  private TestDestinationEnv testEnv;


Looks like both of these fields are not used in the class.

alexandertsukanov · 2021-11-05T14:16:04Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+
+  @Override
+  protected JsonNode getFailCheckConfig() {
+    String ipAddress = db.getContainerInfo().getNetworkSettings().getIpAddress();


What purpose of this variable?

that's a leftover when I was struggling with container's connectivity doing check test, it can be removed. See testcontainers/testcontainers-java#452.

alexandertsukanov · 2021-11-05T14:55:19Z

Hi, @burmecia! I had run the next command:

./gradlew clean :airbyte-integrations:connectors:destination-clickhouse:integrationTest

Despite your screenshot above, I don't have success to passing all Java integration tests:

What OS are you using?
Thanks.

alexandertsukanov

Thanks for PR left some comments.

...ouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java

mkhokh-33 · 2021-11-05T13:54:21Z

...se/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseSqlOperations.java

+
+public class ClickhouseSqlOperations extends JdbcSqlOperations {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseSqlOperations.class);


LOGGER is never used
It can be used to log exception on try/catch block or for record size logging

mkhokh-33 · 2021-11-05T14:44:50Z

...se/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseSqlOperations.java

+                                    final String schemaName,
+                                    final String tmpTableName)
+      throws SQLException {
+    if (records.isEmpty()) {


you could log record size here:
LOGGER.info("actual size of batch: {}", records.size());

mkhokh-33 · 2021-11-05T14:48:26Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+
+  private static final String DB_NAME = "default";
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseDestinationAcceptanceTest.class);


LOGGER and JSON_FORMAT are not used, pls remove them

mkhokh-33 · 2021-11-05T14:49:38Z

...java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationAcceptanceTest.java

+  private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
+
+  private ClickHouseContainer db;
+  private Network network;


network, processFactory, testEnv are not used, pls remove them

...ouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java

...se/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseSqlOperations.java

mkhokh-33 · 2021-11-05T15:43:34Z

.../src/test/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationTest.java

+    db.close();
+  }
+
+  // @Test


pls remove commented // @test and check the test

mkhokh-33 · 2021-11-05T15:55:23Z

Thanks for PR, added some comments

burmecia · 2021-11-06T02:00:15Z

Hi, @burmecia! I had run the next command:

./gradlew clean :airbyte-integrations:connectors:destination-clickhouse:integrationTest

Despite your screenshot above, I don't have success to passing all Java integration tests:

What OS are you using? Thanks.

I am on Mac, but I use Ubuntu VM to run the test because docker's host network is not working well on Mac, see my discussion with them testcontainers/testcontainers-java#452. And also remember to build the destination image before run the test.

burmecia · 2021-11-13T01:37:13Z

I've disabled dbt support and also fixed the reserved keywords bug for #7786, can you try run it again?

alexandertsukanov · 2021-11-15T13:30:57Z

Hi, @burmecia ! Looks like some tests still fail. Please, see #7786

burmecia · 2021-11-17T11:51:29Z

I've disabled dbt in expect test result, can you try it again? I've tried below test command and it is working on my laptop.

./gradlew :airbyte-integrations:connectors:destination-clickhouse-strict-encrypt:integrationTest

alexandertsukanov · 2021-11-17T13:17:10Z

@burmecia
Looks like the build is green for destination-clickhouse-strict-encrypt #7786
Only normalization tests left.

ChristopheDuong · 2021-11-17T13:30:08Z

If it makes it easier, it might be a good idea to separate this PR into two instead?

destination one
normalization one

alexandertsukanov · 2021-11-17T13:55:52Z

If it makes it easier, it might be a good idea to separate this PR into two instead?

destination one

normalization one

@ChristopheDuong I believe this won't make any difference as the main purpose of PR to verify all tests are passed on CI (as we can't to run /test, /publish on forked repos). For me is more easier to have only one branch as I can to fetch and merge contributors commits to it.

@burmecia I see normalization fails here:

> Task :airbyte-integrations:bases:base-normalization:mypyCheck FAILED
	 normalization/transform_catalog/reserved_keywords.py:2536:1: error: Need type
	 annotation for 'CLICKHOUSE' (hint: "CLICKHOUSE: Dict[<type>, <type>] = ...") 
	 [var-annotated]
	     CLICKHOUSE = ***
	     ^

Do we need this variable?

airbyte/airbyte-integrations/bases/base-normalization/normalization/transform_catalog/reserved_keywords.py

Lines 2536 to 2538 in 43e39f6

    
           # In ClickHouse, keywords are not reserved. 
        
           # Ref: https://clickhouse.com/docs/en/sql-reference/syntax/#syntax-keywords 
        
           CLICKHOUSE = set()

alexandertsukanov · 2021-11-25T13:15:32Z

Hi, @burmecia any response here? Looks like some conflicts in this PR.
CC: @marcosmarxm, @sherifnada

burmecia · 2021-11-25T23:44:28Z

alright, I added missing type hints, can you try it again?

alexandertsukanov · 2021-11-26T12:52:31Z

@burmecia could you fix conflicts, please? 🙂 After, I will run tests.

alexandertsukanov · 2021-11-29T12:36:40Z

Looks like destination tests are all in green. But normalization fails https://github.com/airbytehq/airbyte/runs/4352870241?check_suite_focus=true#step:8:13303. Looks like this tests not related to @burmecia changes and normalization tests fails in master as well. Take to the account this fact there are not more comments from my side and approve from my side. Thanks for contribution.
CC: @ChristopheDuong , @sherifnada , @mkhokh-33

…o marcos/test-pr-7620

ChristopheDuong · 2021-12-03T11:32:17Z

normalization tests fails in master as well.

Tests should be passing fine on master (maybe there seems to have some errors with credentials when you ran it)

marcosmarxm

awesome work @burmecia sorry the long delay to merge your contribution.

* add ClickHouse destination * update docs * format code * code improvement as per code review * add ssh tunneling and ssl/tls support and code enhancement * merge from master * disable testCustomDbtTransformationsFailure test * fix string format bug * fix reserved keywords bug and disable dbt * disable dbt in expect result * add type hints * bump connector version Co-authored-by: Alexander Tsukanov <alexander.tsukanovvv@gmail.com> Co-authored-by: Marcos Marx <marcosmarxm@gmail.com>

ChristopheDuong · 2022-01-06T09:55:42Z

airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py

+        dbt_config = {
+            "type": "clickhouse",
+            "host": config["host"],
+            "port": config["port"],


Should this be changed to hardcode 9000 instead?

see this thread in slack: https://airbytehq.slack.com/archives/C01MFR03D5W/p1641461068011100

add ClickHouse destination

59be9e6

github-actions bot added area/connectors Connector related issues area/documentation Improvements or additions to documentation area/platform issues related to the platform area/worker Related to worker normalization labels Nov 4, 2021

octavia-squidington-iii added the community label Nov 4, 2021

update docs

410e9c9

alexandr-shegeda requested a review from alexandertsukanov November 4, 2021 17:26

alexandr-shegeda linked an issue Nov 4, 2021 that may be closed by this pull request

New Destination: ClickHouse #1903

Closed

format code

ff94b62

ChristopheDuong suggested changes Nov 5, 2021

View reviewed changes

alexandertsukanov reviewed Nov 5, 2021

View reviewed changes

...ouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java Show resolved Hide resolved

alexandertsukanov reviewed Nov 5, 2021

View reviewed changes

.../src/test/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestinationTest.java Show resolved Hide resolved

alexandertsukanov reviewed Nov 5, 2021

View reviewed changes

alexandertsukanov suggested changes Nov 5, 2021

View reviewed changes

mkhokh-33 suggested changes Nov 5, 2021

View reviewed changes

disable dbt in expect result

43e39f6

add type hints

d85eb1f

burmecia and others added 2 commits November 27, 2021 00:17

merge from master

2aadea4

Merge branch 'master' into new-dest-clickhouse

9589702

alexandertsukanov approved these changes Nov 29, 2021

View reviewed changes

Merge branch 'new-dest-clickhouse' of github.com:burmecia/airbyte int…

63eff62

…o marcos/test-pr-7620

mkhokh-33 approved these changes Dec 1, 2021

View reviewed changes

marcosmarxm added 2 commits December 13, 2021 17:58

merge master

a6fbcbc

bump connector version

56bbb84

marcosmarxm temporarily deployed to more-secrets December 13, 2021 22:23 Inactive

marcosmarxm approved these changes Dec 13, 2021

View reviewed changes

marcosmarxm merged commit bbcd461 into airbytehq:master Dec 13, 2021

jrhizor mentioned this pull request Dec 14, 2021

Bump Airbyte version from 0.33.11-alpha to 0.33.12-alpha #8787

Merged

ChristopheDuong mentioned this pull request Dec 28, 2021

update snowflake normalization test output #9132

Closed

ChristopheDuong reviewed Jan 6, 2022

View reviewed changes

karinakuz added connectors/destinations-warehouse connectors/destination/clickhouse labels Jan 14, 2022

	.put("airbyte/destination-clickhouse", ImmutablePair.of(BASE_NORMALIZATION_IMAGE_NAME, DestinationType.CLICKHOUSE))
	.put("airbyte/destination-clickhouse", ImmutablePair.of("airbyte/normalization-clickhouse", DestinationType.CLICKHOUSE))

		if (config.has("password")) {
		configBuilder.put("password", config.get("password").asText());


		public class ClickhouseSqlOperations extends JdbcSqlOperations {

		private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseSqlOperations.class);


		private static final String DB_NAME = "default";

		private static final Logger LOGGER = LoggerFactory.getLogger(ClickhouseDestinationAcceptanceTest.class);

		private ProcessFactory processFactory;
		private TestDestinationEnv testEnv;

🎉 New Destination: ClickHouse #7620

🎉 New Destination: ClickHouse #7620

Conversation

burmecia commented Nov 4, 2021 • edited Loading

What

How

Recommended reading order

Test Runs

Integration test

Normalization test

Pre-merge Checklist

Community member or Airbyter

Airbyter

Community member or Airbyter

Airbyter

marcosmarxm commented Nov 4, 2021

burmecia commented Nov 4, 2021

ChristopheDuong left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexandertsukanov commented Nov 5, 2021

alexandertsukanov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mkhokh-33 commented Nov 5, 2021 • edited Loading

burmecia commented Nov 6, 2021

burmecia commented Nov 13, 2021

alexandertsukanov commented Nov 15, 2021

burmecia commented Nov 17, 2021

alexandertsukanov commented Nov 17, 2021 • edited Loading

ChristopheDuong commented Nov 17, 2021

alexandertsukanov commented Nov 17, 2021

alexandertsukanov commented Nov 25, 2021

burmecia commented Nov 25, 2021

alexandertsukanov commented Nov 26, 2021

alexandertsukanov commented Nov 29, 2021 • edited Loading

ChristopheDuong commented Dec 3, 2021

marcosmarxm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

burmecia commented Nov 4, 2021 •

edited

Loading

mkhokh-33 commented Nov 5, 2021 •

edited

Loading

alexandertsukanov commented Nov 17, 2021 •

edited

Loading

alexandertsukanov commented Nov 29, 2021 •

edited

Loading