import random
import time
from multiprocessing.dummy import Pool
import datetime

import pytest
from helpers.client import QueryRuntimeException
from helpers.cluster import ClickHouseCluster

node_options = dict(
    with_zookeeper=True,
    main_configs=[
        "configs/remote_servers.xml",
        "configs/config.d/instant_moves.xml",
        "configs/config.d/part_log.xml",
        "configs/config.d/zookeeper_session_timeout.xml",
        "configs/config.d/storage_configuration.xml",
    ],
    tmpfs=["/external:size=200M", "/internal:size=1M"],
)

cluster = ClickHouseCluster(__file__)
node1 = cluster.add_instance("node1", macros={"shard": 0, "replica": 1}, **node_options)
node2 = cluster.add_instance("node2", macros={"shard": 0, "replica": 2}, **node_options)
node3 = cluster.add_instance("node3", macros={"shard": 1, "replica": 1}, **node_options)
node4 = cluster.add_instance("node4", macros={"shard": 1, "replica": 2}, **node_options)
nodes = [node1, node2, node3, node4]


@pytest.fixture(scope="module")
def started_cluster():
    try:
        cluster.start()
        yield cluster
    except Exception as ex:
        print(ex)
    finally:
        cluster.shutdown()


def drop_table(nodes, table_name):
    for node in nodes:
        node.query("DROP TABLE IF EXISTS {} NO DELAY".format(table_name))


def create_table(
    nodes,
    table_name,
    with_storage_policy=False,
    with_time_column=False,
    with_ttl_move=False,
    with_ttl_delete=False,
):
    extra_columns = ""
    settings = []

    for node in nodes:
        sql = """
            CREATE TABLE {table_name}
                (
                    num UInt32,
                    num2 UInt32 DEFAULT num + 1{extra_columns}
                )
                ENGINE = ReplicatedMergeTree('/clickhouse/tables/test/{table_name}', '{replica}')
                ORDER BY num PARTITION BY num % 100
            """
        if with_ttl_move:
            sql += """
                TTL time + INTERVAL (num2 % 1) SECOND TO DISK 'external'
            """
        if with_ttl_delete:
            sql += """
                TTL time + INTERVAL (num2 % 1) SECOND DELETE
            """
            settings.append("merge_with_ttl_timeout = 1")

        if with_storage_policy:
            settings.append("storage_policy='default_with_external'")

        if settings:
            sql += """
                SETTINGS {}
            """.format(
                ", ".join(settings)
            )

        if with_time_column:
            extra_columns = """,
                    time DateTime
                """
        node.query(
            sql.format(
                table_name=table_name, replica=node.name, extra_columns=extra_columns
            )
        )


def create_distributed_table(node, table_name):
    sql = """
            CREATE TABLE %(table_name)s_replicated ON CLUSTER test_cluster
            (
                num UInt32,
                num2 UInt32 DEFAULT num + 1
            )
            ENGINE = ReplicatedMergeTree('/clickhouse/tables/test/{shard}/%(table_name)s_replicated', '{replica}')
            ORDER BY num PARTITION BY num %% 100;
        """ % dict(
        table_name=table_name
    )
    node.query(sql)
    sql = """
            CREATE TABLE %(table_name)s ON CLUSTER test_cluster AS %(table_name)s_replicated
            ENGINE = Distributed(test_cluster, default, %(table_name)s_replicated, rand())
        """ % dict(
        table_name=table_name
    )
    node.query(sql)


def drop_distributed_table(node, table_name):
    node.query(
        "DROP TABLE IF EXISTS {} ON CLUSTER test_cluster SYNC".format(table_name)
    )
    node.query(
        "DROP TABLE IF EXISTS {}_replicated ON CLUSTER test_cluster SYNC".format(
            table_name
        )
    )
    time.sleep(1)


