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.
1638 lines
63 KiB
Python
1638 lines
63 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import uuid
|
|
from datetime import datetime
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pyarrow as pa
|
|
import pytest
|
|
import pytz
|
|
from datetime import timedelta
|
|
from pyiceberg.catalog import load_catalog
|
|
from pyiceberg.partitioning import PartitionField, PartitionSpec
|
|
from pyiceberg.schema import Schema
|
|
from pyiceberg.table.sorting import SortField, SortOrder
|
|
from pyiceberg.transforms import DayTransform, IdentityTransform
|
|
from helpers.config_cluster import minio_access_key, minio_secret_key
|
|
import decimal
|
|
from pyiceberg.types import (
|
|
DoubleType,
|
|
NestedField,
|
|
StringType,
|
|
StructType,
|
|
TimestampType,
|
|
TimestamptzType,
|
|
MapType,
|
|
DecimalType,
|
|
)
|
|
|
|
from helpers.cluster import DatastoreCluster
|
|
from helpers.mock_servers import start_mock_servers
|
|
|
|
import boto3
|
|
|
|
|
|
def run_s3_mocks(started_cluster, args=[]):
|
|
script_dir = os.path.join(os.path.dirname(__file__), "s3_mocks")
|
|
start_mock_servers(
|
|
started_cluster,
|
|
script_dir,
|
|
[("mock_sts.py", "sts.us-east-1.amazonaws.com", "80", args)],
|
|
)
|
|
|
|
|
|
CATALOG_NAME = "test"
|
|
|
|
BASE_URL = "http://glue:3000"
|
|
|
|
|
|
def get_glue_local_url(cluster):
|
|
return f"http://localhost:{cluster.glue_catalog_port}"
|
|
|
|
def generate_decimal(precision=9, scale=2):
|
|
max_value = 10**(precision - scale) - 1
|
|
value = random.uniform(0, max_value)
|
|
return round(decimal.Decimal(value), scale)
|
|
|
|
DEFAULT_SCHEMA = Schema(
|
|
NestedField(
|
|
field_id=1, name="datetime", field_type=TimestampType(), required=False
|
|
),
|
|
NestedField(field_id=2, name="symbol", field_type=StringType(), required=False),
|
|
NestedField(field_id=3, name="bid", field_type=DoubleType(), required=False),
|
|
NestedField(field_id=4, name="ask", field_type=DoubleType(), required=False),
|
|
NestedField(
|
|
field_id=5,
|
|
name="details",
|
|
field_type=StructType(
|
|
NestedField(
|
|
field_id=4,
|
|
name="created_by",
|
|
field_type=StringType(),
|
|
required=False,
|
|
),
|
|
),
|
|
required=False,
|
|
),
|
|
NestedField(
|
|
field_id=6,
|
|
name="map_string_decimal",
|
|
field_type=MapType(
|
|
key_type=StringType(),
|
|
value_type=DecimalType(9, 2),
|
|
key_id=7,
|
|
value_id=8,
|
|
value_required=False,
|
|
),
|
|
required=False,
|
|
),
|
|
)
|
|
|
|
DEFAULT_CREATE_TABLE = "CREATE TABLE {}.`{}.{}`\\n(\\n `datetime` Nullable(DateTime64(6)),\\n `symbol` Nullable(String),\\n `bid` Nullable(Float64),\\n `ask` Nullable(Float64),\\n `details` Tuple(created_by Nullable(String)),\\n `map_string_decimal` Map(String, Nullable(Decimal(9, 2)))\\n)\\nENGINE = Iceberg(\\'http://minio1:9001/warehouse-glue/data/\\', \\'minio\\', \\'[HIDDEN]\\')\n"
|
|
|
|
DEFAULT_PARTITION_SPEC = PartitionSpec(
|
|
PartitionField(
|
|
source_id=1, field_id=1000, transform=DayTransform(), name="datetime_day"
|
|
)
|
|
)
|
|
|
|
DEFAULT_SORT_ORDER = SortOrder(SortField(source_id=2, transform=IdentityTransform()))
|
|
|
|
|
|
def list_databases(started_cluster):
|
|
client = boto3.client(
|
|
"glue", region_name="us-east-1", endpoint_url=get_glue_local_url(started_cluster)
|
|
)
|
|
databases = client.get_databases()
|
|
return databases
|
|
|
|
|
|
def load_catalog_impl(started_cluster):
|
|
return load_catalog(
|
|
CATALOG_NAME, # name is not important
|
|
**{
|
|
"type": "glue",
|
|
"glue.endpoint": get_glue_local_url(started_cluster),
|
|
"glue.region": "us-east-1",
|
|
"s3.endpoint": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"s3.access-key-id": minio_access_key,
|
|
"s3.secret-access-key": minio_secret_key,
|
|
},
|
|
)
|
|
|
|
|
|
def create_table(
|
|
catalog,
|
|
namespace,
|
|
table,
|
|
schema=DEFAULT_SCHEMA,
|
|
partition_spec=DEFAULT_PARTITION_SPEC,
|
|
sort_order=DEFAULT_SORT_ORDER,
|
|
dir="data"
|
|
):
|
|
return catalog.create_table(
|
|
identifier=f"{namespace}.{table}",
|
|
schema=schema,
|
|
location=f"s3://warehouse-glue/{dir}",
|
|
partition_spec=partition_spec,
|
|
sort_order=sort_order,
|
|
)
|
|
|
|
|
|
|
|
def generate_arrow_data(num_rows=5):
|
|
datetimes = []
|
|
symbols = []
|
|
bids = []
|
|
asks = []
|
|
details_created_by = []
|
|
map_keys = []
|
|
map_values = []
|
|
|
|
offsets = [0]
|
|
|
|
for _ in range(num_rows):
|
|
datetimes.append(datetime.utcnow() - timedelta(minutes=random.randint(0, 60)))
|
|
symbols.append(random.choice(["AAPL", "GOOG", "MSFT"]))
|
|
bids.append(random.uniform(100, 150))
|
|
asks.append(random.uniform(150, 200))
|
|
details_created_by.append(random.choice(["alice", "bob", "carol"]))
|
|
|
|
# map<string, decimal(9,2)>
|
|
keys = []
|
|
values = []
|
|
for i in range(random.randint(1, 3)):
|
|
keys.append(f"key{i}")
|
|
values.append(generate_decimal())
|
|
map_keys.extend(keys)
|
|
map_values.extend(values)
|
|
offsets.append(offsets[-1] + len(keys))
|
|
|
|
# Struct for 'details'
|
|
struct_array = pa.StructArray.from_arrays(
|
|
[pa.array(details_created_by, type=pa.string())],
|
|
names=["created_by"]
|
|
)
|
|
|
|
# Map array
|
|
map_array = pa.MapArray.from_arrays(
|
|
offsets=pa.array(offsets, type=pa.int32()),
|
|
keys=pa.array(map_keys, type=pa.string()),
|
|
items=pa.array(map_values, type=pa.decimal128(9, 2))
|
|
)
|
|
|
|
# Final table
|
|
table = pa.table({
|
|
"datetime": pa.array(datetimes, type=pa.timestamp("us")),
|
|
"symbol": pa.array(symbols, type=pa.string()),
|
|
"bid": pa.array(bids, type=pa.float64()),
|
|
"ask": pa.array(asks, type=pa.float64()),
|
|
"details": struct_array,
|
|
"map_string_decimal": map_array,
|
|
})
|
|
|
|
return table
|
|
|
|
def create_datastore_glue_database(
|
|
started_cluster, node, name, additional_settings={}, query_settings={}, with_credentials=True
|
|
):
|
|
settings = {
|
|
"catalog_type": "glue",
|
|
"warehouse": "test",
|
|
"storage_endpoint": "http://minio1:9001/warehouse-glue",
|
|
"region": "us-east-1",
|
|
}
|
|
|
|
# The Glue catalog API client is subject to the server-managed credential restriction. Unless the session
|
|
# opts in via `s3_allow_server_credentials_in_user_queries`, the creator must pass explicit catalog
|
|
# credentials instead of falling back to the server's AWS identity (moto accepts any non-empty values).
|
|
if not query_settings.get("s3_allow_server_credentials_in_user_queries"):
|
|
settings["aws_access_key_id"] = minio_access_key
|
|
settings["aws_secret_access_key"] = minio_secret_key
|
|
|
|
settings.update(additional_settings)
|
|
|
|
credential_args = f",'{minio_access_key}', '{minio_secret_key}'" if with_credentials else ""
|
|
|
|
node.query(
|
|
f"""
|
|
DROP DATABASE IF EXISTS {name};
|
|
CREATE DATABASE {name} ENGINE = DataLakeCatalog('{BASE_URL}'{credential_args})
|
|
SETTINGS {",".join((k+"="+repr(v) for k, v in settings.items()))}
|
|
""",
|
|
settings={
|
|
"allow_database_glue_catalog": 1,
|
|
"write_full_path_in_iceberg_metadata": 1,
|
|
**query_settings,
|
|
},
|
|
)
|
|
|
|
def create_datastore_glue_table(
|
|
started_cluster, node, database_name, table_name, schema, additional_settings={}
|
|
):
|
|
settings_suffix = "" if len(additional_settings) == 0 else f"SETTINGS {",".join((k+"="+repr(v) for k, v in additional_settings.items()))}"
|
|
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {CATALOG_NAME}.`{database_name}.{table_name}` {schema} ENGINE = IcebergS3('http://minio1:9001/warehouse-glue/{table_name}/', '{minio_access_key}', '{minio_secret_key}')
|
|
{settings_suffix}
|
|
""",
|
|
settings={
|
|
"allow_experimental_database_glue_catalog": 1,
|
|
"write_full_path_in_iceberg_metadata": 1,
|
|
},
|
|
)
|
|
|
|
show_result = node.query(f"SHOW DATABASE {CATALOG_NAME}")
|
|
assert minio_secret_key not in show_result
|
|
|
|
def create_glue_table_directly(
|
|
started_cluster, database_name, table_name, table_type=None, columns=None
|
|
):
|
|
"""Create a raw (non-Iceberg) table directly in Glue via boto3.
|
|
|
|
Such a table - e.g. a Delta Lake table (table_type != 'ICEBERG') or a raw parquet
|
|
file registered by a crawler (no table_type at all) - is visible in the Glue catalog
|
|
but cannot be read by Datastore, whose Glue engine only supports Iceberg tables.
|
|
"""
|
|
client = boto3.client(
|
|
"glue", region_name="us-east-1", endpoint_url=get_glue_local_url(started_cluster)
|
|
)
|
|
if columns is None:
|
|
columns = [
|
|
{"Name": "id", "Type": "bigint"},
|
|
{"Name": "value", "Type": "string"},
|
|
]
|
|
parameters = {} if table_type is None else {"table_type": table_type}
|
|
client.create_table(
|
|
DatabaseName=database_name,
|
|
TableInput={
|
|
"Name": table_name,
|
|
"StorageDescriptor": {
|
|
"Columns": columns,
|
|
"Location": f"s3://warehouse-glue/{table_name}/",
|
|
},
|
|
"TableType": "EXTERNAL_TABLE",
|
|
"Parameters": parameters,
|
|
},
|
|
)
|
|
|
|
|
|
def test_show_tables_hides_unreadable_tables(started_cluster):
|
|
"""
|
|
Regression test: SHOW TABLES and system.tables must agree for a Glue catalog that
|
|
mixes Iceberg tables with non-Iceberg tables (Delta Lake tables, raw parquet files,
|
|
...). Datastore can only read Iceberg tables via the Glue catalog, so the
|
|
non-Iceberg tables must appear in NEITHER listing.
|
|
|
|
Before the fix, the lightweight name-only listing path (used by SHOW TABLES and by
|
|
SELECT name FROM system.tables) did not filter unreadable tables, while the full
|
|
system.tables path dropped them - so the two disagreed and SHOW TABLES listed tables
|
|
that could not be queried. This test fails on master and passes with the fix.
|
|
"""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_hide_unreadable_{uuid.uuid4().hex[:8]}"
|
|
namespace = f"{test_ref}_ns"
|
|
iceberg_table = f"{test_ref}_iceberg"
|
|
delta_table = f"{test_ref}_delta"
|
|
parquet_table = f"{test_ref}_parquet"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(namespace)
|
|
|
|
# Readable Iceberg table (pyiceberg sets table_type=ICEBERG in Glue).
|
|
table = create_table(catalog, namespace, iceberg_table)
|
|
table.append(generate_arrow_data(10))
|
|
|
|
# Unreadable tables registered directly in Glue.
|
|
create_glue_table_directly(started_cluster, namespace, delta_table, table_type="delta")
|
|
create_glue_table_directly(started_cluster, namespace, parquet_table, table_type=None)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
iceberg_full = f"{namespace}.{iceberg_table}"
|
|
delta_full = f"{namespace}.{delta_table}"
|
|
parquet_full = f"{namespace}.{parquet_table}"
|
|
|
|
settings = "SETTINGS show_data_lake_catalogs_in_system_tables = true"
|
|
|
|
# SHOW TABLES (lightweight name-only path): only the Iceberg table is listed.
|
|
show_tables = node.query(
|
|
f"SHOW TABLES FROM {CATALOG_NAME} LIKE '%{test_ref}%'"
|
|
).split()
|
|
assert iceberg_full in show_tables, show_tables
|
|
assert delta_full not in show_tables, show_tables
|
|
assert parquet_full not in show_tables, show_tables
|
|
|
|
# system.tables name-only path (same lightweight iterator as SHOW TABLES).
|
|
names_only = node.query(
|
|
f"SELECT name FROM system.tables WHERE database = '{CATALOG_NAME}' "
|
|
f"AND name LIKE '%{test_ref}%' ORDER BY name {settings}"
|
|
).split()
|
|
assert names_only == [iceberg_full], names_only
|
|
|
|
# system.tables full path (references engine -> fetches per-table metadata).
|
|
full = node.query(
|
|
f"SELECT name FROM system.tables WHERE database = '{CATALOG_NAME}' "
|
|
f"AND name LIKE '%{test_ref}%' AND engine != '' ORDER BY name {settings}"
|
|
).split()
|
|
assert full == [iceberg_full], full
|
|
|
|
# The two listings agree - this is exactly the property that was broken.
|
|
assert sorted(show_tables) == sorted(names_only)
|
|
|
|
|
|
def drop_datastore_glue_table(
|
|
node, database_name, table_name
|
|
):
|
|
node.query(
|
|
f"""
|
|
DROP TABLE {CATALOG_NAME}.`{database_name}.{table_name}`;
|
|
"""
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
# We must add some credentials, otherwise moto (AWS Mock)
|
|
# will reject boto connection
|
|
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
|
|
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
|
|
cluster = DatastoreCluster(__file__)
|
|
cluster.add_instance(
|
|
"node1",
|
|
main_configs=[
|
|
"configs/query_log.xml",
|
|
"configs/text_log.xml",
|
|
],
|
|
user_configs=[],
|
|
stay_alive=True,
|
|
with_glue_catalog=True,
|
|
with_zookeeper=True
|
|
)
|
|
|
|
sts = cluster.add_instance(
|
|
name="sts.us-east-1.amazonaws.com",
|
|
hostname="sts.us-east-1.amazonaws.com",
|
|
image="datastore/python-bottle",
|
|
tag="latest",
|
|
stay_alive=True,
|
|
)
|
|
sts.stop_datastore(kill=True)
|
|
|
|
logging.info("Starting cluster...")
|
|
cluster.start()
|
|
logging.info("Cluster started")
|
|
|
|
run_s3_mocks(cluster)
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_catalog(started_cluster):
|
|
# All tests share one module-scoped Glue catalog. Listing operations such as
|
|
# SHOW TABLES enumerate every namespace and make one sequential Glue call per
|
|
# namespace, so leftover namespaces from earlier tests make these calls grow
|
|
# unbounded and eventually exceed query timeouts. Drop everything each test
|
|
# created so the catalog stays bounded to the running test.
|
|
yield
|
|
catalog = load_catalog_impl(started_cluster)
|
|
glue_client = boto3.client(
|
|
"glue", region_name="us-east-1", endpoint_url=get_glue_local_url(started_cluster)
|
|
)
|
|
for namespace in catalog.list_namespaces():
|
|
# Drop Iceberg tables via pyiceberg so their data and metadata are removed too.
|
|
for table in catalog.list_tables(namespace):
|
|
catalog.drop_table(table)
|
|
# Drop any remaining tables registered directly in Glue (e.g. non-Iceberg tables
|
|
# such as Delta Lake tables or raw data files). pyiceberg's list_tables ignores
|
|
# them, and drop_namespace refuses to drop a namespace that still contains them.
|
|
database_name = ".".join(namespace)
|
|
for table in glue_client.get_tables(DatabaseName=database_name).get("TableList", []):
|
|
glue_client.delete_table(DatabaseName=database_name, Name=table["Name"])
|
|
catalog.drop_namespace(namespace)
|
|
|
|
|
|
def test_no_secrets_in_logs(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
db_name = f"glue_query_log_{uuid.uuid4().hex}"
|
|
root_namespace = f"log_check_ns_{uuid.uuid4().hex}"
|
|
table_name = f"log_check_tbl_{uuid.uuid4().hex}"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
db_settings = {
|
|
"catalog_type": "glue",
|
|
"warehouse": "test",
|
|
"storage_endpoint": "http://minio1:9001/warehouse-glue",
|
|
"region": "us-east-1",
|
|
# The Glue catalog API client is restricted; pass explicit catalog credentials instead of relying on
|
|
# the server's AWS identity (moto accepts any non-empty values).
|
|
"aws_access_key_id": minio_access_key,
|
|
"aws_secret_access_key": minio_secret_key,
|
|
}
|
|
|
|
qid_db = uuid.uuid4().hex
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name}")
|
|
node.query(
|
|
f"""CREATE DATABASE {db_name} ENGINE = DataLakeCatalog('{BASE_URL}', '{minio_access_key}', '{minio_secret_key}')
|
|
SETTINGS {",".join((k + "=" + repr(v) for k, v in db_settings.items()))}""",
|
|
query_id=qid_db,
|
|
settings={
|
|
"allow_database_glue_catalog": 1,
|
|
"write_full_path_in_iceberg_metadata": 1,
|
|
},
|
|
)
|
|
|
|
qid_table = uuid.uuid4().hex
|
|
node.query(
|
|
f"""CREATE TABLE {db_name}.`{root_namespace}.{table_name}` (x String) ENGINE = IcebergS3('http://minio1:9001/warehouse-glue/{table_name}/', '{minio_access_key}', '{minio_secret_key}')""",
|
|
query_id=qid_table,
|
|
settings={
|
|
"allow_experimental_database_glue_catalog": 1,
|
|
"write_full_path_in_iceberg_metadata": 1,
|
|
},
|
|
)
|
|
|
|
qid_show_db = uuid.uuid4().hex
|
|
show_db_result = node.query(
|
|
f"SHOW CREATE DATABASE {db_name}", query_id=qid_show_db
|
|
)
|
|
assert minio_secret_key not in show_db_result
|
|
assert "[HIDDEN]" in show_db_result
|
|
|
|
qid_show_table = uuid.uuid4().hex
|
|
show_table_result = node.query(
|
|
f"SHOW CREATE TABLE {db_name}.`{root_namespace}.{table_name}`",
|
|
query_id=qid_show_table,
|
|
)
|
|
assert minio_secret_key not in show_table_result
|
|
assert "[HIDDEN]" in show_table_result
|
|
|
|
node.query("SYSTEM FLUSH LOGS system.query_log")
|
|
node.query("SYSTEM FLUSH LOGS system.text_log")
|
|
|
|
for qid in (qid_db, qid_table, qid_show_db, qid_show_table):
|
|
assert (
|
|
int(
|
|
node.query(
|
|
f"SELECT count() FROM system.query_log WHERE query_id = '{qid}' AND type = 'QueryFinish'"
|
|
).strip()
|
|
)
|
|
>= 1
|
|
)
|
|
query_text = node.query(
|
|
f"SELECT arrayStringConcat(groupArray(query), '\\n') FROM system.query_log WHERE query_id = '{qid}' AND type = 'QueryFinish'"
|
|
).strip()
|
|
assert minio_secret_key not in query_text
|
|
|
|
text_log_rows = node.query(
|
|
f"""
|
|
SELECT message, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10
|
|
FROM system.text_log
|
|
WHERE query_id IN ('{qid_db}', '{qid_table}', '{qid_show_db}', '{qid_show_table}')
|
|
FORMAT JSONEachRow
|
|
"""
|
|
).strip()
|
|
assert text_log_rows
|
|
for line in text_log_rows.split("\n"):
|
|
row = json.loads(line)
|
|
for val in row.values():
|
|
if isinstance(val, str):
|
|
assert minio_secret_key not in val
|
|
|
|
|
|
def test_list_tables(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
root_namespace = f"datastore_{uuid.uuid4()}"
|
|
namespace_1 = f"{root_namespace}_testA_A"
|
|
namespace_2 = f"{root_namespace}_testB_B"
|
|
namespace_1_tables = ["tableA", "tableB"]
|
|
namespace_2_tables = ["tableC", "tableD"]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in [namespace_1, namespace_2]:
|
|
catalog.create_namespace(namespace)
|
|
|
|
for namespace in [namespace_1, namespace_2]:
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
tables_list = ""
|
|
for table in namespace_1_tables:
|
|
create_table(catalog, namespace_1, table)
|
|
if len(tables_list) > 0:
|
|
tables_list += "\n"
|
|
tables_list += f"{namespace_1}.{table}"
|
|
|
|
for table in namespace_2_tables:
|
|
create_table(catalog, namespace_2, table)
|
|
if len(tables_list) > 0:
|
|
tables_list += "\n"
|
|
tables_list += f"{namespace_2}.{table}"
|
|
|
|
assert (
|
|
tables_list
|
|
== node.query(
|
|
f"SELECT name FROM system.tables WHERE database = '{CATALOG_NAME}' and name ILIKE '{root_namespace}%' ORDER BY name SETTINGS show_data_lake_catalogs_in_system_tables = true"
|
|
).strip()
|
|
)
|
|
node.restart_datastore()
|
|
assert (
|
|
tables_list
|
|
== node.query(
|
|
f"SELECT name FROM system.tables WHERE database = '{CATALOG_NAME}' and name ILIKE '{root_namespace}%' ORDER BY name SETTINGS show_data_lake_catalogs_in_system_tables = true"
|
|
).strip()
|
|
)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace_2, "tableC")
|
|
print("Expected", expected)
|
|
print("Got", node.query(f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace_2}.tableC`"))
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace_2}.tableC`"
|
|
)
|
|
assert int(node.query(f"SELECT count() FROM system.iceberg_history WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%'").strip()) == 0
|
|
|
|
|
|
def test_select(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
f"{root_namespace}_B",
|
|
f"{root_namespace}_C",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace, table_name)
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.{table_name}`"
|
|
)
|
|
|
|
assert num_rows == int(
|
|
node.query(f"SELECT count() FROM {CATALOG_NAME}.`{namespace}.{table_name}`")
|
|
)
|
|
|
|
assert int(node.query(f"SELECT count() FROM system.iceberg_history WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%'").strip()) == 4
|
|
assert int(node.query(f"SELECT count() FROM system.tables WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) == 4
|
|
assert int(node.query(f"SELECT count() FROM system.tables WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = false").strip()) == 0
|
|
|
|
|
|
def test_hide_sensitive_info(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_hide_sensitive_info_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespace = f"{root_namespace}_A"
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(namespace)
|
|
|
|
create_table(catalog, namespace, table_name)
|
|
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
CATALOG_NAME,
|
|
additional_settings={
|
|
"aws_access_key_id": "SECRET_1",
|
|
"aws_secret_access_key": "SECRET_2",
|
|
},
|
|
)
|
|
assert "SECRET_1" not in node.query(f"SHOW CREATE DATABASE {CATALOG_NAME}")
|
|
assert "SECRET_2" not in node.query(f"SHOW CREATE DATABASE {CATALOG_NAME}")
|
|
|
|
|
|
def test_select_after_rename(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
f"{root_namespace}_B",
|
|
f"{root_namespace}_C",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace, table_name)
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.{table_name}`"
|
|
)
|
|
|
|
with table.update_schema() as update:
|
|
update.rename_column("bid", "new_bid")
|
|
|
|
print(node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`"))
|
|
|
|
def test_non_existing_tables(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_non_existing_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace, table_name)
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.{table_name}`"
|
|
)
|
|
|
|
try:
|
|
node.query(f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.wrong_table_name`")
|
|
except Exception as e:
|
|
assert "DB::Exception: Table" in str(e)
|
|
assert "doesn't exist" in str(e)
|
|
|
|
try:
|
|
node.query(f"SHOW CREATE TABLE {CATALOG_NAME}.`fake_namespace.wrong_table_name`")
|
|
except Exception as e:
|
|
assert "DB::Exception: Table" in str(e)
|
|
assert "doesn't exist" in str(e)
|
|
|
|
|
|
def test_empty_table(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
create_table(catalog, root_namespace, table_name)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
assert len(node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`")) == 0
|
|
|
|
|
|
def test_cloud_mode(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_cloud_mode_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespace = f"{root_namespace}_A"
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(namespace)
|
|
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
CATALOG_NAME, query_settings={"cloud_mode": 1}
|
|
)
|
|
|
|
assert int(
|
|
node.query(f"SELECT count() FROM {CATALOG_NAME}.`{namespace}.{table_name}`")
|
|
) == num_rows
|
|
|
|
|
|
def test_timestamps(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
schema = Schema(
|
|
NestedField(
|
|
field_id=1, name="timestamp", field_type=TimestampType(), required=False
|
|
),
|
|
NestedField(
|
|
field_id=2,
|
|
name="timestamptz",
|
|
field_type=TimestamptzType(),
|
|
required=False,
|
|
),
|
|
)
|
|
table = create_table(catalog, root_namespace, table_name, schema)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
data = [
|
|
{
|
|
"timestamp": datetime(2024, 1, 1, hour=12, minute=0, second=0, microsecond=0),
|
|
"timestamptz": datetime(
|
|
2024,
|
|
1,
|
|
1,
|
|
hour=12,
|
|
minute=0,
|
|
second=0,
|
|
microsecond=0,
|
|
tzinfo=pytz.timezone("UTC"),
|
|
)
|
|
}
|
|
]
|
|
df = pa.Table.from_pylist(data)
|
|
table.append(df)
|
|
|
|
assert node.query(f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}`") == f"CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}`\\n(\\n `timestamp` Nullable(DateTime64(6)),\\n `timestamptz` Nullable(DateTime64(6, \\'UTC\\'))\\n)\\nENGINE = Iceberg(\\'http://minio1:9001/warehouse-glue/data/\\', \\'minio\\', \\'[HIDDEN]\\')\n"
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "2024-01-01 12:00:00.000000\t2024-01-01 12:00:00.000000\n"
|
|
|
|
|
|
def test_insert(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
create_table(catalog, root_namespace, table_name, DEFAULT_SCHEMA, PartitionSpec(), DEFAULT_SORT_ORDER, table_name)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
node.query(f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (NULL, 'AAPL', 193.24, 193.31, tuple('bot'), NULL);", settings={"allow_insert_into_iceberg": 1, 'write_full_path_in_iceberg_metadata': 1})
|
|
catalog.load_table(f"{root_namespace}.{table_name}")
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "\\N\tAAPL\t193.24\t193.31\t('bot')\t{}\n"
|
|
|
|
|
|
def test_create(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
create_datastore_glue_table(started_cluster, node, root_namespace, table_name, "(x String)")
|
|
node.query(f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('AAPL');", settings={"allow_insert_into_iceberg": 1, 'write_full_path_in_iceberg_metadata': 1})
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "AAPL\n"
|
|
|
|
|
|
def test_create_gzip_metadata(started_cluster):
|
|
# Regression for issue #109801: a catalog-backed CREATE TABLE from ClickHouse
|
|
# with gzip metadata compression exercises IcebergMetadata::createInitial and
|
|
# the catalog->createTable registration. The first revision of the fix wrote
|
|
# v1.gz.metadata.json on disk but registered a hardcoded v1.metadata.json as
|
|
# the catalog metadata_location, so a reopen through the catalog failed.
|
|
#
|
|
# Glue is the concrete broken path: GlueCatalog::createTable persists the
|
|
# supplied new_metadata_path verbatim as parameters["metadata_location"]
|
|
# (unlike RestCatalog, which ignores new_metadata_path). We therefore read
|
|
# the registered metadata_location straight from Glue immediately after
|
|
# CREATE TABLE, before any INSERT, so the assertion validates createInitial's
|
|
# registration itself rather than a later updateMetadata pointer advance.
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_create_gzip_metadata_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
create_datastore_glue_table(
|
|
started_cluster,
|
|
node,
|
|
root_namespace,
|
|
table_name,
|
|
"(x String)",
|
|
additional_settings={"iceberg_metadata_compression_method": "gzip"},
|
|
)
|
|
|
|
# Validate the catalog registration produced by createInitial, before INSERT.
|
|
glue_client = boto3.client(
|
|
"glue", region_name="us-east-1", endpoint_url=get_glue_local_url(started_cluster)
|
|
)
|
|
table_info = glue_client.get_table(DatabaseName=root_namespace, Name=table_name)["Table"]
|
|
metadata_location = table_info["Parameters"]["metadata_location"]
|
|
assert metadata_location.endswith(".gz.metadata.json"), metadata_location
|
|
assert not metadata_location.endswith(".gzip.metadata.json"), metadata_location
|
|
|
|
node.query(
|
|
f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('AAPL');",
|
|
settings={
|
|
"allow_insert_into_iceberg": 1,
|
|
"write_full_path_in_iceberg_metadata": 1,
|
|
"iceberg_metadata_compression_method": "gzip",
|
|
},
|
|
)
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "AAPL\n"
|
|
|
|
# Reopen through the catalog (fresh database) and confirm read still works.
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "AAPL\n"
|
|
|
|
|
|
def test_schema_evolution(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_schema_evolution_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`"
|
|
write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
create_datastore_glue_table(started_cluster, node, root_namespace, table_name, "(x String, y Int32)")
|
|
|
|
node.query(f"INSERT INTO {table_ref} VALUES ('123', 1);", settings=write_settings)
|
|
|
|
node.query(f"ALTER TABLE {table_ref} ADD COLUMN z Nullable(String);", settings=write_settings)
|
|
assert "z" in node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings)
|
|
assert node.query(f"SELECT x, y, z FROM {table_ref} ORDER BY ALL", settings=write_settings) == "123\t1\t\\N\n"
|
|
|
|
node.query(f"INSERT INTO {table_ref} VALUES ('456', 2, 'hello');", settings=write_settings)
|
|
assert (
|
|
node.query(f"SELECT x, y, z FROM {table_ref} ORDER BY ALL", settings=write_settings)
|
|
== "123\t1\t\\N\n456\t2\thello\n"
|
|
)
|
|
|
|
node.query(f"ALTER TABLE {table_ref} RENAME COLUMN z TO w;", settings=write_settings)
|
|
assert "w" in node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings)
|
|
assert (
|
|
node.query(f"SELECT x, y, w FROM {table_ref} ORDER BY ALL", settings=write_settings)
|
|
== "123\t1\t\\N\n456\t2\thello\n"
|
|
)
|
|
|
|
|
|
def test_writes_schema_evolution_concurrent_add_columns(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_writes_schema_evolution_concurrent_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`"
|
|
write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
create_datastore_glue_table(started_cluster, node, root_namespace, table_name, "(x String, y Int32)")
|
|
|
|
node.query(f"INSERT INTO {table_ref} VALUES ('123', 1);", settings=write_settings)
|
|
|
|
num_columns = 10
|
|
|
|
def add_column(idx):
|
|
node.query(
|
|
f"ALTER TABLE {table_ref} ADD COLUMN col_{idx} Nullable(String);",
|
|
settings=write_settings,
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=num_columns) as executor:
|
|
list(executor.map(add_column, range(num_columns)))
|
|
|
|
description = node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings)
|
|
for idx in range(num_columns):
|
|
assert f"col_{idx}" in description, f"col_{idx} missing from:\n{description}"
|
|
|
|
columns = [line.split("\t")[0] for line in description.strip().split("\n")]
|
|
assert sorted(columns) == sorted(["x", "y"] + [f"col_{idx}" for idx in range(num_columns)])
|
|
|
|
select_cols = ", ".join(["x", "y"] + [f"col_{idx}" for idx in range(num_columns)])
|
|
expected = "123\t1" + "\t\\N" * num_columns + "\n"
|
|
assert node.query(f"SELECT {select_cols} FROM {table_ref} ORDER BY ALL", settings=write_settings) == expected
|
|
|
|
|
|
def test_drop_table(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_list_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
create_datastore_glue_table(started_cluster, node, root_namespace, table_name, "(x String)")
|
|
assert len(catalog.list_tables(root_namespace)) == 1
|
|
assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == ""
|
|
|
|
drop_datastore_glue_table(node, root_namespace, table_name)
|
|
assert len(catalog.list_tables(root_namespace)) == 0
|
|
|
|
|
|
def test_system_tables(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_system_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
f"{root_namespace}_B",
|
|
f"{root_namespace}_C",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace, table_name)
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.{table_name}`"
|
|
)
|
|
|
|
assert num_rows == int(
|
|
node.query(f"SELECT count() FROM {CATALOG_NAME}.`{namespace}.{table_name}`")
|
|
)
|
|
|
|
assert CATALOG_NAME in node.query("SHOW DATABASES")
|
|
assert table_name in node.query(f"SHOW TABLES FROM {CATALOG_NAME}")
|
|
|
|
# system.tables
|
|
assert int(node.query(f"SELECT count() FROM system.tables WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) == 4
|
|
assert int(node.query(f"SELECT count() FROM system.tables WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%'").strip()) == 0
|
|
|
|
# system.iceberg_history - always displays irrespective of show_data_lake_catalogs_in_system_tables
|
|
assert int(node.query(f"SELECT count() FROM system.iceberg_history WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) == 4
|
|
assert int(node.query(f"SELECT count() FROM system.iceberg_history WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = false").strip()) == 4
|
|
|
|
# system.databases always shows data lake catalog databases regardless of show_data_lake_catalogs_in_system_tables,
|
|
# because listing a database name is purely local metadata and never requires calls to an external catalog service.
|
|
assert int(node.query(f"SELECT count() FROM system.databases WHERE name = '{CATALOG_NAME}' SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) == 1
|
|
assert int(node.query(f"SELECT count() FROM system.databases WHERE name = '{CATALOG_NAME}' SETTINGS show_data_lake_catalogs_in_system_tables = false").strip()) == 1
|
|
|
|
# system.columns
|
|
assert int(node.query(f"SELECT count() FROM system.columns WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) == 24
|
|
assert int(node.query(f"SELECT count() FROM system.columns WHERE database = '{CATALOG_NAME}' and table ilike '%{root_namespace}%'").strip()) == 0
|
|
|
|
# system.completions
|
|
assert int(node.query(f"SELECT count() FROM system.completions WHERE startsWith(word, '{test_ref}') SETTINGS show_data_lake_catalogs_in_system_tables = true").strip()) != 0
|
|
assert int(node.query(f"SELECT count() FROM system.completions WHERE startsWith(word, '{test_ref}')").strip()) == 0
|
|
|
|
def test_show_tables_optimization(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_show_tables_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
f"{root_namespace}_B",
|
|
f"{root_namespace}_C",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
assert table_name in node.query(f"SHOW TABLES FROM {CATALOG_NAME}")
|
|
|
|
assert not node.contains_in_log(
|
|
f"Get table information for table {root_namespace}.{table_name}"
|
|
)
|
|
|
|
node.query(f"SELECT * from system.tables where table ilike '%{root_namespace}%' SETTINGS show_data_lake_catalogs_in_system_tables = true")
|
|
assert node.contains_in_log(
|
|
f"Get table information for table {root_namespace}.{table_name}"
|
|
)
|
|
|
|
node.query("SYSTEM ENABLE FAILPOINT lightweight_show_tables")
|
|
node.query(f"SHOW TABLES FROM {CATALOG_NAME}", timeout=5)
|
|
|
|
|
|
def test_typo_hint_does_not_trigger_heavyweight_fetch(started_cluster):
|
|
"""
|
|
Regression test for the high-memory typo-hint path on a DataLake (Glue) database.
|
|
|
|
When a query references a table that does not exist, the UNKNOWN_TABLE error
|
|
path builds a "maybe you meant ...?" suggestion through
|
|
TableNameHints -> getAllRegisteredNames -> IDatabase::getAllTableNames. Before
|
|
this fix DatabaseDataLake did not override getAllTableNames, so the base
|
|
implementation iterated getTablesIterator and fetched per-table metadata from
|
|
the catalog (one heavyweight S3 read per table). With many tables this hint
|
|
lookup alone could exhaust the server's memory. The fix overrides
|
|
getAllTableNames to return getCatalog()->getTables() (table names only).
|
|
|
|
INSERT resolves its target table through the throwing DatabaseCatalog::getTable
|
|
overload independently of the analyzer, so it deterministically drives the
|
|
hint path. show_data_lake_catalogs_in_system_tables=1 is required: otherwise
|
|
getAllRegisteredNames short-circuits for remote databases and getAllTableNames
|
|
is never reached, so the override would not be exercised at all.
|
|
|
|
Signal: the log line emitted once per table inside
|
|
DatabaseDataLake::getTablesIterator. Counting it scoped to this namespace
|
|
makes the measurement exact and immune to other tables in the catalog. A
|
|
global S3GetObject counter is deliberately NOT used: catalog metadata is
|
|
cached across queries, so a repeated heavyweight iteration issues no new S3
|
|
GET requests even though it still walks every table.
|
|
|
|
Pre-fix the marker fires once per table (N). With the fix it stays at 0.
|
|
"""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_hint_heavy_{uuid.uuid4().hex[:8]}"
|
|
namespace = f"{test_ref}_ns"
|
|
N = 20
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(namespace)
|
|
try:
|
|
for i in range(N):
|
|
tbl = create_table(catalog, namespace, f"{test_ref}_t{i:02d}")
|
|
tbl.append(generate_arrow_data(1))
|
|
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
# Emitted once per table inside DatabaseDataLake::getTablesIterator. If the
|
|
# message is ever rephrased, update this marker.
|
|
marker = f"Get table information for table {namespace}."
|
|
baseline = int(node.count_in_log(marker).strip())
|
|
|
|
# Reference a non-existent table; this must raise UNKNOWN_TABLE and on the
|
|
# way build the typo hint that walks getAllTableNames.
|
|
error = node.query_and_get_error(
|
|
f"INSERT INTO {CATALOG_NAME}.`{namespace}.does_not_exist` VALUES (1)",
|
|
settings={"show_data_lake_catalogs_in_system_tables": 1},
|
|
)
|
|
assert "UNKNOWN_TABLE" in error or "does not exist" in error, (
|
|
f"Expected an UNKNOWN_TABLE error for a non-existent table, got: {error}"
|
|
)
|
|
|
|
fetches = int(node.count_in_log(marker).strip()) - baseline
|
|
assert fetches == 0, (
|
|
f"The typo-hint lookup for a non-existent table triggered {fetches} "
|
|
f"per-table metadata fetches (expected 0; the namespace has {N} tables). "
|
|
f"DatabaseDataLake.getAllTableNames is falling through to "
|
|
f"IDatabase::getAllTableNames -> getTablesIterator instead of using "
|
|
f"getCatalog()->getTables() directly."
|
|
)
|
|
finally:
|
|
try:
|
|
for i in range(N):
|
|
catalog.drop_table(f"{namespace}.{test_ref}_t{i:02d}")
|
|
catalog.drop_namespace(namespace)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def test_table_without_metadata_location(started_cluster):
|
|
"""
|
|
Test that Datastore can read Iceberg tables from Glue when 'metadata_location' is not present in Glue parameters.
|
|
Glue's Location field is used to deduce the metadata location.
|
|
"""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_no_metadata_location_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
# Use unique database name to avoid interference with other tests (as we are playing with Glue metadata here)
|
|
db_name = f"db_{test_ref.replace('-', '_')}"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
schema = Schema(
|
|
NestedField(field_id=1, name="id", field_type=StringType(), required=False),
|
|
NestedField(field_id=2, name="value", field_type=DoubleType(), required=False),
|
|
)
|
|
table = create_table(catalog, root_namespace, table_name, schema, PartitionSpec(), DEFAULT_SORT_ORDER, dir=table_name)
|
|
|
|
data = [
|
|
{"id": "row1", "value": 1.5},
|
|
{"id": "row2", "value": 2.5},
|
|
{"id": "row3", "value": 3.5},
|
|
]
|
|
df = pa.Table.from_pylist(data)
|
|
table.append(df)
|
|
|
|
glue_client = boto3.client(
|
|
"glue", region_name="us-east-1", endpoint_url=get_glue_local_url(started_cluster)
|
|
)
|
|
|
|
table_response = glue_client.get_table(
|
|
DatabaseName=root_namespace,
|
|
Name=table_name
|
|
)
|
|
table_info = table_response["Table"]
|
|
|
|
# Verify metadata_location exists before we remove it
|
|
assert "metadata_location" in table_info.get("Parameters", {}), "Test setup failed: metadata_location should exist initially"
|
|
|
|
# Remove metadata_location from parameters but keep other parameters
|
|
new_parameters = {k: v for k, v in table_info.get("Parameters", {}).items() if k != "metadata_location"}
|
|
glue_client.update_table(
|
|
DatabaseName=root_namespace,
|
|
TableInput={
|
|
"Name": table_name,
|
|
"StorageDescriptor": table_info["StorageDescriptor"],
|
|
"Parameters": new_parameters,
|
|
"TableType": table_info.get("TableType", "EXTERNAL_TABLE"),
|
|
}
|
|
)
|
|
|
|
# Verify metadata_location was removed
|
|
updated_table = glue_client.get_table(DatabaseName=root_namespace, Name=table_name)
|
|
assert "metadata_location" not in updated_table["Table"].get("Parameters", {}), "Failed to remove metadata_location"
|
|
assert updated_table["Table"]["StorageDescriptor"]["Location"], "Location should still be present"
|
|
|
|
create_datastore_glue_database(started_cluster, node, db_name)
|
|
|
|
# Verify SHOW TABLES FROM works
|
|
tables_result = node.query(f"SHOW TABLES FROM {db_name} LIKE '%{table_name}%'")
|
|
assert table_name in tables_result, f"Table {table_name} not found in SHOW TABLES: {tables_result}"
|
|
|
|
# Query should work even without metadata_location in Glue
|
|
result = node.query(f"SELECT id, value FROM {db_name}.`{root_namespace}.{table_name}` ORDER BY id")
|
|
assert result == "row1\t1.5\nrow2\t2.5\nrow3\t3.5\n", f"Unexpected result: {result}"
|
|
|
|
# Also verify SHOW CREATE TABLE works
|
|
create_table_result = node.query(f"SHOW CREATE TABLE {db_name}.`{root_namespace}.{table_name}`")
|
|
assert "Iceberg" in create_table_result, f"Expected Iceberg engine in: {create_table_result}"
|
|
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name} SYNC")
|
|
|
|
|
|
def test_check_database(started_cluster):
|
|
"""Test that CHECK DATABASE works with Glue catalog database."""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_check_database_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
namespaces_to_create = [
|
|
root_namespace,
|
|
f"{root_namespace}_A",
|
|
f"{root_namespace}_B",
|
|
f"{root_namespace}_C",
|
|
]
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
|
|
# Create namespaces
|
|
for namespace in namespaces_to_create:
|
|
catalog.create_namespace(namespace)
|
|
assert len(catalog.list_tables(namespace)) == 0
|
|
|
|
# Create Datastore Glue database once
|
|
create_datastore_glue_database(started_cluster, node, CATALOG_NAME)
|
|
|
|
# Create tables in each namespace
|
|
for namespace in namespaces_to_create:
|
|
table = create_table(catalog, namespace, table_name)
|
|
|
|
num_rows = 10
|
|
df = generate_arrow_data(num_rows)
|
|
table.append(df)
|
|
|
|
expected = DEFAULT_CREATE_TABLE.format(CATALOG_NAME, namespace, table_name)
|
|
assert expected == node.query(
|
|
f"SHOW CREATE TABLE {CATALOG_NAME}.`{namespace}.{table_name}`"
|
|
)
|
|
|
|
assert num_rows == int(
|
|
node.query(f"SELECT count() FROM {CATALOG_NAME}.`{namespace}.{table_name}`")
|
|
)
|
|
|
|
# Verify database exists
|
|
assert CATALOG_NAME in node.query("SHOW DATABASES")
|
|
|
|
# Run CHECK DATABASE and verify it completes without error
|
|
node.query(f"CHECK DATABASE {CATALOG_NAME}")
|
|
|
|
try:
|
|
node.query(
|
|
"SYSTEM ENABLE FAILPOINT check_database_datalake_negative"
|
|
)
|
|
|
|
assert "fault when checking database" in node.query_and_get_error(
|
|
f"CHECK DATABASE {CATALOG_NAME}"
|
|
)
|
|
finally:
|
|
node.query(
|
|
"SYSTEM DISABLE FAILPOINT check_database_datalake_negative"
|
|
)
|
|
|
|
def test_sts_smoke(started_cluster):
|
|
"""Test that STS authentication works with Glue catalog using role_arn and role_session_name"""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_sts_smoke_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
schema = Schema(
|
|
NestedField(field_id=1, name="id", field_type=StringType(), required=False),
|
|
NestedField(field_id=2, name="value", field_type=DoubleType(), required=False),
|
|
)
|
|
table = create_table(catalog, root_namespace, table_name, schema, PartitionSpec(), DEFAULT_SORT_ORDER, dir=table_name)
|
|
|
|
data = [
|
|
{"id": "row1", "value": 10.0},
|
|
{"id": "row2", "value": 20.0},
|
|
{"id": "row3", "value": 30.0},
|
|
]
|
|
df = pa.Table.from_pylist(data)
|
|
table.append(df)
|
|
|
|
# Test with wrong role_session_name - should fail
|
|
db_name_fail = f"db_fail_{test_ref.replace('-', '_')}"
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name_fail,
|
|
additional_settings={
|
|
"aws_role_arn": "arn::role",
|
|
"aws_role_session_name": "wrongsession",
|
|
},
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
with_credentials=False,
|
|
)
|
|
|
|
# Query should fail with wrong session name
|
|
try:
|
|
result = node.query(
|
|
f"SELECT sum(value) FROM {db_name_fail}.`{root_namespace}.{table_name}` "
|
|
f"SETTINGS s3_max_single_read_retries = 1, s3_retry_attempts = 1, s3_request_timeout_ms = 1000"
|
|
)
|
|
assert False, f"Expected query to fail with wrong session name but got result: {result}"
|
|
except Exception as e:
|
|
error_str = str(e)
|
|
assert "403" in error_str or "Failed to get object info" in error_str or "HTTP response code: 403" in error_str, \
|
|
f"Expected 403 error but got: {error_str}"
|
|
|
|
# Test with correct role_session_name - should succeed
|
|
db_name_success = f"db_success_{test_ref.replace('-', '_')}"
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name_success,
|
|
additional_settings={
|
|
"aws_role_arn": "arn::role",
|
|
"aws_role_session_name": "miniorole",
|
|
},
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
with_credentials=False,
|
|
)
|
|
|
|
# Query should succeed with correct session name
|
|
result = node.query(f"SELECT sum(value) FROM {db_name_success}.`{root_namespace}.{table_name}`")
|
|
assert result.strip() == "60", f"Expected sum to be 60 but got: {result}"
|
|
|
|
# Cleanup
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name_fail} SYNC")
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name_success} SYNC")
|
|
|
|
|
|
def test_sts_external_id(started_cluster):
|
|
"""Test that `aws_external_id` reaches the STS AssumeRole request from the
|
|
Glue catalog database engine. The mock STS only vends working credentials
|
|
when the supplied ExternalId matches, so the negative case fails purely on
|
|
the external id (the role session name is correct in both cases)."""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"test_sts_external_id_{uuid.uuid4()}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
schema = Schema(
|
|
NestedField(field_id=1, name="id", field_type=StringType(), required=False),
|
|
NestedField(field_id=2, name="value", field_type=DoubleType(), required=False),
|
|
)
|
|
table = create_table(catalog, root_namespace, table_name, schema, PartitionSpec(), DEFAULT_SORT_ORDER, dir=table_name)
|
|
|
|
data = [
|
|
{"id": "row1", "value": 10.0},
|
|
{"id": "row2", "value": 20.0},
|
|
{"id": "row3", "value": 30.0},
|
|
]
|
|
df = pa.Table.from_pylist(data)
|
|
table.append(df)
|
|
|
|
# Negative: correct role session name but wrong external id - should fail.
|
|
db_name_fail = f"db_fail_{test_ref.replace('-', '_')}"
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name_fail,
|
|
additional_settings={
|
|
"aws_role_arn": "arn::role",
|
|
"aws_role_session_name": "miniorole",
|
|
"aws_external_id": "wrong_external_id",
|
|
},
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
with_credentials=False,
|
|
)
|
|
|
|
try:
|
|
result = node.query(
|
|
f"SELECT sum(value) FROM {db_name_fail}.`{root_namespace}.{table_name}` "
|
|
f"SETTINGS s3_max_single_read_retries = 1, s3_retry_attempts = 1, s3_request_timeout_ms = 1000"
|
|
)
|
|
assert False, f"Expected query to fail with wrong external id but got result: {result}"
|
|
except Exception as e:
|
|
error_str = str(e)
|
|
assert "403" in error_str or "Failed to get object info" in error_str or "HTTP response code: 403" in error_str, \
|
|
f"Expected 403 error but got: {error_str}"
|
|
|
|
# Positive: correct role session name and external id - should succeed.
|
|
db_name_success = f"db_success_{test_ref.replace('-', '_')}"
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name_success,
|
|
additional_settings={
|
|
"aws_role_arn": "arn::role",
|
|
"aws_role_session_name": "miniorole",
|
|
"aws_external_id": "miniexternalid",
|
|
},
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
with_credentials=False,
|
|
)
|
|
|
|
result = node.query(f"SELECT sum(value) FROM {db_name_success}.`{root_namespace}.{table_name}`")
|
|
assert result.strip() == "60", f"Expected sum to be 60 but got: {result}"
|
|
|
|
# Cleanup
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name_fail} SYNC")
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name_success} SYNC")
|
|
|
|
|
|
def test_sts_credential_refresh_on_expired_token(started_cluster):
|
|
"""When an S3 read fails mid-query with an `ExpiredToken`-style error,
|
|
`GlueCatalog::getCredentialsConfigurationCallback` should fire and the read
|
|
should recover with freshly-vended STS credentials.
|
|
|
|
Without the override (the previous behavior), the catalog inherits the default
|
|
`std::nullopt` from `ICatalog::getCredentialsConfigurationCallback`,
|
|
`ReadBufferFromS3::processException` skips the refresh branch, and the read
|
|
surfaces the `ExpiredToken` to the user.
|
|
|
|
The `s3_send_request_throw_expired_token` failpoint fires inside
|
|
`ReadBufferFromS3::sendRequest`, so it covers both `nextImpl`-driven
|
|
streaming reads and the `readBigAt` range-read path that Parquet column-chunk
|
|
reads go through. It is `ONCE`, so only the first S3 GET fails — the
|
|
subsequent retry triggered by the credentials refresh callback succeeds.
|
|
"""
|
|
node = started_cluster.instances["node1"]
|
|
|
|
test_ref = f"refresh_token_{uuid.uuid4().hex}"
|
|
table_name = f"{test_ref}_table"
|
|
root_namespace = f"{test_ref}_namespace"
|
|
|
|
catalog = load_catalog_impl(started_cluster)
|
|
catalog.create_namespace(root_namespace)
|
|
|
|
schema = Schema(
|
|
NestedField(field_id=1, name="id", field_type=StringType(), required=False),
|
|
NestedField(field_id=2, name="value", field_type=DoubleType(), required=False),
|
|
)
|
|
|
|
rows = [
|
|
{"id": "row1", "value": 10.0},
|
|
{"id": "row2", "value": 20.0},
|
|
{"id": "row3", "value": 30.0},
|
|
]
|
|
expected_sum = sum(r["value"] for r in rows)
|
|
create_table(
|
|
catalog, root_namespace, table_name, schema, PartitionSpec(), DEFAULT_SORT_ORDER, dir=table_name
|
|
).append(pa.Table.from_pylist(rows))
|
|
|
|
db_name = f"db_{test_ref.replace('-', '_')}"
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name,
|
|
additional_settings={
|
|
"aws_role_arn": "arn::role",
|
|
"aws_role_session_name": "miniorole",
|
|
},
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
with_credentials=False,
|
|
)
|
|
|
|
# Sanity check: query works without the failpoint.
|
|
pre_failpoint = node.query(
|
|
f"SELECT sum(value) FROM {db_name}.`{root_namespace}.{table_name}`"
|
|
).strip()
|
|
assert pre_failpoint == str(int(expected_sum)), (
|
|
f"Baseline query returned {pre_failpoint}, expected {int(expected_sum)}"
|
|
)
|
|
|
|
qid_select = uuid.uuid4().hex
|
|
node.query("SYSTEM ENABLE FAILPOINT s3_send_request_throw_expired_token")
|
|
try:
|
|
result = node.query(
|
|
f"SELECT sum(value) FROM {db_name}.`{root_namespace}.{table_name}` "
|
|
"SETTINGS max_threads = 1",
|
|
query_id=qid_select,
|
|
)
|
|
finally:
|
|
node.query("SYSTEM DISABLE FAILPOINT s3_send_request_throw_expired_token")
|
|
|
|
assert result.strip() == str(int(expected_sum)), (
|
|
f"Expected sum {int(expected_sum)} after refresh-driven retry, got: {result}"
|
|
)
|
|
|
|
node.query("SYSTEM FLUSH LOGS system.text_log")
|
|
|
|
# Diagnostic: confirm the failpoint actually tripped. If this assertion fails,
|
|
# the test environment did not reach the credential-refresh code path at all
|
|
# (e.g. column chunks completed in a single `nextImpl` call); the refresh-log
|
|
# assertion below would be vacuous.
|
|
failpoint_fired = int(
|
|
node.query(
|
|
"SELECT count() FROM system.text_log "
|
|
"WHERE message LIKE '%Caught exception while reading S3 object%' "
|
|
"AND message LIKE '%ExpiredToken%' "
|
|
"AND event_time >= now() - INTERVAL 5 MINUTE"
|
|
).strip()
|
|
)
|
|
assert failpoint_fired >= 1, (
|
|
"Failpoint `s3_send_request_throw_expired_token` did not fire — "
|
|
"`processException` was never invoked, so the credentials-refresh "
|
|
"code path was not exercised."
|
|
)
|
|
|
|
# The `LOG_DEBUG` in `GlueCatalog::getCredentialsConfigurationCallback` must
|
|
# have fired — its absence would mean the override is not in place and the
|
|
# retry was attributable to some other path.
|
|
refresh_log_count = int(
|
|
node.query(
|
|
"SELECT count() FROM system.text_log "
|
|
"WHERE message LIKE '%Refreshing AWS credentials%after expired token%' "
|
|
"AND event_time >= now() - INTERVAL 5 MINUTE"
|
|
).strip()
|
|
)
|
|
assert refresh_log_count >= 1, (
|
|
"The failpoint fired but `GlueCatalog::getCredentialsConfigurationCallback` "
|
|
"was not invoked — the override is missing or not wired into "
|
|
"`StorageS3Configuration::createObjectStorage`"
|
|
)
|
|
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name} SYNC")
|
|
|
|
|
|
def test_glue_catalog_unavailable_after_restart_under_restriction(started_cluster):
|
|
# A Glue `DataLakeCatalog` whose catalog credentials are server-managed (no explicit keys -> resolved from
|
|
# the server environment) is created under the per-session opt-in. After a restart it is reloaded under the
|
|
# default restriction (`s3_allow_server_credentials_in_user_queries = 0`); the server must still start, but
|
|
# the catalog is left unavailable (governed by `s3_load_table_anonymously_if_credentials_restricted`) rather
|
|
# than silently reusing the server identity or aborting startup.
|
|
node = started_cluster.instances["node1"]
|
|
db_name = f"glue_server_cred_restart_{uuid.uuid4().hex}"
|
|
|
|
create_datastore_glue_database(
|
|
started_cluster,
|
|
node,
|
|
db_name,
|
|
with_credentials=False,
|
|
query_settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
)
|
|
# Works while the creating session allows server-managed credentials for the catalog.
|
|
node.query(
|
|
f"SHOW TABLES FROM {db_name}",
|
|
settings={"s3_allow_server_credentials_in_user_queries": 1},
|
|
)
|
|
|
|
node.restart_clickhouse()
|
|
|
|
# The server starts even though the catalog can no longer resolve its (server-managed) credentials.
|
|
assert node.query("SELECT 1").strip() == "1"
|
|
# Reloaded under the default restriction, the catalog is unavailable. Querying a table goes through the
|
|
# catalog client (`getCatalog`) and reports the restriction, rather than silently reusing the server
|
|
# identity. (`SHOW TABLES` intentionally swallows catalog errors so `system.tables` does not fail, so it
|
|
# is not a reliable probe here.)
|
|
error = node.query_and_get_error(f"SELECT * FROM {db_name}.`unavailable.table`")
|
|
assert (
|
|
"ACCESS_DENIED" in error
|
|
or "server-managed" in error
|
|
or "s3_allow_server_credentials_in_user_queries" in error
|
|
), error
|
|
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name} SYNC")
|
|
|
|
|
|
def test_glue_catalog_user_attach_under_restriction_is_rejected(started_cluster):
|
|
# A user ATTACH DATABASE of a catalog that resolves server-managed credentials must stay fail-closed for a
|
|
# restricted user. The catalog is built lazily (on first access, not at ATTACH), so the ATTACH itself
|
|
# succeeds but the database is inaccessible: the first query against it is refused because the catalog
|
|
# resolves server-managed credentials that are restricted for user queries.
|
|
node = started_cluster.instances["node1"]
|
|
db_name = f"glue_user_attach_{uuid.uuid4().hex}"
|
|
allow = {"s3_allow_server_credentials_in_user_queries": 1}
|
|
|
|
create_datastore_glue_database(
|
|
started_cluster, node, db_name, with_credentials=False, query_settings=allow
|
|
)
|
|
node.query(f"DETACH DATABASE {db_name}")
|
|
|
|
# Re-attaching as a user query under the default restriction succeeds (lazy), but the database is
|
|
# fail-closed: accessing a table goes through the catalog client (`getCatalog`), which resolves the
|
|
# restricted server credentials and is refused. (`SHOW TABLES` intentionally swallows catalog errors, so it
|
|
# is not a reliable probe.)
|
|
node.query(f"ATTACH DATABASE {db_name}")
|
|
error = node.query_and_get_error(f"SELECT * FROM {db_name}.`unavailable.table`")
|
|
assert (
|
|
"ACCESS_DENIED" in error
|
|
or "server-managed" in error
|
|
or "s3_allow_server_credentials_in_user_queries" in error
|
|
), error
|
|
|
|
# With the opt-in the database is usable again, so it can be cleaned up.
|
|
node.query(f"DROP DATABASE IF EXISTS {db_name} SYNC", settings=allow)
|