Querying ReplicatedSummingMergeTree table through Distribution table with condition on DateTime64 column with lesser/greater than subquery result returns results only from single shard

[gliter]

Describe what's wrong

When querying a table using ReplicatedSummingMergeTree engine through Distributed table with condition on DateTime64 column with lesser/greater than subquery results ClickHouse returns results only from single shard.

Example query:

select * from gl_test.test_smt_d tsd where dt64 > (select toDateTime64(0, 3, 'UTC'));

Not reproducable in https://fiddle.clickhouse.com/ as at least two shard are needed and ability to connect with specific one

Does it reproduce on recent release?

Unknown I have only access to version 22.8.17.17

How to reproduce

ClickHouse installation with at least 2 shards.

create database gl_test ON CLUSTER '{cluster}' ENGINE = Atomic();

-- drop table gl_test.test_smt ON CLUSTER '{cluster}' sync
CREATE TABLE gl_test.test_smt ON CLUSTER '{cluster}' (
    id String,
    dt DateTime('UTC'),
    dt64 DateTime64(3, 'UTC'),
    val Int32
    ) ENGINE ReplicatedSummingMergeTree('/clickhouse/tables/{shard}/gl_test/test_smt', '{replica}', (val))
    PARTITION BY toYYYYMM(dt)
    ORDER BY (cityHash64(id), toDate(dt), val)
    PRIMARY KEY (cityHash64(id), toDate(dt));

-- drop table gl_test.test_smt_d ON CLUSTER '{cluster}' sync
CREATE TABLE gl_test.test_smt_d ON CLUSTER '{cluster}' AS gl_test.test_smt
    ENGINE = Distributed('{cluster}', 'gl_test', test_smt, cityHash64(id));

-- Insert directly to single shard (Alternatively insert using distributed table but then make sure to switch connection to a shard where row is not present)
INSERT INTO gl_test.test_smt
(id, dt, dt64, val)
VALUES('abc', toDateTime(1686036919, 'UTC'), toDateTime64(1686036919.123, 3, 'UTC'), 1);

-- Verify row was inserted
select * from gl_test.test_smt;
-- Expected: 1 row present / Actual: 1 row present

-- Connect to second shard

-- Verify row is not present
select * from gl_test.test_smt;
-- Expected: 0 row present / Actual: 0 row present

-- Verify row can be fetched using distributed table
select * from gl_test.test_smt_d;
-- Expected: 1 row present / Actual: 1 row present

-- Query with condition
select * from gl_test.test_smt_d tsd where dt64 > toDateTime64(0, 3, 'UTC');
-- Expected: 1 row present / Actual: 1 row present

-- Wrap condition in subquery
select * from gl_test.test_smt_d tsd where dt64 > (select toDateTime64(0, 3, 'UTC'));
-- [FAIL] Expected: 1 row present / Actual: 0 row present

Expected behavior

select * from gl_test.test_smt_d tsd where dt64 > (select toDateTime64(0, 3, 'UTC')); will return row from second shard

Error message and/or stacktrace

N/A

Additional context

I have also tested with different conditional operator and with different column types:

-- Query with = DateTime64
select * from gl_test.test_smt_d tsd where dt64 = (select toDateTime64(1686036919.123, 3, 'UTC'));
-- [PASS] Expected: 1 row present / Actual: 1 row present

-- Query with DateTime in subquery
select * from gl_test.test_smt_d tsd where dt64 > (select toDateTime(0, 'UTC'));
-- [PASS] Expected: 1 row present / Actual: 1 row present

-- Query with Int32 in subquery
select * from gl_test.test_smt_d tsd where val > (select 0);
-- [PASS] Expected: 1 row present / Actual: 1 row present

I have also tested with ReplicatedMergeTree engine:

-- drop table gl_test.test_smt_2 ON CLUSTER '{cluster}' sync
CREATE TABLE gl_test.test_smt_2 ON CLUSTER '{cluster}' (
    id String,
    dt DateTime('UTC'),
    dt64 DateTime64(3, 'UTC'),
    val Int32
    ) ENGINE ReplicatedMergeTree('/clickhouse/tables/{shard}/gl_test/test_smt_2', '{replica}')
    PARTITION BY toYYYYMM(dt)
    ORDER BY (cityHash64(id), toDate(dt), val)
    PRIMARY KEY (cityHash64(id), toDate(dt));

