from __future__ import annotations import json import time from copy import deepcopy from dataclasses import dataclass from typing import Any from pydantic import BaseModel, ConfigDict, ValidationError, field_validator from errors import ConnectorError from metadata import TableName PHASE_ALIGN = "align" PHASE_CATCHUP = "catchup" PHASE_SNAPSHOT = "snapshot" RESYNC_CURRENT_FILE = "current_file" RESYNC_HIGH_WATERMARK = "high_watermark" RESYNC_NEXT_ROW = "next_row" RESYNC_PHASE = "phase" RESYNC_REQUEST_ID = "request_id" RESYNC_SNAPSHOT_PREFIX = "snapshot_prefix" RESYNC_SNAPSHOT_TSO = "snapshot_tso" RESYNC_TABLES = "tables" RESYNC_TRUNCATE_DONE = "truncate_done" SNAPSHOT_CURRENT_FILE = "current_file" SNAPSHOT_DONE = "done" SNAPSHOT_NEXT_ROW = "next_row" SNAPSHOT_TRUNCATE_DONE = "truncate_done" SNAPSHOT_TSO = "snapshot_tso" STATE_INCREMENTAL = "incremental" STATE_NEEDS_CHECKPOINT = "_needs_checkpoint" STATE_RESYNC = "resync" STATE_SNAPSHOT = "snapshot" STATE_SYNC_TASK_ID = "sync_task_id" STATE_VERSION = "version" INCREMENTAL_CHECKPOINT_TS = "checkpoint_ts" INCREMENTAL_DDL_WATERMARK = "ddl_watermark" INCREMENTAL_SCAN = "scan" SCAN_CURRENT_FILE = "current_file" SCAN_HIGH_WATERMARK = "high_watermark" SCAN_NEXT_ROW = "next_row" STATE_KEYS = { STATE_VERSION, STATE_SYNC_TASK_ID, STATE_SNAPSHOT, STATE_INCREMENTAL, STATE_RESYNC, STATE_NEEDS_CHECKPOINT, } SNAPSHOT_KEYS = { SNAPSHOT_DONE, SNAPSHOT_TSO, SNAPSHOT_TRUNCATE_DONE, SNAPSHOT_CURRENT_FILE, SNAPSHOT_NEXT_ROW, } INCREMENTAL_KEYS = {INCREMENTAL_CHECKPOINT_TS, INCREMENTAL_SCAN, INCREMENTAL_DDL_WATERMARK} RESYNC_KEYS = { RESYNC_REQUEST_ID, RESYNC_PHASE, RESYNC_TABLES, RESYNC_SNAPSHOT_TSO, RESYNC_SNAPSHOT_PREFIX, RESYNC_TRUNCATE_DONE, RESYNC_CURRENT_FILE, RESYNC_NEXT_ROW, RESYNC_HIGH_WATERMARK, } def _pydantic_error_message(exc: ValidationError) -> str: error = exc.errors()[0] ctx_error = error.get("ctx", {}).get("error") if ctx_error is not None: return str(ctx_error) return str(error.get("msg", exc)) class ScanState(BaseModel): model_config = ConfigDict(validate_assignment=True) high_watermark: int current_file: str = "" next_row: int = 0 @field_validator("high_watermark", mode="before") @classmethod def _validate_high_watermark(cls, value: Any) -> int: if value is None or not str(value).isdigit(): raise ValueError("state.incremental.scan.high_watermark must be a decimal string") return int(value) @field_validator("current_file", mode="before") @classmethod def _validate_current_file(cls, value: Any) -> str: current_file = value or "" if not isinstance(current_file, str): raise ValueError("state.incremental.scan.current_file must be a string") return _validate_relative_object_path( current_file, "state.incremental.scan.current_file", allow_empty=True, ) @field_validator("next_row", mode="before") @classmethod def _validate_next_row(cls, value: Any) -> int: next_row = value or 0 try: parsed_next_row = int(next_row) except (TypeError, ValueError) as exc: raise ValueError("state.incremental.scan.next_row must be non-negative") from exc if parsed_next_row < 0: raise ValueError("state.incremental.scan.next_row must be non-negative") return parsed_next_row @classmethod def from_dict(cls, raw: Any) -> ScanState: if not isinstance(raw, dict): raise ConnectorError("state.incremental.scan must be an object") _reject_unknown_keys( raw, {SCAN_HIGH_WATERMARK, SCAN_CURRENT_FILE, SCAN_NEXT_ROW}, "state.incremental.scan", ) try: return cls.model_validate(raw) except ValidationError as exc: raise ConnectorError(_pydantic_error_message(exc)) from exc def to_dict(self) -> dict[str, Any]: return { SCAN_HIGH_WATERMARK: str(self.high_watermark), SCAN_CURRENT_FILE: self.current_file, SCAN_NEXT_ROW: self.next_row, } class ResyncState(BaseModel): model_config = ConfigDict(validate_assignment=True) request_id: str phase: str tables: list[str] snapshot_tso: int snapshot_prefix: str | None = None truncate_done: bool = False current_file: str | None = None next_row: int = 0 high_watermark: int | None = None @field_validator("snapshot_tso", mode="before") @classmethod def _validate_snapshot_tso(cls, value: Any) -> int: if value is None or not str(value).isdigit(): raise ValueError("state.resync.snapshot_tso must be a decimal string") return int(value) @field_validator("next_row", mode="before") @classmethod def _validate_next_row(cls, value: Any) -> int: next_row = value or 0 try: parsed_next_row = int(next_row) except (TypeError, ValueError) as exc: raise ValueError("state.resync.next_row must be non-negative") from exc if parsed_next_row < 0: raise ValueError("state.resync.next_row must be non-negative") return parsed_next_row @field_validator("high_watermark", mode="before") @classmethod def _validate_high_watermark(cls, value: Any) -> int | None: if value is None: return None if not str(value).isdigit(): raise ValueError("state.resync.high_watermark must be a decimal string") return int(value) @classmethod def from_dict(cls, raw: Any) -> ResyncState: _validate_resync_marker(raw) snapshot_prefix = None if _raw_value(raw, RESYNC_SNAPSHOT_PREFIX) is not None: snapshot_prefix = _validate_relative_object_path( str(raw[RESYNC_SNAPSHOT_PREFIX]).strip("/"), "state.resync.snapshot_prefix", allow_empty=False, ) current_file = _raw_value(raw, RESYNC_CURRENT_FILE) if current_file is not None: current_file = _validate_relative_object_path( current_file, "state.resync.current_file", allow_empty=True, ) normalized = { RESYNC_REQUEST_ID: str(raw[RESYNC_REQUEST_ID]), RESYNC_PHASE: str(raw[RESYNC_PHASE]), RESYNC_TABLES: [table_name.quoted for table_name in _parse_resync_table_names(raw)], RESYNC_SNAPSHOT_TSO: raw[RESYNC_SNAPSHOT_TSO], RESYNC_SNAPSHOT_PREFIX: snapshot_prefix, RESYNC_TRUNCATE_DONE: bool(_raw_value(raw, RESYNC_TRUNCATE_DONE) or False), RESYNC_CURRENT_FILE: current_file, RESYNC_NEXT_ROW: _raw_value(raw, RESYNC_NEXT_ROW) or 0, RESYNC_HIGH_WATERMARK: _raw_value(raw, RESYNC_HIGH_WATERMARK), } try: return cls.model_validate(normalized) except ValidationError as exc: raise ConnectorError(_pydantic_error_message(exc)) from exc def to_dict(self) -> dict[str, Any]: payload: dict[str, Any] = { RESYNC_REQUEST_ID: self.request_id, RESYNC_PHASE: self.phase, RESYNC_TABLES: list(self.tables), RESYNC_SNAPSHOT_TSO: str(self.snapshot_tso), } if self.snapshot_prefix is not None: payload[RESYNC_SNAPSHOT_PREFIX] = self.snapshot_prefix if self.phase == PHASE_SNAPSHOT: payload[RESYNC_TRUNCATE_DONE] = self.truncate_done payload[RESYNC_CURRENT_FILE] = self.current_file payload[RESYNC_NEXT_ROW] = self.next_row if self.phase == PHASE_CATCHUP and self.high_watermark is not None: payload[RESYNC_HIGH_WATERMARK] = str(self.high_watermark) payload[RESYNC_CURRENT_FILE] = self.current_file payload[RESYNC_NEXT_ROW] = self.next_row return payload def is_snapshot_phase(self) -> bool: return self.phase == PHASE_SNAPSHOT def is_catchup_phase(self) -> bool: return self.phase == PHASE_CATCHUP def is_align_phase(self) -> bool: return self.phase == PHASE_ALIGN @property def table_names(self) -> tuple[TableName, ...]: return tuple(_parse_resync_table_name(table) for table in self.tables) @dataclass class ConnectorState: data: dict[str, Any] was_empty: bool = False resync_state: ResyncState | None = None scan_state: ScanState | None = None snapshot_required: bool = True @property def version(self) -> int: return int(self.data[STATE_VERSION]) @property def sync_task_id(self) -> str: return str(self.data[STATE_SYNC_TASK_ID]) @property def snapshot(self) -> dict[str, Any]: return self.data[STATE_SNAPSHOT] @property def incremental(self) -> dict[str, Any]: return self.data[STATE_INCREMENTAL] @property def snapshot_truncate_done(self) -> bool: return bool(self.data.get(STATE_SNAPSHOT, {}).get(SNAPSHOT_TRUNCATE_DONE)) def mark_snapshot_truncated(self) -> None: self.snapshot[SNAPSHOT_TRUNCATE_DONE] = True @property def snapshot_done(self) -> bool: return bool(self.data.get(STATE_SNAPSHOT, {}).get(SNAPSHOT_DONE)) def mark_snapshot_done(self) -> None: self.snapshot[SNAPSHOT_DONE] = True @property def checkpoint_ts(self) -> int: return int(self.incremental.get(INCREMENTAL_CHECKPOINT_TS) or 0) @checkpoint_ts.setter def checkpoint_ts(self, value: int) -> None: self.incremental[INCREMENTAL_CHECKPOINT_TS] = int(value) @property def ddl_watermark_by_table(self) -> dict[str, Any]: return self.incremental[INCREMENTAL_DDL_WATERMARK] @property def scan(self) -> ScanState | None: return self.scan_state @scan.setter def scan(self, value: ScanState | None) -> None: self.scan_state = value @property def resync(self) -> ResyncState | None: return self.resync_state @resync.setter def resync(self, value: ResyncState | None) -> None: self.resync_state = value @property def needs_checkpoint(self) -> bool: return bool(self.data.get(STATE_NEEDS_CHECKPOINT)) def pop_needs_checkpoint(self) -> bool: return bool(self.data.pop(STATE_NEEDS_CHECKPOINT, False)) def to_dict(self) -> dict[str, Any]: payload = deepcopy(self.data) snapshot = payload.get(STATE_SNAPSHOT) if snapshot is not None: snapshot[SNAPSHOT_TSO] = str(snapshot_tso(self)) if self.snapshot_done: snapshot.pop(SNAPSHOT_TRUNCATE_DONE, None) snapshot.pop(SNAPSHOT_CURRENT_FILE, None) snapshot.pop(SNAPSHOT_NEXT_ROW, None) else: snapshot.setdefault(SNAPSHOT_TRUNCATE_DONE, False) snapshot.setdefault(SNAPSHOT_CURRENT_FILE, None) snapshot.setdefault(SNAPSHOT_NEXT_ROW, 0) incremental = payload[STATE_INCREMENTAL] incremental[INCREMENTAL_CHECKPOINT_TS] = str(self.checkpoint_ts) if self.scan_state is None: incremental.pop(INCREMENTAL_SCAN, None) else: incremental[INCREMENTAL_SCAN] = self.scan_state.to_dict() if self.resync_state is None: payload.pop(STATE_RESYNC, None) else: payload[STATE_RESYNC] = self.resync_state.to_dict() return payload def normalize_state( raw_state: dict[str, Any] | str | None, *, sync_task_id: str, snapshot_required: bool = True, ) -> ConnectorState: state = _decode_raw_state(raw_state) was_empty = not state _normalize_state_header(state, was_empty=was_empty, sync_task_id=sync_task_id) _normalize_snapshot_state(state, was_empty=was_empty, snapshot_required=snapshot_required) resync_state = _pop_resync_state(state) scan_state = _normalize_incremental_state(state) return ConnectorState( state, was_empty=was_empty, resync_state=resync_state, scan_state=scan_state, snapshot_required=snapshot_required, ) def _decode_raw_state(raw_state: dict[str, Any] | str | None) -> dict[str, Any]: if raw_state is None or raw_state == "": return {} if isinstance(raw_state, str): parsed = json.loads(raw_state) if not isinstance(parsed, dict): raise ConnectorError("Fivetran state must decode to a JSON object") return parsed if isinstance(raw_state, dict): return deepcopy(raw_state) raise ConnectorError("Fivetran state must be a dict or JSON string") def _normalize_state_header(state: dict[str, Any], *, was_empty: bool, sync_task_id: str) -> None: if state and STATE_VERSION not in state: raise ConnectorError("non-empty state must include version") _reject_unknown_keys(state, STATE_KEYS, "state") version = state.get(STATE_VERSION, 1) if version != 1: raise ConnectorError(f"unsupported state version: {version!r}") state[STATE_VERSION] = 1 existing_sync_task_id = state.get(STATE_SYNC_TASK_ID) if not was_empty and not existing_sync_task_id: raise ConnectorError("non-empty state must include sync_task_id") if existing_sync_task_id not in (None, "", sync_task_id): raise ConnectorError( f"state sync_task_id {existing_sync_task_id!r} does not match current configuration {sync_task_id!r}" ) state[STATE_SYNC_TASK_ID] = sync_task_id def _normalize_snapshot_state(state: dict[str, Any], *, was_empty: bool, snapshot_required: bool) -> None: if snapshot_required and not was_empty and STATE_SNAPSHOT not in state: raise ConnectorError("non-empty state must include snapshot") if not snapshot_required and STATE_SNAPSHOT not in state: return snapshot = state.setdefault(STATE_SNAPSHOT, {}) if snapshot_required else state[STATE_SNAPSHOT] if not isinstance(snapshot, dict): raise ConnectorError("state.snapshot must be an object") _reject_unknown_keys(snapshot, SNAPSHOT_KEYS, "state.snapshot") if not was_empty and SNAPSHOT_DONE not in snapshot: raise ConnectorError("non-empty state.snapshot must include done") snapshot[SNAPSHOT_DONE] = ( _require_bool(snapshot[SNAPSHOT_DONE], "state.snapshot.done") if SNAPSHOT_DONE in snapshot else False ) if not was_empty and SNAPSHOT_TSO not in snapshot: raise ConnectorError("non-empty state.snapshot must include snapshot_tso") snapshot_tso_value = snapshot.setdefault(SNAPSHOT_TSO, 0) if not str(snapshot_tso_value).isdigit(): raise ConnectorError("state.snapshot.snapshot_tso must be a decimal string") snapshot[SNAPSHOT_TSO] = int(snapshot_tso_value) if snapshot[SNAPSHOT_DONE]: _reject_completed_snapshot_cursor(snapshot) return _normalize_snapshot_cursor(snapshot) def _reject_completed_snapshot_cursor(snapshot: dict[str, Any]) -> None: for cursor_key in (SNAPSHOT_TRUNCATE_DONE, SNAPSHOT_CURRENT_FILE, SNAPSHOT_NEXT_ROW): if cursor_key in snapshot: raise ConnectorError(f"completed state.snapshot must not include {cursor_key}") def _normalize_snapshot_cursor(snapshot: dict[str, Any]) -> None: snapshot[SNAPSHOT_TRUNCATE_DONE] = _require_bool( snapshot.get(SNAPSHOT_TRUNCATE_DONE, False), "state.snapshot.truncate_done", ) snapshot.setdefault(SNAPSHOT_CURRENT_FILE, None) snapshot.setdefault(SNAPSHOT_NEXT_ROW, 0) current_file = _raw_value(snapshot, SNAPSHOT_CURRENT_FILE) if current_file is not None: snapshot[SNAPSHOT_CURRENT_FILE] = _validate_relative_object_path( str(current_file), "state.snapshot.current_file", allow_empty=True, ) next_row = _raw_value(snapshot, SNAPSHOT_NEXT_ROW) if next_row is not None: snapshot[SNAPSHOT_NEXT_ROW] = _parse_non_negative_int(next_row, "state.snapshot.next_row") if not snapshot.get(SNAPSHOT_CURRENT_FILE) and snapshot[SNAPSHOT_NEXT_ROW] != 0: raise ConnectorError("state.snapshot.next_row must be 0 when current_file is null") def _parse_non_negative_int(value: Any, field: str) -> int: try: parsed_value = int(value or 0) except (TypeError, ValueError) as exc: raise ConnectorError(f"{field} must be non-negative") from exc if parsed_value < 0: raise ConnectorError(f"{field} must be non-negative") return parsed_value def _pop_resync_state(state: dict[str, Any]) -> ResyncState | None: if STATE_RESYNC in state: resync_state = ResyncState.from_dict(state[STATE_RESYNC]) del state[STATE_RESYNC] return resync_state return None def _normalize_incremental_state(state: dict[str, Any]) -> ScanState | None: incremental = state.setdefault(STATE_INCREMENTAL, {}) if not isinstance(incremental, dict): raise ConnectorError("state.incremental must be an object") _reject_unknown_keys(incremental, INCREMENTAL_KEYS, "state.incremental") checkpoint_ts = incremental.setdefault(INCREMENTAL_CHECKPOINT_TS, 0) if not str(checkpoint_ts).isdigit(): raise ConnectorError("state.incremental.checkpoint_ts must be a decimal string") incremental[INCREMENTAL_CHECKPOINT_TS] = int(checkpoint_ts) incremental.setdefault(INCREMENTAL_DDL_WATERMARK, {}) if not isinstance(incremental[INCREMENTAL_DDL_WATERMARK], dict): raise ConnectorError("state.incremental.ddl_watermark must be an object") scan_state = None if INCREMENTAL_SCAN in incremental: scan_state = ScanState.from_dict(incremental[INCREMENTAL_SCAN]) del incremental[INCREMENTAL_SCAN] return scan_state def _validate_resync_marker(raw: Any) -> None: if not isinstance(raw, dict): raise ConnectorError("state.resync must be an object") _reject_unknown_keys(raw, RESYNC_KEYS, "state.resync") phase = _validate_resync_phase(raw) _validate_resync_tables(raw) _validate_resync_truncate_done(raw) _validate_resync_cursor(raw) _validate_resync_required_fields(raw) _validate_resync_phase_fields(raw, phase) def _validate_resync_phase(raw: dict[str, Any]) -> str: phase = _raw_value(raw, RESYNC_PHASE) if phase not in {PHASE_SNAPSHOT, PHASE_CATCHUP, PHASE_ALIGN}: raise ConnectorError("state.resync.phase must be one of snapshot/catchup/align") return phase def _validate_resync_tables(raw: dict[str, Any]) -> None: _parse_resync_table_names(raw) def _parse_resync_table_names(raw: dict[str, Any]) -> tuple[TableName, ...]: tables = _raw_value(raw, RESYNC_TABLES) if not isinstance(tables, list) or not tables: raise ConnectorError("state.resync.tables must be a non-empty list") table_names: list[TableName] = [] seen: set[TableName] = set() for table in tables: if not isinstance(table, str) or not table: raise ConnectorError("state.resync.tables must contain source table identifiers") table_name = _parse_resync_table_name(table) if table_name in seen: raise ConnectorError(f"duplicate table in state.resync.tables: {table}") seen.add(table_name) table_names.append(table_name) return tuple(table_names) def _parse_resync_table_name(value: str) -> TableName: try: return TableName.from_qualified(value) except ValueError as exc: raise ConnectorError(f"state.resync.tables {exc}") from exc def _validate_resync_truncate_done(raw: dict[str, Any]) -> None: truncate_done = _raw_value(raw, RESYNC_TRUNCATE_DONE) if truncate_done is not None and not isinstance(truncate_done, bool): raise ConnectorError("state.resync.truncate_done must be a boolean") def _validate_resync_cursor(raw: dict[str, Any]) -> None: next_row = _raw_value(raw, RESYNC_NEXT_ROW) normalized_next_row = 0 if next_row is not None: normalized_next_row = _parse_non_negative_int(next_row, "state.resync.next_row") current_file = _raw_value(raw, RESYNC_CURRENT_FILE) if current_file is not None and not isinstance(current_file, str): raise ConnectorError("state.resync.current_file must be a string or null") if not current_file and normalized_next_row != 0: raise ConnectorError("state.resync.next_row must be 0 when current_file is null") def _validate_resync_required_fields(raw: dict[str, Any]) -> None: request_id = _raw_value(raw, RESYNC_REQUEST_ID) if not isinstance(request_id, str) or not request_id: raise ConnectorError("state.resync.request_id must be a non-empty string") snapshot_tso = _raw_value(raw, RESYNC_SNAPSHOT_TSO) if snapshot_tso is None or not str(snapshot_tso).isdigit(): raise ConnectorError("state.resync.snapshot_tso must be a decimal string") def _validate_resync_phase_fields(raw: dict[str, Any], phase: str) -> None: if phase == PHASE_SNAPSHOT: snapshot_prefix = _raw_value(raw, RESYNC_SNAPSHOT_PREFIX) if not isinstance(snapshot_prefix, str) or not snapshot_prefix.strip("/"): raise ConnectorError("state.resync.snapshot_prefix must be a non-empty string") if phase == PHASE_CATCHUP: high_watermark = _raw_value(raw, RESYNC_HIGH_WATERMARK) if high_watermark is None or not str(high_watermark).isdigit(): raise ConnectorError("state.resync.high_watermark must be a decimal string") def _raw_value(raw: dict[str, Any], key: str) -> Any: return raw.get(key) def _require_bool(value: Any, field: str) -> bool: if not isinstance(value, bool): raise ConnectorError(f"{field} must be a boolean") return value def _reject_unknown_keys(raw: dict[str, Any], allowed: set[str], path: str) -> None: unknown = sorted(set(raw) - allowed) if unknown: raise ConnectorError(f"{path} contains unknown fields: {unknown}") def _validate_relative_object_path(value: str, path: str, *, allow_empty: bool) -> str: if not isinstance(value, str): raise ConnectorError(f"{path} must be a string") if value == "": if allow_empty: return "" raise ConnectorError(f"{path} must be a non-empty relative object path") if value.startswith("/"): raise ConnectorError(f"{path} must be a relative object path") parts = value.split("/") if any(part in {"", ".", ".."} for part in parts): raise ConnectorError(f"{path} must be a normalized relative object path") return value def validate_relative_object_path(value: str, path: str, *, allow_empty: bool) -> str: return _validate_relative_object_path(value, path, allow_empty=allow_empty) def snapshot_tso(state: ConnectorState) -> int: return int(state.data.get(STATE_SNAPSHOT, {}).get(SNAPSHOT_TSO) or 0) def set_snapshot_tso(state: ConnectorState, value: int) -> bool: current = snapshot_tso(state) if current and current != value: raise ConnectorError(f"state snapshot_tso {current} does not match current snapshot metadata {value}") if current == value: return False state.snapshot[SNAPSHOT_TSO] = value return True def table_resync_names(state: ConnectorState) -> set[TableName]: resync = state.resync if resync is None: return set() return set(resync.table_names) def table_resync_truncate_done(state: ConnectorState) -> bool: resync = state.resync return resync is not None and resync.truncate_done is True def mark_table_resync_truncated(state: ConnectorState) -> bool: resync = state.resync if resync is None or resync.truncate_done is True: return False resync.truncate_done = True return True def apply_table_resync_requests(state: ConnectorState) -> bool: resync_tables = table_resync_names(state) if not resync_tables: return False changed = False for table_name in resync_tables: changed |= clear_incremental_table_state(state, table_name) return changed def clear_incremental_table_state(state: ConnectorState, table_name: TableName) -> bool: quoted = table_name.quoted if quoted not in state.ddl_watermark_by_table: return False del state.ddl_watermark_by_table[quoted] return True def finish_table_resync_if_complete(state: ConnectorState) -> bool: resync_tables = table_resync_names(state) if not resync_tables: return False resync = state.resync if resync is None: return False snapshot_tso_value = resync.snapshot_tso checkpoint_ts = state.checkpoint_ts if snapshot_tso_value == checkpoint_ts: state.resync = None return True if snapshot_tso_value < checkpoint_ts: resync.phase = PHASE_CATCHUP resync.high_watermark = checkpoint_ts resync.current_file = "" resync.next_row = 0 return True resync.phase = PHASE_ALIGN return True def snapshots_complete(state: ConnectorState) -> bool: return state.snapshot_done def compact_incremental_state(state: ConnectorState) -> bool: return False def ddl_watermark(state: ConnectorState, quoted_table_name: str) -> int: value = state.ddl_watermark_by_table.get(quoted_table_name, 0) return int(value or 0) def set_ddl_watermark(state: ConnectorState, quoted_table_name: str, table_version: int) -> None: current = ddl_watermark(state, quoted_table_name) if table_version > current: state.ddl_watermark_by_table[quoted_table_name] = table_version class CheckpointPolicy: def __init__(self, rows: int, seconds: int) -> None: self.rows = rows self.seconds = seconds self.rows_since_checkpoint = 0 self.last_checkpoint_monotonic = time.monotonic() def record_row(self) -> bool: self.rows_since_checkpoint += 1 if self.rows_since_checkpoint >= self.rows: return True return time.monotonic() - self.last_checkpoint_monotonic >= self.seconds def mark_checkpointed(self) -> None: self.rows_since_checkpoint = 0 self.last_checkpoint_monotonic = time.monotonic() __all__ = [ "CheckpointPolicy", "ConnectorState", "ResyncState", "ScanState", "apply_table_resync_requests", "clear_incremental_table_state", "compact_incremental_state", "ddl_watermark", "finish_table_resync_if_complete", "mark_table_resync_truncated", "normalize_state", "set_ddl_watermark", "set_snapshot_tso", "snapshot_tso", "snapshots_complete", "table_resync_names", "table_resync_truncate_done", "validate_relative_object_path", ]