DROP TABLE IF EXISTS source_table1;
DROP TABLE IF EXISTS source_table2;
DROP TABLE IF EXISTS distributed_table1;
DROP TABLE IF EXISTS distributed_table2;

CREATE TABLE source_table1 (a Int64, b String) ENGINE = Memory;
CREATE TABLE source_table2 (c Int64, d String) ENGINE = Memory;

INSERT INTO source_table1 VALUES (42, 'qwe');
INSERT INTO source_table2 VALUES (42, 'qwe');

CREATE TABLE distributed_table1 AS source_table1
ENGINE = Distributed('test_shard_localhost', currentDatabase(), source_table1);

CREATE TABLE distributed_table2 AS source_table2
ENGINE = Distributed('test_shard_localhost', currentDatabase(), source_table2);

SET prefer_localhost_replica = 1;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t1.a = t2.c LIMIT 1;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t2.c = t1.a LIMIT 1;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1.a = t2.a LIMIT 1;

SET prefer_localhost_replica = 0;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t1.a = t2.c LIMIT 1;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t2.c = t1.a LIMIT 1;
SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1.a = t2.a LIMIT 1;

SELECT t1.a as t1_a, t2.a as t2_a FROM source_table1 AS t1 JOIN source_table1 AS t2 ON t1_a = t2_a LIMIT 1;
SELECT t1.a as t1_a, t2.a as t2_a FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1_a = t2_a LIMIT 1;

DROP TABLE source_table1;
DROP TABLE source_table2;
DROP TABLE distributed_table1;
DROP TABLE distributed_table2;