ClickHouse/tests/queries/0_stateless/00411_long_accurate_number_comparison.python

#!/usr/bin/env python3

import os, itertools, urllib.request, urllib.parse, urllib.error, urllib.request, urllib.error, urllib.parse, sys


def get_ch_answer(query):
    return (
        urllib.request.urlopen(
            os.environ.get(
                "CLICKHOUSE_URL",
                "http://localhost:" + os.environ.get("CLICKHOUSE_PORT_HTTP", "8123"),
            ),
            data=query.encode(),
        )
        .read()
        .decode()
    )


def check_answers(query, answer):
    ch_answer = get_ch_answer(query)
    if ch_answer.strip() != answer.strip():
        print("FAIL on query:", query)
        print("Expected answer:", answer)
        print("Fetched answer :", ch_answer)
        exit(-1)


def get_values():
    values = [0, 1, -1]
    for bits in [8, 16, 32, 64]:
        values += [2**bits, 2**bits - 1]
        values += [2 ** (bits - 1) - 1, 2 ** (bits - 1), 2 ** (bits - 1) + 1]
        values += [-(2 ** (bits - 1)) - 1, -(2 ** (bits - 1)), -(2 ** (bits - 1)) + 1]
    return values


def is_valid_integer(x):
    return -(2**63) <= x and x <= 2**64 - 1


TEST_WITH_CASTING = True
GENERATE_TEST_FILES = False

TYPES = {
    "UInt8": {"bits": 8, "sign": False, "float": False},
    "Int8": {"bits": 8, "sign": True, "float": False},
    "UInt16": {"bits": 16, "sign": False, "float": False},
    "Int16": {"bits": 16, "sign": True, "float": False},
    "UInt32": {"bits": 32, "sign": False, "float": False},
    "Int32": {"bits": 32, "sign": True, "float": False},
    "UInt64": {"bits": 64, "sign": False, "float": False},
    "Int64": {"bits": 64, "sign": True, "float": False}
    # "Float32" : { "bits" : 32, "sign" : True, "float" : True },
    # "Float64" : { "bits" : 64, "sign" : True, "float" : True }
}


def inside_range(value, type_name):
    bits = TYPES[type_name]["bits"]
    signed = TYPES[type_name]["sign"]
    is_float = TYPES[type_name]["float"]

    if is_float:
        return True

    if signed:
        return -(2 ** (bits - 1)) <= value and value <= 2 ** (bits - 1) - 1
    else:
        return 0 <= value and value <= 2**bits - 1


def test_operators(v1, v2, v1_passed, v2_passed):
    query_str = "{v1} = {v2}, {v1} != {v2}, {v1} < {v2}, {v1} <= {v2}, {v1} > {v2}, {v1} >= {v2},\t".format(
        v1=v1_passed, v2=v2_passed
    )
    query_str += "{v1} = {v2}, {v1} != {v2}, {v1} < {v2}, {v1} <= {v2}, {v1} > {v2}, {v1} >= {v2} ".format(
        v1=v2_passed, v2=v1_passed
    )

    answers = [v1 == v2, v1 != v2, v1 < v2, v1 <= v2, v1 > v2, v1 >= v2]
    answers += [v2 == v1, v2 != v1, v2 < v1, v2 <= v1, v2 > v1, v2 >= v1]

    answers_str = "\t".join([str(int(x)) for x in answers])

    return (query_str, answers_str)


VALUES = [x for x in get_values() if is_valid_integer(x)]


def test_pair(v1, v2):
    query = "SELECT {}, {}, ".format(v1, v2)
    answers = "{}\t{}\t".format(v1, v2)

    q, a = test_operators(v1, v2, str(v1), str(v2))
    query += q
    answers += a

    if TEST_WITH_CASTING:
        for t1 in TYPES.keys():
            if inside_range(v1, t1):
                for t2 in TYPES.keys():
                    if inside_range(v2, t2):
                        q, a = test_operators(
                            v1, v2, "to{}({})".format(t1, v1), "to{}({})".format(t2, v2)
                        )
                        query += ", " + q
                        answers += "\t" + a

    check_answers(query, answers)
    return query, answers


VALUES_INT = [
    0,
    -1,
    1,
    2**64 - 1,
    2**63,
    -(2**63),
    2**63 - 1,
    2**51,
    2**52,
    2**53 - 1,
    2**53,
    2**53 + 1,
    2**53 + 2,
    -(2**53) + 1,
    -(2**53),
    -(2**53) - 1,
    -(2**53) - 2,
    2 * 52,
    -(2**52),
]
VALUES_FLOAT = [
    float(x)
    for x in VALUES_INT
    + [
        -0.5,
        0.5,
        -1.5,
        1.5,
        2**53,
        2**51 - 0.5,
        2**51 + 0.5,
        2**60,
        -(2**60),
        -(2**63) - 10000,
        2**63 + 10000,
    ]
]


def test_float_pair(i, f):
    f_str = "%.9f" % f
    query = "SELECT '{}', '{}', ".format(i, f_str)
    answers = "{}\t{}\t".format(i, f_str)

    q, a = test_operators(i, f, i, f_str)
    query += q
    answers += a

    if TEST_WITH_CASTING:
        for t1 in TYPES.keys():
            if inside_range(i, t1):
                q, a = test_operators(i, f, "to{}({})".format(t1, i), f_str)
                query += ", " + q
                answers += "\t" + a

    check_answers(query, answers)
    return query, answers


def main():
    if GENERATE_TEST_FILES:
        base_name = "00411_accurate_number_comparison"
        sql_file = open(base_name + ".sql", "wt")
        ref_file = open(base_name + ".reference", "wt")

    num_int_tests = len(list(itertools.combinations(VALUES, 2)))

    num_parts = 4
    for part in range(0, num_parts):
        if "int" + str(part + 1) in sys.argv[1:]:
            for v1, v2 in itertools.islice(
                itertools.combinations(VALUES, 2),
                part * num_int_tests // num_parts,
                (part + 1) * num_int_tests // num_parts,
            ):
                q, a = test_pair(v1, v2)
                if GENERATE_TEST_FILES:
                    sql_file.write(q + ";\n")
                    ref_file.write(a + "\n")

    if "float" in sys.argv[1:]:
        for i, f in itertools.product(VALUES_INT, VALUES_FLOAT):
            q, a = test_float_pair(i, f)
            if GENERATE_TEST_FILES:
                sql_file.write(q + ";\n")
                ref_file.write(a + "\n")

    print("PASSED")


if __name__ == "__main__":
    main()