Добрый день, коллеги. Прошу совета у коллективного разума для решения следующей задачи.
Имеется таблица:

CREATE TABLE events (
 event_time DateTime,
 event_id UUID,
 country String,
 state String
) ENGINE = MergeTree()

Пускай в таблицу пишутся некие события, каждое со своим event_id и state=new. Также для некоторых событий приходят обновления значения state (например processing, rejected, approved и т.д.) и для сохранения этого факта в ту же таблицу записывается новая строка с ранее известным event_id (а также country), но свежим event_time и state. Таким образом для каждого уникального event_id актуальное значение state лежит в строке с максимальным event_time.

Требуется выполнить аналитический запрос и получить количество событий со state равным rejected и approved с группировкой по country. Первое, что приходит в голову:

SELECT
 country,
 sum(state='rejected') as rejected,
 sum(state='approved') as approved,
FROM (
 SELECT any(country) AS country, argMax(state, event_time) as state
 FROM events
 GROUP BY event_id
)
GROUP BY country

Но GROUP BY event_id не выглядит хорошим решением. Напрашивается использование ReplacingMergeTree, но он не даёт гарантии отсутствия дубликатов. Есть альтернативные способы построить подобный запрос? Или надо как-то менять структуру хранения событий?

Можно наверное какой-нибудь kubernates job приделать чтоб он на всех подах запускал миграции

ReplacingMergeTree и запрос SELECT ... FINAL ?

Если я правильно понял, документация советует избегать использование FINAL

Привет, при апгрейде с 19.13.3.26 до 20.3.11.97 очень много такого
Таблицы локальные MergeTree, не реплицируемые

https://pastebin.vkpartner.ru/raw/ergJ5aIZb0VAU5vhlmIoQ4CydHWzNksGJDobibwdEsvqRwsEdXvSPgpZ78DQc4xh8Nx2oRL3VoUkhTa8

После 10 минут попытки запуска оно падает c ошибкой Suspiciously many (38) broken parts to remove.

Есть специальный движок для возможных обновлений

Забыл как называется)

Посмотри, может тебе подойдёт для твоей задачи

В итоге запрос станет проще ага:

select
    country,
    state,
    sum(sign) as events_count
from events
group by country, state

Но вставка станет кудрявее. Для каждого нового состояния надо будет вставить ещё и отмену предыдущего. А если порядок вставки не гарантируется, то надо юзать VersionedCollapsingMergeTree. Там ещё версия добавляется.

А если таблица используется не только для этого, не факт что остальные запросы получится переписать под такую схему без последствий.

Доброго всем. Я уже задавал этот вопрос пару дней назад, но все же.

Есть таблица

data_lake (
 order_id nullable(int32),
 order_date nullable(datetime)
 order_product_id nullable(int32)
 order_product_price nullable(decimal)
)

Есть 2 таблицы с engine kafka (для order и для order_product)
и 2 матвьюхи (консюмеры из кафки в таблицу data_lake). В консюмере я перечисляю только те поля, которые есть в соответствующем кафка топике.

Почему при вычитывании из кафки я получаю ошибки

<Error> void DB::StorageKafka::threadFunc(): Code: 8, e.displayText() = DB::Exception: Cannot find column `order_id` in source stream: while pushing to view streams.order_product_consumer

На тестовом сервер работало нормально. Версия 19.15.3.6
Может есть настройка какая, которая требует строгого перечисления ВСЕХ полей?

select
    country,
    state,
    sum(sign) as events_count
from events
group by country, state

Добрый день! А тоже самое, интересно, сделать в MV можно каким-то образом ? У меня тоже есть похожая задача, но хотелось бы её полностью отдать на базу.

data_lake (
 order_id nullable(int32),
 order_date nullable(datetime)
 order_product_id nullable(int32)
 order_product_price nullable(decimal)
)

<Error> void DB::StorageKafka::threadFunc(): Code: 8, e.displayText() = DB::Exception: Cannot find column `order_id` in source stream: while pushing to view streams.order_product_consumer

надо дефолты у таблицы проставить, даже если это NULL

прописаны дефолты вида
default CAST(NULL, 'Nullable(Int32)')

Неа. В MV не получить предыдущие состояния выгружаемых строк. Это всё надо делать перед вставкой.

а если мы точно знаем, что предыдущее состояние строк нам не интересно и мы хотим его в любом случае заменить новым, при этом удалив некоторые из них ? Что то мне подсказывает что для этого надо какой-то другой движок использовать, а не CollapsingMergeTree.

ReplacingMergeTree

не удалит строки