hanzoai/datastore
2
0
Fork 0
mirror of https://github.com/hanzoai/datastore.git synced 2026-09-26 13:55:07 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
datastore/tests/integration/test_storage_postgresql/test.py
T
Alexey MilovidovandGitHub 286e23b1ea Merge pull request #110569 from ClickHouse/ch-fix-62544 
Add regression tests for reading PostgreSQL boolean arrays
2026-07-15 22:36:09 +00:00

1332 lines
55 KiB
Python
Raw Permalink Blame History

import logging
import time
from multiprocessing.dummy import Pool
import pytest
from helpers.cluster import ClickHouseCluster
from helpers.config_cluster import pg_pass
from helpers.postgres_utility import get_postgres_conn
cluster = ClickHouseCluster(__file__)
node1 = cluster.add_instance(
"node1",
main_configs=["configs/named_collections.xml"],
user_configs=["configs/users.xml"],
with_postgres=True,
)
node2 = cluster.add_instance(
"node2",
main_configs=["configs/named_collections.xml"],
user_configs=["configs/settings.xml", "configs/users.xml"],
with_postgres_cluster=True,
)
@pytest.fixture(scope="module")
def started_cluster():
try:
cluster.start()
node1.query("CREATE DATABASE test")
node2.query("CREATE DATABASE test")
# Wait for the PostgreSQL handler to start.
# cluster.start waits until port 9000 becomes accessible.
# Server opens the PostgreSQL compatibility port a bit later.
node1.wait_for_log_line("PostgreSQL compatibility protocol")
yield cluster
finally:
cluster.shutdown()
@pytest.fixture(autouse=True)
def setup_teardown():
print("PostgreSQL is available - running test")
yield # run test
node1.query("DROP DATABASE test")
node1.query("CREATE DATABASE test")
def test_postgres_select_insert(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
table_name = "test_many"
table = f"""postgresql('{started_cluster.postgres_ip}:{started_cluster.postgres_port}', 'postgres', '{table_name}', 'postgres', '{pg_pass}')"""
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
cursor.execute(f"CREATE TABLE {table_name} (a integer, b text, c integer)")
node1.query(
f"""
INSERT INTO TABLE FUNCTION {table}
SELECT number, concat('name_', toString(number)), 3 from numbers(10000)"""
)
check1 = f"SELECT count() FROM {table}"
check2 = f"SELECT Sum(c) FROM {table}"
check3 = f"SELECT count(c) FROM {table} WHERE a % 2 == 0"
check4 = f"SELECT count() FROM {table} WHERE b LIKE concat('name_', toString(1))"
assert (node1.query(check1)).rstrip() == "10000"
assert (node1.query(check2)).rstrip() == "30000"
assert (node1.query(check3)).rstrip() == "5000"
assert (node1.query(check4)).rstrip() == "1"
# Triggers issue https://github.com/ClickHouse/ClickHouse/issues/26088
# for i in range(1, 1000):
# assert (node1.query(check1)).rstrip() == '10000', f"Failed on {i}"
node1.query(
f"""
INSERT INTO TABLE FUNCTION {table}
SELECT number, concat('name_', toString(number)), 3 from numbers(1000000)"""
)
check1 = f"SELECT count() FROM {table}"
check2 = f"SELECT count() FROM (SELECT * FROM {table} LIMIT 10)"
assert (node1.query(check1)).rstrip() == "1010000"
assert (node1.query(check2)).rstrip() == "10"
cursor.execute(f"DROP TABLE {table_name} ")
def test_postgres_addresses_expr(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
table_name = "test_table"
table = """postgresql(`postgres5`)"""
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
cursor.execute(f"CREATE TABLE {table_name} (a integer, b text, c integer)")
node1.query(
f"""
INSERT INTO TABLE FUNCTION {table}
SELECT number, concat('name_', toString(number)), 3 from numbers(10000)"""
)
check1 = f"SELECT count() FROM {table}"
check2 = f"SELECT Sum(c) FROM {table}"
check3 = f"SELECT count(c) FROM {table} WHERE a % 2 == 0"
check4 = f"SELECT count() FROM {table} WHERE b LIKE concat('name_', toString(1))"
assert (node1.query(check1)).rstrip() == "10000"
assert (node1.query(check2)).rstrip() == "30000"
assert (node1.query(check3)).rstrip() == "5000"
assert (node1.query(check4)).rstrip() == "1"
cursor.execute(f"DROP TABLE {table_name} ")
def test_postgres_conversions(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_types")
cursor.execute("DROP TABLE IF EXISTS test_array_dimensions")
cursor.execute(
"""CREATE TABLE test_types (
a smallint, b integer, c bigint, d real, e double precision, f serial, g bigserial,
h timestamp, i date, j decimal(5, 3), k numeric, l boolean, "M" integer)"""
)
node1.query(
f"""
INSERT INTO TABLE FUNCTION postgresql('postgres1:5432', 'postgres', 'test_types', 'postgres', '{pg_pass}') VALUES
(-32768, -2147483648, -9223372036854775808, 1.12345, 1.1234567890, 2147483647, 9223372036854775807, '2000-05-12 12:12:12.012345', '2000-05-12', 22.222, 22.222, 1, 42)"""
)
result = node1.query(
f"""
SELECT a, b, c, d, e, f, g, h, i, j, toDecimal128(k, 3), l, "M" FROM postgresql('postgres1:5432', 'postgres', 'test_types', 'postgres', '{pg_pass}')"""
)
assert (
result
== "-32768\t-2147483648\t-9223372036854775808\t1.12345\t1.123456789\t2147483647\t9223372036854775807\t2000-05-12 12:12:12.012345\t2000-05-12\t22.222\t22.222\t1\t42\n"
)
cursor.execute(
"INSERT INTO test_types (l) VALUES (TRUE), (true), ('yes'), ('y'), ('1');"
)
cursor.execute(
"INSERT INTO test_types (l) VALUES (FALSE), (false), ('no'), ('off'), ('0');"
)
expected = "1\n1\n1\n1\n1\n1\n0\n0\n0\n0\n0\n"
result = node1.query(
f"""SELECT l FROM postgresql('postgres1:5432', 'postgres', 'test_types', 'postgres', '{pg_pass}')"""
)
assert result == expected
cursor.execute(
"""CREATE TABLE IF NOT EXISTS test_array_dimensions
(
a Date[] NOT NULL, -- Date32
b Timestamp[] NOT NULL, -- DateTime64(6)
c real[][] NOT NULL, -- Float32
d double precision[][] NOT NULL, -- Float64
e decimal(5, 5)[][][] NOT NULL, -- Decimal32
f integer[][][] NOT NULL, -- Int32
g Text[][][][][] NOT NULL, -- String
h Integer[][][], -- Nullable(Int32)
i Char(2)[][][][], -- Nullable(String)
j Char(2)[], -- Nullable(String)
k UUID[], -- Nullable(UUID)
l UUID[][], -- Nullable(UUID)
"M" integer[] NOT NULL -- Int32 (mixed-case identifier)
)"""
)
result = node1.query(
f"""
DESCRIBE TABLE postgresql('postgres1:5432', 'postgres', 'test_array_dimensions', 'postgres', '{pg_pass}')"""
)
expected = (
"a\tArray(Date32)\t\t\t\t\t\n"
"b\tArray(DateTime64(6))\t\t\t\t\t\n"
"c\tArray(Array(Float32))\t\t\t\t\t\n"
"d\tArray(Array(Float64))\t\t\t\t\t\n"
"e\tArray(Array(Array(Decimal(5, 5))))\t\t\t\t\t\n"
"f\tArray(Array(Array(Int32)))\t\t\t\t\t\n"
"g\tArray(Array(Array(Array(Array(String)))))\t\t\t\t\t\n"
"h\tArray(Array(Array(Nullable(Int32))))\t\t\t\t\t\n"
"i\tArray(Array(Array(Array(Nullable(String)))))\t\t\t\t\t\n"
"j\tArray(Nullable(String))\t\t\t\t\t\n"
"k\tArray(Nullable(UUID))\t\t\t\t\t\n"
"l\tArray(Array(Nullable(UUID)))\t\t\t\t\t\n"
"M\tArray(Int32)"
""
)
assert result.rstrip() == expected
node1.query(
f"INSERT INTO TABLE FUNCTION postgresql('postgres1:5432', 'postgres', 'test_array_dimensions', 'postgres', '{pg_pass}') "
"VALUES ("
"['2000-05-12', '2000-05-12'], "
"['2000-05-12 12:12:12.012345', '2000-05-12 12:12:12.012345'], "
"[[1.12345], [1.12345], [1.12345]], "
"[[1.1234567891], [1.1234567891], [1.1234567891]], "
"[[[0.11111, 0.11111]], [[0.22222, 0.22222]], [[0.33333, 0.33333]]], "
"[[[1, 1], [1, 1]], [[3, 3], [3, 3]], [[4, 4], [5, 5]]], "
"[[[[['winx', 'winx', 'winx']]]]], "
"[[[1, NULL], [NULL, 1]], [[NULL, NULL], [NULL, NULL]], [[4, 4], [5, 5]]], "
"[[[[NULL]]]], "
"[], "
"['2a0c0bfc-4fec-4e32-ae3a-7fc8eea6626a', '42209d53-d641-4d73-a8b6-c038db1e75d6', NULL], "
"[[NULL, '42209d53-d641-4d73-a8b6-c038db1e75d6'], ['2a0c0bfc-4fec-4e32-ae3a-7fc8eea6626a', NULL], [NULL, NULL]],"
"[42, 42, 42]"
")"
)
result = node1.query(
f"""
SELECT * FROM postgresql('postgres1:5432', 'postgres', 'test_array_dimensions', 'postgres', '{pg_pass}')"""
)
expected = (
"['2000-05-12','2000-05-12']\t"
"['2000-05-12 12:12:12.012345','2000-05-12 12:12:12.012345']\t"
"[[1.12345],[1.12345],[1.12345]]\t"
"[[1.1234567891],[1.1234567891],[1.1234567891]]\t"
"[[[0.11111,0.11111]],[[0.22222,0.22222]],[[0.33333,0.33333]]]\t"
"[[[1,1],[1,1]],[[3,3],[3,3]],[[4,4],[5,5]]]\t"
"[[[[['winx','winx','winx']]]]]\t"
"[[[1,NULL],[NULL,1]],[[NULL,NULL],[NULL,NULL]],[[4,4],[5,5]]]\t"
"[[[[NULL]]]]\t"
"[]\t"
"['2a0c0bfc-4fec-4e32-ae3a-7fc8eea6626a','42209d53-d641-4d73-a8b6-c038db1e75d6',NULL]\t"
"[[NULL,'42209d53-d641-4d73-a8b6-c038db1e75d6'],['2a0c0bfc-4fec-4e32-ae3a-7fc8eea6626a',NULL],[NULL,NULL]]\t"
"[42,42,42]\n"
)
assert result == expected
cursor.execute("DROP TABLE test_types")
cursor.execute("DROP TABLE test_array_dimensions")
def test_postgres_array_ndim_error_messges(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
# cleanup
cursor.execute("DROP VIEW IF EXISTS array_ndim_view;")
cursor.execute("DROP TABLE IF EXISTS array_ndim_table;")
# setup
cursor.execute(
'CREATE TABLE array_ndim_table (x INTEGER, "Mixed-case with spaces" INTEGER[]);'
)
cursor.execute("CREATE VIEW array_ndim_view AS SELECT * FROM array_ndim_table;")
describe_table = f"""
DESCRIBE TABLE postgresql(
'postgres1:5432', 'postgres', 'array_ndim_view',
'postgres', '{pg_pass}'
)
"""
# View with array column cannot be empty. Should throw a useful error message.
# (Cannot infer array dimension.)
try:
node1.query(describe_table)
assert False
except Exception as error:
assert (
"PostgreSQL relation containing arrays cannot be empty: array_ndim_view"
in str(error)
)
# View cannot have empty array. Should throw useful error message.
# (Cannot infer array dimension.)
cursor.execute("TRUNCATE array_ndim_table;")
cursor.execute("INSERT INTO array_ndim_table VALUES (1234, '{}');")
try:
node1.query(describe_table)
assert False
except Exception as error:
assert (
'PostgreSQL cannot infer dimensions of an empty array: array_ndim_view."Mixed-case with spaces". Make sure no empty array values in the first row'
in str(error)
)
# View cannot have NULL array value. Should throw useful error message.
cursor.execute("TRUNCATE array_ndim_table;")
cursor.execute("INSERT INTO array_ndim_table VALUES (1234, NULL);")
try:
node1.query(describe_table)
assert False
except Exception as error:
assert (
'PostgreSQL array cannot be NULL: array_ndim_view."Mixed-case with spaces"'
in str(error)
)
# cleanup
cursor.execute("DROP VIEW IF EXISTS array_ndim_view;")
cursor.execute("DROP TABLE IF EXISTS array_ndim_table;")
def test_non_default_schema(started_cluster):
node1.query("DROP TABLE IF EXISTS test_pg_table_schema")
node1.query("DROP TABLE IF EXISTS test_pg_table_schema_with_dots")
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP SCHEMA IF EXISTS test_schema CASCADE")
cursor.execute('DROP SCHEMA IF EXISTS "test.nice.schema" CASCADE')
cursor.execute("CREATE SCHEMA test_schema")
cursor.execute("CREATE TABLE test_schema.test_table (a integer)")
cursor.execute(
"INSERT INTO test_schema.test_table SELECT i FROM generate_series(0, 99) as t(i)"
)
node1.query(
f"""
CREATE TABLE test.test_pg_table_schema (a UInt32)
ENGINE PostgreSQL('postgres1:5432', 'postgres', 'test_table', 'postgres', '{pg_pass}', 'test_schema');
"""
)
result = node1.query("SELECT * FROM test.test_pg_table_schema")
expected = node1.query("SELECT number FROM numbers(100)")
assert result == expected
parameters = f"'postgres1:5432', 'postgres', 'test_table', 'postgres', '{pg_pass}', 'test_schema'"
table_function = f"postgresql({parameters})"
table_engine = f"PostgreSQL({parameters})"
result = node1.query(f"SELECT * FROM {table_function}")
assert result == expected
cursor.execute('''CREATE SCHEMA "test.nice.schema"''')
cursor.execute("""CREATE TABLE "test.nice.schema"."test.nice.table" (a integer)""")
cursor.execute(
'INSERT INTO "test.nice.schema"."test.nice.table" SELECT i FROM generate_series(0, 99) as t(i)'
)
node1.query(
f"""
CREATE TABLE test.test_pg_table_schema_with_dots (a UInt32)
ENGINE PostgreSQL('postgres1:5432', 'postgres', 'test.nice.table', 'postgres', '{pg_pass}', 'test.nice.schema');
"""
)
result = node1.query("SELECT * FROM test.test_pg_table_schema_with_dots")
assert result == expected
cursor.execute(
'INSERT INTO "test_schema"."test_table" SELECT i FROM generate_series(100, 199) as t(i)'
)
result = node1.query(f"SELECT * FROM {table_function}")
expected = node1.query("SELECT number FROM numbers(200)")
assert result == expected
node1.query(f"CREATE TABLE test.test_pg_auto_schema_engine ENGINE={table_engine}")
node1.query(f"CREATE TABLE test.test_pg_auto_schema_function AS {table_function}")
expected = "a\tNullable(Int32)\t\t\t\t\t\n"
assert node1.query("DESCRIBE TABLE test.test_pg_auto_schema_engine") == expected
assert node1.query("DESCRIBE TABLE test.test_pg_auto_schema_function") == expected
cursor.execute("DROP SCHEMA test_schema CASCADE")
cursor.execute('DROP SCHEMA "test.nice.schema" CASCADE')
node1.query("DROP TABLE test.test_pg_table_schema")
node1.query("DROP TABLE test.test_pg_table_schema_with_dots")
node1.query("DROP TABLE test.test_pg_auto_schema_engine")
node1.query("DROP TABLE test.test_pg_auto_schema_function")
def test_concurrent_queries(started_cluster):
conn = get_postgres_conn(
started_cluster.postgres_ip, started_cluster.postgres_port, database=False
)
cursor = conn.cursor()
database_name = "concurrent_test"
cursor.execute(f"DROP DATABASE IF EXISTS {database_name}")
cursor.execute(f"CREATE DATABASE {database_name}")
conn = get_postgres_conn(
started_cluster.postgres_ip,
started_cluster.postgres_port,
database=True,
database_name=database_name,
)
cursor = conn.cursor()
cursor.execute("CREATE TABLE test_table (key integer, value integer)")
node1.query(
f"""
CREATE TABLE test.test_table (key UInt32, value UInt32)
ENGINE = PostgreSQL(postgres1, database='{database_name}', table='test_table')
"""
)
node1.query(
f"""
CREATE TABLE test.stat (numbackends UInt32, datname String)
ENGINE = PostgreSQL(postgres1, database='{database_name}', table='pg_stat_database')
"""
)
def node_select(_):
for i in range(5):
node1.query("SELECT * FROM test.test_table", user="default")
def node_insert(_):
for i in range(5):
node1.query(
"INSERT INTO test.test_table SELECT number, number FROM numbers(100)",
user="default",
)
def node_insert_select(_):
for i in range(5):
node1.query(
"INSERT INTO test.test_table SELECT number, number FROM numbers(100)",
user="default",
)
node1.query(
"SELECT * FROM test.test_table LIMIT 100", user="default"
)
busy_pool = Pool(30)
p = busy_pool.map_async(node_select, range(30))
p.wait()
count = int(
node1.query(
f"SELECT numbackends FROM test.stat WHERE datname = '{database_name}'"
)
)
print(count)
assert count <= 18 # 16 for test.test_table + 1 for conn + 1 for test.stat
busy_pool = Pool(30)
p = busy_pool.map_async(node_insert, range(30))
p.wait()
count = int(
node1.query(
f"SELECT numbackends FROM test.stat WHERE datname = '{database_name}'"
)
)
print(count)
assert count <= 19 # 16 for test.test_table + 1 for conn + at most 2 for test.stat
busy_pool = Pool(30)
p = busy_pool.map_async(node_insert_select, range(30))
p.wait()
count = int(
node1.query(
f"SELECT numbackends FROM test.stat WHERE datname = '{database_name}'"
)
)
print(count)
assert count <= 20 # 16 for test.test_table + 1 for conn + at most 3 for test.stat
node1.query("DROP TABLE test.test_table;")
node1.query("DROP TABLE test.stat;")
def test_datetime_with_timezone(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_timezone")
node1.query("DROP TABLE IF EXISTS test.test_timezone")
cursor.execute(
"CREATE TABLE test_timezone (ts timestamp without time zone, ts_z timestamp with time zone)"
)
cursor.execute(
"insert into test_timezone select '2014-04-04 20:00:00', '2014-04-04 20:00:00'::timestamptz at time zone 'America/New_York';"
)
cursor.execute("select * from test_timezone")
result = cursor.fetchall()[0]
logging.debug(f"{result[0]}, {str(result[1])[:-6]}")
node1.query(
f"create table test.test_timezone ( ts DateTime, ts_z DateTime('America/New_York')) ENGINE PostgreSQL('postgres1:5432', 'postgres', 'test_timezone', 'postgres', '{pg_pass}');"
)
assert node1.query("select ts from test.test_timezone").strip() == str(result[0])
# [:-6] because 2014-04-04 16:00:00+00:00 -> 2014-04-04 16:00:00
assert (
node1.query("select ts_z from test.test_timezone").strip()
== str(result[1])[:-6]
)
assert (
node1.query("select * from test.test_timezone")
== "2014-04-04 20:00:00\t2014-04-04 16:00:00\n"
)
cursor.execute("DROP TABLE test_timezone")
node1.query("DROP TABLE test.test_timezone")
def test_postgres_ndim(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS arr1, arr2")
cursor.execute("CREATE TABLE arr1 (a Integer[])")
cursor.execute("INSERT INTO arr1 SELECT '{{1}, {2}}'")
# The point is in creating a table via 'as select *', in postgres att_ndim will not be correct in this case.
cursor.execute("CREATE TABLE arr2 AS SELECT * FROM arr1")
cursor.execute(
"SELECT attndims AS dims FROM pg_attribute WHERE attrelid = 'arr2'::regclass; "
)
result = cursor.fetchall()[0]
assert int(result[0]) == 0
result = node1.query(
f"""SELECT toTypeName(a) FROM postgresql('postgres1:5432', 'postgres', 'arr2', 'postgres', '{pg_pass}')"""
)
assert result.strip() == "Array(Array(Nullable(Int32)))"
cursor.execute("DROP TABLE arr1, arr2")
cursor.execute("DROP SCHEMA IF EXISTS ndim_schema CASCADE")
cursor.execute("CREATE SCHEMA ndim_schema")
cursor.execute("CREATE TABLE ndim_schema.arr1 (a integer[])")
cursor.execute("INSERT INTO ndim_schema.arr1 SELECT '{{1}, {2}}'")
# The point is in creating a table via 'as select *', in postgres att_ndim will not be correct in this case.
cursor.execute("CREATE TABLE ndim_schema.arr2 AS SELECT * FROM ndim_schema.arr1")
result = node1.query(
"""SELECT toTypeName(a) FROM postgresql(postgres1, schema='ndim_schema', table='arr2')"""
)
assert result.strip() == "Array(Array(Nullable(Int32)))"
cursor.execute("DROP TABLE ndim_schema.arr1, ndim_schema.arr2")
cursor.execute("DROP SCHEMA ndim_schema CASCADE")
def test_postgres_on_conflict(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
table = "test_conflict"
cursor.execute(f"DROP TABLE IF EXISTS {table}")
cursor.execute(f"CREATE TABLE {table} (a integer PRIMARY KEY, b text, c integer)")
node1.query(
f"""
CREATE TABLE test.test_conflict (a UInt32, b String, c Int32)
ENGINE PostgreSQL('postgres1:5432', 'postgres', 'test_conflict', 'postgres', '{pg_pass}', '', 'ON CONFLICT DO NOTHING');
"""
)
node1.query(
f""" INSERT INTO test.{table} SELECT number, concat('name_', toString(number)), 3 from numbers(100)"""
)
node1.query(
f""" INSERT INTO test.{table} SELECT number, concat('name_', toString(number)), 4 from numbers(100)"""
)
check1 = f"SELECT count() FROM test.{table}"
assert (node1.query(check1)).rstrip() == "100"
table_func = f"""postgresql('{started_cluster.postgres_ip}:{started_cluster.postgres_port}', 'postgres', '{table}', 'postgres', '{pg_pass}', '', 'ON CONFLICT DO NOTHING')"""
node1.query(
f"""INSERT INTO TABLE FUNCTION {table_func} SELECT number, concat('name_', toString(number)), 3 from numbers(100)"""
)
node1.query(
f"""INSERT INTO TABLE FUNCTION {table_func} SELECT number, concat('name_', toString(number)), 3 from numbers(100)"""
)
check1 = f"SELECT count() FROM test.{table}"
assert (node1.query(check1)).rstrip() == "100"
cursor.execute(f"DROP TABLE {table} ")
def test_predefined_connection_configuration(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_table")
cursor.execute("CREATE TABLE test_table (a integer PRIMARY KEY, b integer)")
node1.query(
"""
DROP TABLE IF EXISTS test.test_table;
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres1);
"""
)
node1.query(
""" INSERT INTO test.test_table SELECT number, number from numbers(100)"""
)
assert node1.query("SELECT count() FROM test.test_table").rstrip() == "100"
node1.query(
"""
DROP TABLE test.test_table;
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres1, on_conflict='ON CONFLICT DO NOTHING');
"""
)
node1.query(
""" INSERT INTO test.test_table SELECT number, number from numbers(100)"""
)
node1.query(
""" INSERT INTO test.test_table SELECT number, number from numbers(100)"""
)
assert node1.query("SELECT count() FROM test.test_table").rstrip() == "100"
node1.query("DROP TABLE test.test_table;")
node1.query_and_get_error(
"""
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres1, 'ON CONFLICT DO NOTHING');
"""
)
node1.query_and_get_error(
"""
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres2);
"""
)
node1.query_and_get_error(
"""
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(unknown_collection);
"""
)
node1.query(
"""
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres1, port=5432, database='postgres', table='test_table');
"""
)
assert node1.query("SELECT count() FROM test.test_table").rstrip() == "100"
node1.query(
"""
DROP TABLE test.test_table;
CREATE TABLE test.test_table (a UInt32, b Int32)
ENGINE PostgreSQL(postgres3, port=5432);
"""
)
assert node1.query("SELECT count() FROM test.test_table").rstrip() == "100"
assert node1.query("SELECT count() FROM postgresql(postgres1)").rstrip() == "100"
node1.query(
"INSERT INTO TABLE FUNCTION postgresql(postgres1, on_conflict='ON CONFLICT DO NOTHING') SELECT number, number from numbers(100)"
)
assert node1.query("SELECT count() FROM postgresql(postgres1)").rstrip() == "100"
cursor.execute("DROP SCHEMA IF EXISTS test_schema CASCADE")
cursor.execute("CREATE SCHEMA test_schema")
cursor.execute("CREATE TABLE test_schema.test_table (a integer)")
node1.query(
"INSERT INTO TABLE FUNCTION postgresql(postgres1, schema='test_schema', on_conflict='ON CONFLICT DO NOTHING') SELECT number from numbers(200)"
)
assert (
node1.query(
"SELECT count() FROM postgresql(postgres1, schema='test_schema')"
).rstrip()
== "200"
)
cursor.execute("DROP SCHEMA test_schema CASCADE")
cursor.execute("DROP TABLE test_table ")
def test_where_false(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test")
cursor.execute("CREATE TABLE test (a Integer)")
cursor.execute("INSERT INTO test SELECT 1")
result = node1.query(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', 'test', 'postgres', '{pg_pass}') WHERE 1=0"
)
assert int(result) == 0
result = node1.query(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', 'test', 'postgres', '{pg_pass}') WHERE 0"
)
assert int(result) == 0
result = node1.query(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', 'test', 'postgres', '{pg_pass}') WHERE 1=1"
)
assert int(result) == 1
cursor.execute("DROP TABLE test")
def test_datetime64(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("drop table if exists test")
cursor.execute("create table test (ts timestamp)")
cursor.execute("insert into test select '1960-01-01 20:00:00';")
result = node1.query("select * from postgresql(postgres1, table='test')")
assert result.strip() == "1960-01-01 20:00:00.000000"
def test_uuid(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("drop table if exists test")
cursor.execute("create table test (u uuid)")
cursor.execute("""CREATE EXTENSION IF NOT EXISTS "uuid-ossp";""")
cursor.execute("insert into test select uuid_generate_v1();")
result = node1.query(
"select toTypeName(u) from postgresql(postgres1, table='test')"
)
assert result.strip() == "Nullable(UUID)"
def test_auto_close_connection(started_cluster):
conn = get_postgres_conn(
started_cluster.postgres_ip, started_cluster.postgres_port, database=False
)
cursor = conn.cursor()
database_name = "auto_close_connection_test"
cursor.execute(f"DROP DATABASE IF EXISTS {database_name}")
cursor.execute(f"CREATE DATABASE {database_name}")
conn = get_postgres_conn(
started_cluster.postgres_ip,
started_cluster.postgres_port,
database=True,
database_name=database_name,
)
cursor = conn.cursor()
cursor.execute("CREATE TABLE test_table (key integer, value integer)")
node2.query(
f"""
CREATE TABLE test.test_table (key UInt32, value UInt32)
ENGINE = PostgreSQL(postgres1, database='{database_name}', table='test_table')
"""
)
node2.query(
"INSERT INTO test.test_table SELECT number, number FROM numbers(1000)",
user="default",
)
node2.query("SELECT * FROM test.test_table LIMIT 100", user="default")
node2.query(
f"""
CREATE TABLE test.stat (numbackends UInt32, datname String)
ENGINE = PostgreSQL(postgres1, database='{database_name}', table='pg_stat_database')
"""
)
# Wait for auto-close to take effect (connections may still be closing under TSAN)
for _ in range(20):
count = int(
node2.query(
f"SELECT numbackends FROM test.stat WHERE datname = '{database_name}'"
)
)
if count <= 2:
break
time.sleep(0.5)
# Connection from python + pg_stat table also has a connection at the moment of current query
assert count == 2
node2.query("DROP TABLE test.stat")
node2.query("DROP TABLE test.test_table")
def test_literal_escaping(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS escaping")
cursor.execute("CREATE TABLE escaping(text varchar(255))")
node1.query(
f"CREATE TABLE default.escaping (text String) ENGINE = PostgreSQL('postgres1:5432', 'postgres', 'escaping', 'postgres', '{pg_pass}')"
)
node1.query("SELECT * FROM escaping WHERE text = ''''") # ' -> ''
node1.query("SELECT * FROM escaping WHERE text = '\\''") # ' -> ''
node1.query("SELECT * FROM escaping WHERE text = '\\\\\\''") # \' -> \''
node1.query("SELECT * FROM escaping WHERE text = '\\\\\\''") # \' -> \''
node1.query("SELECT * FROM escaping WHERE text like '%a''a%'") # %a'a% -> %a''a%
node1.query("SELECT * FROM escaping WHERE text like '%a\\'a%'") # %a'a% -> %a''a%
cursor.execute("DROP TABLE escaping")
node1.query("DROP TABLE default.escaping")
def test_filter_pushdown(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("CREATE SCHEMA test_filter_pushdown")
cursor.execute(
"CREATE TABLE test_filter_pushdown.test_table (id integer, value integer)"
)
cursor.execute(
"INSERT INTO test_filter_pushdown.test_table VALUES (1, 10), (1, 110), (2, 0), (3, 33), (4, 0)"
)
node1.query("DROP TABLE IF EXISTS test_filter_pushdown_pg_table")
node1.query(
f"""
CREATE TABLE test_filter_pushdown_pg_table (id UInt32, value UInt32)
ENGINE PostgreSQL('postgres1:5432', 'postgres', 'test_table', 'postgres', '{pg_pass}', 'test_filter_pushdown');
"""
)
node1.query("DROP TABLE IF EXISTS test_filter_pushdown_local_table")
node1.query(
"""
CREATE TABLE test_filter_pushdown_local_table (id UInt32, value UInt32) ENGINE Memory AS SELECT * FROM test_filter_pushdown_pg_table
"""
)
node1.query("DROP TABLE IF EXISTS ch_table")
node1.query(
"CREATE TABLE ch_table (id UInt32, pg_id UInt32) ENGINE MergeTree ORDER BY id"
)
node1.query("INSERT INTO ch_table VALUES (1, 1), (2, 2), (3, 1), (4, 2), (5, 999)")
def compare_results(query, **kwargs):
result1 = node1.query(
query.format(pg_table="test_filter_pushdown_pg_table", **kwargs)
)
result2 = node1.query(
query.format(pg_table="test_filter_pushdown_local_table", **kwargs)
)
assert result1 == result2
for kind in ["INNER", "LEFT", "RIGHT", "FULL", "ANY LEFT", "SEMI RIGHT"]:
for value in [0, 10]:
compare_results(
"SELECT * FROM ch_table {kind} JOIN {pg_table} as p ON ch_table.pg_id = p.id WHERE value = {value} ORDER BY ALL",
kind=kind,
value=value,
)
compare_results(
"SELECT * FROM {pg_table} as p {kind} JOIN ch_table ON ch_table.pg_id = p.id WHERE value = {value} ORDER BY ALL",
kind=kind,
value=value,
)
cursor.execute("DROP SCHEMA test_filter_pushdown CASCADE")
node1.query("DROP TABLE test_filter_pushdown_local_table")
node1.query("DROP TABLE test_filter_pushdown_pg_table")
def test_fixed_string_type(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_fixed_string")
cursor.execute(
"CREATE TABLE test_fixed_string (contact_id numeric NULL, email varchar NULL)"
)
cursor.execute("INSERT INTO test_fixed_string values (1, 'abc')")
node1.query("DROP TABLE IF EXISTS test_fixed_string")
node1.query(
f"CREATE TABLE test_fixed_string(contact_id Int64, email Nullable(FixedString(3))) ENGINE = PostgreSQL('postgres1:5432', 'postgres', 'test_fixed_string', 'postgres', '{pg_pass}')"
)
result = node1.query("SELECT * FROM test_fixed_string format TSV")
assert result.strip() == "1\tabc"
node1.query("DROP TABLE test_fixed_string")
def test_parameters_validation_for_postgresql_function(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
def _create_and_fill_table(table):
cursor.execute(f"DROP TABLE IF EXISTS {table}")
cursor.execute(f"CREATE TABLE {table} (a Integer)")
cursor.execute(f"INSERT INTO {table} SELECT 1")
# Try to do some SQL injection to remove the original table
table = "test_parameters_validation_for_postgresql_function_exception"
_create_and_fill_table(table)
error = node1.query_and_get_error(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', \"whatever')) TO STDOUT; END; DROP TABLE IF EXISTS {table};--\", 'postgres', '{pg_pass}')"
)
assert "PostgreSQL table whatever" in error and "does not exist" in error
result = node1.query(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', '{table}', 'postgres', '{pg_pass}')"
)
assert int(result) == 1
cursor.execute(f"DROP TABLE {table}")
# Check that we can actually work with table names containing single quote
table = "test_parameters_validation_for_postgresql_function_success"
_create_and_fill_table(f'"{table}\'"')
result = node1.query(
f"SELECT count() FROM postgresql('postgres1:5432', 'postgres', '{table}''', 'postgres', '{pg_pass}')"
)
assert int(result) == 1
cursor.execute(f'DROP TABLE "{table}\'"')
def test_postgres_datetime(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_datetime")
cursor.execute("CREATE TABLE test_datetime AS (SELECT '2025-01-02 03:04:05.678900'::timestamptz AS ts, '2025-01-02'::date as d)")
node1.query("DROP TABLE IF EXISTS test_datetime")
node1.query(
f"CREATE TABLE test_datetime (ts DateTime64(6, 'UTC'), d Date) ENGINE = PostgreSQL('postgres1:5432', 'postgres', 'test_datetime', 'postgres', '{pg_pass}')"
)
result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::DateTime")
assert result == "2025-01-02 03:04:05.678900\n"
result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::DateTime64")
assert result == "2025-01-02 03:04:05.678900\n"
result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::Nullable(DateTime)")
assert result == "2025-01-02 03:04:05.678900\n"
result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::Nullable(DateTime64)")
assert result == "2025-01-02 03:04:05.678900\n"
def test_postgres_reading_clone(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_clone")
cursor.execute("CREATE TABLE test_clone AS (SELECT number FROM generate_series(0, 99) AS number)")
node1.query("DROP TABLE IF EXISTS test_clone")
node1.query(
f"CREATE TABLE test_clone ENGINE = PostgreSQL('postgres1:5432', 'postgres', 'test_clone', 'postgres', '{pg_pass}')"
)
result = node1.query(
"SELECT count() FROM (SELECT (SELECT tx.number) = 1 as x FROM test_clone AS tx) WHERE x SETTINGS correlated_subqueries_substitute_equivalent_expressions = 0"
)
assert result.strip() == "1"
def test_postgres_insert_boolean_array(started_cluster):
"""Test for https://github.com/ClickHouse/ClickHouse/issues/72754
Inserting into PostgreSQL BOOLEAN[] was causing a logical error due to
incorrect column type creation for Bool arrays.
"""
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_bool_array")
cursor.execute("CREATE TABLE test_bool_array (id INTEGER, flags BOOLEAN[])")
table_func = f"postgresql('{started_cluster.postgres_ip}:{started_cluster.postgres_port}', 'postgres', 'test_bool_array', 'postgres', '{pg_pass}')"
# Insert boolean arrays using ClickHouse Bool type
node1.query(
f"INSERT INTO TABLE FUNCTION {table_func} VALUES (1, [true, false, true])"
)
node1.query(
f"INSERT INTO TABLE FUNCTION {table_func} SELECT 2, [false, false]"
)
node1.query(
f"INSERT INTO TABLE FUNCTION {table_func} SELECT 3, []"
)
# Verify data was inserted correctly
cursor.execute("SELECT id, flags FROM test_bool_array ORDER BY id")
result = cursor.fetchall()
assert result[0] == (1, [True, False, True])
assert result[1] == (2, [False, False])
assert result[2] == (3, [])
# Verify we can read the data back through ClickHouse
result = node1.query(f"SELECT * FROM {table_func} ORDER BY id")
expected = "1\t[1,0,1]\n2\t[0,0]\n3\t[]\n"
assert result == expected
cursor.execute("DROP TABLE test_bool_array")
def test_postgres_read_boolean_array(started_cluster):
"""Test for https://github.com/ClickHouse/ClickHouse/issues/62544
Reading a PostgreSQL BOOLEAN[] column through the `PostgreSQL` engine used to
fail with `pqxx::conversion_error: Could not convert string to t: 't'` because
the array parser did not understand PostgreSQL's 't'/'f' boolean text format.
The rows below are inserted directly in PostgreSQL (as in the issue), so the
values arrive over the wire exactly as '{t,t,t,f,f,f,t,t}'.
"""
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_bool_read")
cursor.execute(
"CREATE TABLE test_bool_read (id integer, b boolean, booleans boolean[])"
)
cursor.execute(
"INSERT INTO test_bool_read VALUES "
"(1, 't', '{t,t,t,f,f,f,t,t}'), "
"(2, 'f', '{f,f}'), "
"(3, NULL, '{t,NULL,f}'), "
"(4, 't', NULL)"
)
table_func = f"postgresql('{started_cluster.postgres_ip}:{started_cluster.postgres_port}', 'postgres', 'test_bool_read', 'postgres', '{pg_pass}')"
# Reading through a `PostgreSQL` engine table with an explicit Array(Bool)
# schema (the exact form from the issue).
node1.query(
f"""CREATE TABLE test.test_bool_read (id Int32, b Bool, booleans Array(Bool))
ENGINE = PostgreSQL('{started_cluster.postgres_ip}:{started_cluster.postgres_port}', 'postgres', 'test_bool_read', 'postgres', '{pg_pass}')"""
)
result = node1.query("SELECT * FROM test.test_bool_read ORDER BY id")
expected = (
"1\ttrue\t[true,true,true,false,false,false,true,true]\n"
"2\tfalse\t[false,false]\n"
"3\tfalse\t[true,false,false]\n"
"4\ttrue\t[]\n"
)
assert result == expected
# Reading through the table function with automatic schema inference maps the
# boolean array to Array(Nullable(UInt8)) and keeps NULL array elements.
result = node1.query(f"SELECT * FROM {table_func} ORDER BY id")
expected = (
"1\t1\t[1,1,1,0,0,0,1,1]\n"
"2\t0\t[0,0]\n"
"3\t\\N\t[1,NULL,0]\n"
"4\t1\t[]\n"
)
assert result == expected
cursor.execute("DROP TABLE test_bool_read")
def test_postgres_date32(started_cluster):
"""Test that PostgreSQL DATE values outside the Date (UInt16) range are correctly read.
This is a regression test for https://github.com/ClickHouse/ClickHouse/issues/73084
PostgreSQL DATE type supports a much wider range than ClickHouse Date (1970-2149).
Large dates like '2276-11-21' must be read correctly using Date32.
"""
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_date32")
cursor.execute("CREATE TABLE test_date32 (d DATE)")
# Insert dates that would overflow with the old Date (UInt16) type
# Date range is ~1970-2149, these dates are beyond that
cursor.execute(
"INSERT INTO test_date32 VALUES ('2276-11-21'), ('2269-07-01'), ('2200-01-01'), ('1950-06-15')"
)
# Read dates back using the postgresql table function
result = node1.query(
f"SELECT d FROM postgresql('postgres1:5432', 'postgres', 'test_date32', 'postgres', '{pg_pass}') ORDER BY d"
)
expected = "1950-06-15\n2200-01-01\n2269-07-01\n2276-11-21\n"
assert result == expected, f"Expected:\n{expected}\nGot:\n{result}"
# Also verify the column type is Date32
result = node1.query(
f"DESCRIBE TABLE postgresql('postgres1:5432', 'postgres', 'test_date32', 'postgres', '{pg_pass}')"
)
assert "Date32" in result, f"Expected Date32 type, got: {result}"
cursor.execute("DROP TABLE test_date32")
def test_postgres_date32_array(started_cluster):
"""Test that PostgreSQL DATE[] arrays with large dates are correctly read as Array(Date32)."""
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_date32_array")
cursor.execute("CREATE TABLE test_date32_array (dates DATE[] NOT NULL)")
# Insert array with dates that would overflow with Date (UInt16)
cursor.execute(
"INSERT INTO test_date32_array VALUES (ARRAY['2276-11-21'::date, '2269-07-01'::date, '1950-06-15'::date])"
)
# Read dates back (array elements are in insertion order)
result = node1.query(
f"SELECT dates FROM postgresql('postgres1:5432', 'postgres', 'test_date32_array', 'postgres', '{pg_pass}')"
)
expected = "['2276-11-21','2269-07-01','1950-06-15']\n"
assert result == expected, f"Expected:\n{expected}\nGot:\n{result}"
# Verify the column type is Array(Date32)
result = node1.query(
f"DESCRIBE TABLE postgresql('postgres1:5432', 'postgres', 'test_date32_array', 'postgres', '{pg_pass}')"
)
assert "Array(Date32)" in result, f"Expected Array(Date32) type, got: {result}"
cursor.execute("DROP TABLE test_date32_array")
def test_postgres_array_parser_dimension_underflow(started_cluster):
"""Regression test for `size_t` underflow in the PostgreSQL array parser.
When `pqxx::array_parser` emits `row_end` while the parser's `dimension`
counter is 0 (for example, an array text starting with `}`), the previous
code decremented `dimension` past 0 — a `size_t` underflow to `SIZE_MAX` —
and then indexed `dimensions[SIZE_MAX]`, which is out-of-bounds. The fix
throws a `BAD_ARGUMENTS` exception in this case instead.
PostgreSQL itself validates array literals at INSERT time, so the bug is
unreachable via a column declared as `boolean[]`/`integer[]` on the
PostgreSQL side. The reproducer below stores the malformed payload in a
PostgreSQL `text` column and declares the same column as `Array(Int32)` on
the ClickHouse side via the `PostgreSQL` table engine. ClickHouse then
dispatches the raw `'}'` value through the `vtArray` branch of
`insertPostgreSQLValue`, which calls `pqxx::array_parser` on it and
reproduces the bug.
"""
cursor = started_cluster.postgres_conn.cursor()
cursor.execute("DROP TABLE IF EXISTS test_array_underflow")
cursor.execute(
"CREATE TABLE test_array_underflow (id integer, payload text)"
)
cursor.execute("INSERT INTO test_array_underflow VALUES (1, '}')")
started_cluster.postgres_conn.commit()
node1.query("DROP TABLE IF EXISTS pg_array_underflow")
node1.query(
f"CREATE TABLE pg_array_underflow (id Int32, payload Array(Int32)) "
f"ENGINE = PostgreSQL("
f"'{started_cluster.postgres_ip}:{started_cluster.postgres_port}', "
f"'postgres', 'test_array_underflow', 'postgres', '{pg_pass}')"
)
error = node1.query_and_get_error("SELECT id, payload FROM pg_array_underflow")
assert "Unexpected array closing bracket" in error, (
f"Expected BAD_ARGUMENTS('Unexpected array closing bracket'), got: {error}"
)
node1.query("DROP TABLE pg_array_underflow")
cursor.execute("DROP TABLE test_array_underflow")
def test_postgres_query_passing(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
host = f"{started_cluster.postgres_ip}:{started_cluster.postgres_port}"
table_name = "test_query_passing"
dim_name = "test_query_passing_dim"
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
cursor.execute(f"DROP TABLE IF EXISTS {dim_name}")
cursor.execute(f"CREATE TABLE {table_name} (a integer, b text, c integer)")
cursor.execute(f"CREATE TABLE {dim_name} (a integer, factor integer)")
# Populate via the plain (table-name) form of the table function.
node1.query(
f"INSERT INTO TABLE FUNCTION postgresql('{host}', 'postgres', '{table_name}', 'postgres', '{pg_pass}') "
"SELECT number, concat('name_', toString(number)), 3 FROM numbers(100)"
)
cursor.execute(f"INSERT INTO {dim_name} VALUES (1, 10), (2, 100)")
# query('...') form: the WHERE is executed on the PostgreSQL side.
q_func = f"postgresql('{host}', 'postgres', query('SELECT * FROM {table_name} WHERE a % 2 = 0'), 'postgres', '{pg_pass}')"
assert node1.query(f"SELECT count() FROM {q_func}").rstrip() == "50"
assert node1.query(f"SELECT sum(c) FROM {q_func}").rstrip() == "150"
# subquery form.
q_subq = f"postgresql('{host}', 'postgres', (SELECT a, b FROM {table_name} WHERE a < 10), 'postgres', '{pg_pass}')"
assert node1.query(f"SELECT count() FROM {q_subq}").rstrip() == "10"
assert node1.query(f"SELECT b FROM {q_subq} ORDER BY a LIMIT 1").rstrip() == "name_0"
# A JOIN passed to PostgreSQL as is.
q_join = (
f"postgresql('{host}', 'postgres', "
f"query('SELECT t.a AS a, t.c * d.factor AS calculated FROM {table_name} AS t JOIN {dim_name} AS d ON t.a = d.a'), "
f"'postgres', '{pg_pass}')"
)
assert node1.query(f"SELECT sum(calculated) FROM {q_join}").rstrip() == str(3 * 10 + 3 * 100)
# Engine with a query: reading works, writing is rejected.
node1.query("DROP TABLE IF EXISTS pg_engine_query")
node1.query(
f"CREATE TABLE pg_engine_query "
f"ENGINE = PostgreSQL('{host}', 'postgres', query('SELECT a, b FROM {table_name} WHERE a < 5'), 'postgres', '{pg_pass}')"
)
assert node1.query("SELECT count() FROM pg_engine_query").rstrip() == "5"
assert "INCORRECT_QUERY" in node1.query_and_get_error(
"INSERT INTO pg_engine_query VALUES (1, 'x')"
)
node1.query("DROP TABLE pg_engine_query")
# The subquery form is re-serialized in the PostgreSQL dialect, so an identifier that needs quoting
# (here, one containing a space) is emitted with PostgreSQL double quotes and accepted by PostgreSQL.
# With ClickHouse-style backtick quoting (the previous behaviour) PostgreSQL would reject the query.
quoted_name = "test_query_passing_quoted"
cursor.execute(f'DROP TABLE IF EXISTS {quoted_name}')
cursor.execute(f'CREATE TABLE {quoted_name} (id integer, "weird name" text)')
cursor.execute(f"INSERT INTO {quoted_name} VALUES (1, 'quoted_value')")
started_cluster.postgres_conn.commit()
q_quoted = (
f'postgresql(\'{host}\', \'postgres\', '
f'(SELECT id, "weird name" FROM {quoted_name}), \'postgres\', \'{pg_pass}\')'
)
assert node1.query(f'SELECT "weird name" FROM {q_quoted} ORDER BY id').rstrip() == "quoted_value"
cursor.execute(f"DROP TABLE {quoted_name}")
# external_table_strict_query: an outer filter that cannot be pushed down into the passed query is
# applied locally by default, but rejected with INCORRECT_QUERY under external_table_strict_query = 1.
q_strict = f"postgresql('{host}', 'postgres', query('SELECT a, b FROM {table_name}'), 'postgres', '{pg_pass}')"
assert node1.query(f"SELECT count() FROM {q_strict} WHERE a = 1").rstrip() == "1"
assert "INCORRECT_QUERY" in node1.query_and_get_error(
f"SELECT count() FROM {q_strict} WHERE a = 1 SETTINGS external_table_strict_query = 1"
)
# INSERT into a query-backed table function is rejected with INCORRECT_QUERY before the storage is
# constructed, so the remote schema-inference query is never run against PostgreSQL.
assert "INCORRECT_QUERY" in node1.query_and_get_error(
f"INSERT INTO TABLE FUNCTION {q_strict} VALUES (1, 'x')"
)
# Schema inference must preserve PostgreSQL type modifiers and array dimensions of the query result,
# exactly as the plain table-name path does. In particular a wide numeric(78, 0) must be inferred as
# Int256 (the typmod must be carried into format_type), and a multidimensional array must keep its
# dimensions instead of being treated as one-dimensional.
wide_name = "test_query_passing_wide"
cursor.execute(f"DROP TABLE IF EXISTS {wide_name}")
cursor.execute(
f"CREATE TABLE {wide_name} (id integer, big numeric(78, 0), arr1 integer[], arr2 integer[][])"
)
big_value = "123456789012345678901234567890123456789012345678901234567890"
cursor.execute(
f"INSERT INTO {wide_name} VALUES (1, {big_value}, '{{1,2,3}}', '{{{{1,2}},{{3,4}}}}')"
)
started_cluster.postgres_conn.commit()
# Pin external_table_functions_use_nulls = 0 so the inferred types are not wrapped into Nullable,
# which keeps the type-name assertions focused on the typmod and dimension inference.
q_wide = (
f"postgresql('{host}', 'postgres', query('SELECT id, big, arr1, arr2 FROM {wide_name}'), "
f"'postgres', '{pg_pass}') SETTINGS external_table_functions_use_nulls = 0"
)
assert (
node1.query(
f"SELECT toTypeName(big), toTypeName(arr1), toTypeName(arr2) FROM {q_wide}"
).rstrip()
== "Int256\tArray(Int32)\tArray(Array(Int32))"
)
assert node1.query(f"SELECT big FROM {q_wide}").rstrip() == big_value
assert node1.query(f"SELECT arr2 FROM {q_wide}").rstrip() == "[[1,2],[3,4]]"
# When the first sampled row of an array column is NULL or an empty array, the number of dimensions
# cannot be inferred. The query path must fail early during schema inference with a clear error (as the
# table-name path does in its recheck_array step) instead of silently inferring a one-dimensional array
# and then failing at read time with the confusing `Got more dimensions than expected`.
empty_arr_name = "test_query_passing_empty_array"
cursor.execute(f"DROP TABLE IF EXISTS {empty_arr_name}")
cursor.execute(f"CREATE TABLE {empty_arr_name} (id integer, arr integer[][])")
cursor.execute(
f"INSERT INTO {empty_arr_name} VALUES (1, '{{}}'), (2, '{{{{1,2}},{{3,4}}}}')"
)
started_cluster.postgres_conn.commit()
q_empty = (
f"postgresql('{host}', 'postgres', "
f"query('SELECT arr FROM {empty_arr_name} ORDER BY id'), 'postgres', '{pg_pass}')"
)
assert "Cannot infer the number of dimensions" in node1.query_and_get_error(
f"SELECT * FROM {q_empty}"
)
cursor.execute(f"DROP TABLE {empty_arr_name}")
cursor.execute(f"DROP TABLE {wide_name}")
cursor.execute(f"DROP TABLE {table_name}")
cursor.execute(f"DROP TABLE {dim_name}")
def test_postgres_query_passing_edge_cases(started_cluster):
cursor = started_cluster.postgres_conn.cursor()
host = f"{started_cluster.postgres_ip}:{started_cluster.postgres_port}"
table_name = "test_query_passing_edge"
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
cursor.execute(f"CREATE TABLE {table_name} (a integer, b text)")
node1.query(
f"INSERT INTO TABLE FUNCTION postgresql('{host}', 'postgres', '{table_name}', 'postgres', '{pg_pass}') "
"SELECT number, concat('name_', toString(number)) FROM numbers(10)"
)
started_cluster.postgres_conn.commit()
# Empty result set: schema inference runs the query with LIMIT 0, so the structure is still resolved
# even though no rows come back. The types are inferred normally and count() is 0.
q_empty = f"postgresql('{host}', 'postgres', query('SELECT a, b FROM {table_name} WHERE a > 1000'), 'postgres', '{pg_pass}')"
described = [
line.split("\t")[:2]
for line in node1.query(
f"DESCRIBE TABLE {q_empty} SETTINGS external_table_functions_use_nulls = 0"
)
.rstrip()
.split("\n")
]
assert described == [["a", "Int32"], ["b", "String"]]
assert node1.query(f"SELECT count() FROM {q_empty}").rstrip() == "0"
# Missing permissions: the passed query is executed on the PostgreSQL side under the supplied role, so a
# role without SELECT on the referenced table fails with PostgreSQL's permission-denied error surfaced by
# ClickHouse (raised during schema inference, before any rows are read).
cursor.execute("DROP ROLE IF EXISTS query_passing_norole")
cursor.execute(
"CREATE ROLE query_passing_norole WITH LOGIN PASSWORD 'norole_pass'"
)
started_cluster.postgres_conn.commit()
q_noperm = f"postgresql('{host}', 'postgres', query('SELECT a, b FROM {table_name}'), 'query_passing_norole', 'norole_pass')"
assert "permission denied" in node1.query_and_get_error(
f"SELECT count() FROM {q_noperm}"
)
cursor.execute("DROP ROLE query_passing_norole")
started_cluster.postgres_conn.commit()
# Type mismatch on the engine path: the user declares an explicit structure that disagrees with the
# actual query result types. The declared structure wins for the engine, and reading a text value into
# the declared Int32 fails with a query error (BAD_ARGUMENTS) instead of aborting the server.
# The same reader is used for every declared type, so a mismatch against any of them must surface as a
# query error, never as a foreign exception (pqxx or std::runtime_error from the text parsers) that
# would abort the server. Column b holds text like 'name_0', so reading it into each declared type below
# fails to parse.
for declared in ["Int32", "UUID", "Date", "DateTime", "DateTime64(6)", "Decimal(10, 2)"]:
node1.query("DROP TABLE IF EXISTS pg_engine_type_mismatch")
node1.query(
f"CREATE TABLE pg_engine_type_mismatch (a Int32, b {declared}) "
f"ENGINE = PostgreSQL('{host}', 'postgres', query('SELECT a, b FROM {table_name}'), 'postgres', '{pg_pass}')"
)
assert "Cannot parse PostgreSQL value" in node1.query_and_get_error(
"SELECT * FROM pg_engine_type_mismatch"
)
node1.query("DROP TABLE pg_engine_type_mismatch")
# Numeric overflow: a PostgreSQL numeric value parses as a number but does not fit the declared
# ClickHouse Decimal. The Decimal text reader rejects it with a parse error, which the shared value
# layer normalizes to BAD_ARGUMENTS, so it surfaces as a query error (and reaches the
# MaterializedPostgreSQL catch) instead of aborting the server.
# Seed the wide value on the PostgreSQL side so it stays an exact numeric. Inserting the same literal
# through ClickHouse would type it as Float64 and fail the INSERT with DECIMAL_OVERFLOW before reaching
# the read path this case is about.
num_table = "test_query_passing_numeric"
cursor.execute(f"DROP TABLE IF EXISTS {num_table}")
cursor.execute(f"CREATE TABLE {num_table} (a integer, v numeric)")
cursor.execute(
f"INSERT INTO {num_table} VALUES (1, 99999999999999999999999999999999999999)"
)
started_cluster.postgres_conn.commit()
node1.query("DROP TABLE IF EXISTS pg_engine_decimal_overflow")
node1.query(
f"CREATE TABLE pg_engine_decimal_overflow (a Int32, v Decimal(10, 2)) "
f"ENGINE = PostgreSQL('{host}', 'postgres', query('SELECT a, v FROM {num_table}'), 'postgres', '{pg_pass}')"
)
assert "Cannot parse PostgreSQL value" in node1.query_and_get_error(
"SELECT * FROM pg_engine_decimal_overflow"
)
node1.query("DROP TABLE pg_engine_decimal_overflow")
cursor.execute(f"DROP TABLE {num_table}")
cursor.execute(f"DROP TABLE {table_name}")
if __name__ == "__main__":
cluster.start()
input("Cluster created, press any key to destroy...")
cluster.shutdown()
Reference in New Issue View Git Blame Copy Permalink