"""
TiDB Query Plan parser for EXPLAIN output.
Supports multiple TiDB versions (V2, V3, V4).
"""

import re
from typing import Dict, List, Tuple

from .plan_types import (
    ApplyOp,
    BaseOp,
    DeleteOp,
    HashAggOp,
    HashJoinOp,
    IndexJoinOp,
    IndexLookupOp,
    IndexReaderOp,
    IndexScanOp,
    InsertOp,
    JoinType,
    LimitOp,
    MaxOneRowOp,
    MergeJoinOp,
    Operator,
    OpType,
    Plan,
    PlanVersion,
    PointGetOp,
    ProjectionOp,
    SelectionOp,
    SelectLockOp,
    ShowOp,
    SortOp,
    StreamAggOp,
    TableDualOp,
    TableReaderOp,
    TableScanOp,
    TaskType,
    TopNOp,
    UnionScanOp,
    UpdateOp,
    op_type_is_join,
)


class PlanParseError(Exception):
    """Exception raised when plan parsing fails."""


def parse_text(sql: str, plan_text: str, version: str = "v4") -> Plan:
    """Parse EXPLAIN text output into a Plan object."""
    sql = sql.strip().rstrip(";")

    # Normalize different formats to unified row data
    if is_system_table_format(plan_text):
        rows = normalize_system_table_to_rows(plan_text)
    else:
        rows = normalize_explain_to_rows(plan_text)

    # Use unified parsing logic
    return parse_plan(format_version(version), sql, rows)


def normalize_explain_to_rows(explain_text: str) -> List[List[str]]:
    """Normalize standard EXPLAIN format to unified row data."""
    explain_lines = trim_and_split_explain_result(explain_text)
    return split_rows(explain_lines[3:-1])  # Skip header and footer


def normalize_system_table_to_rows(plan_text: str) -> List[List[str]]:
    """Normalize system table format to unified row data."""
    if not plan_text or not plan_text.strip():
        return []

    # Handle both \n and \\n (escaped newlines from database storage)
    if "\\\\n" in plan_text:
        lines = plan_text.strip().split("\\\\n")
    else:
        lines = plan_text.strip().split("\n")

    # Skip header line if present (contains column names)
    if lines and ("id" in lines[0] and "task" in lines[0] and "estRows" in lines[0]):
        lines = lines[1:]

    # Convert tab-separated lines to row format like EXPLAIN
    rows = []
    for line in lines:
        if line.strip():
            # Split by tabs and filter empty parts
            parts = [p.strip() for p in line.split("\t") if p.strip()]
            if len(parts) >= 4:
                # Rearrange to match EXPLAIN format: [id, estRows, task, operator_info]
                row = [
                    parts[0],  # id (with tree structure)
                    parts[2],  # estRows
                    parts[1],  # task
                    parts[3] if len(parts) > 3 else "",  # operator_info
                ]
                rows.append(row)

    return rows


def is_system_table_format(text: str) -> bool:
    """
    Detect if text is system table format.

    System table format characteristics:
    - Contains tabs for column separation
    - No table separator lines (like +---+)
    """
    lines = text.strip().split("\n")
    has_tabs = "\t" in text
    has_separators = any(is_separator_line(line) for line in lines)

    return has_tabs and not has_separators


def parse_plan(version: str, sql: str, explain_rows: List[List[str]]) -> Plan:
    """Parse explain rows into a Plan object."""
    try:
        ver = format_version(version)
        if ver == PlanVersion.V2:
            return parse_v2(sql, explain_rows)
        elif ver == PlanVersion.V3:
            return parse_v3(sql, explain_rows)
        elif ver == PlanVersion.V4:
            return parse_v4(sql, explain_rows)
        else:
            raise PlanParseError(f"Unsupported TiDB version: {version}")
    except Exception as e:
        # Log the error with context
        explain_content = "\n".join(["\t".join(row) for row in explain_rows])
        print(f"Parse error for SQL={sql}, version={version}")
        print(f"EXPLAIN content:\n{explain_content}")
        raise PlanParseError(f"Failed to parse plan: {e}")


def parse_v4(sql: str, rows: List[List[str]]) -> Plan:
    """Parse V4 format explain output."""
    plan = Plan(sql=sql, version=PlanVersion.V4)
    if rows:
        root = parse_v4_op(rows, 0)
        plan.root = root
    return plan