def insert(
    node,
    table_name,
    chunk=1000,
    col_names=None,
    iterations=1,
    ignore_exception=False,
    slow=False,
    with_many_parts=False,
    offset=0,
    with_time_column=False,
):
    if col_names is None:
        col_names = ["num", "num2"]
    for i in range(iterations):
        try:
            query = ["SET max_partitions_per_insert_block = 10000000"]
            if with_many_parts:
                query.append("SET max_insert_block_size = 256")
            if with_time_column:
                query.append(
                    "INSERT INTO {table_name} ({col0}, {col1}, time) SELECT number AS {col0}, number + 1 AS {col1}, now() + 10 AS time FROM numbers_mt({chunk})".format(
                        table_name=table_name,
                        chunk=chunk,
                        col0=col_names[0],
                        col1=col_names[1],
                    )
                )
            elif slow:
                query.append(
                    "INSERT INTO {table_name} ({col0}, {col1}) SELECT number + sleepEachRow(0.001) AS {col0}, number + 1 AS {col1} FROM numbers_mt({chunk})".format(
                        table_name=table_name,
                        chunk=chunk,
                        col0=col_names[0],
                        col1=col_names[1],
                    )
                )
            else:
                query.append(
                    "INSERT INTO {table_name} ({col0},{col1}) SELECT number + {offset} AS {col0}, number + 1 + {offset} AS {col1} FROM numbers_mt({chunk})".format(
                        table_name=table_name,
                        chunk=chunk,
                        col0=col_names[0],
                        col1=col_names[1],
                        offset=str(offset),
                    )
                )
            node.query(";\n".join(query))
        except QueryRuntimeException as ex:
            if not ignore_exception:
                raise


def select(
    node,
    table_name,
    col_name="num",
    expected_result=None,
    iterations=1,
    ignore_exception=False,
    slow=False,
    poll=None,
):
    for i in range(iterations):
        start_time = time.time()
        while True:
            try:
                if slow:
                    r = node.query(
                        "SELECT count() FROM (SELECT num2, sleepEachRow(0.5) FROM {} WHERE {} % 1000 > 0)".format(
                            table_name, col_name
                        )
                    )
                else:
                    r = node.query(
                        "SELECT count() FROM {} WHERE {} % 1000 > 0".format(
                            table_name, col_name
                        )
                    )
                if expected_result:
                    if (
                        r != expected_result
                        and poll
                        and time.time() - start_time < poll
                    ):
                        continue
                    assert r == expected_result
            except QueryRuntimeException as ex:
                if not ignore_exception:
                    raise
            break


def rename_column(
    node, table_name, name, new_name, iterations=1, ignore_exception=False
):
    for i in range(iterations):
        try:
            node.query(
                "ALTER TABLE {table_name} RENAME COLUMN {name} to {new_name}".format(
                    table_name=table_name, name=name, new_name=new_name
                )
            )
        except QueryRuntimeException as ex:
            if not ignore_exception:
                raise


def rename_column_on_cluster(
    node, table_name, name, new_name, iterations=1, ignore_exception=False
):
    for i in range(iterations):
        try:
            node.query(
                "ALTER TABLE {table_name} ON CLUSTER test_cluster RENAME COLUMN {name} to {new_name}".format(
                    table_name=table_name, name=name, new_name=new_name
                )
            )
        except QueryRuntimeException as ex:
            if not ignore_exception:
                raise


def alter_move(node, table_name, iterations=1, ignore_exception=False):
    for i in range(iterations):
        move_part = random.randint(0, 99)
        move_volume = "external"
        try:
            node.query(
                "ALTER TABLE {table_name} MOVE PARTITION '{move_part}' TO VOLUME '{move_volume}'".format(
                    table_name=table_name, move_part=move_part, move_volume=move_volume
                )
            )
        except QueryRuntimeException as ex:
            if not ignore_exception:
                raise


def test_rename_parallel_same_node(started_cluster):
    table_name = "test_rename_parallel_same_node"
    drop_table(nodes, table_name)
    try:
        create_table(nodes, table_name)
        insert(node1, table_name, 1000)

        p = Pool(15)
        tasks = []
        for i in range(1):
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "num2", "foo2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "foo2", "foo3", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "foo3", "num2", 5, True)
                )
            )
        for task in tasks:
            task.get(timeout=240)

        # rename column back to original
        rename_column(node1, table_name, "foo3", "num2", 1, True)
        rename_column(node1, table_name, "foo2", "num2", 1, True)

        # check that select still works
        select(node1, table_name, "num2", "999\n")
    finally:
        drop_table(nodes, table_name)


