from __future__ import annotations from collections.abc import Iterator from typing import Any from conftest import ( table_definition, write_basic_snapshot, write_incremental_schema, write_json, write_snapshot_schema_sql, write_text, ) from fivetran_connector_sdk.constants import DELETE_TYPE, UPSERT_TYPE from config import ConnectorConfig from errors import IncrementalError from fivetran.operations import RecordingOperationSink from fivetran.sdk_records import ( OperationRecordPlan, RawCSVRecordWriter, build_raw_csv_record_writer, build_record_stream, ) from increment.reader import sync_incremental from increment.schema_index import load_storage_schema_index from state import normalize_state from storage.object_store import LocalObjectStore from tests.pytest_compat import pytest class CountingLocalObjectStore(LocalObjectStore): def __init__(self, root: str) -> None: super().__init__(root) self.iter_key_prefixes: list[str] = [] self.list_key_prefixes: list[str] = [] self.common_prefixes: list[str] = [] def iter_keys(self, prefix: str) -> Iterator[str]: self.iter_key_prefixes.append(prefix) return super().iter_keys(prefix) def list_keys(self, prefix: str) -> list[str]: self.list_key_prefixes.append(prefix) return super().list_keys(prefix) def list_common_prefixes(self, prefix: str) -> list[str]: self.common_prefixes.append(prefix) return super().list_common_prefixes(prefix) class RawRecordingOperationSink(RecordingOperationSink): def __init__(self) -> None: super().__init__() self.records: list[Any] = [] self._record_stream = build_record_stream(add_record=self.records.append, add_records=self.records.extend) def raw_csv_record_writer(self, **kwargs: Any) -> RawCSVRecordWriter: return build_raw_csv_record_writer( plan=OperationRecordPlan(**kwargs), record_stream=self._record_stream, ) def _flush_record_stream(self) -> None: self._record_stream.flush() def upsert(self, *, table: str, data: dict[str, Any]) -> None: self._flush_record_stream() super().upsert(table=table, data=data) def delete(self, *, table: str, keys: dict[str, Any]) -> None: self._flush_record_stream() super().delete(table=table, keys=keys) def truncate(self, *, table: str) -> None: self._flush_record_stream() super().truncate(table=table) def checkpoint(self, *, state) -> None: self._flush_record_stream() super().checkpoint(state=state) def _load_context(workspace, *, diagnostic: bool = True, operation_mode: str = "standard"): write_basic_snapshot(workspace, pos=100) config = ConnectorConfig.from_dict( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "enable_diagnostic_columns": diagnostic, "operation_mode": operation_mode, } ) store = LocalObjectStore(str(workspace)) schema_index = load_storage_schema_index(store, config.incremental_prefix) state = normalize_state({}, sync_task_id="task-001") state.snapshot["snapshot_tso"] = 100 state.snapshot["done"] = True return config, store, schema_index, state def test_incremental_consumes_metadata_index_and_data_file(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000002.csv\n") write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,Alice\n") write_text( workspace / "increment/test/worker/120/2026-05-25/CDC000002.csv", "U,worker,test,122,1,Alicia\nD,worker,test,123,1,Alicia\n", ) config, store, schema_index, state = _load_context(workspace) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call[0] for call in operations.calls if call[0] != "checkpoint"] == [ "upsert", "upsert", "delete", ] assert next(call for call in operations.calls if call[0] == "upsert") == ( "upsert", { "table": "test_worker", "data": { "id": 1, "name": "Alice", "_incremental_commit_ts": 121, "_incremental_source_schema": "test", "_incremental_source_table": "worker", }, }, ) final_state = state.to_dict() assert final_state["incremental"]["checkpoint_ts"] == "200" assert "scan" not in final_state["incremental"] def test_incremental_filters_events_outside_scan_range(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 122}) write_incremental_schema(workspace, table_definition(version=80)) write_text(workspace / "increment/test/worker/80/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text( workspace / "increment/test/worker/80/2026-05-25/CDC000001.csv", "I,worker,test,100,1,Old\nI,worker,test,121,1,Alice\nU,worker,test,123,1,Future\n", ) config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "upsert"] == [ ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alice"}}) ] final_state = state.to_dict() assert final_state["incremental"]["checkpoint_ts"] == "122" assert "scan" not in final_state["incremental"] def test_incremental_preserves_file_event_order_without_commit_ts_sort(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text( workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,122,1,Alice\nU,worker,test,121,1,Alicia\n", ) config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "upsert"] == [ ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alice"}}), ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alicia"}}), ] def test_incremental_consumes_same_table_dml_across_physical_paths_in_file_order(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/0/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/0/2026-05-25/CDC000001.csv", "U,worker,test,130,1,Late\n") write_text(workspace / "increment/test/worker/120/1/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/1/2026-05-25/CDC000001.csv", "U,worker,test,120,1,Early\n") config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "upsert"] == [ ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Late"}}), ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Early"}}), ] def test_incremental_future_events_do_not_stop_scan_before_later_files(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/0/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/0/2026-05-25/CDC000001.csv", "U,worker,test,250,1,Future\n") write_text(workspace / "increment/test/worker/120/1/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/1/2026-05-25/CDC000001.csv", "U,worker,test,150,1,Ready\n") config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "upsert"] == [ ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Ready"}}), ] final_state = state.to_dict() assert final_state["incremental"]["checkpoint_ts"] == "200" assert "scan" not in final_state["incremental"] def test_incremental_rejects_dispatcher_index_files(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text( workspace / "increment/test/worker/120/2026-05-25/meta/CDC_dispatcher.index", "CDC_dispatcher_000001.csv\n", ) config, store, schema_index, state = _load_context(workspace) with pytest.raises(IncrementalError, match="table-across-nodes"): sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=RecordingOperationSink(), ) def test_truncate_schema_marker_emits_truncate_after_snapshot_baseline(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=120, ddl_type=11, query="truncate table `test`.`worker`"), ) config, store, schema_index, state = _load_context(workspace) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert ("truncate", {"table": "test_worker"}) in operations.calls assert state.incremental["ddl_watermark"]["`test`.`worker`"] == 120 def test_truncate_schema_marker_checkpoints_before_following_dml(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=120, ddl_type=11, query="truncate table `test`.`worker`"), ) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,AfterTruncate\n") config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) truncate_index = operations.calls.index(("truncate", {"table": "test_worker"})) upsert_index = operations.calls.index( ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "AfterTruncate"}}) ) checkpoint_between = [ call for call in operations.calls[truncate_index + 1 : upsert_index] if call[0] == "checkpoint" ] assert checkpoint_between assert checkpoint_between[0][1]["state"]["incremental"]["ddl_watermark"]["`test`.`worker`"] == 120 def test_truncate_schema_marker_retry_after_checkpoint_does_not_retruncate(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=120, ddl_type=11, query="truncate table `test`.`worker`"), ) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,AfterTruncate\n") config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) truncate_index = operations.calls.index(("truncate", {"table": "test_worker"})) checkpoint_after_truncate = next( call[1]["state"] for call in operations.calls[truncate_index + 1 :] if call[0] == "checkpoint" ) retry_state = normalize_state(checkpoint_after_truncate, sync_task_id="task-001") retry_operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=retry_state, operations=retry_operations, ) assert ("truncate", {"table": "test_worker"}) not in retry_operations.calls assert ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "AfterTruncate"}}) in retry_operations.calls def test_schema_marker_at_snapshot_baseline_does_not_emit_truncate(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=80, ddl_type=11, query="truncate table `test`.`worker`"), ) config, store, schema_index, state = _load_context(workspace) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert ("truncate", {"table": "test_worker"}) not in operations.calls def test_drop_table_schema_marker_soft_deletes_with_truncate_and_checkpoints(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=120, ddl_type=10, query="drop table `test`.`worker`"), ) config, store, schema_index, state = _load_context(workspace) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) truncate_index = operations.calls.index(("truncate", {"table": "test_worker"})) checkpoint_after_truncate = next( call[1]["state"] for call in operations.calls[truncate_index + 1 :] if call[0] == "checkpoint" ) assert checkpoint_after_truncate["incremental"]["ddl_watermark"]["`test`.`worker`"] == 120 def test_schema_update_marker_advances_watermark_without_downstream_ddl(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema( workspace, table_definition(version=120, ddl_type=3, query="alter table `test`.`worker` add column age int"), ) config, store, schema_index, state = _load_context(workspace) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert ("truncate", {"table": "test_worker"}) not in operations.calls assert [call for call in operations.calls if call[0] in {"upsert", "update", "delete"}] == [] assert state.incremental["ddl_watermark"]["`test`.`worker`"] == 120 def test_incremental_wraps_missing_dml_file(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") config, store, schema_index, state = _load_context(workspace) with pytest.raises(IncrementalError, match="DML file referenced by index does not exist"): sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=RecordingOperationSink(), ) def test_incremental_scan_lists_incremental_prefix_once(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,Alice\n") write_basic_snapshot(workspace) config = ConnectorConfig.from_dict( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "operation_mode": "standard", } ) store = CountingLocalObjectStore(str(workspace)) schema_index = load_storage_schema_index(store, config.incremental_prefix) store.iter_key_prefixes.clear() store.list_key_prefixes.clear() store.common_prefixes.clear() state = normalize_state({}, sync_task_id="task-001") state.snapshot["snapshot_tso"] = 100 state.snapshot["done"] = True sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=RecordingOperationSink(), ) assert store.iter_key_prefixes == ["increment"] assert store.list_key_prefixes == [] assert store.common_prefixes == [] def test_incremental_fast_path_skips_listing_when_checkpoint_unchanged(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_basic_snapshot(workspace) config = ConnectorConfig.from_dict( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "operation_mode": "standard", } ) store = CountingLocalObjectStore(str(workspace)) schema_index = load_storage_schema_index(store, config.incremental_prefix) store.iter_key_prefixes.clear() store.list_key_prefixes.clear() store.common_prefixes.clear() state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": { "checkpoint_ts": "200", "ddl_watermark": {"`test`.`worker`": 120}, }, }, sync_task_id="task-001", ) sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=RecordingOperationSink(), ) assert store.iter_key_prefixes == [] assert store.list_key_prefixes == [] assert store.common_prefixes == [] def test_incremental_uses_global_snapshot_tso_for_new_table(workspace): write_basic_snapshot(workspace, pos=100) write_snapshot_schema_sql(workspace, table="new_worker") write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(table="new_worker", version=120)) write_text( workspace / "increment/test/new_worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n", ) write_text( workspace / "increment/test/new_worker/120/2026-05-25/CDC000001.csv", "I,new_worker,test,121,1,TooEarly\nU,new_worker,test,151,1,Ready\n", ) config = ConnectorConfig.from_dict( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "enable_diagnostic_columns": False, "operation_mode": "standard", } ) store = LocalObjectStore(str(workspace)) schema_index = load_storage_schema_index(store, config.incremental_prefix) state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": { "done": True, "snapshot_tso": "100", }, "incremental": { "checkpoint_ts": "0", "ddl_watermark": {}, }, }, sync_task_id="task-001", ) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "upsert"] == [ ("upsert", {"table": "test_new_worker", "data": {"id": 1, "name": "TooEarly"}}), ("upsert", {"table": "test_new_worker", "data": {"id": 1, "name": "Ready"}}), ] def test_incremental_preserves_same_commit_ts_storage_order(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text( workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "D,worker,test,150,1,Alice\nI,worker,test,150,2,Alice\nU,worker,test,150,2,Alicia\n", ) config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call[0] for call in operations.calls if call[0] != "checkpoint"] == [ "delete", "upsert", "upsert", ] def test_incremental_batched_stream_writes_raw_sdk_records_in_event_order(workspace): write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, table_definition(version=120)) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text( workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "D,worker,test,150,1,Alice\nI,worker,test,150,2,Alice\nU,worker,test,150,2,Alicia\n", ) config, store, schema_index, state = _load_context(workspace, operation_mode="batched_stream") operations = RawRecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [record.type for record in operations.records] == [DELETE_TYPE, UPSERT_TYPE, UPSERT_TYPE] assert [record.data["id"].long for record in operations.records] == [1, 2, 2] assert "name" not in operations.records[0].data assert "_incremental_commit_ts" not in operations.records[0].data assert operations.records[1].data["name"].string == "Alice" assert operations.records[1].data["_incremental_commit_ts"].long == 150 assert operations.records[1].data["_incremental_source_schema"].string == "test" assert operations.records[1].data["_incremental_source_table"].string == "worker" assert operations.records[2].data["name"].string == "Alicia" def test_incremental_skips_table_when_multi_bit_column_is_part_of_primary_key(workspace): raw = table_definition(version=120) raw["TableColumns"].append( { "ColumnName": "tenant_bits", "ColumnType": "BIT UNSIGNED", "ColumnPrecision": "8", "ColumnIsPk": "true", } ) raw["TableColumnsTotal"] = 3 write_json(workspace / "increment/metadata", {"checkpoint-ts": 200}) write_incremental_schema(workspace, raw) write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n") write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,Alice,7\n") config, store, schema_index, state = _load_context(workspace, diagnostic=False) operations = RecordingOperationSink() sync_incremental( config=config, store=store, schema_index=schema_index, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] in {"upsert", "update", "delete"}] == []