-- drop table gl_test.test_smt_2_d ON CLUSTER '{cluster}' sync
CREATE TABLE gl_test.test_smt_2_d ON CLUSTER '{cluster}' AS gl_test.test_smt_2
    ENGINE = Distributed('{cluster}', 'gl_test', test_smt, cityHash64(id));

INSERT INTO gl_test.test_smt_2
(id, dt, dt64, val)
VALUES('abc', toDateTime(1686036919, 'UTC'), toDateTime64(1686036919.123, 3, 'UTC'), 1);

-- Connect to second shard

-- Verify row is not present
select * from gl_test.test_smt_2;
-- Expected: 0 row present / Actual: 0 row present

-- Verify row can be fetched using distributed table
select * from gl_test.test_smt_2_d;
-- Expected: 1 row present / Actual: 1 row present

-- Query with subquery
select * from gl_test.test_smt_2_d tsd where dt64 > (select toDateTime64(0, 3, 'UTC'));
-- [PASS] Expected: 1 row present / Actual: 1 row present

[den-crane]

Not related to ReplicatedSummingMergeTree.

simpler:

CREATE TABLE t ( dt64 DateTime64(3, 'UTC') ) ENGINE Memory as select  '1686036919.123';

select * from remote('127.0.0.1', currentDatabase(), t)  where dt64 > (select toDateTime64(0, 3));
┌────────────────────dt64─┐
│ 2023-06-06 07:35:19.123 │
└─────────────────────────┘

set prefer_localhost_replica=0;

select * from remote('127.0.0.1', currentDatabase(), t)  where dt64 > (select toDateTime64(0, 3));
0 rows in set. Elapsed: 0.007 sec.

WA (assumeNotNull):

select * from remote('127.0.0.1', currentDatabase(), t) where dt64 > assumeNotNull((select toDateTime64(0, 3)));
┌────────────────────dt64─┐
│ 2023-06-06 07:35:19.123 │
└─────────────────────────┘

v21.9.1.8000-prestable.md:* Now, scalar subquery always returns Nullable result if it's type can be Nullable. It is needed because in case of empty subquery it's result should be Null. Previously, it was possible to get error about incompatible types (type deduction does not execute scalar subquery, and it could use not-nullable type). Scalar subquery with empty result which can't be converted to Nullable (like Array or Tuple) now throws error. Fixes #25411. #26423 (Nikolai Kochetov).

cc @KochetovNicolai

[den-crane]

Seems the issue is in DateTime64

select *, (select toDateTime64(0, 3)) from remote('127.0.0.1', system.one) settings prefer_localhost_replica=1;
┌─dummy─┬─────────────_subquery13─┐
│     0 │ 1970-01-01 00:00:00.000 │
└───────┴─────────────────────────┘

select *, (select toDateTime64(0, 3)) from remote('127.0.0.1', system.one) settings prefer_localhost_replica=0;
┌─dummy─┬─_subquery14─┐
│     0 │        ᴺᵁᴸᴸ │
└───────┴─────────────┘

select *, (select toDateTime64(-111111111, 3)) from remote('127.0.0.1', system.one) settings prefer_localhost_replica=1;
┌─dummy─┬─────────────_subquery15─┐
│     0 │ 1966-06-24 23:48:09.000 │
└───────┴─────────────────────────┘

select *, (select toDateTime64(-111111111, 3)) from remote('127.0.0.1', system.one) settings prefer_localhost_replica=0;
┌─dummy─┬─_subquery16─┐
│     0 │        ᴺᵁᴸᴸ │
└───────┴─────────────┘

cc @rschu1ze

[gliter]

@den-crane any idea why it seems to be working fine with ReplicatedMergeTree?

[den-crane]

@den-crane any idea why it seems to be working fine with ReplicatedMergeTree?

it's not. Test better.

[zvonand]

It happens because it is forbidden to parse String->DateTime(64) if its length less than 4:

┌─CAST('9999', 'Nullable(DateTime64(3))')─┐
│                                    ᴺᵁᴸᴸ │
└─────────────────────────────────────────┘
┌─CAST('10000', 'Nullable(DateTime64(3))')─┐
│                  1970-01-01 03:46:40.000 │
└──────────────────────────────────────────┘

DateTime64 is generally a modification of Decimal type. Looks like it is implicitly represented as string when querying to/from a remote node. Changing this may affect the precision.

The same queries will work for DateTime because they are built on top of Integer, not Decimal.
And it is allowed to convert any Int to DateTime.