feat: implement strict mode for max_records_per_run to prevent offset overshoot with composite keys

[halconel]

Summary

Implements strict mode for max_records_per_run to fix offset overshoot bug when using composite keys. When max_records_per_run limits the batch size, the offset calculation now uses only the actually processed records instead of all events for the processed key pairs.

Problem

When transform_keys differ from the input table's primary key:

• SQL query with LIMIT returns N specific records with full PK
• These records group by transform_keys into fewer pairs
• get_batch_input_dfs reads ALL events for those pairs (not just N records)
• Offset = MAX(update_ts) from all events → overshoot beyond the limited batch
• Result: intermediate data gets skipped in subsequent runs

Solution

Strict mode behavior when max_records_per_run is set:

1. SQL query includes full PK in result (via _get_additional_idx_columns)
2. Result wrapped in IndexDFWithMetadata(idx_grouped, changed_idx_full_pk)
3. get_batch_input_dfs reads ONLY the specific records from changed_idx (strict mode)
4. Offset calculated from changed_idx instead of querying DB
5. Removed limit_hit guard - safe to update offset with strict mode

Changes

• New file: datapipe/index_metadata.py - wrapper class for passing metadata with idx
• Modified: datapipe/step/batch_transform.py - strict mode implementation
• New test: tests/test_overshoot_composite_keys.py - demonstrates bug and verifies fix
• Updated: tests/test_batch_transform_with_offset_optimization.py - updated for strict mode behavior
• Documentation: docs/max_records_per_run_associativity_plan.md - implementation plan

Test Results

• ✅ test_offset_overshoot_with_composite_keys - PASSED (bug fixed)
• ✅ All 27 batch_transform tests - PASSED
• ✅ No regressions in existing functionality

Assumptions

When max_records_per_run is set, the transformation is assumed to be associative: T(A) + T(B) = T(A+B). No validation mechanism implemented yet.

[halconel]

/codex review

[oneclick-codex-reviewer]

Вердикт: есть блокер.

PR добавляет strict mode для max_records_per_run, обертку индекса, изменения SQL для error records, тесты и CI-пины. Главный риск: частичный батч теперь помечает весь transform_key обработанным, из-за чего последующие записи того же ключа могут быть навсегда отфильтрованы.

Критичные проблемы

Найдена потеря данных при повторных запусках strict mode для ключей, у которых записи выходят за пределы первого LIMIT.

Важные замечания

Есть риск неправильного чтения входов с join_keys: strict branch обходит существующую логику filtered join.

Тестовое покрытие

Новый тест проверяет только первый запуск и offset, но не проверяет последующие прогоны и финальный результат для той же пары ключей. Тесты не запускал: по правилам ревью нельзя выполнять код из репозитория.

Правила репозитория

Repo rules не предоставлены, шаг N/A.

[oneclick-codex-reviewer]

Сценарий: max_records_per_run берет только часть full-PK строк для profile_id/post_id, а здесь они снова сворачиваются до transform_keys.
Влияние: store_batch_result пометит весь transform_key успешным с текущим process_ts; следующие строки того же ключа с update_ts < process_ts больше не попадут в выборку, и агрегаты вроде count будут потеряны/перезаписаны.
Фикс: не отмечать частичный transform_key как полностью обработанный либо хранить прогресс на уровне full PK/watermark и добавить тест с 2-3 run_full, проверяющий финальный результат для ключа, разбитого лимитом.

[oneclick-codex-reviewer]

Сценарий: у второго input есть join_keys={"user_id": "id"}, а changed_idx содержит user_id, но не PK id этой таблицы.
Влияние: inp.dt.get_data(changed_idx) не применит mapping join_keys; для таблицы без пересечения PK это превращается в чтение всей таблицы и может дать неверные join/агрегаты или OOM на больших справочниках.
Фикс: для каждого inp строить отдельный индекс: для join_keys использовать старую filtered-join логику с rename(columns=inp.join_keys), и покрыть это тестом max_records_per_run + JoinSpec.