INNER CODE UNIT · Python

logical_names

Altinity/clickhouse-operator · tools/restore_shared_to_replicated.py:334

def logical_names(name: str) -> tuple[str, str]:
    rel = name[len("metadata/") : -len(".sql")]
    parts = rel.split("/")
    db = unquote(parts[0])
    table = unquote(parts[1]) if len(parts) > 1 else ""
    return db, table


def rewrite_schema(sql: str, kind: str, args: argparse.Namespace) -> str:
    out = sql
    if kind == "table":
        out = SHARED_MERGETREE_RE.sub(r"Replicated\1MergeTree", out)
        args_sql = f"({args.replicated_zk_path}, {args.replicated_replica})"
        out = REPLICATED_MISSING_ARGS_RE.sub(rf"\1{args_sql}", out)
    elif kind == "database":
        out = SHARED_DATABASE_ENGINE_RE.sub(r"\1Atomic", out)
    if CREATE_KIND_RE.search(out) and not IF_NOT_EXISTS_RE.search(out):
        out = CREATE_KIND_RE.sub(lambda m: m.group(0) + " IF NOT EXISTS", out, count=1)

View source record →