Fix materialized column as sharding key

[vitlibar]

I hereby agree to the terms of the CLA available at: https://yandex.ru/legal/cla/?lang=en

Changelog category:

• Bug Fix

Changelog entry:
Allow using a materialized column as the sharding key in a distributed table even if insert_allow_materialized_columns=0:

CREATE TABLE dist ON CLUSTER '{cluster}' (x Int32, date Date MATERIALIZED today()) ENGINE = Distributed('{cluster}', currentDatabase(), local, date);
CREATE TABLE local AS dist ON CLUSTER '{cluster}' ENGINE = MergeTree() ORDER BY tuple();
INSERT INTO dist VALUES (1), (2), (3);

[vitlibar]

Here was a previous solution for avoiding the error Cannot insert column, because it is MATERIALIZED column after a materialized column is calculated and then sent to a remote table. The previous solution just removed materialized columns from blocks before sending them to remote tables. It helped with the error but produced another problem: it made ClickHouse to calculate a materialized column two times - on an initiator host and on a shard, and also made difficult to use a materialized column as a sharding key.

So I decided to try another approach: always send a materialized column to shards and also send the setting insert_allow_materialized_columns forced to true (no matter what the value of insert_allow_materialized_columns on an initiator actually is). Thus a shard will not calculate values for materialized columns by itself, it will always use values calculated on an initiator.

[azat]

Here was a previous solution for avoiding the error Cannot insert column, because it is MATERIALIZED column after a materialized column is calculated and then sent to a remote table. The previous solution just removed materialized columns from blocks before sending them to remote tables. It helped with the error but produced another problem: it made ClickHouse to calculate a materialized column two times - on an initiator host and on a shard, and also made difficult to use a materialized column as a sharding key.

Indeed.

So I decided to try another approach: always send a materialized column to shards and also send the setting insert_allow_materialized_columns forced to true (no matter what the value of insert_allow_materialized_columns on an initiator actually is). Thus a shard will not calculate values for materialized columns by itself, it will always use values calculated on an initiator.

The reason for the previous solution was that some local table may has dictGet that will make sense only on local node regardless was it in the INSERT'ed block or not (#23349 (comment)), and after this patch it will not be possible to restrict these, but personally I fine with this change (pretty specific use-case anyway).

[vitlibar]

So sometimes we may want a materialized column to be calculated in a distributed table, sometimes we may want it to be calculated on a shard. It's getting complicated. Maybe we should add a new setting for distributed tables named calculate_defaults_on_shards to cover more cases.

[azat]

Let me try to write possible cases.

insert_allow_materialized_columns=0

Any write of materialized column to any table (including Distributed) will fail.

But the problem is that if Distributed table has the same MATERIALIZED column as underlying table, then it will materialize it and later it will try to write these into underlying table and will fail, w/o #23349

And so in these case MATERIALIZED column indeed will be calculated 2 times, on initiator and on shards, and the result of the generated column on initiator will be simply discarded.

create table data (key Int, value Int materialized 100) engine=Null();
create table dist (key Int, value Int materialized 100) engine=Distributed(test_cluster_two_shards, currentDatabase(), data, key);

INSERT INTO dist VALUES (1);

[localhost] 2021.09.09 16:57:13.641736 [ 22781 ] {64e81fd5-4f69-48c5-aba3-4f776f47b336} <Debug> DistributedBlockOutputStream: default.dist (876d7f85-aa5a-4c1d-876d-7f85aa5a8c1d): column value will be removed, because it is MATERIALIZED
...

insert_allow_materialized_columns=1

With these setting set, MATERIALIZED column will be passed to the underlying storage and so it will not be calculated multiple times, only on initiator.

But there are some caveats in these (mentioned in the previous comments).

distributed_materialize_defaults_on_shards

You suggested calculate_defaults_on_shards, but I took the liberty and renamed it, since it looks better to me

If these setting is set (distributed_materialize_defaults_on_shards=1) MATERIALIZED column will be calculated on initiator always only, and so the INSERT will fail w/o insert_allow_materialized_columns=1.
Maybe these setting should instead forbid INSERT w/o all columns, including materialized (since it expects that Distributed already filled it), these will make it strict and ensure that Distributed table has the same structure (since if the structure will not match, in terms of some new materialized columns, INSERT will fail), thoughts?

P.S. Actually if someone want strict behavior in this it is better to add MATERIALIZED columns only in Distributed or only in the underlying table.

P.S. 00952_insert_into_distributed_with_materialized_colum should catch you change, but it uses the same default expression in Distributed and underlying table, hence it fails to do this.

[vitlibar]

I've changed my solution to make it more cautious. I put back your code removing materialized columns if insert_allow_materialized_columns=0, but now I do that after the sharding key is calculated. It seems the final version must solve all the current problems without breaking anything.

[vitlibar]

@azat Can you take a look please?

[CLAassistant]

All committers have signed the CLA.

[vitlibar]

The failures (test_merge_tree_s3, test_dictionaries_dependency) are not related to this PR.

[azat]

Oops, looks like the comment is missing now, although it looks useful.

[azat]

This can be done in a better way, using @pytest.mark.parametrize decorator

[azat]

This is just to ensure that partitioning by sharding key indeed works, right?