def test_rename_parallel(started_cluster):
    table_name = "test_rename_parallel"
    drop_table(nodes, table_name)
    try:
        create_table(nodes, table_name)
        insert(node1, table_name, 1000)

        p = Pool(15)
        tasks = []
        for i in range(1):
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "num2", "foo2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node2, table_name, "foo2", "foo3", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node3, table_name, "foo3", "num2", 5, True)
                )
            )
        for task in tasks:
            task.get(timeout=240)

        # rename column back to original
        rename_column(node1, table_name, "foo3", "num2", 1, True)
        rename_column(node1, table_name, "foo2", "num2", 1, True)

        # check that select still works
        select(node1, table_name, "num2", "999\n")
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_select(started_cluster):
    table_name = "test_rename_with_parallel_select"
    drop_table(nodes, table_name)
    try:
        create_table(nodes, table_name)
        insert(node1, table_name, 1000)

        select(node1, table_name, "num2", "999\n", poll=30)
        select(node2, table_name, "num2", "999\n", poll=30)
        select(node3, table_name, "num2", "999\n", poll=30)

        p = Pool(15)
        tasks = []
        for i in range(1):
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "num2", "foo2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node2, table_name, "foo2", "foo3", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node3, table_name, "foo3", "num2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(select, (node1, table_name, "foo3", "999\n", 5, True))
            )
            tasks.append(
                p.apply_async(select, (node2, table_name, "num2", "999\n", 5, True))
            )
            tasks.append(
                p.apply_async(select, (node3, table_name, "foo2", "999\n", 5, True))
            )
        for task in tasks:
            task.get(timeout=240)

        # rename column back to original name
        rename_column(node1, table_name, "foo3", "num2", 1, True)
        rename_column(node1, table_name, "foo2", "num2", 1, True)

        # check that select still works
        select(node1, table_name, "num2", "999\n")
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_insert(started_cluster):
    table_name = "test_rename_with_parallel_insert"
    drop_table(nodes, table_name)
    try:
        create_table(nodes, table_name)
        insert(node1, table_name, 1000)

        p = Pool(15)
        tasks = []
        for i in range(1):
            tasks.append(
                p.apply_async(
                    rename_column, (node1, table_name, "num2", "foo2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node2, table_name, "foo2", "foo3", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column, (node3, table_name, "foo3", "num2", 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    insert, (node1, table_name, 100, ["num", "foo3"], 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    insert, (node2, table_name, 100, ["num", "num2"], 5, True)
                )
            )
            tasks.append(
                p.apply_async(
                    insert, (node3, table_name, 100, ["num", "foo2"], 5, True)
                )
            )
        for task in tasks:
            task.get(timeout=240)

        # rename column back to original
        rename_column(node1, table_name, "foo3", "num2", 1, True)
        rename_column(node1, table_name, "foo2", "num2", 1, True)

        # check that select still works
        select(node1, table_name, "num2")
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_merges(started_cluster):
    table_name = "test_rename_with_parallel_merges"
    drop_table(nodes, table_name)
    try:
        print("Creating tables", datetime.datetime.now())
        create_table(nodes, table_name)
        for i in range(5):
            insert(
                node1,
                table_name,
                100,
                ["num", "num2"],
                1,
                False,
                False,
                True,
                offset=i * 100,
            )

        print("Data inserted", datetime.datetime.now())

        def merge_parts(node, table_name, iterations=1):
            for _ in range(iterations):
                try:
                    node.query("OPTIMIZE TABLE %s FINAL" % table_name)
                except Exception as ex:
                    print("Got an exception while optimizing table", ex)

        print("Creating pool")
        p = Pool(15)
        tasks = []
        tasks.append(
            p.apply_async(rename_column, (node1, table_name, "num2", "foo2", 2, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node2, table_name, "foo2", "foo3", 2, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node3, table_name, "foo3", "num2", 2, True))
        )
        tasks.append(p.apply_async(merge_parts, (node1, table_name, 2)))
        tasks.append(p.apply_async(merge_parts, (node2, table_name, 2)))
        tasks.append(p.apply_async(merge_parts, (node3, table_name, 2)))

        print("Waiting for tasks", datetime.datetime.now())
        for task in tasks:
            task.get(timeout=240)
        print("Finished waiting", datetime.datetime.now())

        print("Renaming columns", datetime.datetime.now())
        # rename column back to the original name
        rename_column(node1, table_name, "foo3", "num2", 1, True)
        rename_column(node1, table_name, "foo2", "num2", 1, True)
        print("Finished renaming", datetime.datetime.now())

        # check that select still works
        select(node1, table_name, "num2", "500\n")
        select(node2, table_name, "num2", "500\n")
        select(node3, table_name, "num2", "500\n")
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_slow_insert(started_cluster):
    table_name = "test_rename_with_parallel_slow_insert"
    drop_table(nodes, table_name)
    try:
        create_table(nodes, table_name)
        insert(node1, table_name, 1000)

        p = Pool(15)
        tasks = []
        tasks.append(
            p.apply_async(
                insert, (node1, table_name, 10000, ["num", "num2"], 1, False, True)
            )
        )
        tasks.append(
            p.apply_async(
                insert, (node1, table_name, 10000, ["num", "num2"], 1, True, True)
            )
        )  # deduplicated
        time.sleep(0.5)
        tasks.append(p.apply_async(rename_column, (node1, table_name, "num2", "foo2")))

        for task in tasks:
            task.get(timeout=240)

        insert(node1, table_name, 100, ["num", "foo2"])

        # rename column back to original
        rename_column(node1, table_name, "foo2", "num2")

        # check that select still works
        select(node1, table_name, "num2", "11089\n")
        select(node2, table_name, "num2", "11089\n", poll=30)
        select(node3, table_name, "num2", "11089\n", poll=30)
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_ttl_move(started_cluster):
    table_name = "test_rename_with_parallel_ttl_move"
    try:
        create_table(
            nodes,
            table_name,
            with_storage_policy=True,
            with_time_column=True,
            with_ttl_move=True,
        )
        rename_column(node1, table_name, "time", "time2", 1, False)
        rename_column(node1, table_name, "time2", "time", 1, False)

        p = Pool(15)
        tasks = []

        tasks.append(
            p.apply_async(
                insert,
                (
                    node1,
                    table_name,
                    10000,
                    ["num", "num2"],
                    1,
                    False,
                    False,
                    True,
                    0,
                    True,
                ),
            )
        )
        time.sleep(5)
        rename_column(node1, table_name, "time", "time2", 1, False)
        time.sleep(4)
        tasks.append(
            p.apply_async(rename_column, (node1, table_name, "num2", "foo2", 5, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node2, table_name, "foo2", "foo3", 5, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node3, table_name, "num3", "num2", 5, True))
        )

        for task in tasks:
            task.get(timeout=240)

        # check some parts got moved
        assert "external" in set(
            node1.query(
                "SELECT disk_name FROM system.parts WHERE table == '{}' AND active=1 ORDER BY modification_time".format(
                    table_name
                )
            )
            .strip()
            .splitlines()
        )

        # rename column back to original
        rename_column(node1, table_name, "foo2", "num2", 1, True)
        rename_column(node1, table_name, "foo3", "num2", 1, True)

        # check that select still works
        select(node1, table_name, "num2", "9990\n")
    finally:
        drop_table(nodes, table_name)


def test_rename_with_parallel_ttl_delete(started_cluster):
    table_name = "test_rename_with_parallel_ttl_delete"
    try:
        create_table(nodes, table_name, with_time_column=True, with_ttl_delete=True)
        rename_column(node1, table_name, "time", "time2", 1, False)
        rename_column(node1, table_name, "time2", "time", 1, False)

        def merge_parts(node, table_name, iterations=1):
            for i in range(iterations):
                node.query("OPTIMIZE TABLE {}".format(table_name))

        p = Pool(15)
        tasks = []

        tasks.append(
            p.apply_async(
                insert,
                (
                    node1,
                    table_name,
                    10000,
                    ["num", "num2"],
                    1,
                    False,
                    False,
                    True,
                    0,
                    True,
                ),
            )
        )
        time.sleep(15)
        tasks.append(
            p.apply_async(rename_column, (node1, table_name, "num2", "foo2", 5, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node2, table_name, "foo2", "foo3", 5, True))
        )
        tasks.append(
            p.apply_async(rename_column, (node3, table_name, "num3", "num2", 5, True))
        )
        tasks.append(p.apply_async(merge_parts, (node1, table_name, 3)))
        tasks.append(p.apply_async(merge_parts, (node2, table_name, 3)))
        tasks.append(p.apply_async(merge_parts, (node3, table_name, 3)))

        for task in tasks:
            task.get(timeout=240)

        # rename column back to original
        rename_column(node1, table_name, "foo2", "num2", 1, True)
        rename_column(node1, table_name, "foo3", "num2", 1, True)

        assert (
            int(node1.query("SELECT count() FROM {}".format(table_name)).strip())
            < 10000
        )
    finally:
        drop_table(nodes, table_name)


def test_rename_distributed(started_cluster):
    table_name = "test_rename_distributed"
    try:
        create_distributed_table(node1, table_name)
        insert(node1, table_name, 1000)

        rename_column_on_cluster(node1, table_name, "num2", "foo2")
        rename_column_on_cluster(node1, "%s_replicated" % table_name, "num2", "foo2")

        insert(node1, table_name, 1000, col_names=["num", "foo2"])

        select(node1, table_name, "foo2", "1998\n", poll=30)
    finally:
        drop_distributed_table(node1, table_name)


def test_rename_distributed_parallel_insert_and_select(started_cluster):
    table_name = "test_rename_distributed_parallel_insert_and_select"
    try:
        create_distributed_table(node1, table_name)
        insert(node1, table_name, 1000)

        p = Pool(15)
        tasks = []
        for i in range(1):
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, table_name, "num2", "foo2", 3, True),
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, "%s_replicated" % table_name, "num2", "foo2", 3, True),
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, table_name, "foo2", "foo3", 3, True),
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, "%s_replicated" % table_name, "foo2", "foo3", 3, True),
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, table_name, "foo3", "num2", 3, True),
                )
            )
            tasks.append(
                p.apply_async(
                    rename_column_on_cluster,
                    (node1, "%s_replicated" % table_name, "foo3", "num2", 3, True),
                )
            )
            tasks.append(
                p.apply_async(insert, (node1, table_name, 10, ["num", "foo3"], 5, True))
            )
            tasks.append(
                p.apply_async(insert, (node2, table_name, 10, ["num", "num2"], 5, True))
            )
            tasks.append(
                p.apply_async(insert, (node3, table_name, 10, ["num", "foo2"], 5, True))
            )
            tasks.append(
                p.apply_async(select, (node1, table_name, "foo2", None, 5, True))
            )
            tasks.append(
                p.apply_async(select, (node2, table_name, "foo3", None, 5, True))
            )
            tasks.append(
                p.apply_async(select, (node3, table_name, "num2", None, 5, True))
            )
        for task in tasks:
            task.get(timeout=240)

        rename_column_on_cluster(node1, table_name, "foo2", "num2", 1, True)
        rename_column_on_cluster(
            node1, "%s_replicated" % table_name, "foo2", "num2", 1, True
        )
        rename_column_on_cluster(node1, table_name, "foo3", "num2", 1, True)
        rename_column_on_cluster(
            node1, "%s_replicated" % table_name, "foo3", "num2", 1, True
        )

        insert(node1, table_name, 1000, col_names=["num", "num2"])
        select(node1, table_name, "num2")
        select(node2, table_name, "num2")
        select(node3, table_name, "num2")
        select(node4, table_name, "num2")
    finally:
        drop_distributed_table(node1, table_name)