fill transaction meta on chunk store and simplify full outer join by swsvc · Pull Request #355 · epoch8/datapipe

swsvc · 2025-10-09T12:18:45Z

No description provided.

elephantum · 2025-10-09T15:48:49Z

datapipe/meta/sql_meta.py

    sa.Column("process_ts", sa.Float),  # Время последней успешной обработки
    sa.Column("is_success", sa.Boolean),  # Успешно ли обработана строка
    sa.Column("priority", sa.Integer),  # Приоритет обработки (чем больше, тем выше)
+    sa.Column("status", sa.String), # Статус исполнения трансформации


Возможно тогда is_success лишняя колонка

удаляю

halconel · 2025-10-10T11:49:04Z

Пока из того что я вижу, тут получится full table scan + cross joins при записи: O(N × T × M) где T - кол-во трансформаций, M - размер связанных таблиц.

Вот в чем суть проблемы:
при каждом вызове store_chunk(), для каждой трансформации зовем get_index_data() для всех связанных таблиц. а get_index_data() читает все данные таблицы. второе, это cross joins при не совпадающих ключах: 1000 profiles х 100K posts = 100M записей в TransformMetaTable

Я бы покрыл новую логику store_chunk() нагрузочными тестами, что бы убедится, что мы не переложили проблему из одного места в другое.

swsvc · 2025-10-13T10:25:01Z

@halconel суть вот в чем. Если на вход трансформации приходит две или больше таблиц, то по transform_keys этот джойн все равно будет исполняться. Ну просто по логике работы трубы. Насколько я понимаю, этот джойн еще нужно делать вручную в коде трансформации, потому что на вход трансформации приходит несколько таблиц и они там внутри джойнятся. И это не перекладывание проблемы из одного места в другое, это просто то, что задается человеком в пайплайне. От этого никуда не уйти.

На примере: если есть таблица картинок и таблица моделей классификации, которые надо прогнать по этим картинкам, то тут будет кросс джойн, просто из-за логики того, что хочет человек. Получается, что раньше в этом месте потенциально было две больших операции: кросс джойн из-за логики и большой full outer join, от которого хотим уйти. Теперь большой full outer join уходит, а кросс джойн переносится на этап записи данных в таблицу

halconel · 2025-10-13T11:49:43Z

@halconel суть вот в чем. Если на вход трансформации приходит две или больше таблиц, то по transform_keys этот джойн все равно будет исполняться. Ну просто по логике работы трубы. Насколько я понимаю, этот джойн еще нужно делать вручную в коде трансформации, потому что на вход трансформации приходит несколько таблиц и они там внутри джойнятся. И это не перекладывание проблемы из одного места в другое, это просто то, что задается человеком в пайплайне. От этого никуда не уйти.

На примере: если есть таблица картинок и таблица моделей классификации, которые надо прогнать по этим картинкам, то тут будет кросс джойн, просто из-за логики того, что хочет человек. Получается, что раньше в этом месте потенциально было две больших операции: кросс джойн из-за логики и большой full outer join, от которого хотим уйти. Теперь большой full outer join уходит, а кросс джойн переносится на этап записи данных в таблицу

Запись происходит намного чаще чем чтение в типичных ETL пайплайнах.
Чтение всех связанных таблиц при записи в store_chunk - это не cross join, это full table scan.

@swsvc Я все еще очень надеюсь, что это будет покрыто нагрузочными тестами и мы увидим, как изменилась производительность при записи.

swsvc · 2025-10-13T17:10:38Z

Очевидно, что производительность при записи упадет, потому что выполняются дополнительные операции. Она упадет для разных типов операций по-разному - это тоже очевидно. Нагрузочные тесты я планирую сделать

tests/test_core_steps1.py

swsvc · 2025-10-21T09:45:19Z

tests/test_image_pipeline.py

    assert len(list(tmp_dir.glob("tbl2/*.png"))) == 10


+@pytest.mark.skip(reason="impossible to trace changes when they happen externally")


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

swsvc · 2025-10-21T09:46:14Z

