from __future__ import annotations from conftest import write_basic_snapshot, write_snapshot_metadata, write_text from config import ConnectorConfig from errors import SnapshotError from fivetran.operations import RecordingOperationSink from snapshot.reader import sync_snapshot from state import apply_table_resync_requests, normalize_state from storage.object_store import LocalObjectStore from tests.pytest_compat import pytest def _standard_config(configuration: dict[str, object]) -> ConnectorConfig: return ConnectorConfig.from_dict({**configuration, "operation_mode": "standard"}) def test_snapshot_upserts_rows_and_checkpoints(workspace): write_basic_snapshot(workspace, pos=100) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,name\n1,Alice\n2,Bob\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "checkpoint_interval_rows": "1", "enable_diagnostic_columns": True, } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call for call in operations.calls if call[0] == "upsert"] assert upserts == [ ( "upsert", { "table": "test_worker", "data": { "id": 1, "name": "Alice", "_incremental_commit_ts": 0, "_incremental_source_schema": "test", "_incremental_source_table": "worker", }, }, ), ( "upsert", { "table": "test_worker", "data": { "id": 2, "name": "Bob", "_incremental_commit_ts": 0, "_incremental_source_schema": "test", "_incremental_source_table": "worker", }, }, ), ] assert state.snapshot["snapshot_tso"] == 100 assert state.snapshot["done"] is True assert state.checkpoint_ts == 100 assert any(call[0] == "checkpoint" for call in operations.calls) checkpoint_states = [call[1]["state"] for call in operations.calls if call[0] == "checkpoint"] assert any( checkpoint_state["snapshot"].get("current_file") == "snapshot/test.worker.000000001.csv" and checkpoint_state["snapshot"].get("next_row") == 2 for checkpoint_state in checkpoint_states ) assert not any( checkpoint_state["snapshot"].get("current_file") == "snapshot/test.worker.000000001.csv" and checkpoint_state["snapshot"].get("next_row") == 1 for checkpoint_state in checkpoint_states ) assert state.to_dict()["snapshot"] == {"done": True, "snapshot_tso": "100"} def test_snapshot_table_without_data_files_completes_as_empty_table(workspace): write_basic_snapshot(workspace, pos=100) config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "truncate_before_snapshot": True, } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "truncate"] == [("truncate", {"table": "test_worker"})] assert [call for call in operations.calls if call[0] == "upsert"] == [] assert state.snapshot["snapshot_tso"] == 100 assert state.snapshot["done"] is True assert "tables" not in state.snapshot assert state.checkpoint_ts == 100 assert state.to_dict()["snapshot"] == {"done": True, "snapshot_tso": "100"} def test_table_resync_table_without_data_files_completes_as_empty_table(workspace): write_basic_snapshot(workspace, pos=100) config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`test`.`worker`"], "snapshot_prefix": "snapshot", "snapshot_tso": "100", "truncate_done": False, }, }, sync_task_id="task-001", ) apply_table_resync_requests(state) operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) assert [call for call in operations.calls if call[0] == "truncate"] == [("truncate", {"table": "test_worker"})] assert [call for call in operations.calls if call[0] == "upsert"] == [] assert "tables" not in state.snapshot assert state.resync is None def test_snapshot_files_use_natural_sort_and_schema_order_header(workspace): write_basic_snapshot(workspace) write_text(workspace / "snapshot/test.worker.10.csv", "id,name\n10,Later\n") write_text(workspace / "snapshot/test.worker.2.csv", "id,name\n2,Earlier\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 2, "name": "Earlier"}, {"id": 10, "name": "Later"}] def test_snapshot_header_order_must_match_sql_schema_order(workspace): write_basic_snapshot(workspace) write_text(workspace / "snapshot/test.worker.000000001.csv", "name,id\nAlice,1\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") with pytest.raises(SnapshotError, match="header order"): sync_snapshot( config=config, store=store, state=state, operations=RecordingOperationSink(), ) def test_snapshot_file_without_header_uses_schema_column_order(workspace): write_basic_snapshot(workspace) write_text(workspace / "snapshot/test.worker.000000001.csv", "1,Alice\n2,Bob\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}] def test_snapshot_resume_from_completed_file_cursor_continues_with_next_file(workspace): write_basic_snapshot(workspace, pos=100) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,name\n1,Alice\n2,Bob\n") write_text(workspace / "snapshot/test.worker.000000002.csv", "id,name\n3,Carol\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": { "done": False, "snapshot_tso": "100", "truncate_done": False, "current_file": "snapshot/test.worker.000000001.csv", "next_row": 2, }, "incremental": {"checkpoint_ts": "0", "ddl_watermark": {}}, }, sync_task_id="task-001", ) operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 3, "name": "Carol"}] def test_snapshot_maps_bit_one_and_skips_multi_bit_columns(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `flag` bit(1) DEFAULT NULL, `bitmap` bit(8) DEFAULT NULL, `name` varchar(64) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,flag,bitmap,name\n1,1,7,Alice\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 1, "flag": True, "name": "Alice"}] def test_snapshot_decodes_timestamp_decimal_and_vector_values(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `amount` decimal(10,5) DEFAULT NULL, `created_at` timestamp(5) NULL DEFAULT NULL, `embedding` vector(3) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text( workspace / "snapshot/test.worker.000000001.csv", 'id,amount,created_at,embedding\n1,13.14520,2023-11-16 04:34:56.12346,"[0.3,0.5,-0.1]"\n', ) config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [ { "id": 1, "amount": "13.14520", "created_at": "2023-11-16T04:34:56.12346Z", "embedding": "[0.3,0.5,-0.1]", } ] def test_snapshot_decodes_dumpling_backslash_escaped_csv_values(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `payload` json DEFAULT NULL, `path` varchar(128) DEFAULT NULL, `note` text DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text( workspace / "snapshot/test.worker.000000001.csv", 'id,payload,path,note\n1,"{\\"type\\": \\"signup\\"}","C:\\\\tmp\\\\file","line\\nbreak"\n2,\\N,\\N,\\N\n', ) config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "snapshot": {"escape_backslash": True}, } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [ { "id": 1, "payload": {"type": "signup"}, "path": r"C:\tmp\file", "note": "line\nbreak", }, { "id": 2, "payload": None, "path": None, "note": None, }, ] def test_snapshot_decodes_binary_with_default_dumpling_utf8_output(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `payload` varbinary(64) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,payload\n1,hello\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 1, "payload": b"hello"}] def test_snapshot_decodes_binary_with_hex_output_dialect(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `payload` blob DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,payload\n1,68656c6c6f\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "snapshot": {"csv_binary_output_dialect": "hex"}, } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call[1]["data"] for call in operations.calls if call[0] == "upsert"] assert upserts == [{"id": 1, "payload": b"hello"}] def test_snapshot_skips_table_when_multi_bit_column_is_part_of_primary_key(workspace): write_snapshot_metadata(workspace) write_text( workspace / "snapshot/test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `tenant_bits` bit(8) NOT NULL, `name` varchar(64) DEFAULT NULL, PRIMARY KEY (`id`, `tenant_bits`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text(workspace / "snapshot/test.worker.000000001.csv", "id,tenant_bits,name\n1,7,Alice\n") write_text( workspace / "snapshot/test.employee-schema.sql", """CREATE TABLE `employee` ( `id` bigint NOT NULL, `name` varchar(64) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) write_text(workspace / "snapshot/test.employee.000000001.csv", "id,name\n2,Bob\n") config = _standard_config( { "sync_task_id": "task-001", "s3_url": f"file://{workspace}", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", } ) store = LocalObjectStore(str(workspace)) state = normalize_state({}, sync_task_id="task-001") operations = RecordingOperationSink() sync_snapshot( config=config, store=store, state=state, operations=operations, ) upserts = [call for call in operations.calls if call[0] == "upsert"] assert upserts == [ ("upsert", {"table": "test_employee", "data": {"id": 2, "name": "Bob"}}), ]