Results from SummingMergeTree MV differ from equivalent query to full table

[Michael Smitasin]

Hoping someone has an idea what may be causing this... I'm creating a
SummingMergeTree MV where I'm trying to count the rows that match a
specific criteria, but I'm finding that it's incorrectly aggregating
two different values from a given column into one. For example, when
querying the "full" table, we might get something back like:

conn_state count()
RST0 2
RSTR 2
S3 5

But when querying the MV, we get back:

conn_state count()
RSTR 4
S3 5

So it seems to be merging RST0 and RSTR counts into RSTR. That said,
this isn't always true, i.e. it doesn't merge every occurrence of RST0
into RSTR, sometimes for a given IP address combination, it will
return the expected results.

####

# Full table schema

$ clickhouse-client --query "describe zeek_conn"
ts String
uid String
orig_h String
orig_p UInt16
resp_h String
resp_p UInt16
proto Enum8(\'icmp\' = 1, \'tcp\' = 6, \'udp\' = 17)
service String
duration Float32
orig_bytes Float32
resp_bytes Float32
conn_state Enum8(\'OTH\' = 0, \'REJ\' = 1, \'RSTO\' = 2,
\'RSTOS0\' = 3, \'RSTR\' = 4, \'RSTRH\' = 5, \'S0\' = 6, \'S1\' = 7,
\'S2\' = 8, \'S3\' = 9, \'SF\' = 10, \'SH\' = 11, \'SHR\' = 12)
local_orig Enum8(\'F\' = 0, \'T\' = 1)
local_resp Enum8(\'F\' = 0, \'T\' = 1)
missed_bytes UInt64
history String
orig_pkts UInt64
orig_ip_bytes UInt64
resp_pkts UInt64
resp_ip_bytes UInt64
tunnel_parents String
peer String
orig_cc FixedString(2)
resp_cc FixedString(2)
file_src String
day Date MATERIALIZED
toDate(toDateTime(round(toFloat64(ts)), \'America/Los_Angeles\'))

# Example data from "full" table

1598897705.919456 CsZuTH2BrrF39W0T5k 192.0.2.29 46296
192.0.2.4 443 tcp - 0.016841 24 0 RSTR
T F 0 DFr 2 128 2 80
(empty) worker-9 \0\0 \0\0 nail7
1598892913.100067 CInR1q4lbiIVICce9j 192.0.2.29 46006
192.0.2.4 443 tcp - 0.016856 24 0 RSTR
T F 0 DFr 2 128 2 80
(empty) worker-1 \0\0 \0\0 nail7
1598885229.768122 CMADvzjCOfB49p55k 192.0.2.29 45756
192.0.2.4 443 tcp - 240.54659 804 4666 RSTO
T F 0 ShADadfR 11 1372 11
5246 (empty) worker-1 \0\0 \0\0 nail7
1598885267.470260 CLYKFr21IxK7LXw6rc 192.0.2.29 45794
192.0.2.4 443 tcp - 240.19713 845 1755 RSTO
T F 0 ShADadfR 9 1309 9
2231 (empty) worker-7 \0\0 \0\0 nail7
1598916647.081880 CaXn3A3enuXaulNjze 192.0.2.29 47342
192.0.2.4 443 tcp - 271.03482 845 1710 S3
T F 0 ShADadf 8 1281 18 2681
(empty) worker-3 \0\0 \0\0 nail7
1598894901.905512 C36g404qrH8be0VAFf 192.0.2.29 46296
192.0.2.4 443 tcp - 240.13982 845 1836 S3
T F 0 ShADadf 9 1321 9 2312
(empty) worker-9 \0\0 \0\0 nail7
1598888580.961462 Cn4PLj1bQwasMVRI2 192.0.2.29 46006
192.0.2.4 443 tcp - 240.22739 845 1753 S3
T F 0 ShADadf 9 1321 9 2229
(empty) worker-1 \0\0 \0\0 nail7
1598899868.071982 Cgk69r4OISTLWOvNak 192.0.2.29 46618
192.0.2.4 443 tcp - 299.16248 845 1798 S3
T F 0 ShADadf 8 1269 18 2742
(empty) worker-2 \0\0 \0\0 nail7
1598897599.121515 CaWBSM32yACfH0OlNj 192.0.2.29 46418
192.0.2.4 443 tcp - 240.2451 845 1710 S3
T F 0 ShADadf 9 1321 9 2186
(empty) worker-10 \0\0 \0\0 nail7

# MV Schema

$ clickhouse-client --query "describe zeek_conn_mv_resp_h_tuples"
day Date
orig_h String
resp_h String
resp_p UInt16
proto Enum8(\'icmp\' = 1, \'tcp\' = 6, \'udp\' = 17)
conn_state Enum8(\'OTH\' = 0, \'REJ\' = 1, \'RSTO\' = 2,
\'RSTOS0\' = 3, \'RSTR\' = 4, \'RSTRH\' = 5, \'S0\' = 6, \'S1\' = 7,
\'S2\' = 8, \'S3\' = 9, \'SF\' = 10, \'SH\' = 11, \'SHR\' = 12)
conn_counts UInt64

# Query against the full table

$ clickhouse-client --query "SELECT day, orig_h, resp_h, resp_p,
proto, conn_state, count() AS conn_counts FROM zeek_conn PREWHERE day
= '2020-08-31' and file_src = 'nail7' where orig_h = '192.0.2.29' and
resp_h = '192.0.2.4' GROUP BY (day, orig_h, resp_h, resp_p, proto,
conn_state) order by orig_h DESC"
2020-08-31 192.0.2.29 192.0.2.4 443 tcp RSTO 2
2020-08-31 192.0.2.29 192.0.2.4 443 tcp RSTR 2
2020-08-31 192.0.2.29 192.0.2.4 443 tcp S3 5

# Query against the MV

$ clickhouse-client --query "SELECT * FROM zeek_conn_mv_resp_h_tuples
where orig_h = '198.128.208.29' and resp_h = '8.8.4.4' order by orig_h
DESC"'192.0.2.4' order by orig_h DESC"
2020-08-31 192.0.2.29 192.0.2.4 443 tcp S3 5
2020-08-31 192.0.2.29 192.0.2.4 443 tcp RSTR 4

### Example that returns as expected:

# different IP, query against the full table

$ clickhouse-client --query "SELECT day, orig_h, resp_h, resp_p,
proto, conn_state, count() AS conn_counts FROM zeek_conn PREWHERE day
= '2020-08-31' and file_src = 'nail7' where orig_h = '192.0.2.207' and
resp_h = '192.0.2.4' GROUP BY (day, orig_h, resp_h, resp_p, proto,
conn_state) order by orig_h DESC"
2020-08-31 192.0.2.207 192.0.2.4 53 udp SF 1
2020-08-31 192.0.2.207 192.0.2.4 443 tcp RSTOS0 78
2020-08-31 192.0.2.207 192.0.2.4 443 tcp RSTO 7

# different IP, query against the MV

$ clickhouse-client --query "SELECT * FROM zeek_conn_mv_resp_h_tuples
where orig_h = '192.0.2.207' and resp_h = '192.0.2.4' order by orig_h
DESC"
2020-08-31 192.0.2.207 192.0.2.4 443 tcp RSTO 7
2020-08-31 192.0.2.207 192.0.2.4 53 udp SF 1
2020-08-31 192.0.2.207 192.0.2.4 443 tcp RSTOS0 78

Michael Smitasin
Cyber Security
Information Technology Division
Lawrence Berkeley National Laboratory

[Denis Zhuravlev]

hard to guess, you did not provide your  `create MV `

SummingMergeTree table and MV are not connected.

The table which stores (MV) data have no knowledge about select in MV.

in 99.99999% cases `order by` section should be equal `group by ` in MV select

SummingMergeTree order by (conn_state ....)

select .... group by conn_state

Another common issue is misunderstanding how deduplication works for MV with replicated engines.

[Michael Smitasin]

Ah, sorry, here's the CREATE:

CREATE MATERIALIZED VIEW zeek_conn_mv_resp_h_tuples
ENGINE SummingMergeTree()
PARTITION BY toYYYYMMDD(day)
ORDER BY (day, resp_h, orig_h, resp_p)
AS SELECT day, orig_h, resp_h, resp_p, proto, conn_state, count() AS conn_counts
FROM zeek_conn

GROUP BY (day, orig_h, resp_h, resp_p, proto, conn_state)

And the query I'm using to populate the MV:

clickhouse-client --max_memory_usage=256000000000 --query "INSERT INTO
zeek_conn_mv_resp_h_tuples SELECT day, orig_h, resp_h, resp_p, proto,

conn_state, count() AS conn_counts FROM zeek_conn PREWHERE day =

'2020-08-31' WHERE file_src = 'nail7' GROUP BY (day, orig_h, resp_h,
resp_p, proto, conn_state)"

I'll check if changing ORDER BY makes any difference.

Michael Smitasin
Cyber Security
Information Technology Division
Lawrence Berkeley National Laboratory

> --
> You received this message because you are subscribed to the Google Groups "ClickHouse" group.
> To unsubscribe from this group and stop receiving emails from it, send an email to clickhouse+...@googlegroups.com.
> To view this discussion on the web visit https://groups.google.com/d/msgid/clickhouse/b2ace35e-c1cc-451a-990a-fad39cfcf187n%40googlegroups.com.

[Denis Zhuravlev]

I would expect

ORDER BY (day, orig_h, resp_h, resp_p, proto, conn_state)

AS SELECT day, orig_h, resp_h, resp_p, proto, conn_state, count() AS conn_counts
FROM zeek_conn

GROUP BY (day, orig_h, resp_h, resp_p, proto, conn_state)

During merges SummingMergeTree recalculate new part from old part and uses own ORDER BY section for collapsing argument, all columns that is not in ORDER BY section will calculated as ANY or SUMMED (if they are numeric)

in your case during merge SummingMergeTree do

 

day, resp_h, orig_h, resp_p, any(conn_state) , any(proto), sum(...)

....

group by (day, resp_h, orig_h, resp_p)

[Michael Smitasin]

You're too good, Denis. I owe you a beer/beverage-of-choice sometime.
I literally just copied the same set of columns from GROUP BY to ORDER
BY and that did it. Even diff'ing a 26M row output of full table vs.
MV and they're identical. Wonderful!

Michael Smitasin
Cyber Security
Information Technology Division
Lawrence Berkeley National Laboratory

On Tue, Sep 1, 2020 at 11:12 AM Denis Zhuravlev

> To view this discussion on the web visit https://groups.google.com/d/msgid/clickhouse/ee6bd9fd-3557-48fa-9329-9a1470438cf7n%40googlegroups.com.

[Denis Zhuravlev]

Good let's go deeper. But all further minor improvements / ideas for next new MV.

1.

PARTITION BY toYYYYMMDD(day)
ORDER BY (day, resp_h, orig_h, resp_p)

your table is partitioned by day so all parts have only one/single value of a day

BTW  PARTITION BY toYYYYMMDD(day)  === PARTITION BY (day)

So you can use

PARTITION BY (day)
ORDER BY (resp_h, orig_h, resp_p)

it's slightly save memory because primary key is shorter (w/o day column)

and queries (day = '2020-08-31' ) are still as fast as they could be because of partition pruning.

2.

You use implicit internal storage for MV. MV creates a table by itself and name the table as .inner.$mvname

This way has some downsides (https://github.com/ClickHouse/ClickHouse/issues/11470) in query performance and not easy for further maintenance. 

You can create the storage explicitly

create table zeek_conn_mv_resp_h_tuples (
day Date,
orig_h String,
resp_h String,
resp_p UInt16,

proto Enum8(\'icmp\' = 1, \'tcp\' = 6, \'udp\' = 17)
conn_state Enum8(\'OTH\' = 0, \'REJ\' = 1, \'RSTO\' = 2,
\'RSTOS0\' = 3, \'RSTR\' = 4, \'RSTRH\' = 5, \'S0\' = 6, \'S1\' = 7,

\'S2\' = 8, \'S3\' = 9, \'SF\' = 10, \'SH\' = 11, \'SHR\' = 12),
conn_counts UInt64 )
SummingMergeTree()
PARTITION BY (day)
ORDER BY (resp_h, orig_h, resp_p, proto, conn_state);

and point MV to store own data in this table (in this case you can name MV as you wish and don't need to define an engine because table already exists).

CREATE MATERIALIZED VIEW zeek_conn_mv_resp_h_tuples_mv to zeek_conn_mv_resp_h_tuples

AS SELECT day, orig_h, resp_h, resp_p, proto, conn_state, count() AS conn_counts
FROM zeek_conn
GROUP BY (day, orig_h, resp_h, resp_p, proto, conn_state)

3. you can create MV with a primary key that differs from the sorting key. It also saves memory. And improves performance in some cases.

https://clickhouse.tech/docs/en/engines/table-engines/mergetree-family/mergetree/#choosing-a-primary-key-that-differs-from-the-sorting-key

For example you never use proto, conn_state in where conditions and probably they are already unusable for index search because they behind high-cardinality columns resp_h, orig_h, resp_p

So you can create your table as

SummingMergeTree()
PARTITION BY (day)

PRIMARY KEY (resp_h, orig_h, resp_p)

ORDER BY (resp_h, orig_h, resp_p, proto, conn_state)

In this case primary_index will contain only columns (resp_h, orig_h, resp_p)

but the merge process will collapse rows over (resp_h, orig_h, resp_p, proto, conn_state) (summing result will be the same).

Also check Altinity articles https://altinity.com/blog/tag/materialized-views/ to shed more light.

On Tuesday, September 1, 2020 at 2:54:45 p.m. UTC-3 mnsmi...@lbl.gov wrote:

[Michael Smitasin]

Excellent info! Will give these a shot and report back. Appreciate it
as always, Denis.

Michael Smitasin
Cyber Security
Information Technology Division
Lawrence Berkeley National Laboratory

> To view this discussion on the web visit https://groups.google.com/d/msgid/clickhouse/c0cce35c-897a-493f-91a1-c38a63567c47n%40googlegroups.com.