tests/test_table_store_json_line.py

        out.write('{"id": "2", "text": "text2"}\n')


+@pytest.mark.skip(reason="impossible to trace changes when they happen externally")


Точто так же как в предыдущем тесте: файл меняется не через трубу а напрямую через file IO. Скипнул

swsvc · 2025-10-21T09:47:03Z

tests/test_table_store_qdrant.py

    yield pd.DataFrame({"id": [1], "embedding": [[0.1]], "str_payload": ["foo"], "int_payload": [42]})


+@pytest.mark.skip(reason="qdrant store cannot read all the rows from the index")


Для корректной работы нужно чтение всего индекса из квадранта, что он делать не умеет. Тест скипнул

Вот это теперь снова работает

elephantum · 2025-10-26T13:54:55Z

datapipe/step/batch_transform.py

+        if not transform_meta_table_exists:
+            meta_index = extract_transformation_meta(self.input_dts, self.transform_keys)
+            if not meta_index.empty:
+                self.meta_table.insert_rows(meta_index)


Я вот здесь не уверен, надо ли это прямо так делать. Может быть лучше это контролируемо через отдельные менеджмент команды?

я убрал в метод класса и напрямую вызываю в тестах. Потом если будет юзкейс, то станет понятно, как лучше сделать

swsvc · 2025-11-12T19:24:10Z

В этом коммите есть реализованный алгоритм записи меты для транзакций полностью на sql. Однако он используется только в одном случае: когда все таблицы, участвующие в трансформации, лежат в одной sql базе данных (postgres, sqlite). Если это не так, то джойн происходит в пандасе (потребляет много памяти, потому что выкачивает таблицы для джойна)

fill transaction meta on chunk store and simplify full outer join

32e4854

swsvc force-pushed the feat/fill-transaction-meta-on-chunk-store branch from 05ed9cc to 32e4854 Compare October 9, 2025 12:40

elephantum reviewed Oct 9, 2025

View reviewed changes

swsvc commented Oct 21, 2025

View reviewed changes

tests/test_core_steps1.py Show resolved Hide resolved

swsvc commented Oct 21, 2025

View reviewed changes

fixes for tests

aaa2756

swsvc force-pushed the feat/fill-transaction-meta-on-chunk-store branch from d7b59d9 to aaa2756 Compare October 21, 2025 09:48

elephantum reviewed Oct 26, 2025

View reviewed changes

elephantum and others added 4 commits October 26, 2025 19:58

fix empty pipeline

9e8d3d5

create initial transform metadata and adjust tests

fbd3005

remove is_success column

1226e19

Merge master into feat/fill-transaction-meta-on-chunk-store

565deb4

swsvc force-pushed the feat/fill-transaction-meta-on-chunk-store branch from 2153276 to 565deb4 Compare October 28, 2025 15:26

calculate transformation meta in db

65d9071

swsvc added 2 commits November 13, 2025 18:03

fix for sqlite

242536b

calculate transaction meta on data meta using sql

bd2dbfb

swsvc force-pushed the feat/fill-transaction-meta-on-chunk-store branch from e9e6b1a to bd2dbfb Compare November 18, 2025 20:22

		assert len(list(tmp_dir.glob("tbl2/*.png"))) == 10


		@pytest.mark.skip(reason="impossible to trace changes when they happen externally")

		out.write('{"id": "2", "text": "text2"}\n')


		@pytest.mark.skip(reason="impossible to trace changes when they happen externally")

		yield pd.DataFrame({"id": [1], "embedding": [[0.1]], "str_payload": ["foo"], "int_payload": [42]})


		@pytest.mark.skip(reason="qdrant store cannot read all the rows from the index")

Conversation

swsvc commented Oct 9, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

halconel commented Oct 10, 2025

Uh oh!

swsvc commented Oct 13, 2025

Uh oh!

halconel commented Oct 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

swsvc commented Oct 13, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

swsvc commented Nov 12, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

halconel commented Oct 13, 2025 •

edited

Loading