def parse_v4_op(rows: List[List[str]], row_no: int) -> Operator:
    """Parse a single operator from V4 format."""
    children = []
    child_row_nos = find_child_row_no(rows, row_no, 0)

    for child_no in child_row_nos:
        child = parse_v4_op(rows, child_no)
        children.append(child)

    return parse_row_v4(rows[row_no], children)


def parse_row_v4(cols: List[str], children: List[Operator]) -> Operator:
    """Parse a single row in V4 format into an operator."""
    if len(cols) < 4:
        raise PlanParseError(f"Invalid row format: {cols}")

    try:
        est_rows = float(cols[1].strip())
    except ValueError:
        est_rows = 0.0

    op_id = extract_operator_id(cols[0])
    op_type = match_op_type(op_id)

    if op_type_is_join(op_type):
        adjust_join_children_v4(children)

    base = BaseOp(
        id=op_id,
        op_type=op_type,
        est_row=est_rows,
        task=parse_task_type(cols[2]),
        children=children,
    )

    # Use unified operator creation
    return create_specific_operator(base, op_type, cols[3])


def parse_v2(sql: str, rows: List[List[str]]) -> Plan:
    """Parse V2 format explain output (placeholder)."""
    # TODO: Implement V2 parsing logic
    raise NotImplementedError("V2 parsing not implemented yet")


def parse_v3(sql: str, rows: List[List[str]]) -> Plan:
    """Parse V3 format explain output (placeholder)."""
    # TODO: Implement V3 parsing logic
    raise NotImplementedError("V3 parsing not implemented yet")


def extract_table_index_v4(info: str) -> Tuple[str, str]:
    """Extract table and index information from V4 format."""
    # Example: table:a, index:idx_notice_config_id_type(user_id, notice_config_id, notice_type)
    table = ""
    index = ""

    # Extract table
    table_prefix = "table:"
    if table_prefix in info:
        start = info.index(table_prefix) + len(table_prefix)
        end = info.find(",", start)
        if end != -1:
            table = info[start:end].strip()
        else:
            table = info[start:].strip()

    # Extract index
    index_prefix = "index:"
    if index_prefix in info:
        start = info.index(index_prefix) + len(index_prefix)
        remaining = info[start:]
        # Find content within parentheses
        paren_start = remaining.find("(")
        if paren_start != -1:
            paren_end = remaining.find(")", paren_start + 1)
            if paren_end != -1:
                index = remaining[paren_start + 1 : paren_end]

    return table, index


def adjust_join_children_v4(children: List[Operator]):
    """Adjust join children order - make children[0] the outer side."""
    if len(children) >= 2:
        if "probe" in children[0].get_id().lower():
            children[0], children[1] = children[1], children[0]


def trim_and_split_explain_result(explain_result: str) -> List[str]:
    """Trim and split EXPLAIN result into lines."""
    lines = explain_result.split("\n")
    separator_indices = []

    for i, line in enumerate(lines):
        if is_separator_line(line):
            separator_indices.append(i)
            if len(separator_indices) == 3:
                break

    if len(separator_indices) != 3:
        raise PlanParseError("Invalid EXPLAIN result format")

    return lines[separator_indices[0] : separator_indices[2] + 1]


def is_separator_line(line: str) -> bool:
    """Check if a line is a separator line (contains only + and -)."""
    line = line.strip()
    if not line:
        return False
    return all(c in "+-" for c in line)


def format_version(version: str) -> PlanVersion:
    """Format version string to PlanVersion enum."""
    version = version.lower()
    if version.startswith("v2"):
        return PlanVersion.V2
    elif version.startswith("v3"):
        return PlanVersion.V3
    else:
        return PlanVersion.V4


def split_rows(rows: List[str]) -> List[List[str]]:
    """Split rows into columns."""
    results = []
    for row in rows:
        cols = row.split("|")
        # Remove first and last empty columns
        cols = cols[1:-1]
        results.append(cols)
    return results


def find_child_row_no(
    rows: List[List[str]], parent_row_no: int, id_col_no: int
) -> List[int]:
    """Find child row numbers for a given parent row."""
    if parent_row_no >= len(rows):
        return []

    parent_field = rows[parent_row_no][id_col_no]
    parent_chars = list(parent_field)

    # Find the column position where the tree structure starts
    col = 0
    for i, c in enumerate(parent_chars):
        if c.isalpha() and c.isupper():
            col = i
            break
    else:
        return []

    child_row_nos = []
    for i in range(parent_row_no + 1, len(rows)):
        field = rows[i][id_col_no]
        if col >= len(field):
            break

        c = field[col] if col < len(field) else ""
        if c in ["├", "└"]:
            child_row_nos.append(i)
        elif c != "│":
            break

    return child_row_nos


