mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
zeekay
9a90943b8f
13,838 upstream commits, forward from the 26.6.1.1 fork point. 129 conflicts: 37 resolved mechanically (take upstream's file, re-apply the brand sweep) and 92 by hand where our fork and upstream had both changed the same code. The rule throughout: upstream owns the engine, we own the naming and a small set of deliberate deviations. Never resolve by discarding an upstream change — a dropped rename or a missed new parameter becomes a compile error a thousand files away. Deviations verified intact after the merge: ENABLE_XRAY still defaults OFF (CMakeLists.txt:105) X-Datastore-User read by the server (authenticateUserByHTTP.cpp) X-ClickHouse-* in src/ 0 hanzo/research present system.*_log retention bound present ArrowFlight rip held against upstream re-adding it Two things worth naming, because neither shows up as a conflict: src/CMakeLists.txt gained an UNGUARDED `add_object_library(datastore_storages_arrowflight Storages/ArrowFlight)` from upstream, for a directory our gRPC rip deleted. It sat outside every conflict hunk, so a resolver reading only the markers would sail straight past it into a CMake configure error. Removed; the Server/ArrowFlight sibling at line 474 stays because it is inside `if (TARGET ch_contrib::arrow_flight)`. The brand sweep had rewritten github.com/ClickHouse/ClickHouse issue links to github.com/Datastore/Datastore — a dead host. Those point at upstream's own tracker and are restored, alongside the contributors list, sql.clickhouse.com and the ClickHouse Inc copyright: attribution is theirs, not ours to rename. base/glibc-compatibility/memcpy/memcpy.h is deleted, not kept: upstream dropped it for LLVM-libc's, and our only change to it was a comment rebrand.
511 lines
18 KiB
Python
511 lines
18 KiB
Python
import os
|
|
import sys
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
from helpers.cluster import ClickHouseCluster
|
|
from helpers.test_tools import TSV
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
node = cluster.add_instance("node", stay_alive=True, main_configs=[])
|
|
|
|
|
|
def skip_test_msan(instance):
|
|
if instance.is_built_with_memory_sanitizer():
|
|
pytest.skip("Memory Sanitizer cannot work with vfork")
|
|
|
|
|
|
def copy_file_to_container(local_path, dist_path, container_id):
|
|
os.system(
|
|
"docker cp {local} {cont_id}:{dist}".format(
|
|
local=local_path, cont_id=container_id, dist=dist_path
|
|
)
|
|
)
|
|
|
|
|
|
config = """<datastore>
|
|
<user_defined_executable_functions_config>/etc/clickhouse-server/functions/test_function_config.xml</user_defined_executable_functions_config>
|
|
</datastore>"""
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.start()
|
|
|
|
node.replace_config(
|
|
"/etc/clickhouse-server/config.d/executable_user_defined_functions_config.xml",
|
|
config,
|
|
)
|
|
|
|
copy_file_to_container(
|
|
os.path.join(SCRIPT_DIR, "functions/."),
|
|
"/etc/clickhouse-server/functions",
|
|
node.docker_id,
|
|
)
|
|
copy_file_to_container(
|
|
os.path.join(SCRIPT_DIR, "user_scripts/."),
|
|
"/var/lib/clickhouse/user_scripts",
|
|
node.docker_id,
|
|
)
|
|
|
|
node.restart_clickhouse()
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
def test_executable_function_bash(started_cluster):
|
|
skip_test_msan(node)
|
|
assert node.query("SELECT test_function_bash(toUInt64(1))") == "Key 1\n"
|
|
assert node.query("SELECT test_function_bash(1)") == "Key 1\n"
|
|
|
|
assert node.query("SELECT test_function_pool_bash(toUInt64(1))") == "Key 1\n"
|
|
assert node.query("SELECT test_function_pool_bash(1)") == "Key 1\n"
|
|
|
|
|
|
def test_executable_function_python(started_cluster):
|
|
skip_test_msan(node)
|
|
assert node.query("SELECT test_function_python(toUInt64(1))") == "Key 1\n"
|
|
assert node.query("SELECT test_function_python(1)") == "Key 1\n"
|
|
|
|
assert node.query("SELECT test_function_pool_python(toUInt64(1))") == "Key 1\n"
|
|
assert node.query("SELECT test_function_pool_python(1)") == "Key 1\n"
|
|
|
|
|
|
def test_executable_function_send_chunk_header_python(started_cluster):
|
|
skip_test_msan(node)
|
|
|
|
for function_name in [
|
|
"test_function_send_chunk_header_python",
|
|
"test_function_send_chunk_header_pool_python",
|
|
]:
|
|
assert node.query(f"SELECT {function_name}(toUInt64(1))") == "Key 1\n"
|
|
assert node.query(f"SELECT {function_name}(1)") == "Key 1\n"
|
|
|
|
assert node.query(f"SELECT {function_name}(toUInt64(1))") == "Key 1\n"
|
|
assert node.query(f"SELECT {function_name}(1)") == "Key 1\n"
|
|
|
|
# Test specifically HTTP protocol
|
|
# This ensures that http_write_exception_in_output_format works as expected
|
|
assert node.http_query(
|
|
f"SELECT {function_name}(number) FROM numbers(10)",
|
|
params={"max_block_size": 3, "http_write_exception_in_output_format": True},
|
|
) == "".join(f"Key {i}\n" for i in range(10))
|
|
|
|
|
|
def test_executable_function_sum_python(started_cluster):
|
|
skip_test_msan(node)
|
|
assert (
|
|
node.query("SELECT test_function_sum_python(toUInt64(1), toUInt64(1))") == "2\n"
|
|
)
|
|
assert node.query("SELECT test_function_sum_python(1, 1)") == "2\n"
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_pool_python(toUInt64(1), toUInt64(1))")
|
|
== "2\n"
|
|
)
|
|
assert node.query("SELECT test_function_sum_pool_python(1, 1)") == "2\n"
|
|
|
|
|
|
def test_executable_function_argument_python(started_cluster):
|
|
skip_test_msan(node)
|
|
assert (
|
|
node.query("SELECT test_function_argument_python(toUInt64(1))") == "Key 1 1\n"
|
|
)
|
|
assert node.query("SELECT test_function_argument_python(1)") == "Key 1 1\n"
|
|
|
|
assert (
|
|
node.query("SELECT test_function_argument_pool_python(toUInt64(1))")
|
|
== "Key 1 1\n"
|
|
)
|
|
assert node.query("SELECT test_function_argument_pool_python(1)") == "Key 1 1\n"
|
|
|
|
|
|
def test_executable_function_signalled_python(started_cluster):
|
|
skip_test_msan(node)
|
|
assert node.query_and_get_error(
|
|
"SELECT test_function_signalled_python(toUInt64(1))"
|
|
)
|
|
assert node.query_and_get_error("SELECT test_function_signalled_python(1)")
|
|
|
|
assert node.query_and_get_error(
|
|
"SELECT test_function_signalled_pool_python(toUInt64(1))"
|
|
)
|
|
assert node.query_and_get_error("SELECT test_function_signalled_pool_python(1)")
|
|
|
|
|
|
def test_executable_function_slow_python(started_cluster):
|
|
skip_test_msan(node)
|
|
assert node.query_and_get_error("SELECT test_function_slow_python(toUInt64(1))")
|
|
assert node.query_and_get_error("SELECT test_function_slow_python(1)")
|
|
|
|
assert node.query_and_get_error(
|
|
"SELECT test_function_slow_pool_python(toUInt64(1))"
|
|
)
|
|
assert node.query_and_get_error("SELECT test_function_slow_pool_python(1)")
|
|
|
|
|
|
def test_executable_function_non_direct_bash(started_cluster):
|
|
skip_test_msan(node)
|
|
assert node.query("SELECT test_function_non_direct_bash(toUInt64(1))") == "Key 1\n"
|
|
assert node.query("SELECT test_function_non_direct_bash(1)") == "Key 1\n"
|
|
|
|
assert (
|
|
node.query("SELECT test_function_non_direct_pool_bash(toUInt64(1))")
|
|
== "Key 1\n"
|
|
)
|
|
assert node.query("SELECT test_function_non_direct_pool_bash(1)") == "Key 1\n"
|
|
|
|
|
|
def test_executable_function_sum_json_python(started_cluster):
|
|
skip_test_msan(node)
|
|
|
|
node.query("DROP TABLE IF EXISTS test_table;")
|
|
node.query("CREATE TABLE test_table (lhs UInt64, rhs UInt64) ENGINE=TinyLog;")
|
|
node.query("INSERT INTO test_table VALUES (0, 0), (1, 1), (2, 2);")
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_json_unnamed_args_python(1, 2);") == "3\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_unnamed_args_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_json_partially_named_args_python(1, 2);")
|
|
== "3\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_partially_named_args_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
assert node.query("SELECT test_function_sum_json_named_args_python(1, 2);") == "3\n"
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_named_args_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_json_unnamed_args_pool_python(1, 2);")
|
|
== "3\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_unnamed_args_pool_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_json_partially_named_args_python(1, 2);")
|
|
== "3\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_partially_named_args_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
assert (
|
|
node.query("SELECT test_function_sum_json_named_args_pool_python(1, 2);")
|
|
== "3\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_sum_json_named_args_pool_python(lhs, rhs) FROM test_table;"
|
|
)
|
|
== "0\n2\n4\n"
|
|
)
|
|
|
|
node.query("DROP TABLE test_table;")
|
|
|
|
|
|
def test_executable_function_input_nullable_python(started_cluster):
|
|
skip_test_msan(node)
|
|
|
|
node.query("DROP TABLE IF EXISTS test_table_nullable;")
|
|
node.query(
|
|
"CREATE TABLE test_table_nullable (value Nullable(UInt64)) ENGINE=TinyLog;"
|
|
)
|
|
node.query("INSERT INTO test_table_nullable VALUES (0), (NULL), (2);")
|
|
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_nullable_python(1), test_function_nullable_python(NULL)"
|
|
)
|
|
== "Key 1\tKey Nullable\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_nullable_python(value) FROM test_table_nullable;"
|
|
)
|
|
== "Key 0\nKey Nullable\nKey 2\n"
|
|
)
|
|
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_nullable_pool_python(1), test_function_nullable_pool_python(NULL)"
|
|
)
|
|
== "Key 1\tKey Nullable\n"
|
|
)
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_nullable_pool_python(value) FROM test_table_nullable;"
|
|
)
|
|
== "Key 0\nKey Nullable\nKey 2\n"
|
|
)
|
|
|
|
node.query("DROP TABLE test_table_nullable;")
|
|
|
|
|
|
def test_executable_function_parameter_python(started_cluster):
|
|
skip_test_msan(node)
|
|
|
|
assert node.query_and_get_error(
|
|
"SELECT test_function_parameter_python(2,2)(toUInt64(1))"
|
|
)
|
|
assert node.query_and_get_error("SELECT test_function_parameter_python(2,2)(1)")
|
|
assert node.query_and_get_error("SELECT test_function_parameter_python(1)")
|
|
assert node.query_and_get_error(
|
|
"SELECT test_function_parameter_python('test')(toUInt64(1))"
|
|
)
|
|
|
|
assert (
|
|
node.query("SELECT test_function_parameter_python('2')(toUInt64(1))")
|
|
== "Parameter 2 key 1\n"
|
|
)
|
|
assert (
|
|
node.query("SELECT test_function_parameter_python(2)(toUInt64(1))")
|
|
== "Parameter 2 key 1\n"
|
|
)
|
|
|
|
# Placeholders with invalid parameter names must not be registered as
|
|
# command parameters, so each of these functions takes zero parameters and
|
|
# passing one fails the parameter-count check with a specific error.
|
|
for function_name in (
|
|
"test_function_invalid_parameter_name_python", # name with a space: {test parameter:UInt64}
|
|
"test_function_invalid_empty_parameter_name_python", # empty name: {:UInt64}
|
|
"test_function_invalid_blank_parameter_name_python", # blank name: { :UInt64}
|
|
):
|
|
assert (
|
|
"number of parameters does not match. Expected 0. Actual 1"
|
|
in node.query_and_get_error(
|
|
f"SELECT {function_name}(2)(toUInt64(1))"
|
|
)
|
|
)
|
|
|
|
|
|
def test_executable_function_always_error_python(started_cluster):
|
|
skip_test_msan(node)
|
|
try:
|
|
node.query("SELECT test_function_always_error_throw_python(1)")
|
|
assert False, "Exception have to be thrown"
|
|
except Exception as ex:
|
|
assert "DB::Exception: User defined function 'test_function_always_error_throw_python' failed" in str(ex)
|
|
assert "DB::Exception: Executable generates stderr: Fake error" in str(ex)
|
|
|
|
query_id = uuid.uuid4().hex
|
|
assert (
|
|
node.query("SELECT test_function_always_error_log_python(1)", query_id=query_id)
|
|
== "Key 1\n"
|
|
)
|
|
assert node.contains_in_log(
|
|
f"{{{query_id}}} <Warning> TimeoutReadBufferFromFileDescriptor: Executable generates stderr: Fake error"
|
|
)
|
|
|
|
query_id = uuid.uuid4().hex
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_always_error_log_first_python(1)", query_id=query_id
|
|
)
|
|
== "Key 1\n"
|
|
)
|
|
assert node.contains_in_log(
|
|
f"{{{query_id}}} <Warning> TimeoutReadBufferFromFileDescriptor: Executable generates stderr at the beginning: {'a' * (3 * 1024)}{'b' * 1024}\n"
|
|
)
|
|
|
|
query_id = uuid.uuid4().hex
|
|
assert (
|
|
node.query(
|
|
"SELECT test_function_always_error_log_last_python(1)", query_id=query_id
|
|
)
|
|
== "Key 1\n"
|
|
)
|
|
assert node.contains_in_log(
|
|
f"{{{query_id}}} <Warning> TimeoutReadBufferFromFileDescriptor: Executable generates stderr at the end: {'b' * 1024}{'c' * (3 * 1024)}\n"
|
|
)
|
|
|
|
assert node.query("SELECT test_function_exit_error_ignore_python(1)") == "Key 1\n"
|
|
|
|
try:
|
|
node.query("SELECT test_function_exit_error_fail_python(1)")
|
|
assert False, "Exception have to be thrown"
|
|
except Exception as ex:
|
|
assert "DB::Exception: User defined function 'test_function_exit_error_fail_python' failed" in str(ex)
|
|
assert "DB::Exception: Child process was exited with return code 1" in str(ex)
|
|
|
|
def test_executable_function_query_cache(started_cluster):
|
|
'''Test for issues #77553 and #59988: Users should be able to specify if externally-defined are non-deterministic, and the query cache should treat them correspondingly.'''
|
|
'''Also see tests/0_stateless/test_query_cache_udf_sql.sql'''
|
|
skip_test_msan(node)
|
|
|
|
node.query("SYSTEM CLEAR QUERY CACHE");
|
|
|
|
# we are each testing an UDF without explicit <deterministic> tag (to check the default behavior) and two queries with <deterministic> true respectively false </deterministic>.
|
|
|
|
# query_cache_nondeterministic_function_handling = throw
|
|
|
|
assert node.query_and_get_error("SELECT test_function_bash(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'throw'")
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "0\n"
|
|
|
|
assert node.query("SELECT test_function_bash_deterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'throw'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "1\n"
|
|
|
|
assert node.query_and_get_error("SELECT test_function_bash_nondeterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'throw'")
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "1\n"
|
|
|
|
node.query("SYSTEM CLEAR QUERY CACHE");
|
|
|
|
# query_cache_nondeterministic_function_handling = save
|
|
|
|
assert node.query("SELECT test_function_bash(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'save'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "1\n"
|
|
|
|
assert node.query("SELECT test_function_bash_deterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'save'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "2\n"
|
|
|
|
assert node.query("SELECT test_function_bash_nondeterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'save'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "3\n"
|
|
|
|
node.query("SYSTEM CLEAR QUERY CACHE");
|
|
|
|
# query_cache_nondeterministic_function_handling = ignore
|
|
|
|
assert node.query("SELECT test_function_bash(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'ignore'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "0\n"
|
|
|
|
assert node.query("SELECT test_function_bash_deterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'ignore'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "1\n"
|
|
|
|
assert node.query("SELECT test_function_bash_nondeterministic(1) SETTINGS use_query_cache = true, query_cache_nondeterministic_function_handling = 'ignore'") == "Key 1\n"
|
|
assert node.query("SELECT count(*) FROM system.query_cache") == "1\n"
|
|
|
|
node.query("SYSTEM CLEAR QUERY CACHE");
|
|
|
|
def test_executable_function_python_exception_in_query_log(started_cluster):
|
|
'''Test that Python exceptions with tracebacks appear in query_log when stderr_reaction is configured as throw'''
|
|
skip_test_msan(node)
|
|
|
|
# Clear query log
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Generate a unique query_id for tracking
|
|
query_id = uuid.uuid4().hex
|
|
|
|
# Try to execute UDF that will raise Python exception
|
|
try:
|
|
node.query("SELECT test_function_python_exception_default(1)", query_id=query_id)
|
|
assert False, "Exception should have been thrown"
|
|
except Exception as ex:
|
|
# Verify exception is thrown
|
|
assert "DB::Exception" in str(ex)
|
|
assert "Executable generates stderr" in str(ex)
|
|
|
|
# Flush logs to ensure query_log is updated
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Check query_log for the exception
|
|
# Note: type is 'ExceptionBeforeStart' because exception occurs during prepare(), not during block processing
|
|
result = node.query(f"""
|
|
SELECT exception
|
|
FROM system.query_log
|
|
WHERE query_id = '{query_id}'
|
|
AND type = 'ExceptionBeforeStart'
|
|
FORMAT TabSeparated
|
|
""")
|
|
|
|
# Parse result with TSV to ensure proper formatting
|
|
exception_text = TSV(result).lines[0]
|
|
|
|
# Verify specific exception components are present
|
|
# UDF stderr must contain complete Python traceback
|
|
required_components = [
|
|
"Executable generates stderr: Traceback (most recent call last):",
|
|
"in process_data",
|
|
"result = int(value) / 0",
|
|
"ZeroDivisionError: division by zero",
|
|
]
|
|
|
|
for component in required_components:
|
|
assert component in exception_text, f"Missing required component: {component}"
|
|
|
|
|
|
@pytest.mark.parametrize("func_name", [
|
|
"test_function_stderr_log_last_reaction",
|
|
"test_function_stderr_log_first_reaction",
|
|
"test_function_stderr_none_reaction",
|
|
])
|
|
def test_executable_function_stderr_no_throw_on_success(started_cluster, func_name):
|
|
'''Test that UDFs writing to stderr succeed under log_last/log_first/none when exit code is 0'''
|
|
skip_test_msan(node)
|
|
|
|
assert node.query(f"SELECT {func_name}('abc')") == "Key abc\n"
|
|
|
|
|
|
@pytest.mark.parametrize("func_name,mode", [
|
|
("test_function_python_exception_log_last", "log_last"),
|
|
("test_function_python_exception_log_first", "log_first"),
|
|
])
|
|
def test_executable_function_stderr_in_exception_on_failure(started_cluster, func_name, mode):
|
|
'''Test that stderr content appears in exception when exit code != 0 under log_last/log_first'''
|
|
skip_test_msan(node)
|
|
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
|
|
query_id = uuid.uuid4().hex
|
|
|
|
try:
|
|
node.query(f"SELECT {func_name}(1)", query_id=query_id)
|
|
assert False, "Exception should have been thrown"
|
|
except Exception as ex:
|
|
assert "DB::Exception" in str(ex)
|
|
assert "Child process was exited with return code 1" in str(ex)
|
|
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
|
|
result = node.query(f"""
|
|
SELECT exception
|
|
FROM system.query_log
|
|
WHERE query_id = '{query_id}'
|
|
AND type IN ('ExceptionBeforeStart', 'ExceptionWhileProcessing')
|
|
FORMAT TabSeparated
|
|
""")
|
|
|
|
exception_text = TSV(result).lines[0]
|
|
|
|
required_components = [
|
|
"Stderr:",
|
|
"in process_data",
|
|
"result = int(value) / 0",
|
|
"ZeroDivisionError: division by zero",
|
|
]
|
|
|
|
for component in required_components:
|
|
assert component in exception_text, f"Missing required component in {mode}: {component}"
|