from __future__ import annotations

import csv
import json
import re
from collections.abc import ItemsView
from dataclasses import dataclass, field
from datetime import date, datetime, time
from decimal import Decimal
from pathlib import Path
from typing import Any

import sqlglot
from sqlglot import exp
from sqlglot.errors import ParseError

from increment.schema import compute_table_definition_checksum
from metadata import TableName
from snapshot.schema import SnapshotTable, parse_create_table_sql, quote_object_path_segment
from type_mapping import coerce_value

SNAPSHOT_TSO = 100
NULL_MARKER = r"\N"
DEFAULT_SCHEMA = "test"
MARKER_RE = re.compile(r"fivetran\s*:\s*(?:phase\s*=\s*)?(snapshot|phase1|incremental-1|phase2|incremental-2)\b", re.I)

type RowKey = tuple[str | None, ...]
type RowPayload = dict[str, str | None]


@dataclass
class SnapshotTableRows:
    by_key: dict[RowKey, RowPayload] = field(default_factory=dict)

    def __bool__(self) -> bool:
        return bool(self.by_key)

    def items(self) -> ItemsView[RowKey, RowPayload]:
        return self.by_key.items()

    def row(self, key: RowKey) -> RowPayload:
        return self.by_key[key]

    def upsert(self, key: RowKey, row: RowPayload) -> None:
        self.by_key[key] = dict(row)

    def pop(self, key: RowKey) -> RowPayload:
        return self.by_key.pop(key)

    def clone(self) -> SnapshotTableRows:
        return SnapshotTableRows({key: dict(row) for key, row in self.by_key.items()})


@dataclass(frozen=True)
class ReplayEntry:
    row: RowPayload
    commit_ts: int

    def clone(self) -> ReplayEntry:
        return ReplayEntry(dict(self.row), self.commit_ts)


@dataclass
class ReplayTableRows:
    by_key: dict[RowKey, ReplayEntry] = field(default_factory=dict)

    @classmethod
    def from_snapshot_table_rows(cls, rows: SnapshotTableRows) -> ReplayTableRows:
        return cls({key: ReplayEntry(dict(row), 0) for key, row in rows.items()})

    def items(self) -> ItemsView[RowKey, ReplayEntry]:
        return self.by_key.items()

    def clone(self) -> ReplayTableRows:
        return ReplayTableRows({key: entry.clone() for key, entry in self.by_key.items()})

    def delete(self, key: RowKey) -> None:
        self.by_key.pop(key, None)

    def upsert(self, key: RowKey, row: RowPayload, commit_ts: int) -> None:
        self.by_key[key] = ReplayEntry(dict(row), commit_ts)


@dataclass
class SnapshotRows:
    by_table_name: dict[TableName, SnapshotTableRows] = field(default_factory=dict)

    def table_rows(self, table_name: TableName) -> SnapshotTableRows:
        return self.by_table_name.get(table_name, SnapshotTableRows())


@dataclass
class ReplayState:
    by_table_name: dict[TableName, ReplayTableRows] = field(default_factory=dict)

    @classmethod
    def from_snapshot_rows(cls, rows: SnapshotRows) -> ReplayState:
        return cls(
            {
                table_name: ReplayTableRows.from_snapshot_table_rows(table_rows)
                for table_name, table_rows in rows.by_table_name.items()
            }
        )

    def clone(self) -> ReplayState:
        return ReplayState({table_name: table_rows.clone() for table_name, table_rows in self.by_table_name.items()})

    def table_rows(self, table_name: TableName) -> ReplayTableRows:
        return self.by_table_name.get(table_name, ReplayTableRows())

    def delete(self, table_name: TableName, key: RowKey) -> None:
        self.by_table_name.setdefault(table_name, ReplayTableRows()).delete(key)

    def upsert(self, table_name: TableName, key: RowKey, row: RowPayload, commit_ts: int) -> None:
        self.by_table_name.setdefault(table_name, ReplayTableRows()).upsert(key, row, commit_ts)


@dataclass(frozen=True)
class SqlStatement:
    sql: str
    phase: str


@dataclass
class TableRuntime:
    snapshot_table: SnapshotTable
    create_sql: str
    rows: SnapshotTableRows = field(default_factory=dict)


@dataclass(frozen=True)
class DmlEvent:
    operation: str
    table_name: TableName
    row: RowPayload


@dataclass(frozen=True)
class PreparedSqlFixture:
    has_phase2: bool
    expected_path: Path