def extract_operator_id(field: str) -> str:
    """Extract operator ID from the field by removing tree characters."""
    tree_chars = {"└", "─", "│", "├", " "}
    return "".join(c for c in field if c not in tree_chars).strip()


def split_kvs(kv_str: str) -> Dict[str, str]:
    """Split key-value string into dictionary."""
    kv_map = {}
    kvs = kv_str.split(",")
    for kv in kvs:
        parts = kv.split(":", 1)
        if len(parts) == 2:
            key = parts[0].strip()
            value = parts[1].strip()
            kv_map[key] = value
    return kv_map


def parse_task_type(task_str: str) -> TaskType:
    """Parse task type from string."""
    task = task_str.strip().lower()
    if task == "root":
        return TaskType.ROOT
    elif "tiflash" in task:
        return TaskType.TIFLASH
    else:
        return TaskType.TIKV


def match_op_type(op_id: str) -> OpType:
    """Match operator ID to OpType (compatible with Golang logic)."""
    # Remove probe/build suffixes that appear in TiDB explain output
    base_id = op_id.replace("(Probe)", "").replace("(Build)", "")
    x = base_id.lower()

    if "agg" in x:
        if "hash" in x:
            return OpType.HASH_AGG
        elif "stream" in x:
            return OpType.STREAM_AGG
        return OpType.UNKNOWN

    if "join" in x:
        if "hash" in x:
            return OpType.HASH_JOIN
        elif "merge" in x:
            return OpType.MERGE_JOIN
        elif "index" in x:
            return OpType.INDEX_JOIN
        return OpType.UNKNOWN

    if "table" in x:
        if "reader" in x:
            return OpType.TABLE_READER
        elif "scan" in x:
            return OpType.TABLE_SCAN
        elif "dual" in x:
            return OpType.TABLE_DUAL
        return OpType.UNKNOWN

    if "index" in x:
        if "reader" in x:
            return OpType.INDEX_READER
        elif "scan" in x:
            return OpType.INDEX_SCAN
        elif "lookup" in x:
            return OpType.INDEX_LOOKUP
        return OpType.UNKNOWN

    if "selection" in x:
        return OpType.SELECTION
    if "projection" in x:
        return OpType.PROJECTION
    if "unionscan" in x:
        return OpType.UNION_SCAN
    if "point" in x:
        return OpType.POINT_GET
    if "maxonerow" in x:
        return OpType.MAX_ONE_ROW
    if "apply" in x:
        return OpType.APPLY
    if "limit" in x:
        return OpType.LIMIT
    if "sort" in x:
        return OpType.SORT
    if "topn" in x:
        return OpType.TOP_N
    if "update" in x:
        return OpType.UPDATE
    if "delete" in x:
        return OpType.DELETE
    if "selectlock" in x:
        return OpType.SELECT_LOCK
    if "show" in x:
        return OpType.SHOW
    if "insert" in x:
        return OpType.INSERT  # NOTE: Not implemented in Golang reference

    return OpType.UNKNOWN


def format_explain_rows(rows: List[List[str]]) -> str:
    """Format explain rows for display."""
    if not rows:
        return ""

    n_rows = len(rows)
    n_cols = len(rows[0]) if rows else 0
    fmt_rows = [""] * n_rows

    for col in range(n_cols):
        # Find the longest content in this column so far
        longest = max(len(fmt_rows[i]) for i in range(n_rows))

        for i in range(n_rows):
            # Pad to align columns
            gap = longest - len(fmt_rows[i])
            fmt_rows[i] += " " * gap

            # Add separator except for first and last columns
            if col != 0 and col != n_cols - 1:
                fmt_rows[i] += "  |  "

            # Add the column content
            if col < len(rows[i]):
                fmt_rows[i] += rows[i][col]

    return "\n".join(fmt_rows)


