from __future__ import annotations import io from config import ConnectorConfig from increment.csv import iter_events from increment.paths import SchemaPath from increment.schema import compute_table_definition_checksum, load_table_definition def _config(*, output_old_value: bool, binary_encoding: str = "base64") -> ConnectorConfig: return ConnectorConfig.from_dict( { "sync_task_id": "task-001", "s3_url": "file:///tmp/source-sync-csv-test", "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "incremental": {"csv_output_old_value": output_old_value, "csv_binary_encoding": binary_encoding}, } ) def _table_definition(): raw = { "Table": "t", "Schema": "abc", "Version": 0, "TableVersion": 120, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "a", "ColumnType": "INT", "ColumnIsPk": "true"}, {"ColumnName": "b", "ColumnType": "INT"}, ], "TableColumnsTotal": 2, } return load_table_definition( raw, SchemaPath( key="increment/abc/t/meta/schema_120.json", source_schema="abc", source_table="t", table_version=120, checksum=compute_table_definition_checksum(raw), ), ) def test_csv_output_old_value_decodes_storage_events() -> None: table_def = _table_definition() events = list( iter_events( io.StringIO( "\n".join( [ '"I","t","abc",466765397536276493,3,3', '"D","t","abc",466765401442222092,3,3', '"I","t","abc",466765401442222092,5,3', '"U","t","abc",466765401442222093,5,4', ] ) + "\n" ), table_def=table_def, config=_config(output_old_value=True), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ("I", 466765397536276493, {"a": 3, "b": 3}), ("D", 466765401442222092, {"a": 3, "b": 3}), ("I", 466765401442222092, {"a": 5, "b": 3}), ("U", 466765401442222093, {"a": 5, "b": 4}), ] def test_csv_defaults_to_events_without_commit_ts() -> None: table_def = _table_definition() events = list( iter_events( io.StringIO( "\n".join( [ '"I","t","abc",3,3', '"D","t","abc",3,3', '"I","t","abc",5,3', '"U","t","abc",5,4', ] ) + "\n" ), table_def=table_def, config=_config(output_old_value=True), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ("I", None, {"a": 3, "b": 3}), ("D", None, {"a": 3, "b": 3}), ("I", None, {"a": 5, "b": 3}), ("U", None, {"a": 5, "b": 4}), ] def test_csv_maps_bit_one_and_skips_multi_bit_columns() -> None: raw = { "Table": "t", "Schema": "abc", "Version": 0, "TableVersion": 120, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "a", "ColumnType": "INT", "ColumnIsPk": "true"}, {"ColumnName": "flag", "ColumnType": "BIT UNSIGNED", "ColumnPrecision": "1"}, {"ColumnName": "bitmap", "ColumnType": "BIT UNSIGNED", "ColumnPrecision": "8"}, ], "TableColumnsTotal": 3, } table_def = load_table_definition( raw, SchemaPath( key="increment/abc/t/meta/schema_120.json", source_schema="abc", source_table="t", table_version=120, checksum=compute_table_definition_checksum(raw), ), ) events = list( iter_events( io.StringIO('"I","t","abc",121,3,1,7\n'), table_def=table_def, config=_config(output_old_value=True), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ("I", 121, {"a": 3, "flag": True}), ] def test_csv_decodes_timestamp_decimal_and_vector_values() -> None: raw = { "Table": "t", "Schema": "abc", "Version": 0, "TableVersion": 120, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "a", "ColumnType": "INT", "ColumnIsPk": "true"}, {"ColumnName": "amount", "ColumnType": "DECIMAL", "ColumnPrecision": "10", "ColumnScale": "5"}, {"ColumnName": "created_at", "ColumnType": "TIMESTAMP"}, {"ColumnName": "embedding", "ColumnType": "VECTOR"}, ], "TableColumnsTotal": 4, } table_def = load_table_definition( raw, SchemaPath( key="increment/abc/t/meta/schema_120.json", source_schema="abc", source_table="t", table_version=120, checksum=compute_table_definition_checksum(raw), ), ) events = list( iter_events( io.StringIO('"I","t","abc",121,1,"13.14520","2023-11-16 04:34:56.12346","[0.3,0.5,-0.1]"\n'), table_def=table_def, config=_config(output_old_value=True), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ( "I", 121, { "a": 1, "amount": "13.14520", "created_at": "2023-11-16T04:34:56.12346Z", "embedding": "[0.3,0.5,-0.1]", }, ), ] def test_csv_uses_epoch_for_invalid_timestamp_primary_key() -> None: raw = { "Table": "t", "Schema": "abc", "Version": 0, "TableVersion": 120, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "created_at", "ColumnType": "TIMESTAMP", "ColumnIsPk": "true"}, {"ColumnName": "name", "ColumnType": "VARCHAR"}, ], "TableColumnsTotal": 2, } table_def = load_table_definition( raw, SchemaPath( key="increment/abc/t/meta/schema_120.json", source_schema="abc", source_table="t", table_version=120, checksum=compute_table_definition_checksum(raw), ), ) events = list( iter_events( io.StringIO('"I","t","abc",121,"0000-00-00 00:00:00","Alice"\n'), table_def=table_def, config=_config(output_old_value=True), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ("I", 121, {"created_at": "1970-01-01T00:00:00Z", "name": "Alice"}), ] def test_csv_decodes_binary_with_incremental_csv_binary_encoding() -> None: raw = { "Table": "t", "Schema": "abc", "Version": 0, "TableVersion": 120, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "a", "ColumnType": "INT", "ColumnIsPk": "true"}, {"ColumnName": "payload", "ColumnType": "VARBINARY"}, ], "TableColumnsTotal": 2, } table_def = load_table_definition( raw, SchemaPath( key="increment/abc/t/meta/schema_120.json", source_schema="abc", source_table="t", table_version=120, checksum=compute_table_definition_checksum(raw), ), ) events = list( iter_events( io.StringIO('"I","t","abc",121,1,"68656c6c6f"\n'), table_def=table_def, config=_config(output_old_value=True, binary_encoding="hex"), ) ) assert [(event.operation, event.commit_ts, event.data) for event in events] == [ ("I", 121, {"a": 1, "payload": b"hello"}), ]