from __future__ import annotations from dataclasses import dataclass from typing import Any from checkpointing import CheckpointController from config import OPERATION_MODE_STANDARD, ConnectorConfig from errors import IncrementalError from fivetran.operations import OperationSink, TimedOperationSink, raw_csv_record_writer_factory from fivetran.sdk_records import ( DiagnosticColumnSource, RawCSVOperationWriters, RawValueRef, build_raw_csv_operation_writers, ) from metadata import TableName from observability import OperationStats, elapsed_ms, info, info_with_memory, timer_start, warning, warning_with_memory from pipeline import DecodedSink, RawSink, Sink from row_decoder import CSVRowDecoder as EventDecoder from snapshot.schema import fivetran_table_name from state import ( CheckpointPolicy, ConnectorState, ResyncState, ScanState, compact_incremental_state, ddl_watermark, set_ddl_watermark, snapshot_tso, snapshots_complete, table_resync_names, ) from storage.object_store import ObjectStore, join_key from type_mapping import is_source_type_syncable from .csv import Event, RawEvent, add_metadata_columns, iter_raw_events from .paths import IndexPath, SchemaPath, is_index_key, parse_index_path, parse_schema_path from .schema import TableDefinition, classify_ddl_action, load_table_definition from .schema_index import StorageSchemaIndex @dataclass(frozen=True) class SchemaMarker: path: SchemaPath table_def: TableDefinition @property def sort_key(self) -> tuple[int, int, str, str, str, int]: return ( self.path.table_version, 0, self.path.source_schema, self.path.source_table, "", 0, ) @dataclass(frozen=True) class DMLRange: index_path: IndexPath @property def sort_key(self) -> tuple[int, int, str, str, str, int]: return ( self.index_path.table_version, 1, self.index_path.source_schema, self.index_path.source_table, self.index_path.date, self.index_path.partition_id, ) @dataclass(frozen=True) class ScanResult: schema_paths: list[SchemaPath] index_paths: list[IndexPath] listed_prefixes: int listed_keys: int @dataclass(frozen=True) class DMLBatchResult: files_processed: int events_emitted: int events_skipped_lower_bound: int stats: OperationStats @dataclass class IncrementalProgress: files_processed: int = 0 events_emitted: int = 0 events_skipped_lower_bound: int = 0 def record_dml_batch(self, result: DMLBatchResult) -> None: self.files_processed += result.files_processed self.events_emitted += result.events_emitted self.events_skipped_lower_bound += result.events_skipped_lower_bound def read_increment_metadata(store: ObjectStore, increment_prefix: str) -> int | None: metadata_key = join_key(increment_prefix, "metadata") if not store.exists(metadata_key): return None raw = store.read_json(metadata_key) if not isinstance(raw, dict): raise IncrementalError("increment/metadata must be a JSON object") if "checkpoint-ts" not in raw: raise IncrementalError("increment/metadata missing checkpoint-ts") try: return int(raw["checkpoint-ts"]) except (TypeError, ValueError) as exc: raise IncrementalError("increment/metadata checkpoint-ts must be an integer") from exc def _scan_visible_inputs( *, config: ConnectorConfig, store: ObjectStore, high_watermark: int, ) -> ScanResult: schema_paths: list[SchemaPath] = [] index_paths: list[IndexPath] = [] unknown_sample: list[str] = [] listed_prefixes = 1 listed_keys = 0 skipped: dict[str, int] = { "future_schema": 0, "future_index": 0, "unknown": 0, } def record_unknown(key: str) -> None: skipped["unknown"] += 1 if len(unknown_sample) < 20: unknown_sample.append(key) for key in store.iter_keys(config.incremental_prefix): listed_keys += 1 schema_path = parse_schema_path(key, config.incremental_prefix) if schema_path is not None: if schema_path.table_version <= high_watermark: schema_paths.append(schema_path) else: skipped["future_schema"] += 1 continue if is_index_key(key): index_path = parse_index_path(key, config.incremental_prefix, store.read_text(key)) if index_path is None: record_unknown(key) continue if index_path.table_version <= high_watermark: index_paths.append(index_path) else: skipped["future_index"] += 1 continue if _is_unknown_meta_key(key, config.incremental_prefix): record_unknown(key) skipped = {name: count for name, count in skipped.items() if count} if skipped: info( "skipped incremental inputs", phase="incremental", metadata_checkpoint_ts=high_watermark, skipped=skipped, sample_keys=unknown_sample[:5], ) return ScanResult( schema_paths=schema_paths, index_paths=index_paths, listed_prefixes=listed_prefixes, listed_keys=listed_keys, ) def _is_unknown_meta_key(key: str, incremental_prefix: str) -> bool: prefix = incremental_prefix.strip("/") relative = key.strip("/") if prefix and relative.startswith(prefix + "/"): relative = relative[len(prefix) + 1 :] parts = relative.split("/") return len(parts) >= 3 and parts[-2] == "meta" def _looks_like_date(value: str) -> bool: parts = value.split("-") if len(parts) == 1: return len(parts[0]) == 4 and parts[0].isdigit() if len(parts) == 2: return len(parts[0]) == 4 and len(parts[1]) == 2 and all(part.isdigit() for part in parts) if len(parts) == 3: return ( len(parts[0]) == 4 and len(parts[1]) == 2 and len(parts[2]) == 2 and all(part.isdigit() for part in parts) ) return False def _load_table_definitions( store: ObjectStore, schema_paths: list[SchemaPath] ) -> dict[tuple[TableName, int], TableDefinition]: table_defs: dict[tuple[TableName, int], TableDefinition] = {} for path in schema_paths: raw = store.read_json(path.key) table_def = load_table_definition(raw, path) key = (table_def.name, table_def.table_version) if key in table_defs: warning( "duplicate schema file for same table version", phase="incremental", source_schema=table_def.source_schema, source_table=table_def.source_table, tableVersion=table_def.table_version, file_path=path.key, ) table_defs[key] = table_def return table_defs def _build_work_items( schema_paths: list[SchemaPath], table_defs: dict[tuple[TableName, int], TableDefinition], index_paths: list[IndexPath], ) -> list[SchemaMarker | DMLRange]: items: list[SchemaMarker | DMLRange] = [] for path in schema_paths: items.append(SchemaMarker(path=path, table_def=table_defs[(path.table_name, path.table_version)])) for index_path in index_paths: items.append(DMLRange(index_path=index_path)) return sorted(items, key=lambda item: item.sort_key) def _decode_event(event: RawEvent, event_decoder: EventDecoder) -> Event: data: dict[str, Any] = {} event_decoder.decode_into(event.raw_values, data) return Event( operation=event.operation, source_schema=event.source_schema, source_table=event.source_table, commit_ts=event.commit_ts, data=data, event_offset=event.event_offset, ) def _new_sink( *, config: ConnectorConfig, operations: OperationSink, table_def: TableDefinition, ) -> Sink[RawEvent]: if config.operation_mode == OPERATION_MODE_STANDARD: event_decoders: dict[int, EventDecoder] = {} def event_decoder(input_offset: int) -> EventDecoder: decoder = event_decoders.get(input_offset) if decoder is not None: return decoder decoder = EventDecoder.new( table_def.columns, null_marker=config.incremental.csv_null_value, binary_encoding=config.incremental.csv_binary_encoding, input_offset=input_offset, ) event_decoders[input_offset] = decoder return decoder def write_decoded(event: Event) -> None: table = fivetran_table_name(event.source_schema, event.source_table) if event.operation == "D": operations.delete(table=table, keys={name: event.data[name] for name in table_def.primary_key}) return data = add_metadata_columns(event, event.data) if config.enable_diagnostic_columns else event.data operations.upsert(table=table, data=data) return DecodedSink( decode=lambda event: _decode_event(event, event_decoder(event.input_offset)), write_decoded=write_decoded, ) operation_writers: dict[int, RawCSVOperationWriters] = {} def diagnostic_columns(input_offset: int) -> DiagnosticColumnSource | None: if not config.enable_diagnostic_columns: return None return DiagnosticColumnSource( commit_ts=RawValueRef(3) if input_offset == 4 else 0, source_schema=RawValueRef(2), source_table=RawValueRef(1), ) def operation_writer(input_offset: int) -> RawCSVOperationWriters: writers = operation_writers.get(input_offset) if writers is not None: return writers writers = build_raw_csv_operation_writers( writer_factory=raw_csv_record_writer_factory(operations), table=fivetran_table_name(table_def.source_schema, table_def.source_table), columns=table_def.columns, null_marker=config.incremental.csv_null_value, binary_encoding=config.incremental.csv_binary_encoding, input_offset=input_offset, primary_key=table_def.primary_key, diagnostic_columns=diagnostic_columns(input_offset), include_delete=True, ) operation_writers[input_offset] = writers return writers def write_raw(event: RawEvent) -> None: writers = operation_writer(event.input_offset) if event.operation == "D": writers.write_delete(event.raw_values) return writers.write_upsert(event.raw_values) return RawSink(write_raw=write_raw) def _unsupported_primary_key_columns(table_def: TableDefinition) -> list[str]: primary_key = set(table_def.primary_key) return [ column.name for column in table_def.columns if column.name in primary_key and not is_source_type_syncable(column.source_type, precision=column.precision) ] def _syncable_table_names(schema_index: StorageSchemaIndex) -> set[TableName]: table_names: set[TableName] = set() for table_def in schema_index.table_defs.values(): if not table_def.primary_key: continue unsupported_primary_key_columns = _unsupported_primary_key_columns(table_def) if unsupported_primary_key_columns: warning( "skipping incremental table because primary key contains unsupported column type", phase="incremental", source_schema=table_def.source_schema, source_table=table_def.source_table, tableVersion=table_def.table_version, columns=unsupported_primary_key_columns, ) continue table_names.add(table_def.name) return table_names def _handle_schema_marker( *, operations: OperationSink, state: ConnectorState, table_def: TableDefinition, baseline_versions: dict[TableName, int], ) -> None: current = max( ddl_watermark(state, table_def.quoted), baseline_versions.get(table_def.name, 0), ) if table_def.table_version <= current: return action = classify_ddl_action(table_def) if action.operation == "truncate": operations.truncate(table=fivetran_table_name(table_def.source_schema, table_def.source_table)) set_ddl_watermark(state, table_def.quoted, table_def.table_version) operations.checkpoint(state=state) return if action.should_warn: warning( "schema marker advanced with warning", phase="incremental", source_schema=table_def.source_schema, source_table=table_def.source_table, tableVersion=table_def.table_version, ddl_type=table_def.ddl_type, ddl_action=action.name, query=table_def.query, ) set_ddl_watermark(state, table_def.quoted, table_def.table_version) operations.checkpoint(state=state) def _baseline_versions( table_defs: dict[tuple[TableName, int], TableDefinition], lower_bounds: dict[TableName, int], ) -> dict[TableName, int]: baselines: dict[TableName, int] = {} for table_name, table_version in table_defs: if table_version <= lower_bounds.get(table_name, 0): baselines[table_name] = max(baselines.get(table_name, 0), table_version) return baselines def _data_file_order_key(config: ConnectorConfig, file_key: str) -> tuple[Any, ...]: prefix = config.incremental_prefix.strip("/") key = file_key.strip("/") if prefix and key.startswith(prefix + "/"): key = key[len(prefix) + 1 :] parts = key.split("/") if len(parts) < 4: return (file_key,) source_schema = parts[0] source_table = parts[1] try: table_version = int(parts[2]) except ValueError: table_version = -1 filename = parts[-1] digits = "".join(ch for ch in filename if ch.isdigit()) file_index = int(digits or 0) tail = parts[3:-1] date = "" partition_id = 0 if tail and _looks_like_date(tail[-1]): date = tail[-1] tail = tail[:-1] if tail: try: partition_id = int(tail[-1]) except ValueError: partition_id = 0 return (table_version, source_schema, source_table, date, partition_id, file_index, file_key) def _compare_with_cursor(config: ConnectorConfig, file_key: str, current_file: str) -> int: if not current_file: return 1 current_key = _data_file_order_key(config, current_file) file_order = _data_file_order_key(config, file_key) if file_order < current_key: return -1 if file_order == current_key: return 0 return 1 def _cursor_current_file(cursor_state: ScanState | ResyncState | None) -> str: return str(getattr(cursor_state, "current_file", "") or "") def _cursor_next_event(cursor_state: ScanState | ResyncState | None) -> int: return int(getattr(cursor_state, "next_row", 0) or 0) def _advance_dml_cursor(cursor_state: ScanState | ResyncState | None, *, file_key: str, next_event: int) -> None: if cursor_state is None: return cursor_state.current_file = file_key cursor_state.next_row = next_event def _record_event(checkpoint: CheckpointController) -> None: checkpoint.record_row() def _has_pending_events(checkpoint: CheckpointController) -> bool: return checkpoint.has_pending_rows def _process_dml_range_group( *, config: ConnectorConfig, store: ObjectStore, state: ConnectorState, operations: OperationSink, policy: CheckpointPolicy, table_def: TableDefinition, index_paths: list[IndexPath], lower_bound: int, high_watermark: int, cursor_state: ScanState | ResyncState | None, ) -> DMLBatchResult: batch_stats = OperationStats() batch_operations = TimedOperationSink(operations, batch_stats) checkpoint = CheckpointController(batch_operations, state, policy) files_processed = 0 events_emitted = 0 events_skipped_lower_bound = 0 sink = _new_sink(config=config, operations=batch_operations, table_def=table_def) for index_path in sorted(index_paths, key=lambda path: path.sort_key): for file_index in range(1, index_path.latest_file_index + 1): file_key = index_path.data_file_key(file_index) if not store.exists(file_key): raise IncrementalError( f"DML file referenced by index does not exist: index={index_path.key} file={file_key}" ) cursor_compare = _compare_with_cursor(config, file_key, _cursor_current_file(cursor_state)) if cursor_compare < 0: continue start_event = _cursor_next_event(cursor_state) if cursor_compare == 0 else 0 _advance_dml_cursor(cursor_state, file_key=file_key, next_event=start_event) file_events_skipped_lower_bound = 0 file_events_skipped_high_watermark = 0 file_events_emitted = 0 with store.open_text(file_key) as handle: for event in iter_raw_events( handle, table_def=table_def, start_event=start_event, ): if event.commit_ts is not None and event.commit_ts <= lower_bound: _advance_dml_cursor(cursor_state, file_key=file_key, next_event=event.event_offset + 1) file_events_skipped_lower_bound += 1 _record_event(checkpoint) continue if event.commit_ts is not None and event.commit_ts > high_watermark: _advance_dml_cursor(cursor_state, file_key=file_key, next_event=event.event_offset + 1) file_events_skipped_high_watermark += 1 _record_event(checkpoint) continue sink.write(event) events_emitted += 1 file_events_emitted += 1 _advance_dml_cursor(cursor_state, file_key=file_key, next_event=event.event_offset + 1) _record_event(checkpoint) events_skipped_lower_bound += file_events_skipped_lower_bound files_processed += 1 info( "incremental file consumed", phase="incremental", source_schema=index_path.source_schema, source_table=index_path.source_table, tableVersion=index_path.table_version, file_path=file_key, file_index=file_index, events_emitted=file_events_emitted, events_skipped_lower_bound=file_events_skipped_lower_bound, events_skipped_high_watermark=file_events_skipped_high_watermark, start_event=start_event, ) if events_emitted or _has_pending_events(checkpoint): checkpoint.checkpoint() return DMLBatchResult( files_processed=files_processed, events_emitted=events_emitted, events_skipped_lower_bound=events_skipped_lower_bound, stats=batch_stats, ) def _build_incremental_lower_bounds( *, config: ConnectorConfig, state: ConnectorState, ready_tables: set[TableName], resync: ResyncState | None, resync_tables: set[TableName], ) -> dict[TableName, int]: lower_bounds = { table_name: (max(snapshot_tso(state), state.checkpoint_ts) if config.snapshot_required else state.checkpoint_ts) for table_name in ready_tables } if resync is not None and resync.is_catchup_phase(): lower_bounds = {table_name: resync.snapshot_tso for table_name in ready_tables} elif resync is not None and resync.is_align_phase(): for table_name in resync_tables: if table_name in lower_bounds: lower_bounds[table_name] = max(lower_bounds[table_name], resync.snapshot_tso) if config.snapshot_required and any(value <= 0 for value in lower_bounds.values()): raise IncrementalError("snapshot.snapshot_tso is required before incremental sync") return lower_bounds def _next_dml_group( items: list[SchemaMarker | DMLRange], start_index: int, ) -> tuple[list[IndexPath], int]: first_item = items[start_index] if not isinstance(first_item, DMLRange): raise TypeError("_next_dml_group must start at a DML range") table_name = first_item.index_path.table_name table_version = first_item.index_path.table_version dml_group = [first_item.index_path] item_index = start_index + 1 while item_index < len(items): next_item = items[item_index] if not isinstance(next_item, DMLRange): break if next_item.index_path.table_name != table_name or next_item.index_path.table_version != table_version: break dml_group.append(next_item.index_path) item_index += 1 return dml_group, item_index @dataclass class IncrementalWorkProcessor: config: ConnectorConfig store: ObjectStore state: ConnectorState operations: OperationSink phase_operations: OperationSink phase_stats: OperationStats policy: CheckpointPolicy table_defs: dict[tuple[TableName, int], TableDefinition] baseline_versions: dict[TableName, int] lower_bounds: dict[TableName, int] ready_tables: set[TableName] high_watermark: int cursor_state: ScanState | ResyncState | None progress: IncrementalProgress def process(self, schema_paths: list[SchemaPath], index_paths: list[IndexPath]) -> None: items = _build_work_items(schema_paths, self.table_defs, index_paths) item_index = 0 while item_index < len(items): item = items[item_index] if isinstance(item, SchemaMarker): item_index += 1 self._handle_schema_marker_item(item) continue dml_group, item_index = _next_dml_group(items, item_index) result = self._process_dml_work_group(dml_group) if result is None: continue self.phase_stats.merge(result.stats) self.progress.record_dml_batch(result) def _handle_schema_marker_item(self, item: SchemaMarker) -> None: if item.table_def.name not in self.ready_tables: return _handle_schema_marker( operations=self.phase_operations, state=self.state, table_def=item.table_def, baseline_versions=self.baseline_versions, ) self.policy.mark_checkpointed() def _process_dml_work_group(self, dml_group: list[IndexPath]) -> DMLBatchResult | None: first_index_path = dml_group[0] table_name = first_index_path.table_name table_version = first_index_path.table_version if table_name not in self.ready_tables: return None table_def = self.table_defs.get((table_name, table_version)) if table_def is None: raise IncrementalError(f"missing visible schema file for DML index {first_index_path.key}") if not table_def.primary_key: warning( "skipping DML range because table has no primary key", phase="incremental", source_schema=first_index_path.source_schema, source_table=first_index_path.source_table, tableVersion=table_version, ) return None unsupported_primary_key_columns = _unsupported_primary_key_columns(table_def) if unsupported_primary_key_columns: warning( "skipping DML range because primary key contains unsupported column type", phase="incremental", source_schema=first_index_path.source_schema, source_table=first_index_path.source_table, tableVersion=table_version, columns=unsupported_primary_key_columns, ) return None effective_watermark = max( ddl_watermark(self.state, table_def.quoted), self.baseline_versions.get(table_name, 0), ) if table_version < effective_watermark: info( "skipping DML range older than DDL watermark", phase="incremental", source_schema=first_index_path.source_schema, source_table=first_index_path.source_table, tableVersion=table_version, ddl_watermark=effective_watermark, ) return None return _process_dml_range_group( config=self.config, store=self.store, state=self.state, operations=self.operations, policy=self.policy, table_def=table_def, index_paths=dml_group, lower_bound=self.lower_bounds[table_name], high_watermark=self.high_watermark, cursor_state=self.cursor_state, ) def _finish_incremental_scan( *, state: ConnectorState, operations: OperationSink, resync: ResyncState | None, high_watermark: int, needs_checkpoint: bool, ) -> None: if compact_incremental_state(state) or needs_checkpoint: operations.checkpoint(state=state) info("incremental state compacted", phase="incremental") if resync is not None and resync.is_catchup_phase() and state.resync is not None: state.resync = None operations.checkpoint(state=state) return state.checkpoint_ts = high_watermark state.scan = None if ( resync is not None and resync.is_align_phase() and state.resync is not None and high_watermark >= resync.snapshot_tso ): state.resync = None operations.checkpoint(state=state) def _skip_incremental_phase( *, message: str, state: ConnectorState, operations: OperationSink, needs_checkpoint: bool, phase_start: float, phase_stats: OperationStats, **fields: Any, ) -> None: if needs_checkpoint: operations.checkpoint(state=state) info( message, phase="incremental", duration_ms=elapsed_ms(phase_start), **fields, **phase_stats.as_log_fields(), ) def sync_incremental( *, config: ConnectorConfig, store: ObjectStore, schema_index: StorageSchemaIndex, state: ConnectorState, operations: OperationSink, ) -> None: phase_start = timer_start() phase_stats = OperationStats() phase_operations = TimedOperationSink(operations, phase_stats) progress = IncrementalProgress() needs_checkpoint = state.pop_needs_checkpoint() visible_tables = _syncable_table_names(schema_index) snapshot_ready = snapshots_complete(state) if config.snapshot_required else True ready_tables = set(visible_tables) if snapshot_ready else set() if not ready_tables: _skip_incremental_phase( message="incremental skipped because no table snapshot is ready", state=state, operations=phase_operations, needs_checkpoint=needs_checkpoint, phase_start=phase_start, phase_stats=phase_stats, ) return resync = state.resync resync_phase = resync.phase if resync is not None else "" resync_tables = table_resync_names(state) scan = state.scan metadata_start = timer_start() metadata_checkpoint_ts = read_increment_metadata(store, config.incremental_prefix) metadata_read_duration_ms = elapsed_ms(metadata_start) cursor_state: ScanState | ResyncState | None = scan if resync is not None and resync.is_catchup_phase(): if resync.high_watermark is None: raise IncrementalError("resync.high_watermark is required in catchup phase") high_watermark = resync.high_watermark cursor_state = resync ready_tables &= resync_tables elif scan is not None: high_watermark = scan.high_watermark else: if metadata_checkpoint_ts is None: _skip_incremental_phase( message="incremental metadata not found", state=state, operations=phase_operations, needs_checkpoint=needs_checkpoint, phase_start=phase_start, phase_stats=phase_stats, increment_prefix=config.incremental_prefix, metadata_read_duration_ms=metadata_read_duration_ms, ) return if metadata_checkpoint_ts <= state.checkpoint_ts: _skip_incremental_phase( message="incremental fast path skipped scan", state=state, operations=phase_operations, needs_checkpoint=needs_checkpoint, phase_start=phase_start, phase_stats=phase_stats, metadata_checkpoint_ts=metadata_checkpoint_ts, metadata_read_duration_ms=metadata_read_duration_ms, ) return state.scan = ScanState(high_watermark=metadata_checkpoint_ts) cursor_state = state.scan high_watermark = metadata_checkpoint_ts phase_operations.checkpoint(state=state) needs_checkpoint = False if not ready_tables: _skip_incremental_phase( message="incremental skipped because no ready tables are in scope", state=state, operations=phase_operations, needs_checkpoint=needs_checkpoint, phase_start=phase_start, phase_stats=phase_stats, resync_phase=resync_phase, ) return lower_bounds = _build_incremental_lower_bounds( config=config, state=state, ready_tables=ready_tables, resync=resync, resync_tables=resync_tables, ) min_lower_bound = min(lower_bounds.values()) info_with_memory( "starting incremental phase", phase="incremental", snapshot_tso=min_lower_bound, ready_table_count=len(ready_tables), pending_table_count=len(visible_tables - ready_tables), metadata_checkpoint_ts=high_watermark, state_checkpoint_ts=state.checkpoint_ts, metadata_read_duration_ms=metadata_read_duration_ms, resync_phase=resync_phase, ) try: scan_start = timer_start() scan_result = _scan_visible_inputs( config=config, store=store, high_watermark=high_watermark, ) scan_duration_ms = elapsed_ms(scan_start) except Exception as exc: warning( "incremental phase failed", phase="incremental", duration_ms=elapsed_ms(phase_start), metadata_checkpoint_ts=high_watermark, metadata_read_duration_ms=metadata_read_duration_ms, error_type=type(exc).__name__, error_message=str(exc), **phase_stats.as_log_fields(), ) raise info_with_memory( "incremental scan completed", phase="incremental", metadata_checkpoint_ts=high_watermark, scan_duration_ms=scan_duration_ms, listed_prefixes=scan_result.listed_prefixes, listed_keys=scan_result.listed_keys, schema_files=len(scan_result.schema_paths), index_files=len(scan_result.index_paths), ) schema_paths = scan_result.schema_paths index_paths = scan_result.index_paths try: schema_load_start = timer_start() table_defs = _load_table_definitions(store, schema_paths) schema_load_duration_ms = elapsed_ms(schema_load_start) info_with_memory( "incremental schema definitions loaded", phase="incremental", metadata_checkpoint_ts=high_watermark, schema_load_duration_ms=schema_load_duration_ms, schema_definitions=len(table_defs), ) baseline_versions = _baseline_versions(table_defs, lower_bounds) policy = CheckpointPolicy(config.checkpoint.interval_rows, config.checkpoint.interval_seconds) work_processor = IncrementalWorkProcessor( config=config, store=store, state=state, operations=operations, phase_operations=phase_operations, phase_stats=phase_stats, policy=policy, table_defs=table_defs, baseline_versions=baseline_versions, lower_bounds=lower_bounds, ready_tables=ready_tables, high_watermark=high_watermark, cursor_state=cursor_state, progress=progress, ) work_processor.process(schema_paths, index_paths) _finish_incremental_scan( state=state, operations=phase_operations, resync=resync, high_watermark=high_watermark, needs_checkpoint=needs_checkpoint, ) except Exception as exc: warning_with_memory( "incremental phase failed", phase="incremental", duration_ms=elapsed_ms(phase_start), metadata_checkpoint_ts=high_watermark, metadata_read_duration_ms=metadata_read_duration_ms, error_type=type(exc).__name__, error_message=str(exc), events_emitted=progress.events_emitted, events_skipped_lower_bound=progress.events_skipped_lower_bound, files_processed=progress.files_processed, **phase_stats.as_log_fields(), ) raise phase_duration_ms = elapsed_ms(phase_start) info_with_memory( "incremental phase completed", phase="incremental", metadata_checkpoint_ts=high_watermark, resync_phase=resync_phase, duration_ms=phase_duration_ms, metadata_read_duration_ms=metadata_read_duration_ms, scan_duration_ms=scan_duration_ms, schema_load_duration_ms=schema_load_duration_ms, read_parse_duration_ms=round(max(phase_duration_ms - phase_stats.total_duration_ms, 0), 3), events_emitted=progress.events_emitted, events_skipped_lower_bound=progress.events_skipped_lower_bound, files_processed=progress.files_processed, **phase_stats.as_log_fields(), ) __all__ = ["read_increment_metadata", "sync_incremental"]