def prepare_from_sql(
    *,
    input_sql: Path,
    workspace: Path,
    expected_path: Path,
    include_phase2_increment: bool = False,
) -> PreparedSqlFixture:
    sql_text = input_sql.read_text(encoding="utf-8")
    statements = _split_sql_statements(sql_text)
    if not statements:
        raise SystemExit(f"input SQL has no statements: {input_sql}")

    tables: dict[TableName, TableRuntime] = {}
    snapshot_rows: SnapshotRows | None = None
    phase1_events: list[DmlEvent] = []
    phase2_events: list[DmlEvent] = []
    default_schema = DEFAULT_SCHEMA
    auto_snapshot_open = True

    def freeze_snapshot() -> None:
        nonlocal snapshot_rows
        if snapshot_rows is None:
            snapshot_rows = _copy_table_rows(tables)

    for statement in statements:
        try:
            parsed = sqlglot.parse_one(statement.sql, read="mysql")
        except ParseError as exc:
            if _is_ignorable_statement(statement.sql):
                continue
            raise SystemExit(f"unsupported SQL statement: {statement.sql[:200]}\n{exc}") from exc
        if parsed is None or _is_ignorable_expression(parsed):
            continue

        if isinstance(parsed, exp.Use):
            default_schema = _use_schema(parsed) or default_schema
            continue
        if isinstance(parsed, exp.Create):
            if _create_kind(parsed) != "TABLE":
                continue
            if snapshot_rows is not None:
                raise SystemExit("CREATE TABLE after incremental data is not supported by local SQL fixture mode")
            table = _create_table(parsed, default_schema)
            tables[table.name] = TableRuntime(snapshot_table=table, create_sql=parsed.sql(dialect="mysql"))
            continue
        if isinstance(parsed, exp.Insert):
            phase = statement.phase
            if phase == "auto":
                phase = "snapshot" if auto_snapshot_open else "phase1"
            events = _apply_insert(parsed, tables, default_schema)
            if phase == "snapshot":
                continue
            freeze_snapshot()
            auto_snapshot_open = False
            _append_events(phase, events, phase1_events, phase2_events)
            continue
        if isinstance(parsed, exp.Update):
            phase = statement.phase
            if phase == "auto":
                phase = "phase1"
            if phase != "snapshot":
                freeze_snapshot()
                auto_snapshot_open = False
            events = _apply_update(parsed, tables, default_schema)
            if phase != "snapshot":
                _append_events(phase, events, phase1_events, phase2_events)
            continue
        if isinstance(parsed, exp.Delete):
            phase = statement.phase
            if phase == "auto":
                phase = "phase1"
            if phase != "snapshot":
                freeze_snapshot()
                auto_snapshot_open = False
            events = _apply_delete(parsed, tables, default_schema)
            if phase != "snapshot":
                _append_events(phase, events, phase1_events, phase2_events)
            continue

        if not _is_ignorable_expression(parsed):
            raise SystemExit(f"unsupported SQL statement type for fixture generation: {type(parsed).__name__}")

    if not tables:
        raise SystemExit("input SQL did not define any CREATE TABLE statements")
    if snapshot_rows is None:
        snapshot_rows = _copy_table_rows(tables)

    phase1_checkpoint = max(200, SNAPSHOT_TSO + len(phase1_events) + 10)
    phase2_checkpoint = max(300, phase1_checkpoint + len(phase2_events) + 10)
    _write_snapshot(workspace, tables, snapshot_rows)
    _write_increment(
        workspace=workspace,
        tables=tables,
        events=phase1_events,
        file_index=1,
        start_commit_ts=SNAPSHOT_TSO + 1,
        checkpoint_ts=phase1_checkpoint,
    )

    phase1_state = _replay_expected_from_snapshot(tables, snapshot_rows, phase1_events, SNAPSHOT_TSO + 1)
    expected: dict[str, Any] = {
        "phase1": {
            "checkpoint_ts": str(phase1_checkpoint),
            "tables": _expected_tables(tables, phase1_state),
        }
    }
    if phase2_events:
        if include_phase2_increment:
            _write_increment(
                workspace=workspace,
                tables=tables,
                events=phase2_events,
                file_index=2,
                start_commit_ts=phase1_checkpoint + 1,
                checkpoint_ts=phase2_checkpoint,
            )
        phase2_state = _replay_expected_from_state(
            tables,
            phase1_state,
            phase2_events,
            phase1_checkpoint + 1,
        )
        expected["phase2"] = {
            "checkpoint_ts": str(phase2_checkpoint),
            "tables": _expected_tables(tables, phase2_state),
        }

    expected_path.parent.mkdir(parents=True, exist_ok=True)
    expected_path.write_text(json.dumps(expected, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return PreparedSqlFixture(has_phase2=bool(phase2_events), expected_path=expected_path)


def add_phase2_from_sql(*, input_sql: Path, workspace: Path, expected_path: Path) -> bool:
    # Regenerate from scratch into memory, then write only phase2 incremental files.
    temp_workspace = workspace.parent / f".{workspace.name}.tmp_sql_parse"
    if temp_workspace.exists():
        import shutil

        shutil.rmtree(temp_workspace)
    prepared = prepare_from_sql(
        input_sql=input_sql,
        workspace=temp_workspace,
        expected_path=expected_path,
        include_phase2_increment=True,
    )
    if not prepared.has_phase2:
        import shutil

        shutil.rmtree(temp_workspace, ignore_errors=True)
        return False

    source_increment = temp_workspace / "increment"
    target_increment = workspace / "increment"
    for path in source_increment.rglob("*"):
        if path.is_file():
            relative = path.relative_to(source_increment)
            if (
                "CDC000002.csv" in path.name
                or path.name == "CDC.index"
                or path.name == "metadata"
                or path.name.startswith("schema_")
            ):
                target = target_increment / relative
                target.parent.mkdir(parents=True, exist_ok=True)
                target.write_text(path.read_text(encoding="utf-8"), encoding="utf-8")
    import shutil

    shutil.rmtree(temp_workspace, ignore_errors=True)
    return True


def _split_sql_statements(sql_text: str) -> list[SqlStatement]:
    statements: list[SqlStatement] = []
    current_phase = "auto"
    buf: list[str] = []
    quote: str | None = None
    i = 0
    while i < len(sql_text):
        ch = sql_text[i]
        nxt = sql_text[i : i + 2]

        if quote is None and nxt == "--" and (i + 2 == len(sql_text) or sql_text[i + 2].isspace()):
            end = sql_text.find("\n", i)
            if end == -1:
                end = len(sql_text)
            current_phase = _phase_from_comment(sql_text[i:end], current_phase)
            buf.append("\n")
            i = end
            continue
        if quote is None and ch == "#":
            end = sql_text.find("\n", i)
            if end == -1:
                end = len(sql_text)
            current_phase = _phase_from_comment(sql_text[i:end], current_phase)
            buf.append("\n")
            i = end
            continue
        if quote is None and nxt == "/*":
            end = sql_text.find("*/", i + 2)
            if end == -1:
                end = len(sql_text) - 2
            current_phase = _phase_from_comment(sql_text[i : end + 2], current_phase)
            buf.append(" ")
            i = end + 2
            continue

        if quote is None and ch in {"'", '"', "`"}:
            quote = ch
            buf.append(ch)
            i += 1
            continue
        if quote is not None:
            buf.append(ch)
            if ch == "\\" and quote in {"'", '"'} and i + 1 < len(sql_text):
                i += 1
                buf.append(sql_text[i])
            elif ch == quote:
                quote = None
            i += 1
            continue

        if ch == ";":
            raw = "".join(buf).strip()
            if raw:
                statements.append(SqlStatement(raw, current_phase))
            buf.clear()
            i += 1
            continue
        buf.append(ch)
        i += 1

    raw = "".join(buf).strip()
    if raw:
        statements.append(SqlStatement(raw, current_phase))
    return statements


def _phase_from_comment(comment: str, current_phase: str) -> str:
    match = MARKER_RE.search(comment)
    if not match:
        return current_phase
    marker = match.group(1).lower()
    if marker in {"snapshot"}:
        return "snapshot"
    if marker in {"phase1", "incremental-1"}:
        return "phase1"
    if marker in {"phase2", "incremental-2"}:
        return "phase2"
    return current_phase


def _append_events(phase: str, events: list[DmlEvent], phase1: list[DmlEvent], phase2: list[DmlEvent]) -> None:
    if phase == "phase1":
        phase1.extend(events)
        return
    if phase == "phase2":
        phase2.extend(events)
        return
    raise SystemExit(f"unsupported fivetran SQL phase marker: {phase}")


def _create_kind(create: exp.Create) -> str:
    kind = create.args.get("kind")
    return str(kind or "").upper()


def _create_table(create: exp.Create, default_schema: str) -> SnapshotTable:
    schema = create.this
    if not isinstance(schema, exp.Schema) or not isinstance(schema.this, exp.Table):
        raise SystemExit(f"unsupported CREATE TABLE form: {create.sql(dialect='mysql')}")
    source_schema, source_table = _table_parts(schema.this, default_schema)
    stem = _snapshot_stem(source_schema, source_table)
    table = parse_create_table_sql(
        create.sql(dialect="mysql"),
        source_schema=source_schema,
        source_table=source_table,
        snapshot_file_stem=stem,
    )
    if not table.primary_key:
        raise SystemExit(f"table {table.quoted} has no primary key; Fivetran delete/upsert tests require one")
    return table


def _table_parts(table: exp.Table, default_schema: str) -> tuple[str, str]:
    name = table.name
    db = table.db
    if not name:
        raise SystemExit(f"SQL table name is empty: {table.sql(dialect='mysql')}")
    return str(db or default_schema), str(name)


def _use_schema(use: exp.Use) -> str | None:
    table = use.this
    if isinstance(table, exp.Table):
        return table.name
    return None


def _table_for_expression(
    table_expr: exp.Expression, tables: dict[TableName, TableRuntime], default_schema: str
) -> TableRuntime:
    if isinstance(table_expr, exp.Schema) and isinstance(table_expr.this, exp.Table):
        source_schema, source_table = _table_parts(table_expr.this, default_schema)
    elif isinstance(table_expr, exp.Table):
        source_schema, source_table = _table_parts(table_expr, default_schema)
    else:
        raise SystemExit(f"unsupported DML table expression: {table_expr.sql(dialect='mysql')}")
    table_name = TableName(source_schema, source_table)
    if table_name not in tables:
        raise SystemExit(f"DML references table without CREATE TABLE: {table_name.quoted}")
    return tables[table_name]


def _apply_insert(insert: exp.Insert, tables: dict[TableName, TableRuntime], default_schema: str) -> list[DmlEvent]:
    target = _table_for_expression(insert.this, tables, default_schema)
    insert_schema = insert.this
    if isinstance(insert_schema, exp.Schema) and insert_schema.expressions:
        insert_columns = [_identifier_name(column) for column in insert_schema.expressions]
    else:
        insert_columns = target.snapshot_table.column_names
    values = insert.args.get("expression")
    if not isinstance(values, exp.Values):
        raise SystemExit("INSERT ... SELECT is not supported by local SQL fixture mode")
    events: list[DmlEvent] = []
    for row_expr in values.expressions:
        if not isinstance(row_expr, exp.Tuple):
            raise SystemExit(f"unsupported INSERT row expression: {row_expr.sql(dialect='mysql')}")
        if len(row_expr.expressions) != len(insert_columns):
            raise SystemExit("INSERT row width does not match column list")
        row = {column: None for column in target.snapshot_table.column_names}
        for column, value_expr in zip(insert_columns, row_expr.expressions, strict=True):
            row[column] = _literal_raw(value_expr)
        key = _pk_key(target.snapshot_table, row)
        target.rows.upsert(key, row)
        events.append(DmlEvent("I", target.snapshot_table.name, dict(row)))
    return events


def _apply_update(update: exp.Update, tables: dict[TableName, TableRuntime], default_schema: str) -> list[DmlEvent]:
    target = _table_for_expression(update.this, tables, default_schema)
    assignments: dict[str, str | None] = {}
    for assignment in update.expressions:
        if not isinstance(assignment, exp.EQ) or not isinstance(assignment.this, exp.Column):
            raise SystemExit(f"unsupported UPDATE assignment: {assignment.sql(dialect='mysql')}")
        assignments[_column_name(assignment.this)] = _literal_raw(assignment.expression)
    matched = _matching_keys(target, update.args.get("where"))
    events: list[DmlEvent] = []
    for key in matched:
        row = dict(target.rows.row(key))
        row.update(assignments)
        new_key = _pk_key(target.snapshot_table, row)
        if new_key != key:
            target.rows.pop(key)
        target.rows.upsert(new_key, row)
        events.append(DmlEvent("U", target.snapshot_table.name, dict(row)))
    return events


def _apply_delete(delete: exp.Delete, tables: dict[TableName, TableRuntime], default_schema: str) -> list[DmlEvent]:
    target = _table_for_expression(delete.this, tables, default_schema)
    matched = _matching_keys(target, delete.args.get("where"))
    events: list[DmlEvent] = []
    for key in matched:
        row = target.rows.pop(key)
        events.append(DmlEvent("D", target.snapshot_table.name, dict(row)))
    return events


def _matching_keys(target: TableRuntime, where: exp.Expression | None) -> list[RowKey]:
    return [key for key, row in list(target.rows.items()) if _matches_where(row, where)]


def _matches_where(row: dict[str, str | None], where: exp.Expression | None) -> bool:
    if where is None:
        return True
    expression = where.this if isinstance(where, exp.Where) else where
    if isinstance(expression, exp.And):
        return _matches_where(row, expression.this) and _matches_where(row, expression.expression)
    if isinstance(expression, exp.Or):
        return _matches_where(row, expression.this) or _matches_where(row, expression.expression)
    if isinstance(expression, exp.EQ) and isinstance(expression.this, exp.Column):
        return row.get(_column_name(expression.this)) == _literal_raw(expression.expression)
    raise SystemExit(f"unsupported WHERE expression in local SQL fixture mode: {expression.sql(dialect='mysql')}")


def _identifier_name(identifier: exp.Expression) -> str:
    if isinstance(identifier, exp.Identifier):
        return identifier.name
    if isinstance(identifier, exp.Column):
        return _column_name(identifier)
    return identifier.name


def _column_name(column: exp.Column) -> str:
    return column.name


def _literal_raw(value: exp.Expression) -> str | None:
    if isinstance(value, exp.Null):
        return None
    if isinstance(value, exp.Boolean):
        return "1" if value.this else "0"
    if isinstance(value, exp.Neg):
        inner = _literal_raw(value.this)
        return f"-{inner}" if inner is not None else None
    if isinstance(value, exp.Literal):
        return str(value.this)
    raise SystemExit(f"unsupported SQL value expression: {value.sql(dialect='mysql')}")


def _pk_key(table: SnapshotTable, row: RowPayload) -> RowKey:
    key = tuple(row.get(column) for column in table.primary_key)
    if any(value is None for value in key):
        raise SystemExit(f"row for {table.quoted} is missing primary key values: {row}")
    return key


def _copy_table_rows(tables: dict[TableName, TableRuntime]) -> SnapshotRows:
    return SnapshotRows({table_name: table.rows.clone() for table_name, table in tables.items()})


def _write_snapshot(
    workspace: Path,
    tables: dict[TableName, TableRuntime],
    snapshot_rows: SnapshotRows,
) -> None:
    snapshot_dir = workspace / "snapshot"
    snapshot_dir.mkdir(parents=True, exist_ok=True)
    _write_text(
        snapshot_dir / "metadata",
        "\n".join(
            [
                "Started dump at: 2026-06-04 10:00:00",
                "SHOW MASTER STATUS:",
                "    Log: tidb-binlog",
                f"    Pos: {SNAPSHOT_TSO}",
                "    GTID:",
                "",
                "Finished dump at: 2026-06-04 10:00:10",
                "",
            ]
        ),
    )
    for table_name, runtime in sorted(tables.items(), key=lambda item: item[0].quoted):
        stem = runtime.snapshot_table.snapshot_file_stem
        _write_text(snapshot_dir / f"{stem}-schema.sql", runtime.create_sql + ";\n")
        rows = snapshot_rows.table_rows(table_name)
        if rows:
            _write_csv(
                snapshot_dir / f"{stem}.000000001.csv",
                runtime.snapshot_table.column_names,
                [row for _key, row in sorted(rows.items())],
            )


def _write_increment(
    *,
    workspace: Path,
    tables: dict[TableName, TableRuntime],
    events: list[DmlEvent],
    file_index: int,
    start_commit_ts: int,
    checkpoint_ts: int,
) -> None:
    increment_dir = workspace / "increment"
    _write_json(increment_dir / "metadata", {"checkpoint-ts": checkpoint_ts})
    for runtime in sorted(tables.values(), key=lambda item: item.snapshot_table.quoted):
        _write_schema_marker(increment_dir, runtime.snapshot_table)

    events_by_table: dict[TableName, list[tuple[int, DmlEvent]]] = {}
    for offset, event in enumerate(events):
        events_by_table.setdefault(event.table_name, []).append((start_commit_ts + offset, event))

    for table_name, runtime in sorted(tables.items(), key=lambda item: item[0].quoted):
        table_events = events_by_table.get(table_name, [])
        base = increment_dir / runtime.snapshot_table.path_key / "120" / "2026-06-04"
        _write_text(base / "meta" / "CDC.index", f"CDC{file_index:06d}.csv\n")
        rows = []
        for commit_ts, event in table_events:
            rows.append(
                [
                    event.operation,
                    runtime.snapshot_table.source_table,
                    runtime.snapshot_table.source_schema,
                    str(commit_ts),
                    *[_csv_value(event.row.get(column)) for column in runtime.snapshot_table.column_names],
                ]
            )
        _write_csv_rows(base / f"CDC{file_index:06d}.csv", rows)


def _write_schema_marker(increment_dir: Path, table: SnapshotTable) -> None:
    payload = {
        "Table": table.source_table,
        "Schema": table.source_schema,
        "Version": 0,
        "TableVersion": 120,
        "Query": "",
        "Type": 0,
        "TableColumns": [{"ColumnName": column.name, "ColumnType": column.source_type} for column in table.columns],
        "TableColumnsTotal": len(table.columns),
    }
    checksum = compute_table_definition_checksum(payload)
    _write_json(increment_dir / table.path_key / "meta" / f"schema_120_{checksum}.json", payload)


def _replay_expected_from_snapshot(
    tables: dict[TableName, TableRuntime],
    base_rows: SnapshotRows,
    events: list[DmlEvent],
    start_commit_ts: int,
) -> ReplayState:
    state = ReplayState.from_snapshot_rows(base_rows)
    return _apply_replay_events(tables, state, events, start_commit_ts)


def _replay_expected_from_state(
    tables: dict[TableName, TableRuntime],
    base_state: ReplayState,
    events: list[DmlEvent],
    start_commit_ts: int,
) -> ReplayState:
    state = base_state.clone()
    return _apply_replay_events(tables, state, events, start_commit_ts)


def _apply_replay_events(
    tables: dict[TableName, TableRuntime],
    state: ReplayState,
    events: list[DmlEvent],
    start_commit_ts: int,
) -> ReplayState:
    for offset, event in enumerate(events):
        runtime = tables[event.table_name]
        key = _pk_key(runtime.snapshot_table, event.row)
        if event.operation == "D":
            state.delete(event.table_name, key)
        else:
            state.upsert(event.table_name, key, event.row, start_commit_ts + offset)
    return state


def _expected_tables(
    tables: dict[TableName, TableRuntime],
    state: ReplayState,
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for table_name, runtime in sorted(tables.items(), key=lambda item: item[0].quoted):
        rows = []
        for _key, entry in sorted(state.table_rows(table_name).items()):
            payload = {
                column.name: _json_safe(
                    coerce_value(
                        entry.row.get(column.name),
                        column.source_type,
                        null_marker=NULL_MARKER,
                        binary_encoding="base64",
                    )
                )
                for column in runtime.snapshot_table.columns
            }
            payload["_incremental_commit_ts"] = entry.commit_ts
            payload["_incremental_source_schema"] = runtime.snapshot_table.source_schema
            payload["_incremental_source_table"] = runtime.snapshot_table.source_table
            payload["_fivetran_deleted"] = False
            rows.append(payload)
        result[runtime.snapshot_table.fivetran_table] = {
            "columns": runtime.snapshot_table.column_names,
            "primary_key": list(runtime.snapshot_table.primary_key),
            "rows": rows,
        }
    return result


def _json_safe(value: Any) -> Any:
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, (datetime, date, time)):
        return value.isoformat()
    if isinstance(value, bytes):
        return value.hex()
    return value


def _write_csv(path: Path, columns: list[str], rows: list[dict[str, str | None]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(columns)
        for row in rows:
            writer.writerow([_csv_value(row.get(column)) for column in columns])


def _write_csv_rows(path: Path, rows: list[list[str]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerows(rows)


def _csv_value(value: str | None) -> str:
    return NULL_MARKER if value is None else str(value)


def _write_text(path: Path, payload: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(payload, encoding="utf-8")


def _write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def _snapshot_stem(source_schema: str, source_table: str) -> str:
    return f"{quote_object_path_segment(source_schema)}.{quote_object_path_segment(source_table)}"


def _is_ignorable_statement(sql: str) -> bool:
    lowered = sql.strip().lower()
    return lowered.startswith(
        (
            "set ",
            "start transaction",
            "begin",
            "commit",
            "rollback",
            "lock tables",
            "unlock tables",
            "delimiter ",
        )
    )


def _is_ignorable_expression(expression: exp.Expression) -> bool:
    if isinstance(expression, (exp.Command, exp.Transaction, exp.Commit, exp.Rollback)):
        return True
    if isinstance(expression, exp.Drop):
        return _create_kind(expression) != "TABLE"
    return False