Optimize BatchTransformStep incremental updates with offset-based query optimization

datapipe/cli.py

@@ -498,6 +498,73 @@ def migrate_transform_tables(ctx: click.Context, labels: str, name: str) -> None
     return migrations_v013.migrate_transform_tables(app, batch_transforms_steps)
 
 
+@cli.command()
+@click.option("--step", type=click.STRING, help="Step name to initialize offsets for (optional)")
+@click.pass_context
+def init_offsets(ctx, step: Optional[str]):
+    """
+    Инициализировать таблицу offset'ов из существующих данных TransformMetaTable.
+
+    Команда сканирует уже обработанные данные и устанавливает начальные значения offset'ов,
+    чтобы обеспечить плавную миграцию на оптимизацию через offset'ы (метод v2).
+
+    Если указан --step, инициализирует только этот шаг. Иначе инициализирует
+    все экземпляры BatchTransformStep в пайплайне.
+    """
+    from datapipe.meta.sql_meta import initialize_offsets_from_transform_meta
+
+    app: DatapipeApp = ctx.obj["app"]
+
+    # Collect all BatchTransformStep instances
+    transform_steps = []
+    for compute_step in app.steps:
+        if isinstance(compute_step, BaseBatchTransformStep):
+            if step is None or compute_step.get_name() == step:
+                transform_steps.append(compute_step)
+
+    if not transform_steps:
+        if step:
+            rprint(f"[red]Step '{step}' not found or is not a BatchTransformStep[/red]")
+        else:
+            rprint("[yellow]No BatchTransformStep instances found in pipeline[/yellow]")
+        return
+
+    rprint(f"[cyan]Found {len(transform_steps)} transform step(s) to initialize[/cyan]")
+
+    # Initialize offsets for each step
+    results = {}
+    for transform_step in transform_steps:
+        step_name = transform_step.get_name()
+        rprint(f"\n[cyan]Initializing offsets for: {step_name}[/cyan]")
+
+        try:
+            offsets = initialize_offsets_from_transform_meta(app.ds, transform_step)
+
+            if offsets:
+                rprint(f"[green]✓ Initialized {len(offsets)} offset(s):[/green]")
+                for input_name, offset_value in offsets.items():
+                    rprint(f"  - {input_name}: {offset_value}")
+                results[step_name] = offsets
+            else:
+                rprint("[yellow]No offsets initialized (no processed data found)[/yellow]")
+                results[step_name] = {}
+
+        except Exception as e:
+            rprint(f"[red]✗ Failed to initialize: {e}[/red]")
+            results[step_name] = {}
+
+    # Summary
+    rprint("\n[cyan]═══ Summary ═══[/cyan]")
+    success_count = sum(1 for v in results.values() if v is not None and len(v) > 0)
+    empty_count = sum(1 for v in results.values() if v is not None and len(v) == 0)
+    failed_count = sum(1 for v in results.values() if v is None)
+
+    rprint(f"[green]Successful: {success_count}[/green]")
+    rprint(f"[yellow]Empty (no data): {empty_count}[/yellow]")
+    if failed_count > 0:
+        rprint(f"[red]Failed: {failed_count}[/red]")
+
+
 try:
     entry_points = metadata.entry_points(group="datapipe.cli")  # type: ignore
 except TypeError:

datapipe/compute.py

@@ -85,6 +85,9 @@ class StepStatus:
 class ComputeInput:
     dt: DataTable
     join_type: Literal["inner", "full"] = "full"
+    # Filtered join optimization: mapping from idx columns to dt columns
+    # Example: {"user_id": "id"} means filter dt by dt.id IN (idx.user_id)
+    join_keys: Optional[Dict[str, str]] = None
 
 
 class ComputeStep:

datapipe/datatable.py

@@ -5,7 +5,7 @@
 from opentelemetry import trace
 
 from datapipe.event_logger import EventLogger
-from datapipe.meta.sql_meta import MetaTable
+from datapipe.meta.sql_meta import MetaTable, TransformInputOffsetTable
 from datapipe.run_config import RunConfig
 from datapipe.store.database import DBConn
 from datapipe.store.table_store import TableStore
@@ -165,6 +165,9 @@ def __init__(
 
         self.create_meta_table = create_meta_table
 
+        # Создать таблицу offset'ов (используем тот же флаг create_meta_table)
+        self.offset_table = TransformInputOffsetTable(meta_dbconn, create_table=create_meta_table)
+
     def create_table(self, name: str, table_store: TableStore) -> DataTable:
         assert name not in self.tables