Clickhouse: Distributed JOIN over engine Merge fails with 'Missing columns' errors

[Namnd-gem]

When I am trying to run this query:

SELECT * FROM products_d AS products GLOBAL ALL LEFT JOIN categories_d USING (categoryId);

The clickhouse throws an exception:
[2020-06-18 17:20:01] Code: 47, e.displayText() = DB::Exception: Missing columns: 'categories_d.dateTime' 'categories_d.errorCode' 'categories_d.processingTimestampNs' 'categoryName' 'productIds' while processing query: 'SELECT dateTime, categoryId, productId, name, errorCode, processingTimestampNs, productIds, categoryName, categories_d.dateTime, categories_d.errorCode, categories_d.processingTimestampNs FROM products_d AS products GLOBAL ALL LEFT JOIN categories_d AS categories_d USING (categoryId)', required columns: 'dateTime' 'errorCode' 'categoryId' 'productId' 'productIds' 'name' 'categoryName' 'categories_d.processingTimestampNs' 'processingTimestampNs' 'categories_d.errorCode' 'categories_d.dateTime', source columns: 'processingTimestampNs' 'name' 'productId' 'categoryId' 'errorCode' 'dateTime', joined columns: (version 20.4.5.36 (official build))

Here are the queries that create 2 tables:

CREATE TABLE IF NOT EXISTS products_data_hist (
   dateTime                  DateTime('UTC'),
   categoryId                LowCardinality(String),
   productId                 UUID,
   name                      LowCardinality(String),
   errorCode                 UInt8,
   processingTimestampNs     UInt64
) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/%%DATABASE%%/products_data_hist', '{replica}', processingTimestampNs)
  PARTITION BY toYYYYMM(dateTime)
  ORDER BY (cityHash64(name), toDate(dateTime), productId)
  PRIMARY KEY (cityHash64(name), toDate(dateTime));

CREATE TABLE IF NOT EXISTS products_data_d_hist AS products_data_hist ENGINE = Distributed('kafka-cluster', %%DATABASE%%, products_data_hist, sipHash64(toString(productId)));

CREATE TABLE IF NOT EXISTS products_data_d_daily AS products_data_hist ENGINE = Distributed('kafka-cluster', %%DATABASE%%, products_data_hist, sipHash64(toString(productId)));

CREATE TABLE products_d as products_data_d_hist ENGINE=Merge('%%DATABASE%%', '^products_data_d_');

CREATE TABLE if NOT EXISTS categories_data_hist (
    productIds              Array(String),
    categoryId              UUID,
    categoryName            String,
    dateTime                DateTime('UTC'),
    errorCode               UInt8,
    processingTimestampNs   UInt64
) ENGINE ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/%%DATABASE%%/categories_data_hist', '{replica}', processingTimestampNs)
PARTITION BY toYYYYMM(dateTime)
ORDER BY (cityHash64(categoryName), toDate(dateTime))
PRIMARY KEY (cityHash64(categoryName), toDate(dateTime));

CREATE TABLE IF NOT EXISTS categories_data_d_hist AS categories_data_hist ENGINE = Distributed('kafka-cluster', %%DATABASE%%, categories_data_hist, sipHash64(toString(categoryId)));

CREATE TABLE IF NOT EXISTS categories_data_d_daily AS categories_data_hist ENGINE = Distributed('kafka-cluster', %%DATABASE%%, categories_data_hist, sipHash64(toString(categoryId)));

CREATE TABLE categories_d as categories_data_hist ENGINE=Merge('%%DATABASE%%', '^categories_data_d_');

I think the issue is about it is not adding the columns from the right table to the joined table when executing the query.

[4ertus2]

Does the query fail if the JOIN is not GLOBAL? Does it fail if you change ENGINE = Kafka to MergeTree or Memory?

[Namnd-gem]

Yes, the query is failed even in the case without GLOBAL. Change ENGINE to Memory but it still gets this error

[4ertus2]

Could you update the queries in issue when? We need as simple sample as possible. The harder sample is to understand and reproduce the less probablitity it would be fixed.

[Namnd-gem]

I've updated queries on first comment

[4ertus2]

Am I right that if you use GLOBAL JOIN not over engine Merge but directly over enigne Distributed it works?

[xmariachi]

Hi @4ertus2 , @namnd-gem

The issue only happens when the left part of the JOIN is the Merge Engine table.

Left side Distributed, Right side Merge works:

SELECT sessionId
FROM orders_data_d_hist AS orders
GLOBAL ALL LEFT JOIN orderAttributions_d AS orderAttributions USING (frontendId, orderId)

Ok.

Left side Distributed, Right side Distributed works:

SELECT sessionId
FROM orders_data_d_hist AS orders
GLOBAL ALL LEFT JOIN orderAttributions_data_d_hist AS orderAttributions USING (frontendId, orderId)

Ok.

Left side Merge, Right side Distributed fails:

SELECT sessionId
FROM orders_d AS orders
GLOBAL ALL LEFT JOIN orderAttributions_data_d_hist AS orderAttributions USING (frontendId, orderId)


Received exception from server (version 20.4.5):
Code: 47. DB::Exception: Received from clickhouse:9000. DB::Exception: Missing columns: 'sessionId' while processing query: 'SELECT sessionId FROM orders_d AS orders GLOBAL ALL LEFT JOIN orderAttributions AS orderAttributions USING (frontendId, orderId)', required columns: 'sessionId' 'frontendId' 'orderId', source columns: 'frontendId' 'orderId', joined columns:. 

0 rows in set. Elapsed: 0.009 sec. 

[xmariachi]

Hi @4ertus2 , was my previous comment answering your question well enough?
I'd love to collaborate if anything else can be done from our end.

[4ertus2]

It could be related to #15242

[4ertus2]

Sample to reproduce

CREATE TABLE cat_hist (categoryId UUID, categoryName String) ENGINE Memory;
CREATE TABLE prod_hist (categoryId UUID, productId UUID) ENGINE = MergeTree ORDER BY productId;

CREATE TABLE products_l AS prod_hist ENGINE = Distributed(test_shard_localhost, currentDatabase(), prod_hist);
CREATE TABLE products as prod_hist ENGINE = Merge(currentDatabase(), '^products_');

SELECT * FROM products AS p LEFT JOIN cat_hist AS c USING (categoryId);
SELECT * FROM products AS p GLOBAL LEFT JOIN cat_hist AS c USING (categoryId);

[xmariachi]

Hi @4ertus2 ,
Thanks.
Is the error confirmed to still happen in a version that already contains the #15242 fix?

[den-crane]

still relevant 20.11.1

And it worked before. At least in 19.13.7.57

[lnurullina]

are there any updates on this issue? really waiting for this fix, we need to update to 20 version and just can't because of it :(