def create_specific_operator(
    base: BaseOp, op_type: OpType, operator_info: str
) -> Operator:
    """Create specific operator instance based on type (unified for all formats)."""
    if op_type == OpType.HASH_JOIN:
        return HashJoinOp(**base.__dict__, join_type=JoinType.UNKNOWN)
    elif op_type == OpType.MERGE_JOIN:
        return MergeJoinOp(**base.__dict__, join_type=JoinType.UNKNOWN)
    elif op_type == OpType.INDEX_JOIN:
        return IndexJoinOp(**base.__dict__, join_type=JoinType.UNKNOWN)
    elif op_type == OpType.TABLE_READER:
        return TableReaderOp(**base.__dict__)
    elif op_type == OpType.TABLE_SCAN:
        # Try both formats for table extraction
        table = extract_table_from_operator_info(operator_info)
        return TableScanOp(**base.__dict__, table=table)
    elif op_type == OpType.INDEX_READER:
        return IndexReaderOp(**base.__dict__)
    elif op_type == OpType.INDEX_SCAN:
        # Try both formats for table and index extraction
        table, index = extract_table_index_from_operator_info(operator_info)
        return IndexScanOp(**base.__dict__, table=table, index=index)
    elif op_type == OpType.INDEX_LOOKUP:
        return IndexLookupOp(**base.__dict__)
    elif op_type == OpType.SELECTION:
        return SelectionOp(**base.__dict__)
    elif op_type == OpType.PROJECTION:
        return ProjectionOp(**base.__dict__)
    elif op_type == OpType.POINT_GET:
        table = extract_table_from_operator_info(operator_info)
        return PointGetOp(**base.__dict__, batch="Batch" in base.id, table=table)
    elif op_type == OpType.HASH_AGG:
        return HashAggOp(**base.__dict__)
    elif op_type == OpType.STREAM_AGG:
        return StreamAggOp(**base.__dict__)
    elif op_type == OpType.MAX_ONE_ROW:
        return MaxOneRowOp(**base.__dict__)
    elif op_type == OpType.APPLY:
        return ApplyOp(**base.__dict__)
    elif op_type == OpType.LIMIT:
        return LimitOp(**base.__dict__)
    elif op_type == OpType.SORT:
        return SortOp(**base.__dict__)
    elif op_type == OpType.TOP_N:
        return TopNOp(**base.__dict__)
    elif op_type == OpType.TABLE_DUAL:
        return TableDualOp(**base.__dict__)
    elif op_type == OpType.UNION_SCAN:
        return UnionScanOp(**base.__dict__)
    elif op_type == OpType.SELECT_LOCK:
        return SelectLockOp(**base.__dict__)
    elif op_type == OpType.SHOW:
        return ShowOp(**base.__dict__)
    elif op_type == OpType.INSERT:
        # NOTE: Insert operator not implemented in Golang reference
        table = extract_table_from_operator_info(operator_info)
        return InsertOp(**base.__dict__, table=table)
    elif op_type == OpType.UPDATE:
        table = extract_table_from_operator_info(operator_info)
        return UpdateOp(**base.__dict__, table=table)
    elif op_type == OpType.DELETE:
        table = extract_table_from_operator_info(operator_info)
        return DeleteOp(**base.__dict__, table=table)
    else:
        # Truly unknown operators should raise error to expose issues
        raise PlanParseError(f"Unknown operator type: {base.id} (OpType: {op_type})")


def extract_table_from_operator_info(operator_info: str) -> str:
    """Extract table name from operator info (handles both EXPLAIN and system table formats)."""
    # Try key-value format first (EXPLAIN format): "table:tablename"
    kvs = split_kvs(operator_info)
    if "table" in kvs:
        return kvs["table"]

    # Try regex format (system table format): "table:tablename"
    match = re.search(r"table:([^,\s]+)", operator_info)
    if match:
        return match.group(1)

    return ""


def extract_table_index_from_operator_info(operator_info: str) -> Tuple[str, str]:
    """Extract table and index from operator info (handles both formats)."""
    # Try EXPLAIN v4 format first
    try:
        table, index = extract_table_index_v4(operator_info)
        if table or index:  # If we got something, use it
            return table, index
    except Exception:
        pass

    # Try system table format
    table = ""
    index = ""

    # Extract table
    table_match = re.search(r"table:([^,\s]+)", operator_info)
    if table_match:
        table = table_match.group(1)

    # Extract index
    index_match = re.search(r"index:([^,\s(]+)", operator_info)
    if index_match:
        index = index_match.group(1)

    return table, index