Cluster setting and Distributed Table tests

[gfunc]

Summary

• added attribute should_on_cluster for ClickHouseRelation class to indicate whether theon_cluster_clause macro should return content.
• added cluster setting for tests.
• debug marcos.
• added tests for distributed_table materialization and replicated table engines.
• changed github workflow test to use docker-compose which starts a 3 nodes ClickHouse cluster

Checklist

Delete items not relevant to your PR:

• Unit and integration tests covering the common scenarios were added

Caveats

• tests for distributed_incremental is not added yet.

[CLAassistant]

All committers have signed the CLA.

[zli06160]

👍 The modifications in dbt/include/clickhouse/macros/materializations/distributed_table.sql do make sense.
We should indeed

• define the default sharding_key, cluster, etc;
• or show the error messages(instructions) if they are not defined in template.sql and/or dbt_project.yml.

[gfunc]

Hi @genzgd any update on the review of this MR, or any thoughts/suggestions?

[zli06160]

Maybe join in lower case. Same comments for other sql keywords in several files.

[zli06160]

Maybe add more comments to describe the tests more clearly 1) at the beginning of the .py file, or 2) at the beginning of each test.

[gfunc]

sure, will do.

[gfunc]

I am afraid most tests are self explainatory by names, and mostly base off dbt's sample tests. I will try to add more comments for distributed materialization tests.

[zli06160]

Maybe it will be better if test at least once with multiple shards x multiple replicas?
e.g. ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/table_name', '{replica}') (example from https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/replication)

[gfunc]

for integration tests, if we managed to set a multiple shards * multple replicas matrix, data will always exist in one shard (same zk path), maybe it is not worth adding another test?

[zli06160]

For me it is an usual usage.
But it is more managed by dbt-core & the clickhouse cluster, so any integration test(in the current project) on the cluster mode is enough.

[zli06160]

BTW, I did not get the reason why the data persist in one shard.
Do you use drop table distributed_table_name on cluster cluster_name or create or replace table, or something similar?

[gfunc]

I meant for the Replicated Engine tables with non-distributed materializations, like this seed. If I understand it correctly, during our tests, insert into query will be performed by one node and data will be replicated among nodes with the same shard (same zk path).

[zli06160]

Yep, if create (non-distributed)table xxx on cluster xxx then insert into the non-distributed table, the data will on only one shard(and its replicated shards); and the next insert into can target to another shard😿.

For the seeding step, I see 3 possible ways:

• insert into a distributed table, which is based on a table with any replicated engine;
• or use a view as interface;
• or use File Table Engine <= the best in my opinion.

[gfunc]

My approach is to make sure seed are using the same shard and different replica, i.e. to broadcast the whole dataset.
I have never used File Table Engine before, could you explain a bit more? How it would help in this situation?

[zli06160]

File Table Engine(or just file function) reads the csv files directly, not sure if it will help.
https://clickhouse.com/docs/en/engines/table-engines/special/file
https://clickhouse.com/docs/en/sql-reference/table-functions/file

[zli06160]

I think now I understand how you test the distributed_table. I still recommend a separate .py for each main materialization, or, if not applicable, document more.

[zli06160]

• The distributed_table works well on the cluster with my test case (several dbt run, no error, correct count(*) );
• The distributed_incremental works well on the cluster with my test case (several dbt run, no error, correct count(*) ).

[gfunc]

Hi @zli06160 any thoughts on the logic for should_on_cluster attribute in relation.py?

    @property
    def should_on_cluster(self) -> bool:
        if self.include_policy.identifier:
            return self.can_on_cluster
        else:
            # create database/schema on cluster by default
            return True
    @classmethod
    def get_on_cluster(
        cls: Type[Self], cluster: str = '', materialized: str = '', engine: str = ''
    ) -> bool:
        if cluster:
            return 'view' == materialized or 'distributed' in materialized or 'Replicated' in engine
        else:
            return False

I think we should start the discussion early since this is changing the behavior of this adapter once a cluster is set. Also, I tried to make logic compatible with the previous version by adding the attribute can_on_cluster for existing models.

Maybe it is worth mentioning in README?

[zli06160]

For me, the query must contain the on cluster clause: 1) if the connection is to a cluster; 2)or if any keyword like distributed/Replicated/etc detected in the .sql file.

The logic of get_on_cluster seems correct, just need to include MATERIALIZED VIEW (of course this materialization is not implemented yet), and prevent the case cluster = ' '.

Other contributors do need more detailed comments and README to understand the project-specific notions.

[gfunc]

I will be available starting tomorrow (Oct 09) for further modifications.

[zli06160]

I think it is better to set "DBT_CH_TEST_CLUSTER"(I guess the value should be 'test_shard'?) somewhere for integration testing.

After I cloned the project, checkout to the branch, build + up the docker containers, there is no such variable, so many tests are ignored.

Then, in this file, in order to fully run the tests, with os.environ["DBT_CH_TEST_CLUSTER"] = 'test_shard' added, there is an error related to BaseSnapshotCheckCols:

09:11:29  Database Error in snapshot cc_all_snapshot (snapshots/cc_all_snapshot.sql)
09:11:29    :HTTPDriver for http://localhost:8123 returned response code 400)
09:11:29     Code: 15. DB::Exception: Column dbt_change_type specified more than once. (DUPLICATE_COLUMN) (version 23.9.1.1854 (official build))

[gfunc]

Yes, I am using test_shard for unit tests. And I added the env var in github workflow file.
And tests returned no error. I will take a look at this BaseSnapshotCheckCols.

[gfunc]

I am not seeing any problem with this test.env file

DBT_CH_TEST_USE_DOCKER=true
DBT_MACRO_DEBUGGING=true
DBT_CH_TEST_INCLUDE_S3=false
DBT_CH_TEST_CLUSTER=test_shard

and command

pytest tests/integration/adapter/test_basic.py::TestSnapshotCheckCols

result attached:

============================= test session starts ==============================
platform linux -- Python 3.9.18, pytest-7.4.0, pluggy-1.2.0
rootdir: /home/xxx/projects/dbt-clickhouse
configfile: pytest.ini
plugins: dotenv-0.5.2
collected 1 item

tests/integration/adapter/test_basic.py .                                [100%]

============================== 1 passed in 19.58s ==============================

Please share your env file for me to reproduce

[zli06160]

The reason is that I didn't have the test.env.

With the same env file as yours,

pytest tests/integration/adapter/test_basic.py -c pytest.ini -s

=> everything is ok after double check test by test.

pytest tests/integration/ -c pytest.ini

=> 107 passed, 2 skipped, 2 warnings .

[zli06160]

OK for me 👍

[genzgd]

Thanks for all the work here, @gfunc and @zli06160. I'll do a 1.4.9 release shortly.