feat: stream reset for incremental syncs

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Stream reset for incremental syncs**: re-fetch an incremental stream in full and *replace* the destination table for one run, then resume incremental from it. Previously the only way to rebuild a drifted table was to delete backend rows by hand — the watermark (`last_run`, taken from the last succeeded job) had no escape hatch. Request one with `bizon run config.yml --reset`, with `source.reset: true` in the config, or with `bizon stream reset <config>` (`--cancel` to withdraw, `--stream` to target a stream other than the config's for templated configs). Requests are scoped to a single stream — keyed on `(name, source_name, stream_name)`, the same triple as the watermark they override — so resetting one stream never affects another under the same pipeline name. The last form records the request in the backend and the next run consumes it, so pipelines whose command line is fixed by a scheduler need no change. During a reset the producer skips the watermark and calls `get()` instead of `get_records_after()`, and the run reaches destinations as `sync_mode: full_refresh` so they replace their table through their existing full-refresh path (for `bigquery`: staging into `{table}_temp`, then a `WRITE_TRUNCATE` copy job). The job row stays `incremental`, so the reset run becomes the next run's watermark. The request stays bound to the job running it, so a crashed reset is retried as a reset rather than silently degrading into an append. Only meaningful for `sync_mode: incremental` (ignored with a warning otherwise) and supported by every destination with a working full-refresh path — the exception is `bigquery_streaming`, which has no staging table and appends even on a full refresh, so a reset there is rejected at validation instead of duplicating data. Adds a `stream_resets` table, created automatically alongside the existing ones (no migration needed).
+
 ## [0.4.1] - 2026-06-29
 
 ### Fixed

CLAUDE.md

@@ -29,6 +29,7 @@ uv run ruff check --fix .       # Lint and auto-fix
 uv run bizon run config.yml     # Run a pipeline from YAML config
 uv run bizon source list        # List available sources
 uv run bizon stream list <source>  # List streams for a source
+uv run bizon stream reset config.yml  # Queue a stream reset for the next run
 ```
 
 ## Releasing
@@ -213,6 +214,39 @@ Then register in:
 - `INCREMENTAL` - Only new/updated records since last run (append-only)
 - `STREAM` - Continuous streaming mode
 
+### Stream Reset
+
+One incremental run that ignores the watermark, re-fetches everything, and **replaces** the
+destination table — then incremental resumes from that run. See `README.md#stream-reset` for the
+user-facing docs.
+
+The whole feature hangs off a single config field, `source.reset`, so it needs no new plumbing:
+`init_job()` (`bizon/engine/runner/runner.py`) runs in the parent before the producer and consumer
+are submitted, and both are handed the same `bizon_config` / `config` objects.
+
+- **Trigger** — `bizon run --reset`, `source.reset: true`, or a pending row in `stream_resets`
+  written by `bizon stream reset <config>` (the only form that reaches a run whose command line a
+  scheduler owns; `--stream` overrides the config's stream). `AbstractRunner.resolve_reset()`
+  collapses all three into one bool.
+- **Granularity** — keyed on `(name, source_name, stream_name)`, the same triple as
+  `get_last_successful_stream_job`, so a reset is exactly as scoped as the watermark it overrides.
+  Multi-stream configs (the `streams:` block) can never be reset: they require `sync_mode: stream`.
+- **Producer** (`pipeline/producer.py`) — skips the `get_last_successful_stream_job` lookup and falls
+  through to `source.get()`.
+- **Destination** — needs no reset-specific code. `SyncMetadata.from_bizon_config()` maps a reset onto
+  `sync_mode: full_refresh`, so the existing full-refresh path is reused with no new finalize branch
+  and every destination that can replace its table supports reset for free. Note this is the sync mode
+  of the *materialization*, not of the job. Destinations that append even on a full refresh (only
+  `bigquery_streaming`, which has no `finalize()`) are listed in `RESET_UNSUPPORTED_DESTINATIONS` in
+  `bizon/common/models.py` and rejected at config validation.
+- **Job row** — stays `incremental`, so `get_last_successful_stream_job` picks the reset run up as the
+  next watermark automatically.
+- **Crash safety** — every reset job has a consumed `stream_resets` row pointing at it
+  (`bind_stream_reset_to_job`). That is how a retry knows the in-flight job is a reset instead of
+  degrading into an append. `BigQueryDestination._ensure_clean_temp_table()` only drops the stale temp
+  table when the job has written no destination cursor yet — otherwise it would discard iterations the
+  resuming producer will not re-fetch.
+
 ### Implementing Incremental Sync
 
 Incremental sync requires implementation in both **sources** and **destinations**.

README.md

@@ -23,6 +23,7 @@ staying small enough to read end to end.
   - [Destinations](#destinations)
 - [Sync Modes](#sync-modes)
   - [Incremental Sync](#incremental-sync)
+  - [Stream Reset](#stream-reset)
 - [Engine Configuration](#engine-configuration)
   - [Backends](#backends-state-storage)
   - [Queues](#queues)
@@ -166,6 +167,7 @@ The CLI entry point is `bizon` (`bizon.cli.main:cli`).
 | `bizon run <config.yml>` | Run a pipeline from a YAML config |
 | `bizon source list` | List available sources and their streams |
 | `bizon stream list <source>` | List a source's streams, flagged `[Supports incremental]` / `[Full refresh only]` |
+| `bizon stream reset <config.yml>` | Queue a [stream reset](#stream-reset) for the next run of that pipeline |
 | `bizon secrets check <config.yml>` | Dry-run every `gsm://` / `env://` reference and report (masked) results |
 | `bizon destination` | Subcommand group (no subcommands yet) |
 
@@ -220,6 +222,7 @@ Common `SourceConfig` fields (`bizon/source/config.py`); each connector adds its
 | `cursor_field` | `None` | Timestamp field for incremental filtering (e.g. `updated_at`) |
 | `authentication` | `None` | Auth block (`type` + `params`); connector-specific |
 | `force_ignore_checkpoint` | `false` | Ignore existing checkpoints and restart from iteration 0 |
+| `reset` | `false` | Re-fetch the whole stream and replace the destination table, then resume incremental ([details](#stream-reset)) |
 | `max_iterations` | `None` | Cap iterations per run (default: run until source is exhausted) |
 | `api_config.retry_limit` | `10` | Retries before giving up on an API call |
 | `source_file_path` | `None` | Path to a custom source file (same as `--custom-source`) |
@@ -417,6 +420,52 @@ On the first incremental run (no previous successful job):
 - The job is marked successful
 - Subsequent runs use `get_records_after()` with the `last_run` timestamp
 
+#### Stream Reset
+
+A reset re-fetches the whole stream once and **replaces** the destination table, then resumes
+incremental from that run. Use it when the table has drifted — a backfill, a bug in a transform, or
+records the source changed without bumping its cursor field.
+
+There are three ways to ask for one; all do the same thing:
+
+```bash
+# One-shot, for a run you launch yourself
+bizon run config.yml --reset
+
+# Queued in the backend, for a pipeline whose command line is fixed by a scheduler.
+# The next `bizon run config.yml` picks it up — no change to the cron/Airflow job.
+bizon stream reset config.yml
+bizon stream reset config.yml --cancel   # changed your mind
+
+# One config templated across streams? Pick which stream to reset.
+bizon stream reset config.yml --stream deals
+```
+
+```yaml
+source:
+  sync_mode: incremental
+  reset: true      # same effect, set in the config
+```
+
+What the run does differently:
+- **Producer** ignores the watermark and calls `get()` instead of `get_records_after()`
+- **Destination** materializes the run as a full refresh, replacing the table rather than appending to
+  it (for `bigquery`: staging into `{table}_temp`, then a `WRITE_TRUNCATE` copy job)
+- The job row stays `incremental`, so the next run uses *this* run as its new watermark
+
+Notes:
+- **Scoped to a single stream.** The request is keyed on `(name, source_name, stream_name)` — the same
+  triple as the watermark it overrides — so resetting one stream never affects another, even under the
+  same pipeline name. `bizon stream reset` takes the stream from the config; `--stream` overrides it.
+- Only meaningful with `sync_mode: incremental`; ignored (with a warning) for the other modes.
+- Supported by every destination with a working full-refresh path, since that is what a reset
+  materializes as. The exception is `bigquery_streaming`, which has no staging table and appends even
+  on a full refresh; a reset there is rejected at config validation rather than duplicating your data.
+- A reset that crashes is retried as a reset — the request stays bound to its job, so a rerun cannot
+  silently degrade into an append.
+- `bizon stream reset` writes to the backend, so it must reach the same one the pipeline uses. With
+  the default file-based `sqlite` backend that means the same machine and file.
+
 ## Engine Configuration
 
 The `engine` block configures three pluggable subsystems. All have defaults, so `engine` is

bizon/cli/main.py

@@ -1,19 +1,28 @@
 import click
 from dotenv import find_dotenv, load_dotenv
 
-from bizon.engine.engine import RunnerFactory, replace_env_variables_in_config
+from bizon.common.models import BizonConfig
+from bizon.engine.backend.backend import BackendFactory
+from bizon.engine.backend.config import BackendTypes
+from bizon.engine.engine import (
+    RunnerFactory,
+    replace_env_variables_in_config,
+    resolve_config,
+)
 from bizon.engine.resolvers import (
     ReferenceResolutionError,
     ResolverRegistry,
     collect_references_in_config,
 )
 from bizon.engine.runner.config import LoggerLevel
+from bizon.source.config import SourceSyncModes
 from bizon.source.discover import discover_all_sources
 
 from .utils import (
     parse_from_yaml,
     set_custom_source_path_in_config,
     set_log_level,
+    set_reset_in_config,
     set_runner_in_config,
 )
 
@@ -71,6 +80,105 @@ def list(source_name: str):  # noqa
         click.echo(f"{stream_mode} - {stream.name}")
 
 
+@stream.command()
+@click.argument("filename", type=click.Path(exists=True))
+@click.option(
+    "--env-file",
+    required=False,
+    type=click.Path(exists=True),
+    help="Path to .env file to load environment variables from.",
+)
+@click.option("--cancel", is_flag=True, default=False, help="Cancel pending reset requests instead of adding one.")
+@click.option(
+    "--stream",
+    "stream_name",
+    required=False,
+    help="Reset this stream instead of the one named in the config, for configs templated across streams.",
+)
+def reset(filename: str, env_file: str, cancel: bool, stream_name: str):
+    """Request a reset of the incremental stream defined by a config file.
+
+    The request is stored in the backend and consumed by the next run of that pipeline, so scheduled
+    pipelines pick it up without any change to their command line. It is scoped to a single stream:
+    resetting one stream never affects another, even under the same pipeline name.
+    """
+
+    if env_file:
+        load_dotenv(env_file)
+    else:
+        load_dotenv(find_dotenv(".env"))
+
+    config = resolve_config(parse_from_yaml(filename))
+
+    if stream_name:
+        config["source"]["stream"] = stream_name
+
+    bizon_config = BizonConfig.model_validate(obj=config)
+
+    if bizon_config.source.sync_mode != SourceSyncModes.INCREMENTAL:
+        raise click.exceptions.ClickException(
+            f"Only incremental streams can be reset, but sync_mode is '{bizon_config.source.sync_mode.value}'. "
+            f"A '{bizon_config.source.sync_mode.value}' stream already rebuilds its destination on every run."
+        )
+
+    backend = BackendFactory.get_backend(config=bizon_config.engine.backend)
+    backend.check_prerequisites()
+    backend.create_all_tables()
+
+    if bizon_config.engine.backend.type == BackendTypes.SQLITE:
+        click.secho(
+            "Warning: the sqlite backend is a local file, so this request is only visible to runs using the same file.",
+            fg="yellow",
+        )
+
+    stream_label = f"{bizon_config.source.name} - {bizon_config.source.stream}"
+
+    if cancel:
+        cancelled = backend.cancel_pending_stream_resets(
+            name=bizon_config.name,
+            source_name=bizon_config.source.name,
+            stream_name=bizon_config.source.stream,
+        )
+
+        if cancelled:
+            click.secho(f"Cancelled {cancelled} pending reset request(s) for {stream_label}.", fg="green")
+        else:
+            click.echo(f"No pending reset request for {stream_label}.")
+        return
+
+    if backend.get_pending_stream_reset(
+        name=bizon_config.name,
+        source_name=bizon_config.source.name,
+        stream_name=bizon_config.source.stream,
+    ):
+        click.echo(f"A reset is already pending for {stream_label}, nothing to do.")
+        return
+
+    # Nothing validates the stream name here (the source is never instantiated), so a typo — most
+    # likely via --stream — would otherwise queue a reset that silently never fires.
+    if not backend.get_last_successful_stream_job(
+        name=bizon_config.name,
+        source_name=bizon_config.source.name,
+        stream_name=bizon_config.source.stream,
+    ):
+        click.secho(
+            f"Warning: no previous successful run found for {stream_label}. Check the stream name — "
+            f"a stream that has never run already fetches everything on its next run.",
+            fg="yellow",
+        )
+
+    backend.create_stream_reset(
+        name=bizon_config.name,
+        source_name=bizon_config.source.name,
+        stream_name=bizon_config.source.stream,
+    )
+    click.secho(
+        f"Reset requested for {stream_label}. The next run will re-fetch the full stream, replace the "
+        f"destination table, and resume incremental from there.",
+        fg="green",
+    )
+
+
 # Create a 'destination' group under 'bizon'
 @cli.group()
 def destination():
@@ -160,12 +268,20 @@ def check(filename: str, env_file: str):
     type=click.Path(exists=True),
     help="Path to .env file to load environment variables from.",
 )
+@click.option(
+    "--reset",
+    is_flag=True,
+    default=False,
+    help="Reset the incremental stream: re-fetch it in full and replace the destination table, "
+    "then resume incremental from this run.",
+)
 def run(
     filename: str,
     custom_source: str,
     runner: str,
     log_level: LoggerLevel,
     env_file: str,
+    reset: bool,
     help="Run a bizon pipeline from a YAML file.",
 ):
     """Run a bizon pipeline from a YAML file."""
@@ -188,6 +304,9 @@ def run(
     # Override runner param in config
     set_runner_in_config(config=config, runner=runner)
 
+    # Override reset param in config
+    set_reset_in_config(config=config, reset=reset)
+
     runner = RunnerFactory.create_from_config_dict(config=config)
     result = runner.run()
 

bizon/cli/utils.py

@@ -25,6 +25,12 @@ def set_custom_source_path_in_config(config: dict, custom_source: str):
         config["source"]["source_file_path"] = custom_source
 
 
+def set_reset_in_config(config: dict, reset: bool):
+    # Only ever written when the flag is passed: writing False would clobber `reset: true` set in YAML.
+    if reset:
+        config["source"]["reset"] = True
+
+
 # TODO: Refacto
 def set_runner_in_config(config: dict, runner: str):
     if runner:

bizon/common/models.py

@@ -15,12 +15,19 @@
 )
 from bizon.connectors.destinations.file.src.config import FileDestinationConfig
 from bizon.connectors.destinations.logger.src.config import LoggerConfig
+from bizon.destination.config import DestinationTypes
 from bizon.engine.config import EngineConfig
 from bizon.engine.resolvers.config import SecretsConfig
 from bizon.monitoring.config import MonitoringConfig
 from bizon.source.config import SourceConfig, SourceSyncModes
 from bizon.transform.config import TransformModel
 
+# A reset reaches destinations as `sync_mode: full_refresh` (see SyncMetadata.from_bizon_config), so
+# any destination with a working full-refresh path supports it for free. `bigquery_streaming` is the
+# exception: it has no `finalize()` and no staging table, so even a plain full refresh appends to the
+# final table instead of replacing it.
+RESET_UNSUPPORTED_DESTINATIONS = {DestinationTypes.BIGQUERY_STREAMING}
+
 
 class StreamSourceConfig(BaseModel):
     """Source-specific stream routing configuration.
@@ -176,6 +183,26 @@ def validate_streams_config(cls, v: Optional[list[StreamConfig]], info) -> Optio
 
         return v
 
+    @model_validator(mode="after")
+    def validate_reset_is_supported_by_destination(self) -> "BizonConfig":
+        """Reject a reset on destinations that cannot replace their table.
+
+        A reset re-fetches the whole stream, so a destination that appends instead of replacing would
+        silently duplicate the data. Fail loudly instead.
+        """
+        # Only incremental runs act on the flag (see AbstractRunner.resolve_reset), so a reset that is
+        # already going to be ignored must not be rejected here.
+        if not self.source.reset or self.source.sync_mode != SourceSyncModes.INCREMENTAL:
+            return self
+
+        if self.destination.name in RESET_UNSUPPORTED_DESTINATIONS:
+            raise ValueError(
+                f"Configuration Error: source.reset is not supported by destination "
+                f"'{self.destination.name}', which appends to its table instead of replacing it."
+            )
+
+        return self
+
     @model_validator(mode="before")
     @classmethod
     def inject_config_from_streams(cls, data: Any) -> Any:
@@ -260,15 +287,27 @@ class SyncMetadata(BaseModel):
     sync_mode: SourceSyncModes
     destination_name: str
     destination_alias: str
+    reset: bool = False
 
     @classmethod
     def from_bizon_config(cls, job_id: str, config: BizonConfig) -> "SyncMetadata":
+        sync_mode = config.source.sync_mode
+
+        # A reset materializes as a full refresh: it re-fetches the whole stream and replaces the
+        # table. Mapping it here means every destination with a working full-refresh path supports
+        # reset with no changes of its own. Note this is the sync mode of the *materialization*, not
+        # of the job: the job row is written from bizon_config and stays `incremental`, so
+        # get_last_successful_stream_job keeps using the reset run as the next watermark.
+        if config.source.reset and sync_mode == SourceSyncModes.INCREMENTAL:
+            sync_mode = SourceSyncModes.FULL_REFRESH
+
         return cls(
             name=config.name,
             job_id=job_id,
             source_name=config.source.name,
             stream_name=config.source.stream,
-            sync_mode=config.source.sync_mode,
+            sync_mode=sync_mode,
             destination_name=config.destination.name,
             destination_alias=config.destination.alias,
+            reset=config.source.reset,
         )