mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2026-10-28 16:18:23 +08:00
Groene AIandClaude Opus 5.5
f8cbc530ec
Versions before 25.12 serialized implicit min-max indices into the table definition (after any ALTER) and into Keeper metadata. The loader analyzed such a stored index as a regular one and then re-marked every index with the reserved name as implicit, so: - once the dictionary, `Join` table or function used by an ALIAS was gone (or, for an executable function, not yet registered at startup), the table could not be loaded, and a replica stayed read-only because `parseAndNormalize` analyzed the stored definition too; - for an ALIAS expression column (`b ALIAS key * 2`), `addImplicitIndicesForColumn` did not recognize the stored index (its `column_names` holds the columns of the expression) and generated it a second time, so loading failed with `Index with name auto_minmax_index_b already exists` even with every object present (measured on 25.8, 26.3, 26.8 and 26.9 with a table written and altered by 25.8); - a stored implicit index over a non-deterministic `joinGet` / `dictGet` ALIAS was kept, although such an index is never generated; - an index with the reserved name but another shape (e.g. `GRANULARITY 8` from a full-definition `ATTACH`) was re-marked implicit, so the next ALTER dropped it from the definition and the next load generated it with granularity 1, after which reading it failed with `Cannot read all marks`. Now a stored index identical to the one the generator builds is skipped without analysis, in `registerStorageMergeTree` (it is generated below exactly as for a table created now) and in `parseAndNormalize` (stripped from Keeper metadata); any other index is a regular index on both sides. The upgrade test creates and alters the table on 25.10; the executable function case attaches the stored form of the definition. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
637 lines
24 KiB
Python
637 lines
24 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 = """<clickhouse>
|
|
<user_defined_executable_functions_config>/etc/clickhouse-server/functions/test_function_config.xml</user_defined_executable_functions_config>
|
|
</clickhouse>"""
|
|
|
|
|
|
@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_determinism_in_distributed_predicate_push_down(
|
|
started_cluster,
|
|
):
|
|
skip_test_msan(node)
|
|
|
|
# `ReadFromRemote` pushes a predicate over a column of a distributed subquery into the `HAVING`
|
|
# of the query sent to the shard, unless `hasNonRewritableFunction` finds a non-deterministic
|
|
# function in the shard's `SELECT` list. `ExpressionInfoVisitor` reads the `deterministic` flag
|
|
# of an executable UDF from its configuration: a parametric UDF must not be instantiated there
|
|
# (an empty parameter list raises `BAD_ARGUMENTS`), and a UDF declared deterministic must still
|
|
# be pushed down.
|
|
node.query("DROP TABLE IF EXISTS test_table_distributed_predicate")
|
|
node.query(
|
|
"CREATE TABLE test_table_distributed_predicate (k UInt64) ENGINE = MergeTree ORDER BY k"
|
|
)
|
|
node.query(
|
|
"INSERT INTO test_table_distributed_predicate SELECT number FROM numbers(4)"
|
|
)
|
|
|
|
settings = {
|
|
"serialize_query_plan": 0,
|
|
"allow_push_predicate_ast_for_distributed_subqueries": 1,
|
|
}
|
|
cases = {
|
|
"test_function_parameter_python(2)(k)": False,
|
|
"test_function_bash_nondeterministic(k)": False,
|
|
"test_function_bash_deterministic(k)": True,
|
|
}
|
|
for expression, pushed_down in cases.items():
|
|
log_comment = "distributed_predicate_" + expression.split("(")[0]
|
|
assert (
|
|
node.query(
|
|
f"SELECT count() FROM (SELECT {expression} AS v"
|
|
" FROM remote('127.0.0.2', default, test_table_distributed_predicate))"
|
|
" WHERE v != ''",
|
|
settings={**settings, "log_comment": log_comment},
|
|
)
|
|
== "4\n"
|
|
)
|
|
node.query("SYSTEM FLUSH LOGS query_log")
|
|
shard_query = node.query(
|
|
"SELECT query FROM system.query_log WHERE type = 'QueryFinish'"
|
|
f" AND is_initial_query = 0 AND log_comment = '{log_comment}'"
|
|
" AND query LIKE '%test_table_distributed_predicate%'"
|
|
)
|
|
assert expression.split("(")[0] in shard_query
|
|
assert ("HAVING" in shard_query) == pushed_down, shard_query
|
|
|
|
node.query("DROP TABLE test_table_distributed_predicate")
|
|
|
|
|
|
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_deterministic_declaration_deduplicates(started_cluster):
|
|
'''A function declared deterministic is entitled to run once per distinct value of a
|
|
LowCardinality argument instead of once per row.'''
|
|
skip_test_msan(node)
|
|
|
|
node.query("DROP TABLE IF EXISTS low_cardinality_argument")
|
|
node.query(
|
|
"CREATE TABLE low_cardinality_argument (v LowCardinality(UInt64)) ENGINE = MergeTree ORDER BY tuple()",
|
|
settings={"allow_suspicious_low_cardinality_types": 1},
|
|
)
|
|
node.query("INSERT INTO low_cardinality_argument SELECT number % 2 FROM numbers(100)")
|
|
|
|
def run(function_name):
|
|
query_id = uuid.uuid4().hex
|
|
# `sum` over the result keeps the call from being pruned as an unused column.
|
|
result = node.query(
|
|
f"SELECT sum(length({function_name}(v))) FROM low_cardinality_argument",
|
|
query_id=query_id,
|
|
)
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
input_bytes = node.query(
|
|
f"""SELECT ProfileEvents['ExecutableUserDefinedFunctionInputBytes']
|
|
FROM system.query_log
|
|
WHERE query_id = '{query_id}' AND type = 'QueryFinish'"""
|
|
)
|
|
return result.strip(), int(input_bytes.strip())
|
|
|
|
deterministic_result, deterministic_bytes = run("test_function_bash_deterministic")
|
|
nondeterministic_result, nondeterministic_bytes = run("test_function_bash_nondeterministic")
|
|
|
|
# The table holds 100 rows over 2 distinct values, and the argument is one line per row on
|
|
# the child's stdin, so the declaration decides how much reaches the child.
|
|
assert nondeterministic_bytes >= 100
|
|
assert deterministic_bytes * 10 < nondeterministic_bytes
|
|
# Deduplicating must not change the answer for a function that is deterministic in fact.
|
|
assert deterministic_result == nondeterministic_result
|
|
|
|
node.query("DROP TABLE low_cardinality_argument")
|
|
|
|
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}"
|
|
|
|
|
|
def test_executable_function_in_alias_with_implicit_index_after_restart(started_cluster):
|
|
skip_test_msan(node)
|
|
node.query("DROP TABLE IF EXISTS t_alias_implicit_index SYNC")
|
|
node.query(
|
|
"CREATE TABLE t_alias_implicit_index (id UInt64, a String ALIAS test_function_bash_deterministic(id)) "
|
|
"ENGINE = MergeTree ORDER BY id SETTINGS add_minmax_index_for_string_columns = 1"
|
|
)
|
|
node.query("INSERT INTO t_alias_implicit_index VALUES (1)")
|
|
assert (
|
|
node.query(
|
|
"SELECT name FROM system.data_skipping_indices WHERE table = 't_alias_implicit_index' ORDER BY name"
|
|
)
|
|
== "auto_minmax_index_a\n"
|
|
)
|
|
|
|
# Versions before 25.12 stored the implicit index in the table definition.
|
|
node.query("DROP TABLE IF EXISTS t_alias_implicit_index_legacy SYNC")
|
|
node.query(
|
|
"ATTACH TABLE t_alias_implicit_index_legacy UUID '6d1f0c3e-5b8a-4c2e-9f47-3a0b8e2d1c75' "
|
|
"(id UInt64, a String ALIAS test_function_bash_deterministic(id), INDEX auto_minmax_index_a a TYPE minmax() GRANULARITY 1) "
|
|
"ENGINE = MergeTree ORDER BY id SETTINGS add_minmax_index_for_string_columns = 1"
|
|
)
|
|
node.query("INSERT INTO t_alias_implicit_index_legacy VALUES (1)")
|
|
|
|
# Executable functions are registered only after the tables are loaded.
|
|
node.replace_config(
|
|
"/etc/clickhouse-server/config.d/async_load_databases.xml",
|
|
"<clickhouse><async_load_databases>false</async_load_databases></clickhouse>",
|
|
)
|
|
node.restart_clickhouse()
|
|
assert node.query("SELECT id, a FROM t_alias_implicit_index") == "1\tKey 1\n"
|
|
assert node.query("SELECT id, a FROM t_alias_implicit_index_legacy") == "1\tKey 1\n"
|
|
node.query("DROP TABLE t_alias_implicit_index SYNC")
|
|
node.query("DROP TABLE t_alias_implicit_index_legacy SYNC")
|