mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
5796 lines
205 KiB
Python
5796 lines
205 KiB
Python
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import string
|
|
import time
|
|
import uuid
|
|
from datetime import datetime
|
|
from multiprocessing.dummy import Pool
|
|
|
|
import pyarrow as pa
|
|
import pyarrow.parquet as pq
|
|
import pyspark
|
|
import pytest
|
|
from delta import DeltaTable
|
|
from deltalake.writer import write_deltalake
|
|
from minio.deleteobjects import DeleteObject
|
|
from pyspark.sql.functions import (
|
|
col,
|
|
monotonically_increasing_id,
|
|
row_number,
|
|
)
|
|
from pyspark.sql.types import (
|
|
ArrayType,
|
|
BooleanType,
|
|
DateType,
|
|
IntegerType,
|
|
ShortType,
|
|
StringType,
|
|
DecimalType,
|
|
StructField,
|
|
StructType,
|
|
TimestampType,
|
|
)
|
|
from decimal import Decimal
|
|
from pyspark.sql.window import Window
|
|
|
|
from helpers.cluster import CLICKHOUSE_CI_MIN_TESTED_VERSION, ClickHouseCluster
|
|
from helpers.config_cluster import minio_access_key, minio_secret_key
|
|
from helpers.network import PartitionManager
|
|
from helpers.s3_tools import (
|
|
AzureUploader,
|
|
S3Uploader,
|
|
get_file_contents,
|
|
list_s3_objects,
|
|
upload_directory,
|
|
LocalDownloader,
|
|
LocalUploader,
|
|
)
|
|
from helpers.test_tools import TSV
|
|
from helpers.spark_tools import ResilientSparkSession, write_spark_log_config
|
|
|
|
|
|
SCRIPT_DIR = "/var/lib/clickhouse/user_files" + os.path.join(
|
|
os.path.dirname(os.path.realpath(__file__))
|
|
)
|
|
cluster = ClickHouseCluster(__file__, with_spark=True)
|
|
|
|
S3_DATA = [
|
|
"field_ids_struct_test/data/00000-1-7cad83a6-af90-42a9-8a10-114cbc862a42-0-00001.parquet",
|
|
]
|
|
|
|
|
|
def get_spark(log_dir=None):
|
|
builder = (
|
|
pyspark.sql.SparkSession.builder.appName("test_storage_delta")
|
|
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
|
|
.config(
|
|
"spark.sql.catalog.spark_catalog",
|
|
"org.apache.spark.sql.delta.catalog.DeltaCatalog",
|
|
)
|
|
.config(
|
|
"spark.sql.catalog.spark_catalog.warehouse",
|
|
"/var/lib/clickhouse/user_files/test_storage_delta",
|
|
)
|
|
.config("spark.driver.memory", "2g")
|
|
.config("spark.executor.memory", "2g")
|
|
.master("local")
|
|
)
|
|
|
|
if log_dir:
|
|
props_path = write_spark_log_config(log_dir)
|
|
builder = builder.config(
|
|
"spark.driver.extraJavaOptions",
|
|
f"-Dlog4j2.configurationFile=file:{props_path}",
|
|
)
|
|
|
|
return builder.master("local").getOrCreate()
|
|
|
|
|
|
def randomize_table_name(table_name, random_suffix_length=10):
|
|
letters = string.ascii_letters + string.digits
|
|
return f"{table_name}{''.join(random.choice(letters) for _ in range(random_suffix_length))}"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.add_instance(
|
|
"node1",
|
|
main_configs=[
|
|
"configs/config.d/named_collections.xml",
|
|
"configs/config.d/filesystem_caches.xml",
|
|
"configs/config.d/remote_servers.xml",
|
|
"configs/config.d/metadata_log.xml",
|
|
"configs/config.d/disable_s3_retries.xml",
|
|
],
|
|
user_configs=[
|
|
"configs/users.d/users.xml",
|
|
"configs/users.d/enable_writes.xml",
|
|
],
|
|
env_variables={
|
|
"RUST_BACKTRACE": "1",
|
|
},
|
|
with_minio=True,
|
|
with_azurite=True,
|
|
stay_alive=True,
|
|
with_zookeeper=True,
|
|
macros={"shard": 0, "replica": 1},
|
|
)
|
|
cluster.add_instance(
|
|
"node2",
|
|
main_configs=[
|
|
"configs/config.d/named_collections.xml",
|
|
"configs/config.d/remote_servers.xml",
|
|
"configs/config.d/metadata_log.xml",
|
|
"configs/config.d/disable_s3_retries.xml",
|
|
],
|
|
user_configs=[
|
|
"configs/users.d/users.xml",
|
|
"configs/users.d/enable_writes.xml",
|
|
],
|
|
with_minio=True,
|
|
stay_alive=True,
|
|
with_zookeeper=True,
|
|
with_remote_database_disk=False, # Disable `with_remote_database_disk` as in `test_replicated_database_and_unavailable_s3``, minIO rejects node2 connections
|
|
macros={"shard": 0, "replica": 2},
|
|
)
|
|
cluster.add_instance(
|
|
"node_with_environment_credentials",
|
|
with_minio=True,
|
|
main_configs=[
|
|
"configs/config.d/named_collections.xml",
|
|
"configs/config.d/use_environment_credentials.xml",
|
|
"configs/config.d/metadata_log.xml",
|
|
],
|
|
user_configs=[
|
|
"configs/allow_server_credentials.xml",
|
|
],
|
|
env_variables={
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
},
|
|
with_remote_database_disk=False,
|
|
)
|
|
cluster.add_instance(
|
|
"node_old",
|
|
main_configs=[
|
|
"configs/config.d/named_collections.xml",
|
|
"configs/config.d/filesystem_caches.xml",
|
|
"configs/config.d/remote_servers.xml",
|
|
"configs/config.d/metadata_log.xml",
|
|
"configs/config.d/remove_masking_rules.xml",
|
|
],
|
|
user_configs=["configs/users.d/users.xml"],
|
|
with_installed_binary=True,
|
|
image="clickhouse/clickhouse-server",
|
|
tag=CLICKHOUSE_CI_MIN_TESTED_VERSION,
|
|
with_minio=True,
|
|
with_azurite=True,
|
|
stay_alive=True,
|
|
with_zookeeper=True,
|
|
)
|
|
cluster.add_instance(
|
|
"node_with_disabled_delta_kernel",
|
|
main_configs=[
|
|
"configs/config.d/named_collections.xml",
|
|
"configs/config.d/filesystem_caches.xml",
|
|
"configs/config.d/remote_servers.xml",
|
|
"configs/config.d/metadata_log.xml",
|
|
],
|
|
user_configs=[
|
|
"configs/users.d/users.xml",
|
|
"configs/users.d/disabled_delta_kernel.xml",
|
|
],
|
|
with_minio=True,
|
|
with_azurite=True,
|
|
stay_alive=True,
|
|
with_zookeeper=True,
|
|
)
|
|
|
|
logging.info("Starting cluster...")
|
|
cluster.start()
|
|
|
|
if int(cluster.instances["node1"].query("SELECT count() FROM system.table_engines WHERE name = 'DeltaLake'").strip()) == 0:
|
|
pytest.skip(
|
|
"DeltaLake engine is not available"
|
|
)
|
|
|
|
cluster.default_s3_uploader = S3Uploader(
|
|
cluster.minio_client, cluster.minio_bucket
|
|
)
|
|
|
|
cluster.minio_restricted_bucket = "{}-with-auth".format(cluster.minio_bucket)
|
|
if cluster.minio_client.bucket_exists(cluster.minio_restricted_bucket):
|
|
cluster.minio_client.remove_bucket(cluster.minio_restricted_bucket)
|
|
|
|
cluster.minio_client.make_bucket(cluster.minio_restricted_bucket)
|
|
|
|
cluster.azure_container_name = "mycontainer"
|
|
cluster.blob_service_client = cluster.blob_service_client
|
|
container_client = cluster.blob_service_client.create_container(
|
|
cluster.azure_container_name
|
|
)
|
|
cluster.container_client = container_client
|
|
cluster.default_azure_uploader = AzureUploader(
|
|
cluster.blob_service_client, cluster.azure_container_name
|
|
)
|
|
|
|
# Only support local delta tables on the first node for now
|
|
# extend this if testing on other nodes becomes necessary
|
|
cluster.local_uploader = LocalUploader(cluster.instances["node1"])
|
|
|
|
cluster.spark_session = ResilientSparkSession(
|
|
lambda: get_spark(cluster.instances_dir)
|
|
)
|
|
|
|
for file in S3_DATA:
|
|
print(f"Copying object {file}")
|
|
cluster.minio_client.fput_object(
|
|
bucket_name=cluster.minio_bucket,
|
|
object_name=file,
|
|
file_path=os.path.join(
|
|
os.path.join(os.path.dirname(os.path.realpath(__file__))), file
|
|
),
|
|
)
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
def write_delta_from_file(spark, path, result_path, mode="overwrite"):
|
|
spark.read.load(path).write.mode(mode).option("compression", "none").format(
|
|
"delta"
|
|
).option("delta.columnMapping.mode", "name").save(result_path)
|
|
|
|
|
|
def write_delta_from_df(
|
|
spark, df, result_path, mode="overwrite", partition_by=None, column_mapping="name"
|
|
):
|
|
if partition_by is None:
|
|
df.write.mode(mode).option("compression", "none").option(
|
|
"delta.columnMapping.mode", column_mapping
|
|
).format("delta").save(result_path)
|
|
else:
|
|
df.write.mode(mode).option("compression", "none").format("delta").option(
|
|
"delta.columnMapping.mode", column_mapping
|
|
).partitionBy("a").save(result_path)
|
|
|
|
|
|
def generate_data(spark, start, end):
|
|
a = spark.range(start, end, 1).toDF("a")
|
|
b = spark.range(start + 1, end + 1, 1).toDF("b")
|
|
b = b.withColumn("b", b["b"].cast(StringType()))
|
|
|
|
a = a.withColumn(
|
|
"row_index", row_number().over(Window.orderBy(monotonically_increasing_id()))
|
|
)
|
|
b = b.withColumn(
|
|
"row_index", row_number().over(Window.orderBy(monotonically_increasing_id()))
|
|
)
|
|
|
|
df = a.join(b, on=["row_index"]).drop("row_index")
|
|
return df
|
|
|
|
|
|
def get_storage_options(cluster):
|
|
return {
|
|
"AWS_ENDPOINT_URL": f"http://{cluster.minio_ip}:{cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": "minio",
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
|
|
|
|
def write_deltalake_with_retry(
|
|
table_uri, data, storage_options, retries=3, delay=5, **kwargs
|
|
):
|
|
"""
|
|
Wrapper around write_deltalake with test-level retries.
|
|
The deltalake library has internal retries, but they're too fast (~2s total)
|
|
when minio is temporarily overloaded. This adds longer delays between attempts.
|
|
"""
|
|
last_exception = None
|
|
for attempt in range(retries):
|
|
try:
|
|
write_deltalake(table_uri, data, storage_options=storage_options, **kwargs)
|
|
return
|
|
except OSError as e:
|
|
last_exception = e
|
|
if attempt < retries - 1:
|
|
logging.warning(
|
|
f"write_deltalake failed (attempt {attempt + 1}/{retries}): {e}. Retrying in {delay}s..."
|
|
)
|
|
time.sleep(delay)
|
|
raise last_exception
|
|
|
|
|
|
def get_delta_metadata(delta_metadata_file):
|
|
jsons = [json.loads(x) for x in delta_metadata_file.splitlines()]
|
|
combined_json = {}
|
|
for d in jsons:
|
|
combined_json.update(d)
|
|
return combined_json
|
|
|
|
|
|
def get_node(cluster, use_delta_kernel):
|
|
if use_delta_kernel == "1":
|
|
return cluster.instances["node1"]
|
|
elif use_delta_kernel == "0":
|
|
return cluster.instances["node_with_disabled_delta_kernel"]
|
|
else:
|
|
assert False
|
|
|
|
|
|
def create_delta_table(
|
|
instance,
|
|
storage_type,
|
|
table_name,
|
|
cluster,
|
|
format="Parquet",
|
|
**kwargs,
|
|
):
|
|
|
|
if storage_type == "s3":
|
|
if "bucket" in kwargs:
|
|
bucket = kwargs["bucket"]
|
|
else:
|
|
bucket = cluster.minio_bucket
|
|
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {table_name};
|
|
CREATE TABLE {table_name}
|
|
ENGINE=DeltaLake(s3, filename = '{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')
|
|
"""
|
|
)
|
|
|
|
elif storage_type == "azure":
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {table_name};
|
|
CREATE TABLE {table_name}
|
|
ENGINE=DeltaLakeAzure(azure, container = {cluster.azure_container_name}, storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '{table_name}', format={format})
|
|
"""
|
|
)
|
|
elif storage_type == "local":
|
|
# For local storage, we need to use the absolute path
|
|
user_files_path = os.path.join(
|
|
SCRIPT_DIR, f"{cluster.instances_dir_name}/node1/database/user_files"
|
|
)
|
|
table_path = os.path.join(user_files_path, table_name)
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {table_name};
|
|
CREATE TABLE {table_name}
|
|
ENGINE=DeltaLakeLocal('{table_path}', {format})
|
|
"""
|
|
)
|
|
else:
|
|
raise Exception(f"Unknown delta lake storage type: {storage_type}")
|
|
|
|
|
|
def default_upload_directory(
|
|
started_cluster, storage_type, local_path, remote_path, **kwargs
|
|
):
|
|
if storage_type == "s3":
|
|
print(kwargs)
|
|
return started_cluster.default_s3_uploader.upload_directory(
|
|
local_path, remote_path, **kwargs
|
|
)
|
|
elif storage_type == "azure":
|
|
# Azure blob storage preserves leading slashes in blob names, unlike S3/MinIO which strips
|
|
# them. Use relative-path mode with a stripped remote prefix so that blob names match what
|
|
# delta-kernel-rs expects (no leading slash in the az:// table location path component).
|
|
effective_remote = remote_path.lstrip("/") if remote_path else local_path.lstrip("/")
|
|
azure_uploader = AzureUploader(
|
|
started_cluster.blob_service_client,
|
|
started_cluster.azure_container_name,
|
|
use_relpath=True,
|
|
)
|
|
return azure_uploader.upload_directory(local_path, effective_remote, **kwargs)
|
|
elif storage_type == "local":
|
|
return started_cluster.local_uploader.upload_directory(
|
|
local_path, remote_path, **kwargs
|
|
)
|
|
else:
|
|
raise Exception(f"Unknown delta storage type: {storage_type}")
|
|
|
|
|
|
def create_initial_data_file(
|
|
cluster,
|
|
node,
|
|
query,
|
|
table_name,
|
|
compression_method="none",
|
|
node_name="node1",
|
|
settings=None,
|
|
):
|
|
settings = settings or {}
|
|
|
|
merged_settings = {
|
|
"output_format_parquet_compression_method": compression_method,
|
|
"s3_truncate_on_insert": 1,
|
|
**settings,
|
|
}
|
|
|
|
settings_sql = ",\n ".join(
|
|
f"{k}='{v}'" if isinstance(v, str) else f"{k}={v}"
|
|
for k, v in merged_settings.items()
|
|
)
|
|
|
|
node.query(
|
|
f"""
|
|
INSERT INTO TABLE FUNCTION
|
|
file('{table_name}.parquet')
|
|
SETTINGS
|
|
{settings_sql}
|
|
{query}
|
|
FORMAT Parquet"""
|
|
)
|
|
|
|
user_files_path = os.path.join(
|
|
os.path.join(os.path.dirname(os.path.realpath(__file__))),
|
|
f"{cluster.instances_dir_name}/{node_name}/database/user_files",
|
|
)
|
|
result_path = f"{user_files_path}/{table_name}.parquet"
|
|
return result_path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel, storage_type",
|
|
[("1", "s3"), ("0", "s3"), ("0", "azure"), ("1", "azure"), ("1", "local")],
|
|
)
|
|
def test_single_log_file(started_cluster, use_delta_kernel, storage_type):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_single_log_file")
|
|
|
|
inserted_data = "SELECT number as a, toString(number + 1) as b FROM numbers(100)"
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster, instance, inserted_data, TABLE_NAME, node_name=instance.name
|
|
)
|
|
|
|
# For local storage, we need to use the absolute path
|
|
user_files_path = os.path.join(
|
|
SCRIPT_DIR, f"{cluster.instances_dir_name}/{instance.name}/database/user_files"
|
|
)
|
|
table_path = os.path.join(user_files_path, TABLE_NAME)
|
|
|
|
# We need to exclude the leading slash for local storage protocol file://
|
|
delta_path = table_path if storage_type == "local" else f"/{TABLE_NAME}"
|
|
write_delta_from_file(spark, parquet_data_path, delta_path)
|
|
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
storage_type,
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
assert len(files) == 2 # 1 metadata files + 1 data file
|
|
|
|
create_delta_table(
|
|
instance,
|
|
storage_type,
|
|
TABLE_NAME,
|
|
started_cluster,
|
|
)
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
assert instance.query(f"SELECT * FROM {TABLE_NAME}") == instance.query(
|
|
inserted_data
|
|
)
|
|
|
|
|
|
def test_single_log_file_azure_connection_string(started_cluster):
|
|
"""Test DeltaLakeAzure with connection string authentication and delta kernel enabled."""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_single_log_file_azure_cs")
|
|
|
|
inserted_data = "SELECT number as a, toString(number + 1) as b FROM numbers(100)"
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster, instance, inserted_data, TABLE_NAME, node_name=instance.name
|
|
)
|
|
|
|
delta_path = f"/{TABLE_NAME}"
|
|
write_delta_from_file(spark, parquet_data_path, delta_path)
|
|
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
"azure",
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
assert len(files) == 2 # 1 metadata file + 1 data file
|
|
|
|
connection_string = started_cluster.env_variables["AZURITE_CONNECTION_STRING"]
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME}
|
|
ENGINE=DeltaLakeAzure('{connection_string}', '{started_cluster.azure_container_name}', '{TABLE_NAME}', Parquet)
|
|
"""
|
|
)
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
assert instance.query(f"SELECT * FROM {TABLE_NAME}") == instance.query(
|
|
inserted_data
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel, storage_type",
|
|
[("1", "s3"), ("0", "s3"), ("0", "azure"), ("1", "azure"), ("1", "local")],
|
|
)
|
|
def test_partition_by(started_cluster, use_delta_kernel, storage_type):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
|
|
TABLE_NAME = randomize_table_name("test_partition_by")
|
|
|
|
# For local storage, we need to use the absolute path
|
|
user_files_path = os.path.join(
|
|
SCRIPT_DIR, f"{cluster.instances_dir_name}/{instance.name}/database/user_files"
|
|
)
|
|
table_path = os.path.join(user_files_path, TABLE_NAME)
|
|
|
|
# We need to exclude the leading slash for local storage protocol file://
|
|
delta_path = table_path if storage_type == "local" else f"/{TABLE_NAME}"
|
|
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, 0, 10),
|
|
delta_path,
|
|
mode="overwrite",
|
|
partition_by="a",
|
|
)
|
|
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
storage_type,
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
assert len(files) == 11 # 10 partitions and 1 metadata file
|
|
|
|
create_delta_table(
|
|
instance,
|
|
storage_type,
|
|
TABLE_NAME,
|
|
started_cluster,
|
|
)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 10
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel, storage_type",
|
|
[("1", "s3"), ("0", "s3"), ("0", "azure"), ("1", "azure"), ("1", "local")],
|
|
)
|
|
def test_checkpoint(started_cluster, use_delta_kernel, storage_type):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_checkpoint")
|
|
|
|
# For local storage, we need to use the absolute path
|
|
user_files_path = os.path.join(
|
|
SCRIPT_DIR, f"{cluster.instances_dir_name}/{instance.name}/database/user_files"
|
|
)
|
|
table_path = os.path.join(user_files_path, TABLE_NAME)
|
|
# We need to exclude the leading slash for local storage protocol file://
|
|
delta_path = table_path if storage_type == "local" else f"/{TABLE_NAME}"
|
|
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, 0, 1),
|
|
delta_path,
|
|
mode="overwrite",
|
|
)
|
|
for i in range(1, 25):
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, i, i + 1),
|
|
delta_path,
|
|
mode="append",
|
|
)
|
|
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
storage_type,
|
|
delta_path,
|
|
"",
|
|
)
|
|
# 25 data files
|
|
# 25 metadata files
|
|
# 1 last_metadata file
|
|
# 2 checkpoints
|
|
assert len(files) == 25 * 2 + 3
|
|
|
|
ok = False
|
|
for file in files:
|
|
if file.endswith("last_checkpoint"):
|
|
ok = True
|
|
assert ok
|
|
|
|
create_delta_table(
|
|
instance,
|
|
storage_type,
|
|
TABLE_NAME,
|
|
started_cluster,
|
|
)
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} SETTINGS input_format_parquet_allow_missing_columns=1"
|
|
)
|
|
)
|
|
== 25
|
|
)
|
|
|
|
table = DeltaTable.forPath(spark, delta_path)
|
|
table.delete("a < 10")
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
storage_type,
|
|
delta_path,
|
|
"",
|
|
)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 15
|
|
|
|
for i in range(0, 5):
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, i, i + 1),
|
|
delta_path,
|
|
mode="append",
|
|
)
|
|
# + 1 metadata files (for delete)
|
|
# + 5 data files
|
|
# + 5 metadata files
|
|
# + 1 checkpoint file
|
|
# + 1 ?
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
storage_type,
|
|
delta_path,
|
|
"",
|
|
)
|
|
assert len(files) == 53 + 1 + 5 * 2 + 1 + 1
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 20
|
|
|
|
assert (
|
|
instance.query(f"SELECT * FROM {TABLE_NAME} ORDER BY 1").strip()
|
|
== instance.query(
|
|
"SELECT * FROM ("
|
|
"SELECT number, toString(number + 1) FROM numbers(5) "
|
|
"UNION ALL SELECT number, toString(number + 1) FROM numbers(10, 15) "
|
|
") ORDER BY 1"
|
|
).strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_multiple_log_files(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_multiple_log_files")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) == 2 # 1 metadata files + 1 data file
|
|
|
|
s3_objects = list(
|
|
minio_client.list_objects(bucket, f"{TABLE_NAME}/_delta_log/", recursive=True)
|
|
)
|
|
assert len(s3_objects) == 1
|
|
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 100, 200), f"/{TABLE_NAME}", mode="append"
|
|
)
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) == 4 # 2 metadata files + 2 data files
|
|
|
|
s3_objects = list(
|
|
minio_client.list_objects(bucket, f"{TABLE_NAME}/_delta_log/", recursive=True)
|
|
)
|
|
assert len(s3_objects) == 2
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 200
|
|
assert instance.query(f"SELECT * FROM {TABLE_NAME} ORDER BY 1") == instance.query(
|
|
"SELECT number, toString(number + 1) FROM numbers(200)"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_metadata(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_metadata")
|
|
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT number, toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=instance.name,
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
|
|
data = get_file_contents(
|
|
minio_client,
|
|
bucket,
|
|
f"/{TABLE_NAME}/_delta_log/00000000000000000000.json",
|
|
)
|
|
delta_metadata = get_delta_metadata(data)
|
|
|
|
stats = json.loads(delta_metadata["add"]["stats"])
|
|
assert stats["numRecords"] == 100
|
|
assert next(iter(stats["minValues"].values())) == 0
|
|
assert next(iter(stats["maxValues"].values())) == 99
|
|
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_types(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
TABLE_NAME = randomize_table_name("test_types")
|
|
spark = started_cluster.spark_session
|
|
result_file = randomize_table_name(f"{TABLE_NAME}_result_2")
|
|
|
|
(
|
|
DeltaTable.create(spark)
|
|
.tableName(TABLE_NAME)
|
|
.location(f"/{result_file}")
|
|
.addColumn("a", "INT", nullable=True)
|
|
.addColumn("b", "STRING", nullable=False)
|
|
.addColumn("c", "DATE", nullable=False)
|
|
.addColumn("d", "ARRAY<STRING>", nullable=False)
|
|
.addColumn("e", "BOOLEAN", nullable=True)
|
|
.addColumn("f", ArrayType(StringType(), containsNull=False), nullable=False)
|
|
.execute()
|
|
)
|
|
data = [
|
|
(
|
|
123,
|
|
"string",
|
|
datetime.strptime("2000-01-01", "%Y-%m-%d"),
|
|
["str1", "str2"],
|
|
True,
|
|
["str1", "str4"],
|
|
)
|
|
]
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", IntegerType(), nullable=True),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", DateType(), nullable=False),
|
|
StructField("d", ArrayType(StringType())),
|
|
StructField("e", BooleanType(), nullable=False),
|
|
StructField("f", ArrayType(StringType(), containsNull=False)),
|
|
]
|
|
)
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.printSchema()
|
|
df.write.mode("append").format("delta").saveAsTable(TABLE_NAME)
|
|
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
upload_directory(minio_client, bucket, f"/{result_file}", "")
|
|
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME} ENGINE=DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"""
|
|
)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 1
|
|
assert (
|
|
instance.query(f"SELECT * FROM {TABLE_NAME}").strip()
|
|
== "123\tstring\t2000-01-01\t['str1','str2']\ttrue\t['str1','str4']"
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
assert (
|
|
instance.query(f"SELECT * FROM {table_function}").strip()
|
|
== "123\tstring\t2000-01-01\t['str1','str2']\ttrue\t['str1','str4']"
|
|
)
|
|
|
|
assert instance.query(f"DESCRIBE {table_function} FORMAT TSV") == TSV(
|
|
[
|
|
["a", "Nullable(Int32)"],
|
|
["b", "String"],
|
|
["c", "Date32"],
|
|
["d", "Array(Nullable(String))"],
|
|
["e", "Nullable(Bool)"],
|
|
["f", "Array(String)"],
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_varchar_char_types(started_cluster, use_delta_kernel):
|
|
"""
|
|
VARCHAR(n) and CHAR(n) are valid Delta Lake column types emitted by Spark/Databricks
|
|
when tables originate from relational databases. ClickHouse must map them to String.
|
|
"""
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
TABLE_NAME = randomize_table_name("test_varchar_char_types")
|
|
spark = started_cluster.spark_session
|
|
result_file = randomize_table_name(f"{TABLE_NAME}_result")
|
|
|
|
# Use the DeltaTable builder with DDL-style type names so that the Delta Lake
|
|
# schema metadata records varchar/char column types. Using VarcharType/CharType
|
|
# directly in a StructType would be rejected by Spark's Catalyst planner.
|
|
(
|
|
DeltaTable.create(spark)
|
|
.tableName(TABLE_NAME)
|
|
.location(f"/{result_file}")
|
|
.addColumn("id", "INT", nullable=False)
|
|
.addColumn("varchar_col", "VARCHAR(256)", nullable=True)
|
|
.addColumn("char_col", "CHAR(10)", nullable=True)
|
|
.execute()
|
|
)
|
|
spark.sql(
|
|
f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello varchar', 'hello char')"
|
|
)
|
|
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
upload_directory(minio_client, bucket, f"/{result_file}", "")
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
# Both varchar and char columns must be readable and resolve to String
|
|
assert instance.query(f"DESCRIBE {table_function} FORMAT TSV") == TSV(
|
|
[
|
|
["id", "Int32"],
|
|
["varchar_col", "Nullable(String)"],
|
|
["char_col", "Nullable(String)"],
|
|
]
|
|
)
|
|
assert (
|
|
instance.query(f"SELECT id, varchar_col, char_col FROM {table_function}").strip()
|
|
== "1\thello varchar\thello char"
|
|
)
|
|
|
|
|
|
def test_varchar_char_types_in_delta_log(started_cluster):
|
|
"""
|
|
Regression test for: `Unsupported DeltaLake type: varchar(n)`.
|
|
|
|
Spark/Delta normalises VARCHAR/CHAR DDL down to the physical `string` type in
|
|
the Delta log, so a Spark-driven test never exercises the broken code path.
|
|
Some real-world emitters (notably the Unity Catalog REST API in `type_json`)
|
|
do leave `varchar(n)` / `char(n)` in the schema string, which is what hits
|
|
`DeltaLakeMetadata::getSimpleTypeByName` and previously threw.
|
|
|
|
This test constructs a Delta log by hand with such non-normalised type names
|
|
so the regression path is exercised under both the bugfix-validation run
|
|
(must fail without the fix) and the patched build (must succeed).
|
|
The bug only ever lived in the non-kernel path, so this test does not
|
|
parametrise on `use_delta_kernel`.
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_varchar_char_types_in_delta_log")
|
|
|
|
schema = pa.schema([
|
|
pa.field("id", pa.int32()),
|
|
pa.field("varchar_col", pa.string()),
|
|
pa.field("char_col", pa.string()),
|
|
])
|
|
table_data = pa.table(
|
|
{
|
|
"id": pa.array([1], type=pa.int32()),
|
|
"varchar_col": pa.array(["hello varchar"], type=pa.string()),
|
|
"char_col": pa.array(["hello char"], type=pa.string()),
|
|
},
|
|
schema=schema,
|
|
)
|
|
buf = io.BytesIO()
|
|
pq.write_table(table_data, buf, compression=None)
|
|
parquet_bytes = buf.getvalue()
|
|
|
|
parquet_object_name = f"{TABLE_NAME}/part-0.parquet"
|
|
minio_client.put_object(
|
|
bucket_name=bucket,
|
|
object_name=parquet_object_name,
|
|
data=io.BytesIO(parquet_bytes),
|
|
length=len(parquet_bytes),
|
|
)
|
|
|
|
protocol = '{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}'
|
|
metadata = json.dumps(
|
|
{
|
|
"metaData": {
|
|
"id": str(uuid.uuid4()),
|
|
"format": {"provider": "parquet", "options": {}},
|
|
"schemaString": json.dumps(
|
|
{
|
|
"type": "struct",
|
|
"fields": [
|
|
{
|
|
"name": "id",
|
|
"type": "integer",
|
|
"nullable": True,
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"name": "varchar_col",
|
|
"type": "varchar(256)",
|
|
"nullable": True,
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"name": "char_col",
|
|
"type": "char(10)",
|
|
"nullable": True,
|
|
"metadata": {},
|
|
},
|
|
],
|
|
}
|
|
),
|
|
"partitionColumns": [],
|
|
"configuration": {},
|
|
"createdTime": 1600000000000,
|
|
}
|
|
}
|
|
)
|
|
add = json.dumps(
|
|
{
|
|
"add": {
|
|
"path": "part-0.parquet",
|
|
"partitionValues": {},
|
|
"size": len(parquet_bytes),
|
|
"modificationTime": 1600000000000,
|
|
"dataChange": True,
|
|
"stats": json.dumps({"numRecords": 1}),
|
|
}
|
|
}
|
|
)
|
|
|
|
log_content = ("\n".join([protocol, metadata, add])).encode()
|
|
minio_client.put_object(
|
|
bucket_name=bucket,
|
|
object_name=f"{TABLE_NAME}/_delta_log/00000000000000000000.json",
|
|
data=io.BytesIO(log_content),
|
|
length=len(log_content),
|
|
)
|
|
|
|
table_function = (
|
|
f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{TABLE_NAME}/', "
|
|
f"'minio', '{minio_secret_key}')"
|
|
)
|
|
|
|
# `allow_experimental_delta_kernel_rs=0` exercises the C++ schema parser,
|
|
# which is where `getSimpleTypeByName` was throwing on `varchar(n)`/`char(n)`.
|
|
settings = {"allow_experimental_delta_kernel_rs": "0"}
|
|
assert instance.query(f"DESCRIBE {table_function} FORMAT TSV", settings=settings) == TSV(
|
|
[
|
|
["id", "Nullable(Int32)"],
|
|
["varchar_col", "Nullable(String)"],
|
|
["char_col", "Nullable(String)"],
|
|
]
|
|
)
|
|
assert (
|
|
instance.query(
|
|
f"SELECT id, varchar_col, char_col FROM {table_function}",
|
|
settings=settings,
|
|
).strip()
|
|
== "1\thello varchar\thello char"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_restart_broken(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = "broken"
|
|
TABLE_NAME = randomize_table_name("test_restart_broken")
|
|
|
|
if not minio_client.bucket_exists(bucket):
|
|
minio_client.make_bucket(bucket)
|
|
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT number, toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=instance.name,
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
|
|
create_delta_table(
|
|
instance,
|
|
"s3",
|
|
TABLE_NAME,
|
|
started_cluster,
|
|
bucket=bucket,
|
|
)
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
s3_objects = list_s3_objects(minio_client, bucket, prefix="")
|
|
assert (
|
|
len(
|
|
list(
|
|
minio_client.remove_objects(
|
|
bucket,
|
|
[DeleteObject(obj) for obj in s3_objects],
|
|
)
|
|
)
|
|
)
|
|
== 0
|
|
)
|
|
minio_client.remove_bucket(bucket)
|
|
|
|
instance.restart_clickhouse()
|
|
|
|
assert "NoSuchBucket" in instance.query_and_get_error(
|
|
f"SELECT count() FROM {TABLE_NAME}"
|
|
)
|
|
|
|
s3_disk_no_key_errors_metric_value = int(
|
|
instance.query(
|
|
"""
|
|
SELECT value
|
|
FROM system.metrics
|
|
WHERE metric = 'DiskS3NoSuchKeyErrors'
|
|
"""
|
|
).strip()
|
|
)
|
|
|
|
assert s3_disk_no_key_errors_metric_value == 0
|
|
|
|
minio_client.make_bucket(bucket)
|
|
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_restart_broken_table_function(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = "broken2"
|
|
TABLE_NAME = randomize_table_name("test_restart_broken_table_function")
|
|
|
|
if not minio_client.bucket_exists(bucket):
|
|
minio_client.make_bucket(bucket)
|
|
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT number, toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=instance.name,
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME}
|
|
AS deltaLake(s3, filename = '{TABLE_NAME}/', url = 'http://minio1:9001/{bucket}/')"""
|
|
)
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
s3_objects = list_s3_objects(minio_client, bucket, prefix="")
|
|
assert (
|
|
len(
|
|
list(
|
|
minio_client.remove_objects(
|
|
bucket,
|
|
[DeleteObject(obj) for obj in s3_objects],
|
|
)
|
|
)
|
|
)
|
|
== 0
|
|
)
|
|
minio_client.remove_bucket(bucket)
|
|
|
|
instance.restart_clickhouse()
|
|
|
|
assert "NoSuchBucket" in instance.query_and_get_error(
|
|
f"SELECT count() FROM {TABLE_NAME}"
|
|
)
|
|
|
|
minio_client.make_bucket(bucket)
|
|
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel, cluster",
|
|
[("1", False), ("1", True)],
|
|
# [("1", False), ("1", True), ("0", False)], /// FIXME: this does not work with disabled delta-kernel
|
|
)
|
|
def test_partition_columns(started_cluster, use_delta_kernel, cluster):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_partition_columns")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = ["b", "c", "d", "e", "f", "g"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
num_rows = 9
|
|
|
|
schema = pa.schema(
|
|
[
|
|
pa.field("a", pa.int32(), nullable=True),
|
|
pa.field("b", pa.string(), nullable=True),
|
|
pa.field("c", pa.date32(), nullable=True),
|
|
pa.field("d", pa.int32(), nullable=True),
|
|
pa.field("e", pa.timestamp("us"), nullable=True),
|
|
pa.field("f", pa.bool_(), nullable=True),
|
|
pa.field("g", pa.decimal128(10, 2), nullable=True),
|
|
pa.field("h", pa.bool_(), nullable=True),
|
|
]
|
|
)
|
|
|
|
now = datetime.now()
|
|
for i in range(1, num_rows + 1):
|
|
data = {
|
|
"a": [i],
|
|
"b": [f"test{i}"],
|
|
"c": [datetime.strptime(f"2000-01-0{i}", "%Y-%m-%d").date()],
|
|
"d": [i],
|
|
"e": [
|
|
(
|
|
now
|
|
if i % 2 == 0
|
|
else datetime.strptime(
|
|
f"2012-01-0{i} 12:34:56.789123", "%Y-%m-%d %H:%M:%S.%f"
|
|
)
|
|
)
|
|
],
|
|
"f": [True if i % 2 == 0 else False],
|
|
"g": [Decimal(f"{i * 1.11:.2f}")],
|
|
"h": [False if i % 2 == 0 else True],
|
|
}
|
|
write_deltalake_with_retry(
|
|
f"s3://{bucket}/{result_file}",
|
|
pa.Table.from_pydict(data, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
partition_by=partition_columns,
|
|
mode="append",
|
|
)
|
|
|
|
if cluster:
|
|
table_function = f"deltaLakeCluster(cluster, 'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
else:
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
result = instance.query(f"describe table {table_function}").strip()
|
|
assert (
|
|
result == "a\tNullable(Int32)\t\t\t\t\t\n"
|
|
"b\tNullable(String)\t\t\t\t\t\n"
|
|
"c\tNullable(Date32)\t\t\t\t\t\n"
|
|
"d\tNullable(Int32)\t\t\t\t\t\n"
|
|
"e\tNullable(DateTime64(6))\t\t\t\t\t\n"
|
|
"f\tNullable(Bool)\t\t\t\t\t\n"
|
|
"g\tNullable(Decimal(10, 2))\t\t\t\t\t\n"
|
|
"h\tNullable(Bool)"
|
|
)
|
|
|
|
result = int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert result == num_rows
|
|
|
|
expected_output = f"""1 test1 2000-01-01 1 2012-01-01 12:34:56.789123 false 1.11 true
|
|
2 test2 2000-01-02 2 {now} true 2.22 false
|
|
3 test3 2000-01-03 3 2012-01-03 12:34:56.789123 false 3.33 true
|
|
4 test4 2000-01-04 4 {now} true 4.44 false
|
|
5 test5 2000-01-05 5 2012-01-05 12:34:56.789123 false 5.55 true
|
|
6 test6 2000-01-06 6 {now} true 6.66 false
|
|
7 test7 2000-01-07 7 2012-01-07 12:34:56.789123 false 7.77 true
|
|
8 test8 2000-01-08 8 {now} true 8.88 false
|
|
9 test9 2000-01-09 9 2012-01-09 12:34:56.789123 false 9.99 true"""
|
|
|
|
assert (
|
|
expected_output
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY b").strip()
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE c == toDateTime('2000/01/05')
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
assert result == 1
|
|
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE e = toDateTime64('{now}', 6)
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
assert result == 4
|
|
|
|
if use_delta_kernel == 1:
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert num_rows - 1 == int(
|
|
instance.query(
|
|
f"""
|
|
SELECT ProfileEvents['DeltaLakePartitionPrunedFiles']
|
|
FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'
|
|
"""
|
|
)
|
|
)
|
|
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME} (a Nullable(Int32), b Nullable(String), c Nullable(Date32), d Nullable(Int32), h Nullable(Bool))
|
|
ENGINE=DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')
|
|
"""
|
|
)
|
|
assert (
|
|
"""1 test1 2000-01-01 1 true
|
|
2 test2 2000-01-02 2 false
|
|
3 test3 2000-01-03 3 true
|
|
4 test4 2000-01-04 4 false
|
|
5 test5 2000-01-05 5 true
|
|
6 test6 2000-01-06 6 false
|
|
7 test7 2000-01-07 7 true
|
|
8 test8 2000-01-08 8 false
|
|
9 test9 2000-01-09 9 true"""
|
|
== instance.query(f"SELECT * FROM {TABLE_NAME} ORDER BY b").strip()
|
|
)
|
|
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} WHERE c == toDateTime('2000/01/05')"
|
|
)
|
|
)
|
|
== 1
|
|
)
|
|
|
|
# Subset of columns should work.
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME} (b Nullable(String), c Nullable(Date32), d Nullable(Int32))
|
|
ENGINE=DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')
|
|
"""
|
|
)
|
|
assert (
|
|
"""test1 2000-01-01 1
|
|
test2 2000-01-02 2
|
|
test3 2000-01-03 3
|
|
test4 2000-01-04 4
|
|
test5 2000-01-05 5
|
|
test6 2000-01-06 6
|
|
test7 2000-01-07 7
|
|
test8 2000-01-08 8
|
|
test9 2000-01-09 9"""
|
|
== instance.query(f"SELECT * FROM {TABLE_NAME} ORDER BY b").strip()
|
|
)
|
|
|
|
num_commits = 100 # to create a checkpoint
|
|
for i in range(num_rows + 1, num_rows + 1 + num_commits):
|
|
data = {
|
|
"a": [i],
|
|
"b": [f"test{i}"],
|
|
"c": [
|
|
datetime.strptime(
|
|
f"2000-01-{i%30 if i%30 != 0 else 1}", "%Y-%m-%d"
|
|
).date()
|
|
],
|
|
"d": [i],
|
|
"e": [
|
|
(
|
|
now
|
|
if i % 2 == 0
|
|
else datetime.strptime(
|
|
f"2012-01-{i%30 if i%30 != 0 else 1} 12:34:56.789123",
|
|
"%Y-%m-%d %H:%M:%S.%f",
|
|
)
|
|
)
|
|
],
|
|
"f": [True if i % 2 == 0 else False],
|
|
"g": [Decimal(f"{i * 1.1:.2f}")],
|
|
"h": [False if i % 2 == 0 else True],
|
|
}
|
|
write_deltalake_with_retry(
|
|
f"s3://{bucket}/{result_file}",
|
|
pa.Table.from_pydict(data, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
partition_by=partition_columns,
|
|
mode="append",
|
|
)
|
|
|
|
files = [
|
|
obj.object_name
|
|
for obj in minio_client.list_objects(bucket, result_file, recursive=True)
|
|
]
|
|
ok = False
|
|
for file in files:
|
|
if file.endswith("last_checkpoint"):
|
|
ok = True
|
|
assert ok, f"Files: {files}"
|
|
|
|
result = int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert result == num_rows + num_commits
|
|
|
|
assert (
|
|
f"""1 test1 2000-01-01 1 2012-01-01 12:34:56.789123 false 1.11 true
|
|
2 test2 2000-01-02 2 {now} true 2.22 false
|
|
3 test3 2000-01-03 3 2012-01-03 12:34:56.789123 false 3.33 true
|
|
4 test4 2000-01-04 4 {now} true 4.44 false
|
|
5 test5 2000-01-05 5 2012-01-05 12:34:56.789123 false 5.55 true
|
|
6 test6 2000-01-06 6 {now} true 6.66 false
|
|
7 test7 2000-01-07 7 2012-01-07 12:34:56.789123 false 7.77 true
|
|
8 test8 2000-01-08 8 {now} true 8.88 false
|
|
9 test9 2000-01-09 9 2012-01-09 12:34:56.789123 false 9.99 true
|
|
10 test10 2000-01-10 10 {now} true 11 false
|
|
11 test11 2000-01-11 11 2012-01-11 12:34:56.789123 false 12.1 true
|
|
12 test12 2000-01-12 12 {now} true 13.2 false
|
|
13 test13 2000-01-13 13 2012-01-13 12:34:56.789123 false 14.3 true
|
|
14 test14 2000-01-14 14 {now} true 15.4 false
|
|
15 test15 2000-01-15 15 2012-01-15 12:34:56.789123 false 16.5 true
|
|
16 test16 2000-01-16 16 {now} true 17.6 false
|
|
17 test17 2000-01-17 17 2012-01-17 12:34:56.789123 false 18.7 true
|
|
18 test18 2000-01-18 18 {now} true 19.8 false"""
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY a LIMIT 18").strip()
|
|
)
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} WHERE c == toDateTime('2000/01/15') and d < 20"
|
|
)
|
|
)
|
|
== 1
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_complex_types(started_cluster, use_delta_kernel):
|
|
node = get_node(started_cluster, use_delta_kernel)
|
|
|
|
schema = pa.schema(
|
|
[
|
|
pa.field("id", pa.int32(), nullable=False),
|
|
pa.field("name", pa.string(), nullable=False),
|
|
(
|
|
"address",
|
|
pa.struct(
|
|
[
|
|
("street", pa.string()),
|
|
("city", pa.string()),
|
|
("state", pa.string()),
|
|
]
|
|
),
|
|
),
|
|
("interests", pa.list_(pa.string())),
|
|
(
|
|
"metadata",
|
|
pa.map_(
|
|
pa.string(), pa.string()
|
|
), # Map with string keys and string values
|
|
),
|
|
]
|
|
)
|
|
|
|
# Create sample data
|
|
data = [
|
|
pa.array([1, 2, 3], type=pa.int32()),
|
|
pa.array(["John Doe", "Jane Smith", "Jake Johnson"], type=pa.string()),
|
|
pa.array(
|
|
[
|
|
{"street": "123 Elm St", "city": "Springfield", "state": "IL"},
|
|
{"street": "456 Maple St", "city": "Shelbyville", "state": "IL"},
|
|
{"street": "789 Oak St", "city": "Ogdenville", "state": "IL"},
|
|
],
|
|
type=schema.field("address").type,
|
|
),
|
|
pa.array(
|
|
[
|
|
pa.array(["dancing", "coding", "hiking"]),
|
|
pa.array(["dancing", "coding", "hiking"]),
|
|
pa.array(["dancing", "coding", "hiking"]),
|
|
],
|
|
type=schema.field("interests").type,
|
|
),
|
|
pa.array(
|
|
[
|
|
{"key1": "value1", "key2": "value2"},
|
|
{"key1": "value3", "key2": "value4"},
|
|
{"key1": "value5", "key2": "value6"},
|
|
],
|
|
type=schema.field("metadata").type,
|
|
),
|
|
]
|
|
|
|
endpoint_url = f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}"
|
|
aws_access_key_id = "minio"
|
|
aws_secret_access_key = "ClickHouse_Minio_P@ssw0rd"
|
|
table_name = randomize_table_name("test_complex_types")
|
|
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": endpoint_url,
|
|
"AWS_ACCESS_KEY_ID": aws_access_key_id,
|
|
"AWS_SECRET_ACCESS_KEY": aws_secret_access_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(path, table, storage_options=storage_options)
|
|
|
|
assert "1\n2\n3\n" in node.query(
|
|
f"SELECT id FROM deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' , 'minio', '{minio_secret_key}')"
|
|
)
|
|
assert (
|
|
"('123 Elm St','Springfield','IL')\n('456 Maple St','Shelbyville','IL')\n('789 Oak St','Ogdenville','IL')"
|
|
in node.query(
|
|
f"SELECT address FROM deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' , 'minio', '{minio_secret_key}')"
|
|
)
|
|
)
|
|
assert (
|
|
"{'key1':'value1','key2':'value2'}\n{'key1':'value3','key2':'value4'}\n{'key1':'value5','key2':'value6'}"
|
|
in node.query(
|
|
f"SELECT metadata FROM deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' , 'minio', '{minio_secret_key}')"
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_filesystem_cache(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
TABLE_NAME = randomize_table_name("test_filesystem_cache")
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
if not minio_client.bucket_exists(bucket):
|
|
minio_client.make_bucket(bucket)
|
|
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT toUInt64(number), toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=instance.name
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
query_id = f"{TABLE_NAME}-{uuid.uuid4()}"
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS filesystem_cache_name = 'cache1'",
|
|
query_id=query_id,
|
|
)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
count = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['CachedReadBufferCacheWriteBytes'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
assert 0 < int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['S3GetObject'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
query_id = f"{TABLE_NAME}-{uuid.uuid4()}"
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS filesystem_cache_name = 'cache1'",
|
|
query_id=query_id,
|
|
)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Parquet reader v3 reads very small files suboptimally when readBigAt is enabled
|
|
# (it reads last 64kb for parquet metadata unconditionally
|
|
# assuming most of it will be metadata, see Reader::readFileMetaData)
|
|
# So we end up reading the same data two times here with parquet reader v3.
|
|
# So we cannot check count == CachedReadBufferReadFromCacheBytes,
|
|
# but instead check that CachedReadBufferReadFromCacheBytes is no more than 2 times more :(
|
|
assert count * 2 > int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['CachedReadBufferReadFromCacheBytes'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
assert 0 == int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['S3GetObject'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_filesystem_cache_azure(started_cluster, use_delta_kernel):
|
|
# Regression test for https://github.com/ClickHouse/ClickHouse/issues/106090:
|
|
# AzureObjectStorage::getObjectMetadata used to return an empty etag, which
|
|
# silently disabled the object-storage filesystem cache for Azure-backed
|
|
# tables (logging "Cannot use filesystem cache, no etag specified"). Without
|
|
# the fix the second read below still hits Azure (AzureGetObject > 0) and
|
|
# nothing is written to the cache (CachedReadBufferCacheWriteBytes == 0).
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_filesystem_cache_azure")
|
|
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT toUInt64(number), toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=instance.name,
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
default_upload_directory(started_cluster, "azure", f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "azure", TABLE_NAME, started_cluster)
|
|
|
|
query_id = f"{TABLE_NAME}-{uuid.uuid4()}"
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS filesystem_cache_name = 'cache1'",
|
|
query_id=query_id,
|
|
)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
count = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['CachedReadBufferCacheWriteBytes'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
# The first read must populate the cache (this is 0 without the etag fix).
|
|
assert count > 0
|
|
assert 0 < int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['AzureGetObject'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
query_id = f"{TABLE_NAME}-{uuid.uuid4()}"
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS filesystem_cache_name = 'cache1'",
|
|
query_id=query_id,
|
|
)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# See the comment in test_filesystem_cache about parquet reader v3 reading
|
|
# small files twice, hence the "no more than 2x" check instead of equality.
|
|
assert count * 2 > int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['CachedReadBufferReadFromCacheBytes'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
# The second read must be served entirely from the filesystem cache.
|
|
assert 0 == int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['AzureGetObject'] FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_replicated_database_and_unavailable_s3(started_cluster, use_delta_kernel):
|
|
node1 = started_cluster.instances["node1"]
|
|
node2 = started_cluster.instances["node2"]
|
|
|
|
DB_NAME = randomize_table_name("db")
|
|
TABLE_NAME = randomize_table_name("test_replicated_database_and_unavailable_s3")
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_restricted_bucket
|
|
|
|
if not minio_client.bucket_exists(bucket):
|
|
minio_client.make_bucket(bucket)
|
|
|
|
node1.query(
|
|
f"CREATE DATABASE {DB_NAME} ENGINE=Replicated('/clickhouse/databases/{DB_NAME}', 'shard1', 'node1')"
|
|
)
|
|
node2.query(
|
|
f"CREATE DATABASE {DB_NAME} ENGINE=Replicated('/clickhouse/databases/{DB_NAME}', 'shard1', 'node2')"
|
|
)
|
|
|
|
create_initial_data_file(
|
|
started_cluster,
|
|
node1,
|
|
"SELECT number, toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=node1.name,
|
|
)
|
|
|
|
endpoint_url = f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}"
|
|
aws_access_key_id = "minio"
|
|
aws_secret_access_key = "ClickHouse_Minio_P@ssw0rd"
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("id", pa.int32()),
|
|
("name", pa.string()),
|
|
]
|
|
)
|
|
|
|
data = [
|
|
pa.array([1, 2, 3], type=pa.int32()),
|
|
pa.array(["John Doe", "Jane Smith", "Jake Johnson"], type=pa.string()),
|
|
]
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": endpoint_url,
|
|
"AWS_ACCESS_KEY_ID": aws_access_key_id,
|
|
"AWS_SECRET_ACCESS_KEY": aws_secret_access_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{TABLE_NAME}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(path, table, storage_options=storage_options)
|
|
|
|
with PartitionManager() as pm:
|
|
pm_rule_reject = {
|
|
"instance": node2,
|
|
"probability": 1,
|
|
"destination": node2.ip_address,
|
|
"source_port": started_cluster.minio_port,
|
|
"action": "REJECT --reject-with tcp-reset",
|
|
"protocol": "tcp",
|
|
}
|
|
pm.add_rule(pm_rule_reject)
|
|
|
|
node1.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {DB_NAME}.{TABLE_NAME};
|
|
CREATE TABLE {DB_NAME}.{TABLE_NAME}
|
|
AS deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{TABLE_NAME}' , 'minio', '{minio_secret_key}')
|
|
"""
|
|
)
|
|
|
|
assert TABLE_NAME in node1.query(
|
|
f"select name from system.tables where database = '{DB_NAME}'"
|
|
)
|
|
assert TABLE_NAME in node2.query(
|
|
f"select name from system.tables where database = '{DB_NAME}'"
|
|
)
|
|
|
|
replica_path = f"/clickhouse/databases/{DB_NAME}/replicas/shard1|node2"
|
|
zk = started_cluster.get_kazoo_client("zoo1")
|
|
zk.set(replica_path + "/digest", "123456".encode())
|
|
|
|
# Compare the `digest` value exactly instead of substring-matching the
|
|
# whole dump of all znodes: the recomputed digest is a 64-bit hash whose
|
|
# decimal representation can incidentally contain "123456" as a substring.
|
|
assert (
|
|
node2.query(
|
|
f"SELECT value FROM system.zookeeper WHERE path = '{replica_path}' AND name = 'digest'"
|
|
).strip()
|
|
== "123456"
|
|
)
|
|
|
|
node2.restart_clickhouse()
|
|
|
|
assert (
|
|
node2.query(
|
|
f"SELECT value FROM system.zookeeper WHERE path = '{replica_path}' AND name = 'digest'"
|
|
).strip()
|
|
!= "123456"
|
|
)
|
|
|
|
|
|
def test_session_token(started_cluster):
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
TABLE_NAME = randomize_table_name("test_session_token")
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
if not minio_client.bucket_exists(bucket):
|
|
minio_client.make_bucket(bucket)
|
|
|
|
node_name = "node_with_environment_credentials"
|
|
instance = started_cluster.instances[node_name]
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster,
|
|
instance,
|
|
"SELECT toUInt64(number), toString(number) FROM numbers(100)",
|
|
TABLE_NAME,
|
|
node_name=node_name,
|
|
)
|
|
|
|
write_delta_from_file(spark, parquet_data_path, f"/{TABLE_NAME}")
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
|
|
assert 0 < int(
|
|
instance.query(
|
|
f"""
|
|
SELECT count() FROM deltaLake(
|
|
'http://{started_cluster.minio_host}:{started_cluster.minio_port}/{started_cluster.minio_bucket}/{TABLE_NAME}/',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
)
|
|
)
|
|
|
|
instance2 = started_cluster.instances["node1"]
|
|
|
|
assert 0 < int(
|
|
instance2.query(
|
|
f"""
|
|
SELECT count() FROM deltaLake(
|
|
'http://{started_cluster.minio_host}:{started_cluster.minio_port}/{started_cluster.minio_bucket}/{TABLE_NAME}/', '{minio_access_key}', '{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
)
|
|
)
|
|
|
|
assert (
|
|
"Received DeltaLake kernel error ObjectStoreError: Error interacting with object store"
|
|
in instance2.query_and_get_error(
|
|
f"""
|
|
SELECT count() FROM deltaLake(
|
|
'http://{started_cluster.minio_host}:{started_cluster.minio_port}/{started_cluster.minio_bucket}/{TABLE_NAME}/', '{minio_access_key}', '{minio_secret_key}', 'fake-token',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("cluster", [False, True])
|
|
def test_partition_columns_2(started_cluster, cluster):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_partition_columns_2")
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("a", pa.int32()),
|
|
("b", pa.int32()),
|
|
("c", pa.int32()),
|
|
("d", pa.string()),
|
|
("e", pa.string()),
|
|
]
|
|
)
|
|
data = [
|
|
pa.array([1, 2, 3, 4, 5], type=pa.int32()),
|
|
pa.array([4, 5, 6, 7, 8], type=pa.int32()),
|
|
pa.array([7, 7, 8, 9, 10], type=pa.int32()),
|
|
pa.array(["aa", "bb", "cc", "aa", "bb"], type=pa.string()),
|
|
pa.array(["aa", "bb", "cc", "aa", "cc"], type=pa.string()),
|
|
]
|
|
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(
|
|
path, table, storage_options=storage_options, partition_by=["c", "d"]
|
|
)
|
|
|
|
if cluster:
|
|
delta_function = f"""
|
|
deltaLakeCluster(
|
|
cluster,
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
else:
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=0)
|
|
"""
|
|
|
|
num_files = int(
|
|
node.query(
|
|
f"SELECT uniqExact(_path) FROM {delta_function}",
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
)
|
|
)
|
|
assert num_files == 5
|
|
|
|
new_data = [
|
|
pa.array([2], type=pa.int32()),
|
|
pa.array([3], type=pa.int32()),
|
|
pa.array([7], type=pa.int32()),
|
|
pa.array(["aa"], type=pa.string()),
|
|
pa.array(["cc"], type=pa.string()),
|
|
]
|
|
new_table_data = pa.Table.from_arrays(new_data, schema=schema)
|
|
|
|
write_deltalake_with_retry(
|
|
path, new_table_data, storage_options=storage_options, mode="append"
|
|
)
|
|
|
|
assert (
|
|
"a\tNullable(Int32)\t\t\t\t\t\n"
|
|
"b\tNullable(Int32)\t\t\t\t\t\n"
|
|
"c\tNullable(Int32)\t\t\t\t\t\n"
|
|
"d\tNullable(String)\t\t\t\t\t\n"
|
|
"e\tNullable(String)"
|
|
== node.query(
|
|
f"DESCRIBE TABLE {delta_function}",
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
).strip()
|
|
)
|
|
|
|
num_files = int(
|
|
node.query(
|
|
f"SELECT uniqExact(_path) FROM {delta_function}",
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
)
|
|
)
|
|
assert num_files == 6
|
|
|
|
query_id = f"{table_name}-{uuid.uuid4()}"
|
|
|
|
assert (
|
|
"1\t4\t7\taa\taa\n2\t3\t7\taa\tcc\n2\t5\t7\tbb\tbb\n3\t6\t8\tcc\tcc\n4\t7\t9\taa\taa\n5\t8\t10\tbb\tcc"
|
|
== node.query(
|
|
f" SELECT * FROM {delta_function} ORDER BY all",
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"use_hive_partitioning": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
|
|
assert (
|
|
"1"
|
|
in node.query(
|
|
f" SELECT a FROM {delta_function} WHERE c = 7 and d = 'aa'",
|
|
query_id=query_id,
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"delta_lake_enable_engine_predicate": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
|
|
def check_pruned(count, query_id):
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
assert count == int(
|
|
node.query(
|
|
f"""
|
|
SELECT ProfileEvents['DeltaLakePartitionPrunedFiles']
|
|
FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'
|
|
"""
|
|
)
|
|
)
|
|
|
|
check_pruned(num_files - 2, query_id)
|
|
|
|
query_id = f"{table_name}-{uuid.uuid4()}"
|
|
assert (
|
|
"2"
|
|
in node.query(
|
|
f"SELECT a FROM {delta_function} WHERE c = 7 and d = 'bb'",
|
|
query_id=query_id,
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"delta_lake_enable_engine_predicate": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
|
|
check_pruned(num_files - 1, query_id)
|
|
|
|
|
|
def test_partition_by_nullable_bool(started_cluster):
|
|
# Nullable(Bool) partition column with true / false / NULL in separate files, exercising
|
|
# both the Bool literal visitor and the null visitor; all must yield Nullable(Bool).
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_partition_by_nullable_bool")
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("id", pa.int32()),
|
|
("flag", pa.bool_()),
|
|
]
|
|
)
|
|
data = [
|
|
pa.array([1, 2, 3], type=pa.int32()),
|
|
pa.array([True, False, None], type=pa.bool_()),
|
|
]
|
|
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(
|
|
path, table, storage_options=storage_options, partition_by=["flag"]
|
|
)
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}',
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
# The partition column must be advertised as Nullable(Bool), not Nullable(UInt8).
|
|
assert (
|
|
"id\tNullable(Int32)\t\t\t\t\t\n"
|
|
"flag\tNullable(Bool)"
|
|
== node.query(
|
|
f"DESCRIBE TABLE {delta_function}",
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
).strip()
|
|
)
|
|
|
|
# true / false / NULL are materialized correctly across the separate data files.
|
|
assert (
|
|
"1\ttrue\n2\tfalse\n3\t\\N"
|
|
== node.query(
|
|
f"SELECT id, flag FROM {delta_function} ORDER BY id",
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"use_hive_partitioning": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
|
|
|
|
def test_partition_by_binary(started_cluster):
|
|
# Delta `binary` partition column with two non-NULL values and a NULL in separate files,
|
|
# exercising both the binary literal visitor and the null visitor. `binary` is advertised as
|
|
# Nullable(String), so every per-file transform must yield Nullable(String); a stray FixedString
|
|
# on the non-NULL files would fail to unify with the schema's Nullable(String).
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_partition_by_binary")
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("id", pa.int32()),
|
|
("data", pa.binary()),
|
|
]
|
|
)
|
|
data = [
|
|
pa.array([1, 2, 3], type=pa.int32()),
|
|
pa.array([b"aa", b"bb", None], type=pa.binary()),
|
|
]
|
|
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(
|
|
path, table, storage_options=storage_options, partition_by=["data"]
|
|
)
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}',
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
# The partition column must be advertised as Nullable(String), not Nullable(FixedString(N)).
|
|
assert (
|
|
"id\tNullable(Int32)\t\t\t\t\t\n"
|
|
"data\tNullable(String)"
|
|
== node.query(
|
|
f"DESCRIBE TABLE {delta_function}",
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
).strip()
|
|
)
|
|
|
|
# The non-NULL and NULL partition files must unify: the transform type is Nullable(String) for
|
|
# every file. We assert the runtime type and null-ness rather than the exact bytes, since delta-rs
|
|
# escapes binary partition values in the log.
|
|
assert (
|
|
"1\tNullable(String)\t0\n"
|
|
"2\tNullable(String)\t0\n"
|
|
"3\tNullable(String)\t1"
|
|
== node.query(
|
|
f"SELECT id, toTypeName(data), isNull(data) FROM {delta_function} ORDER BY id",
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"use_hive_partitioning": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"column_mapping", ["", "name"]
|
|
) # "id" is not supported by delta-kernel at the moment
|
|
def test_rename_and_add_column(started_cluster, column_mapping):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_rename_column")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
|
|
df = spark.createDataFrame([("alice", 47), ("anora", 23), ("aelin", 51)]).toDF(
|
|
"first_name", "age"
|
|
)
|
|
|
|
if column_mapping == "":
|
|
df.write.format("delta").partitionBy("age").save(path)
|
|
else:
|
|
df.write.format("delta").partitionBy("age").option(
|
|
"delta.minReaderVersion", "2"
|
|
).option("delta.minWriterVersion", "5").option(
|
|
"delta.columnMapping.mode", column_mapping
|
|
).save(
|
|
path
|
|
)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
|
|
def check_schema(expected):
|
|
assert expected == node.query(f"DESCRIBE TABLE {delta_function}").strip()
|
|
|
|
def check_data(expected):
|
|
assert (
|
|
expected
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
def append_data(df):
|
|
df.write.option("mergeSchema", "true").mode("append").format(
|
|
"delta"
|
|
).partitionBy("age").save(path)
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
def check_pruned_files(expected, query_id):
|
|
node.query("SYSTEM FLUSH LOGS")
|
|
assert expected == int(
|
|
node.query(
|
|
f"""
|
|
SELECT ProfileEvents['DeltaLakePartitionPrunedFiles']
|
|
FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'
|
|
"""
|
|
)
|
|
)
|
|
|
|
check_schema("first_name\tNullable(String)\t\t\t\t\t\nage\tNullable(Int64)")
|
|
check_data("aelin\t51\n" "alice\t47\n" "anora\t23")
|
|
|
|
spark.sql(f"CREATE TABLE {table_name} USING DELTA LOCATION '{path}'")
|
|
|
|
if column_mapping == "":
|
|
# To allow column rename
|
|
spark.sql(
|
|
f"""
|
|
ALTER TABLE {table_name}
|
|
SET TBLPROPERTIES ('delta.minReaderVersion'='2', 'delta.minWriterVersion'='5', 'delta.columnMapping.mode' = 'name')
|
|
"""
|
|
)
|
|
|
|
spark.sql(f"ALTER TABLE {table_name} RENAME COLUMN first_name TO naam")
|
|
|
|
df = spark.createDataFrame([("bob", 12), ("bill", 33), ("bober", 49)]).toDF(
|
|
"naam", "age"
|
|
)
|
|
append_data(df)
|
|
|
|
assert "Unknown expression identifier `first_name`" in node.query_and_get_error(
|
|
f"SELECT first_name FROM {delta_function} WHERE age = 51"
|
|
)
|
|
|
|
check_schema("naam\tNullable(String)\t\t\t\t\t\nage\tNullable(Int64)")
|
|
check_data(
|
|
"aelin\t51\n" "alice\t47\n" "anora\t23\n" "bill\t33\n" "bob\t12\n" "bober\t49"
|
|
)
|
|
|
|
query_id = f"{table_name}-{uuid.uuid4()}"
|
|
assert (
|
|
"bob"
|
|
== node.query(
|
|
f"SELECT naam FROM {delta_function} WHERE age = 12",
|
|
query_id=query_id,
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"delta_lake_enable_engine_predicate": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
check_pruned_files(5, query_id)
|
|
|
|
query_id = f"{table_name}-{uuid.uuid4()}"
|
|
assert (
|
|
"aelin"
|
|
== node.query(
|
|
f"SELECT naam FROM {delta_function} WHERE age = 51",
|
|
query_id=query_id,
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"delta_lake_enable_engine_predicate": 0,
|
|
},
|
|
).strip()
|
|
)
|
|
check_pruned_files(5, query_id)
|
|
|
|
df = spark.createDataFrame([("cicil", 68, "usa"), ("corsha", 26, "chaol")]).toDF(
|
|
"naam", "age", "country"
|
|
)
|
|
|
|
df.write.option("mergeSchema", "true").mode("append").format("delta").partitionBy(
|
|
"age"
|
|
).save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
assert (
|
|
"naam\tNullable(String)\t\t\t\t\t\n"
|
|
"age\tNullable(Int64)\t\t\t\t\t\n"
|
|
"country\tNullable(String)"
|
|
== node.query(f"DESCRIBE TABLE {delta_function}").strip()
|
|
)
|
|
|
|
assert (
|
|
"aelin\t51\t\\N\n"
|
|
"alice\t47\t\\N\n"
|
|
"anora\t23\t\\N\n"
|
|
"bill\t33\t\\N\n"
|
|
"bob\t12\t\\N\n"
|
|
"bober\t49\t\\N\n"
|
|
"cicil\t68\tusa\n"
|
|
"corsha\t26\tchaol"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
df = spark.createDataFrame([("engineer", 32)]).toDF("profession", "age")
|
|
|
|
df.write.option("mergeSchema", "true").mode("append").format("delta").partitionBy(
|
|
"age"
|
|
).save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
assert (
|
|
"aelin\t51\t\\N\t\\N\n"
|
|
"alice\t47\t\\N\t\\N\n"
|
|
"anora\t23\t\\N\t\\N\n"
|
|
"bill\t33\t\\N\t\\N\n"
|
|
"bob\t12\t\\N\t\\N\n"
|
|
"bober\t49\t\\N\t\\N\n"
|
|
"cicil\t68\tusa\t\\N\n"
|
|
"corsha\t26\tchaol\t\\N\n"
|
|
"\\N\t32\t\\N\tengineer"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
paths = (
|
|
node.query(f"SELECT _path FROM {delta_function} ORDER BY all")
|
|
.strip()
|
|
.splitlines()
|
|
)
|
|
|
|
def s3_function(path):
|
|
return f""" s3(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{path}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
assert len(paths) == 9
|
|
|
|
schemas = dict()
|
|
for path in paths:
|
|
schema = node.query(f"DESCRIBE TABLE {s3_function(path)}").strip()
|
|
if schema in schemas:
|
|
schemas[schema].append(path)
|
|
else:
|
|
schemas[schema] = [path]
|
|
|
|
assert len(schemas) == 3
|
|
counts = []
|
|
for schema, schema_paths in schemas.items():
|
|
counts.append(len(schema_paths))
|
|
counts.sort()
|
|
|
|
assert counts == [1, 2, 6]
|
|
|
|
|
|
def test_alter_column_type(started_cluster):
|
|
## Delta lake supports a very limited set of type changes:
|
|
## https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-alter-table-manage-column#parameters-1
|
|
## What is done in this test:
|
|
## Alter Short -> Int
|
|
## Alter Int -> Nullable(Int)
|
|
##
|
|
## Complex type changes are supported only with data overwrite
|
|
## https://docs.delta.io/latest/delta-batch.html#change-column-type-or-name
|
|
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_rename_column")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
|
|
def check_schema(expected):
|
|
assert node.query(f"DESCRIBE TABLE {delta_function} FORMAT TSV") == TSV(
|
|
expected
|
|
)
|
|
|
|
def check_data(expected):
|
|
assert (
|
|
expected
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
def append_data(df):
|
|
df.write.option("mergeSchema", "true").mode("append").format(
|
|
"delta"
|
|
).partitionBy("age").save(path)
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
(
|
|
DeltaTable.create(spark)
|
|
.tableName(table_name)
|
|
.location(path)
|
|
.addColumn("a", "SHORT", nullable=False)
|
|
.addColumn("b", "STRING", nullable=False)
|
|
.addColumn("c", "DATE", nullable=False)
|
|
.addColumn("d", "ARRAY<STRING>", nullable=False)
|
|
.addColumn("e", "BOOLEAN", nullable=True)
|
|
.addColumn("f", ArrayType(StringType(), containsNull=False), nullable=False)
|
|
.partitionedBy("c")
|
|
.property("delta.minReaderVersion", "2")
|
|
.property("delta.minWriterVersion", "5")
|
|
.property("delta.columnMapping.mode", "name")
|
|
.execute()
|
|
)
|
|
|
|
data = [
|
|
(
|
|
1,
|
|
"a",
|
|
datetime.strptime("2000-01-01", "%Y-%m-%d"),
|
|
["aa", "aa"],
|
|
True,
|
|
["aaa", "aaa"],
|
|
)
|
|
]
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", ShortType(), nullable=True),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", DateType(), nullable=False),
|
|
StructField("d", ArrayType(StringType())),
|
|
StructField("e", BooleanType(), nullable=False),
|
|
StructField("f", ArrayType(StringType(), containsNull=False)),
|
|
]
|
|
)
|
|
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.write.format("delta").partitionBy("c").mode("overwrite").save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
check_schema(
|
|
[
|
|
["a", "Int16"],
|
|
["b", "String"],
|
|
["c", "Date32"],
|
|
["d", "Array(Nullable(String))"],
|
|
["e", "Nullable(Bool)"],
|
|
["f", "Array(String)"],
|
|
]
|
|
)
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", IntegerType(), nullable=False),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", DateType(), nullable=False),
|
|
StructField("d", ArrayType(StringType())),
|
|
StructField("e", BooleanType(), nullable=False),
|
|
StructField("f", ArrayType(StringType(), containsNull=False)),
|
|
]
|
|
)
|
|
|
|
data = [
|
|
(
|
|
214748364,
|
|
"b",
|
|
datetime.strptime("2000-02-02", "%Y-%m-%d"),
|
|
["bb", "bb"],
|
|
False,
|
|
["bbb", "bbb"],
|
|
)
|
|
]
|
|
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.write.option("mergeSchema", "true").mode("append").format("delta").partitionBy(
|
|
"c"
|
|
).save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
check_schema(
|
|
[
|
|
["a", "Int32"],
|
|
["b", "String"],
|
|
["c", "Date32"],
|
|
["d", "Array(Nullable(String))"],
|
|
["e", "Nullable(Bool)"],
|
|
["f", "Array(String)"],
|
|
]
|
|
)
|
|
|
|
assert (
|
|
"1\ta\t2000-01-01\t['aa','aa']\ttrue\t['aaa','aaa']\n214748364\tb\t2000-02-02\t['bb','bb']\tfalse\t['bbb','bbb']\n"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all")
|
|
)
|
|
|
|
spark.sql(f"ALTER TABLE {table_name} CHANGE COLUMN a DROP NOT NULL;")
|
|
schema = StructType(
|
|
[
|
|
StructField("a", IntegerType(), nullable=True),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", DateType(), nullable=False),
|
|
StructField("d", ArrayType(StringType())),
|
|
StructField("e", BooleanType(), nullable=False),
|
|
StructField("f", ArrayType(StringType(), containsNull=False)),
|
|
]
|
|
)
|
|
|
|
data = [
|
|
(
|
|
None,
|
|
"c",
|
|
datetime.strptime("2000-03-03", "%Y-%m-%d"),
|
|
["cc", "cc"],
|
|
False,
|
|
["ccc", "ccc"],
|
|
)
|
|
]
|
|
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.write.option("mergeSchema", "true").mode("append").format("delta").partitionBy(
|
|
"c"
|
|
).save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
check_schema(
|
|
[
|
|
["a", "Nullable(Int32)"],
|
|
["b", "String"],
|
|
["c", "Date32"],
|
|
["d", "Array(Nullable(String))"],
|
|
["e", "Nullable(Bool)"],
|
|
["f", "Array(String)"],
|
|
]
|
|
)
|
|
assert (
|
|
"1\ta\t2000-01-01\t['aa','aa']\ttrue\t['aaa','aaa']\n214748364\tb\t2000-02-02\t['bb','bb']\tfalse\t['bbb','bbb']\n\\N\tc\t2000-03-03\t['cc','cc']\tfalse\t['ccc','ccc']\n"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all")
|
|
)
|
|
|
|
paths = (
|
|
node.query(f"SELECT _path FROM {delta_function} ORDER BY all")
|
|
.strip()
|
|
.splitlines()
|
|
)
|
|
|
|
def s3_function(path):
|
|
return f""" s3(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{path}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
assert len(paths) == 3
|
|
|
|
assert "Nullable(Int16)" in node.query(
|
|
f"DESCRIBE TABLE {s3_function(paths[0])}"
|
|
) or "Nullable(Int16)" in node.query(f"DESCRIBE TABLE {s3_function(paths[1])}")
|
|
|
|
assert "Nullable(Int32)" in node.query(
|
|
f"DESCRIBE TABLE {s3_function(paths[0])}"
|
|
) or "Nullable(Int32)" in node.query(f"DESCRIBE TABLE {s3_function(paths[1])}")
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", StringType(), nullable=True),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", DateType(), nullable=False),
|
|
StructField("d", ArrayType(StringType())),
|
|
StructField("e", BooleanType(), nullable=False),
|
|
StructField("f", ArrayType(StringType(), containsNull=False)),
|
|
]
|
|
)
|
|
|
|
data = [
|
|
(
|
|
"123",
|
|
"d",
|
|
datetime.strptime("2000-04-04", "%Y-%m-%d"),
|
|
["ddd", "dd"],
|
|
False,
|
|
["ddd", "ddd"],
|
|
)
|
|
]
|
|
|
|
spark.read.table(table_name).withColumn("a", col("a").cast("String")).write.format(
|
|
"delta"
|
|
).mode("overwrite").option("overwriteSchema", "true").partitionBy("c").save(path)
|
|
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.write.mode("append").format("delta").partitionBy("c").save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
# spark.read.table(table_name).printSchema()
|
|
check_schema(
|
|
[
|
|
["a", "Nullable(String)"],
|
|
["b", "Nullable(String)"],
|
|
["c", "Nullable(Date32)"],
|
|
["d", "Array(Nullable(String))"],
|
|
["e", "Nullable(Bool)"],
|
|
["f", "Array(Nullable(String))"],
|
|
]
|
|
)
|
|
assert (
|
|
"1\ta\t2000-01-01\t['aa','aa']\ttrue\t['aaa','aaa']\n123\td\t2000-04-04\t['ddd','dd']\tfalse\t['ddd','ddd']\n214748364\tb\t2000-02-02\t['bb','bb']\tfalse\t['bbb','bbb']\n\\N\tc\t2000-03-03\t['cc','cc']\tfalse\t['ccc','ccc']\n"
|
|
== node.query(
|
|
f"SELECT * FROM {delta_function} ORDER BY all settings input_format_parquet_allow_missing_columns=0 "
|
|
)
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
"new_analyzer, storage_type", [["1", "s3"], ["0", "s3"]]
|
|
)
|
|
def test_cluster_function(started_cluster, new_analyzer, storage_type):
|
|
instance = started_cluster.instances["node1"]
|
|
started_cluster.instances["node_old"]
|
|
table_name = randomize_table_name("test_cluster_function")
|
|
|
|
schema = pa.schema([("a", pa.int32()), ("b", pa.string())])
|
|
data = [
|
|
pa.array([1, 2, 3, 4, 5], type=pa.int32()),
|
|
pa.array(["aa", "bb", "cc", "aa", "bb"], type=pa.string()),
|
|
]
|
|
|
|
if storage_type == "s3":
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
write_deltalake_with_retry(
|
|
path, table, storage_options=storage_options, partition_by=["b"]
|
|
)
|
|
|
|
table_function = f"""
|
|
deltaLakeCluster(cluster,
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=1)
|
|
"""
|
|
instance.query(
|
|
f"SELECT * FROM {table_function} SETTINGS allow_experimental_analyzer={new_analyzer}"
|
|
)
|
|
assert 5 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function} SETTINGS allow_experimental_analyzer={new_analyzer}"
|
|
)
|
|
)
|
|
assert "1\taa\n"
|
|
"2\tbb\n"
|
|
"3\tcc\n"
|
|
"4\taa\n"
|
|
"5\tbb\n" == instance.query(
|
|
f"SELECT * FROM {table_function} ORDER BY a SETTINGS allow_experimental_analyzer={new_analyzer}"
|
|
)
|
|
|
|
|
|
|
|
def test_partition_columns_3(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_partition_columns_3")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = ["year"]
|
|
bucket = started_cluster.minio_bucket
|
|
num_rows = 10
|
|
|
|
schema = pa.schema(
|
|
[
|
|
pa.field("id", pa.int32(), nullable=False),
|
|
pa.field("name", pa.string(), nullable=False),
|
|
pa.field("age", pa.int32(), nullable=False),
|
|
pa.field("country", pa.string(), nullable=False),
|
|
pa.field("year", pa.string(), nullable=False),
|
|
]
|
|
)
|
|
data = {
|
|
"id": list(range(num_rows)),
|
|
"name": [f"name_{i}" for i in range(num_rows)],
|
|
"age": [32] * num_rows,
|
|
"country": ["US"] * num_rows,
|
|
"year": ["2025"] * num_rows,
|
|
}
|
|
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{TABLE_NAME}",
|
|
pa.Table.from_pydict(data, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="append",
|
|
partition_by=partition_columns,
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
result = int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert result == num_rows
|
|
|
|
assert (
|
|
"0\tname_0\t32\tUS\t2025\n"
|
|
"1\tname_1\t32\tUS\t2025\n"
|
|
"2\tname_2\t32\tUS\t2025\n"
|
|
"3\tname_3\t32\tUS\t2025\n"
|
|
"4\tname_4\t32\tUS\t2025\n"
|
|
"5\tname_5\t32\tUS\t2025\n"
|
|
"6\tname_6\t32\tUS\t2025\n"
|
|
"7\tname_7\t32\tUS\t2025\n"
|
|
"8\tname_8\t32\tUS\t2025\n"
|
|
"9\tname_9\t32\tUS\t2025"
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY all").strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
|
|
def test_filtering_by_virtual_columns(started_cluster, use_delta_kernel):
|
|
instance = started_cluster.instances["node1"]
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_filtering_by_virtual_columns")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = ["year"]
|
|
bucket = started_cluster.minio_bucket
|
|
num_rows = 10
|
|
|
|
schema = pa.schema(
|
|
[
|
|
pa.field("id", pa.int32(), nullable=False),
|
|
pa.field("name", pa.string(), nullable=False),
|
|
pa.field("age", pa.int32(), nullable=False),
|
|
pa.field("country", pa.string(), nullable=False),
|
|
pa.field("year", pa.string(), nullable=False),
|
|
]
|
|
)
|
|
data = {
|
|
"id": list(range(num_rows)),
|
|
"name": [f"name_{i}" for i in range(num_rows)],
|
|
"age": [32] * num_rows,
|
|
"country": ["US"] * num_rows,
|
|
"year": [f"202{i}" for i in range(num_rows)],
|
|
}
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{TABLE_NAME}",
|
|
pa.Table.from_pydict(data, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="append",
|
|
partition_by=partition_columns,
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
result = int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function} SETTINGS allow_experimental_delta_kernel_rs={use_delta_kernel}"
|
|
)
|
|
)
|
|
assert result == num_rows
|
|
|
|
assert (
|
|
"0\tname_0\t32\tUS\t2020\n"
|
|
"1\tname_1\t32\tUS\t2021\n"
|
|
"2\tname_2\t32\tUS\t2022\n"
|
|
"3\tname_3\t32\tUS\t2023\n"
|
|
"4\tname_4\t32\tUS\t2024\n"
|
|
"5\tname_5\t32\tUS\t2025\n"
|
|
"6\tname_6\t32\tUS\t2026\n"
|
|
"7\tname_7\t32\tUS\t2027\n"
|
|
"8\tname_8\t32\tUS\t2028\n"
|
|
"9\tname_9\t32\tUS\t2029"
|
|
== instance.query(
|
|
f"SELECT * FROM {table_function} ORDER BY all SETTINGS allow_experimental_delta_kernel_rs={use_delta_kernel}"
|
|
).strip()
|
|
)
|
|
|
|
query_id = f"query_{TABLE_NAME}_1"
|
|
result = int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function} WHERE _path ILIKE '%year=2024%' SETTINGS allow_experimental_delta_kernel_rs={use_delta_kernel}",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert result == int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['EngineFileLikeReadFiles'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
if use_delta_kernel == "0":
|
|
assert 1 < int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and logger_name = 'DeltaLakeMetadataParser'"
|
|
)
|
|
)
|
|
elif use_delta_kernel == "1":
|
|
assert 0 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and logger_name = 'DeltaLakeMetadataParser'"
|
|
)
|
|
)
|
|
else:
|
|
assert False
|
|
|
|
|
|
def test_column_pruning(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_column_pruning")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = []
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("id", IntegerType(), nullable=False),
|
|
StructField("name", StringType(), nullable=False),
|
|
StructField("age", IntegerType(), nullable=False),
|
|
StructField("country", StringType(), nullable=False),
|
|
StructField("year", StringType(), nullable=False),
|
|
]
|
|
)
|
|
|
|
num_rows = 10000
|
|
datetime.now()
|
|
data = [
|
|
(i, f"name_{i}", 32, "".join("a" for _ in range(100)), "2025")
|
|
for i in range(num_rows)
|
|
]
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.printSchema()
|
|
df.write.mode("append").format("delta").partitionBy(partition_columns).save(
|
|
f"/{TABLE_NAME}"
|
|
)
|
|
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) > 0
|
|
print(f"Uploaded files: {files}")
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
query_id = f"query_{TABLE_NAME}_1"
|
|
sum = int(
|
|
instance.query(
|
|
f"SELECT sum(id) FROM {table_function} SETTINGS allow_experimental_delta_kernel_rs=0, max_read_buffer_size_remote_fs=100, remote_read_min_bytes_for_seek=1",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
bytes_read = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['ReadBufferFromS3Bytes'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
|
|
)
|
|
)
|
|
# Slightly different number depending on reader implementation.
|
|
assert 107120 <= bytes_read <= 107232
|
|
|
|
query_id = f"query_{TABLE_NAME}_2"
|
|
assert sum == int(
|
|
instance.query(
|
|
f"SELECT sum(id) FROM {table_function} SETTINGS enable_filesystem_cache=0, max_read_buffer_size_remote_fs=100, remote_read_min_bytes_for_seek=1, use_parquet_metadata_cache=0",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['EngineFileLikeReadFiles'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
|
|
)
|
|
)
|
|
bytes_read = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['ReadBufferFromS3Bytes'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
|
|
)
|
|
)
|
|
# Small diff because in case of delta-kernel metadata reading is not counted in the metric.
|
|
assert 105580 <= bytes_read <= 105689
|
|
|
|
|
|
def test_concurrent_reads(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_concurrent_reads")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = []
|
|
bucket = started_cluster.minio_bucket
|
|
num_rows = 500000
|
|
|
|
schema = pa.schema(
|
|
[
|
|
pa.field("id", pa.int32(), nullable=False),
|
|
pa.field("name", pa.string(), nullable=False),
|
|
pa.field("age", pa.int32(), nullable=False),
|
|
pa.field("country", pa.string(), nullable=False),
|
|
pa.field("year", pa.string(), nullable=False),
|
|
]
|
|
)
|
|
data = {
|
|
"id": list(range(num_rows)),
|
|
"name": [f"name_{i}" for i in range(num_rows)],
|
|
"age": [32] * num_rows,
|
|
"country": ["a" * 100] * num_rows,
|
|
"year": ["2025"] * num_rows,
|
|
}
|
|
write_deltalake_with_retry(
|
|
f"s3://{bucket}/{result_file}",
|
|
pa.Table.from_pydict(data, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
partition_by=partition_columns,
|
|
mode="append",
|
|
)
|
|
|
|
instance.query(
|
|
f"create table {TABLE_NAME} (id Int32, name String, age Int32, country String, year String) engine = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
|
|
def select(_):
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS max_read_buffer_size_remote_fs=100",
|
|
)
|
|
|
|
busy_pool = Pool(10)
|
|
p = busy_pool.map_async(select, range(10))
|
|
p.wait()
|
|
|
|
select(0)
|
|
|
|
|
|
def test_snapshot_version(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_snapshot_version")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
column_mapping = "name"
|
|
|
|
df = spark.createDataFrame([("alice", 47), ("anora", 23), ("aelin", 51)]).toDF(
|
|
"first_name", "age"
|
|
)
|
|
|
|
df.write.format("delta").partitionBy("age").option(
|
|
"delta.minReaderVersion", "2"
|
|
).option("delta.minWriterVersion", "5").option(
|
|
"delta.columnMapping.mode", column_mapping
|
|
).save(
|
|
path
|
|
)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
def check_schema(expected, version):
|
|
assert (
|
|
expected
|
|
== node.query(
|
|
f"DESCRIBE TABLE {delta_function} SETTINGS delta_lake_snapshot_version = {version}"
|
|
).strip()
|
|
)
|
|
|
|
def check_data(expected, version, table=None):
|
|
if table is None:
|
|
table = delta_function
|
|
if version is not None:
|
|
assert (
|
|
expected
|
|
== node.query(
|
|
f"SELECT * FROM {table} ORDER BY all SETTINGS delta_lake_snapshot_version = {version}"
|
|
).strip()
|
|
)
|
|
else:
|
|
assert expected == node.query(f"SELECT * FROM {table} ORDER BY all").strip()
|
|
|
|
def append_data(df):
|
|
df.write.option("mergeSchema", "true").mode("append").format(
|
|
"delta"
|
|
).partitionBy("age").save(path)
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
check_schema("first_name\tNullable(String)\t\t\t\t\t\nage\tNullable(Int64)", 0)
|
|
check_data("aelin\t51\n" "alice\t47\n" "anora\t23", 0)
|
|
|
|
spark.sql(f"CREATE TABLE {table_name} USING DELTA LOCATION '{path}'")
|
|
spark.sql(f"ALTER TABLE {table_name} RENAME COLUMN first_name TO naam")
|
|
|
|
df = spark.createDataFrame([("bob", 12), ("bill", 33), ("bober", 49)]).toDF(
|
|
"naam", "age"
|
|
)
|
|
append_data(df)
|
|
|
|
check_schema("first_name\tNullable(String)\t\t\t\t\t\nage\tNullable(Int64)", 0)
|
|
check_data("aelin\t51\n" "alice\t47\n" "anora\t23", 0)
|
|
|
|
check_schema("naam\tNullable(String)\t\t\t\t\t\nage\tNullable(Int64)", 2)
|
|
check_data("aelin\t51\nalice\t47\nanora\t23\nbill\t33\nbob\t12\nbober\t49", 2)
|
|
|
|
assert "Unknown expression identifier `first_name`" in node.query_and_get_error(
|
|
f"SELECT first_name FROM {delta_function} WHERE age = 51"
|
|
)
|
|
|
|
assert "Unknown expression identifier `first_name`" in node.query_and_get_error(
|
|
f"SELECT first_name FROM {delta_function} WHERE age = 51",
|
|
settings={"delta_lake_snapshot_version": 2},
|
|
)
|
|
|
|
assert (
|
|
"aelin"
|
|
in node.query(
|
|
f"SELECT first_name FROM {delta_function} WHERE age = 51",
|
|
settings={"delta_lake_snapshot_version": 0},
|
|
).strip()
|
|
)
|
|
|
|
assert "Unknown expression identifier `naam`" in node.query_and_get_error(
|
|
f"SELECT naam FROM {delta_function} WHERE age = 51",
|
|
settings={"delta_lake_snapshot_version": 0},
|
|
)
|
|
|
|
assert (
|
|
"aelin"
|
|
in node.query(
|
|
f"SELECT naam FROM {delta_function} WHERE age = 51",
|
|
settings={"delta_lake_snapshot_version": 2},
|
|
).strip()
|
|
)
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {table_name} (naam String, age Int32) ENGINE = DeltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
)
|
|
df = spark.createDataFrame([("cap", 91), ("cip", 92), ("cop", 93)]).toDF(
|
|
"naam", "age"
|
|
)
|
|
append_data(df)
|
|
check_data(
|
|
"aelin\t51\nalice\t47\nanora\t23\nbill\t33\nbob\t12\nbober\t49\ncap\t91\ncip\t92\ncop\t93",
|
|
3,
|
|
table_name,
|
|
)
|
|
check_data(
|
|
"aelin\t51\nalice\t47\nanora\t23\nbill\t33\nbob\t12\nbober\t49", 2, table_name
|
|
)
|
|
check_data(
|
|
"aelin\t51\nalice\t47\nanora\t23\nbill\t33\nbob\t12\nbober\t49\ncap\t91\ncip\t92\ncop\t93",
|
|
None,
|
|
table_name,
|
|
)
|
|
|
|
|
|
def test_join_with_distributed(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_join_with_distributed")
|
|
result_file = f"{TABLE_NAME}"
|
|
|
|
df = spark.createDataFrame(
|
|
[
|
|
(1, "a"),
|
|
(2, "b"),
|
|
(3, "c"),
|
|
(4, "d"),
|
|
(5, "e"),
|
|
(6, "f"),
|
|
(7, "g"),
|
|
(8, "h"),
|
|
(9, "i"),
|
|
],
|
|
["id", "val"],
|
|
)
|
|
|
|
df.write.format("delta").save(f"/{TABLE_NAME}")
|
|
|
|
clickhouse_table_name = f"test_join_with_distributed_{uuid.uuid4().hex}"
|
|
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
instance.query(
|
|
f"create table {clickhouse_table_name} on cluster cluster (id UInt8, val char) engine = ReplicatedMergeTree('/clickhouse/tables/{{shard}}/{clickhouse_table_name}', '{{replica}}') order by id"
|
|
)
|
|
instance.query(
|
|
f"create table {clickhouse_table_name}_dist on cluster cluster AS {clickhouse_table_name} engine = Distributed(cluster, default, {clickhouse_table_name}, rand())"
|
|
)
|
|
instance.query(
|
|
f"insert into {clickhouse_table_name}_dist values (1, 'A'),(2, 'B'),(3, 'C'),(4, 'D'),(5, 'E'),(6, 'F'),(7, 'G'),(8, 'H'),(9, 'I');"
|
|
)
|
|
|
|
table_function_cluster = f"deltaLakeCluster(cluster, 'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
# All cases which were reported as faulty
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function_cluster} SETTINGS prefer_localhost_replica = 0"
|
|
).strip()
|
|
)
|
|
== 9
|
|
)
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function} SETTINGS cluster_for_parallel_replicas='cluster', max_parallel_replicas=2, allow_experimental_parallel_reading_from_replicas=2, parallel_replicas_for_cluster_engines=1"
|
|
).strip()
|
|
)
|
|
== 9
|
|
)
|
|
|
|
assert (
|
|
len(
|
|
instance.query(
|
|
f"with b as (select * from {table_function}) select {clickhouse_table_name}_dist.val, b.val from {clickhouse_table_name}_dist join b on {clickhouse_table_name}_dist.id = b.id;"
|
|
).split("\n")
|
|
)
|
|
== 10
|
|
)
|
|
assert (
|
|
len(
|
|
instance.query(
|
|
f"with b as (select * from {table_function}) select {clickhouse_table_name}_dist.val, b.val from b join {clickhouse_table_name}_dist on {clickhouse_table_name}_dist.id = b.id;"
|
|
).split("\n")
|
|
)
|
|
== 10
|
|
)
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM remote('localhost', {table_function}) SETTINGS prefer_localhost_replica = 0"
|
|
).strip()
|
|
)
|
|
== 9
|
|
)
|
|
|
|
|
|
def test_delta_kernel_internal_pruning(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_partition_columns")
|
|
result_file = f"{TABLE_NAME}"
|
|
partition_columns = ["b", "c", "d", "e", "f", "g", "h"]
|
|
|
|
(
|
|
DeltaTable.create(spark)
|
|
.tableName(TABLE_NAME)
|
|
.location(f"/{result_file}")
|
|
.addColumn("a", "INT")
|
|
.addColumn("b", "STRING")
|
|
.addColumn("c", "DATE")
|
|
.addColumn("d", "INT")
|
|
.addColumn("e", "TIMESTAMP")
|
|
.addColumn("f", "BOOLEAN")
|
|
.addColumn("g", "DECIMAL(10,2)")
|
|
.addColumn("h", "BOOLEAN")
|
|
.partitionedBy(partition_columns)
|
|
.execute()
|
|
)
|
|
num_rows = 9
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", IntegerType()),
|
|
StructField("b", StringType()),
|
|
StructField("c", DateType()),
|
|
StructField("d", IntegerType()),
|
|
StructField("e", TimestampType()),
|
|
StructField("f", BooleanType()),
|
|
StructField("g", DecimalType(10, 2)),
|
|
StructField("h", BooleanType()),
|
|
]
|
|
)
|
|
|
|
now = datetime.now()
|
|
for i in range(1, num_rows + 1):
|
|
data = [
|
|
(
|
|
i,
|
|
"test" + str(i % 3),
|
|
datetime.strptime(f"2000-01-0{i}", "%Y-%m-%d"),
|
|
i % 2,
|
|
(
|
|
now
|
|
if i % 2 == 0
|
|
else datetime.strptime(
|
|
f"2012-01-0{i} 12:34:56.789123", "%Y-%m-%d %H:%M:%S.%f"
|
|
)
|
|
),
|
|
True if i % 2 == 0 else False,
|
|
Decimal(f"{i * 1.11:.2f}"),
|
|
False if i % 2 == 0 else True,
|
|
)
|
|
]
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.printSchema()
|
|
df.write.mode("append").format("delta").partitionBy(partition_columns).save(
|
|
f"/{TABLE_NAME}"
|
|
)
|
|
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) > 0
|
|
print(f"Uploaded files: {files}")
|
|
|
|
cluster = False
|
|
if cluster:
|
|
table_function = f"deltaLakeCluster(cluster, 'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
else:
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
result = instance.query(f"describe table {table_function}").strip()
|
|
assert (
|
|
result == "a\tNullable(Int32)\t\t\t\t\t\n"
|
|
"b\tNullable(String)\t\t\t\t\t\n"
|
|
"c\tNullable(Date32)\t\t\t\t\t\n"
|
|
"d\tNullable(Int32)\t\t\t\t\t\n"
|
|
"e\tNullable(DateTime64(6))\t\t\t\t\t\n"
|
|
"f\tNullable(Bool)\t\t\t\t\t\n"
|
|
"g\tNullable(Decimal(10, 2))\t\t\t\t\t\n"
|
|
"h\tNullable(Bool)"
|
|
)
|
|
|
|
result = int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert result == num_rows
|
|
|
|
expected_output = f"""1 test1 2000-01-01 1 2012-01-01 12:34:56.789123 false 1.11 true
|
|
2 test2 2000-01-02 0 {now} true 2.22 false
|
|
3 test0 2000-01-03 1 2012-01-03 12:34:56.789123 false 3.33 true
|
|
4 test1 2000-01-04 0 {now} true 4.44 false
|
|
5 test2 2000-01-05 1 2012-01-05 12:34:56.789123 false 5.55 true
|
|
6 test0 2000-01-06 0 {now} true 6.66 false
|
|
7 test1 2000-01-07 1 2012-01-07 12:34:56.789123 false 7.77 true
|
|
8 test2 2000-01-08 0 {now} true 8.88 false
|
|
9 test0 2000-01-09 1 2012-01-09 12:34:56.789123 false 9.99 true"""
|
|
|
|
assert (
|
|
expected_output
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY a").strip()
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_1"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE 'test2' == b
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
assert result == 3
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 3 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 3 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_2"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE 'test2' == b AND d == 1
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_3"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE 'test2' == b AND d != 1
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 2
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_4"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE 'test2' == b AND d > 0
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_5"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2' AND d < 1
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 2
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_6"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2' AND 1 >= d
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_7"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2' AND d <= 0
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 2
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_8"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE c == toDate('2000-01-08')
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_9"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE c == '2000-01-08'
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 1
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 1 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_10"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2' AND not (h == 1)
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 2
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_11"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2' AND not h
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
assert result == 2
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
assert 2 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file: {TABLE_NAME}/b=test2%'"
|
|
)
|
|
)
|
|
|
|
instance.query("SYSTEM ENABLE FAILPOINT delta_kernel_fail_literal_visitor")
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_12"
|
|
assert "Injecting fault for visitLiteralValue" in instance.query_and_get_error(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2'
|
|
""",
|
|
query_id=query_id,
|
|
settings={"delta_lake_throw_on_engine_predicate_error": 1},
|
|
)
|
|
|
|
query_id = f"query_with_filter_{TABLE_NAME}_13"
|
|
result = int(
|
|
instance.query(
|
|
f"""SELECT count() FROM {table_function} WHERE b == 'test2'
|
|
""",
|
|
query_id=query_id,
|
|
)
|
|
)
|
|
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_kernel_fail_literal_visitor")
|
|
|
|
assert result == 3
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 3 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' and message ILIKE '%Scanned file%'"
|
|
)
|
|
)
|
|
|
|
|
|
def test_count_from_cache(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_empty_format_header")
|
|
result_file = f"{TABLE_NAME}"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {TABLE_NAME} (id Int32, name String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
|
|
for i in range(3):
|
|
instance.query(
|
|
f"INSERT INTO TABLE FUNCTION deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}') SELECT {i + 1} as name, toString({i + 1}) as id from numbers(1)"
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
# Sleep 1 second because cache works by timestamps.
|
|
time.sleep(1)
|
|
assert 3 == int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert 3 == int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
query_id = f"{TABLE_NAME}_query"
|
|
assert 3 == int(
|
|
instance.query(f"SELECT count() FROM {table_function}", query_id=query_id)
|
|
)
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
assert 3 == int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['SchemaInferenceCacheNumRowsHits'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
|
|
)
|
|
)
|
|
assert (
|
|
"3\t3"
|
|
== instance.query(f"SELECT count(), count() FROM {table_function}").strip()
|
|
)
|
|
assert (
|
|
"3\t6\t3"
|
|
== instance.query(
|
|
f"SELECT count(), sum(id), uniqExact(_path) FROM {table_function}"
|
|
).strip()
|
|
)
|
|
|
|
|
|
def test_writes(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
instance_disabled_kernel = cluster.instances["node_with_disabled_delta_kernel"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
table_name = randomize_table_name("test_writes")
|
|
result_file = f"{table_name}_data"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
instance_disabled_kernel.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
|
|
instance.query(
|
|
f"INSERT INTO TABLE FUNCTION deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}', settings allow_experimental_delta_kernel_rs=1) SELECT number as name, toString(number) as id from numbers(10)"
|
|
)
|
|
|
|
s3_objects = list(minio_client.list_objects(bucket, result_file, recursive=True))
|
|
file_name = None
|
|
for obj in s3_objects:
|
|
print(f"File: {obj.object_name}")
|
|
if obj.object_name.endswith("parquet") and not obj.object_name.endswith(
|
|
"snappy.parquet"
|
|
):
|
|
file_name = obj.object_name
|
|
|
|
assert (
|
|
"0\t0\n1\t1\n2\t2\n3\t3\n4\t4\n5\t5\n6\t6\n7\t7\n8\t8\n9\t9"
|
|
== instance.query(
|
|
f"SELECT * FROM s3('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{file_name}', 'minio', '{minio_secret_key}')"
|
|
).strip()
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
def check_count(expected):
|
|
assert expected == int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert expected == int(
|
|
instance_disabled_kernel.query(f"SELECT count() FROM {table_name}")
|
|
)
|
|
assert expected == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
|
|
def check_data(expected):
|
|
assert (
|
|
expected
|
|
== instance.query(f"SELECT * FROM {table_name} ORDER BY all").strip()
|
|
)
|
|
assert (
|
|
expected
|
|
== instance_disabled_kernel.query(
|
|
f"SELECT * FROM {table_name} ORDER BY all"
|
|
).strip()
|
|
)
|
|
assert (
|
|
expected
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY all").strip()
|
|
)
|
|
|
|
check_count(10)
|
|
check_data("0\t0\n1\t1\n2\t2\n3\t3\n4\t4\n5\t5\n6\t6\n7\t7\n8\t8\n9\t9")
|
|
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number, toString(number) FROM numbers(10, 10)"
|
|
)
|
|
check_count(20)
|
|
check_data(
|
|
"0\t0\n1\t1\n2\t2\n3\t3\n4\t4\n5\t5\n6\t6\n7\t7\n8\t8\n9\t9\n10\t10\n11\t11\n12\t12\n13\t13\n14\t14\n15\t15\n16\t16\n17\t17\n18\t18\n19\t19"
|
|
)
|
|
|
|
|
|
def test_partitioned_writes(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
instance_disabled_kernel = cluster.instances["node_with_disabled_delta_kernel"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
table_name = randomize_table_name("test_partitioned_writes")
|
|
result_file = f"{table_name}_data"
|
|
partition_columns = ["id", "comment"]
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("id", pa.int32(), False),
|
|
("name", pa.string(), False),
|
|
("comment", pa.string(), False),
|
|
]
|
|
)
|
|
empty_arrays = [
|
|
pa.array([], type=pa.int32()),
|
|
pa.array([], type=pa.string()),
|
|
pa.array([], type=pa.string()),
|
|
]
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
partition_by=partition_columns,
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String, comment String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
instance_disabled_kernel.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String, comment String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number, toString(number), concat('comment-', toString(number % 2)) FROM numbers(10)"
|
|
)
|
|
|
|
def check_files(expected):
|
|
s3_objects = list(
|
|
minio_client.list_objects(bucket, result_file, recursive=True)
|
|
)
|
|
file_names = []
|
|
for obj in s3_objects:
|
|
print(f"File: {obj.object_name}")
|
|
if obj.object_name.endswith("parquet") and not obj.object_name.endswith(
|
|
"snappy.parquet"
|
|
):
|
|
file_names.append(obj.object_name)
|
|
|
|
assert len(file_names) == expected
|
|
file_names.sort()
|
|
expected_ids = [str(i) for i in range(expected)]
|
|
expected_ids.sort()
|
|
for i in range(expected):
|
|
expected_id = expected_ids[i]
|
|
comment_id = int(expected_ids[i]) % 2
|
|
assert file_names[i].startswith(
|
|
f"{result_file}/id={expected_id}/comment=comment-{comment_id}/"
|
|
)
|
|
assert (
|
|
f"{expected_id}\t{expected_id}\tcomment-{comment_id}"
|
|
== instance.query(
|
|
f"SELECT id, name, comment FROM s3('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{file_names[i]}', 'minio', '{minio_secret_key}') ORDER BY all"
|
|
).strip()
|
|
)
|
|
assert (
|
|
f"{expected_id}"
|
|
== instance.query(
|
|
f"SELECT * FROM s3('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{file_names[i]}', 'minio', '{minio_secret_key}')",
|
|
settings={"use_hive_partitioning": 0},
|
|
).strip()
|
|
)
|
|
|
|
table_function = f"deltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
|
|
def check_count(expected):
|
|
assert expected == int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_name} settings use_cache_for_count_from_files=0"
|
|
)
|
|
)
|
|
assert expected == int(
|
|
instance.query(
|
|
f"SELECT count() FROM {table_function} settings use_cache_for_count_from_files=0"
|
|
)
|
|
)
|
|
assert expected == int(
|
|
instance_disabled_kernel.query(
|
|
f"SELECT count() FROM {table_name} settings use_cache_for_count_from_files=0"
|
|
)
|
|
)
|
|
|
|
def check_data(expected):
|
|
assert (
|
|
expected
|
|
== instance.query(f"SELECT * FROM {table_name} ORDER BY all").strip()
|
|
)
|
|
# TODO: Fix read with disabled delta-kernel?
|
|
# assert (
|
|
# expected
|
|
# == instance_disabled_kernel.query(
|
|
# f"SELECT * FROM {table_name} ORDER BY all"
|
|
# ).strip()
|
|
# )
|
|
assert (
|
|
expected
|
|
== instance.query(f"SELECT * FROM {table_function} ORDER BY all").strip()
|
|
)
|
|
|
|
check_files(10)
|
|
check_count(10)
|
|
check_data(
|
|
"0\t0\tcomment-0\n1\t1\tcomment-1\n2\t2\tcomment-0\n3\t3\tcomment-1\n4\t4\tcomment-0\n5\t5\tcomment-1\n6\t6\tcomment-0\n7\t7\tcomment-1\n8\t8\tcomment-0\n9\t9\tcomment-1"
|
|
)
|
|
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number, toString(number), concat('comment-', toString(number % 2)) FROM numbers(10, 10)"
|
|
)
|
|
check_count(20)
|
|
check_files(20)
|
|
check_data(
|
|
"0\t0\tcomment-0\n1\t1\tcomment-1\n2\t2\tcomment-0\n3\t3\tcomment-1\n4\t4\tcomment-0\n5\t5\tcomment-1\n6\t6\tcomment-0\n7\t7\tcomment-1\n8\t8\tcomment-0\n9\t9\tcomment-1\n10\t10\tcomment-0\n11\t11\tcomment-1\n12\t12\tcomment-0\n13\t13\tcomment-1\n14\t14\tcomment-0\n15\t15\tcomment-1\n16\t16\tcomment-0\n17\t17\tcomment-1\n18\t18\tcomment-0\n19\t19\tcomment-1"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("partitioned", [False, True])
|
|
def test_concurrent_queries(started_cluster, partitioned):
|
|
instance = started_cluster.instances["node1"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_concurrent_queries")
|
|
result_file = f"{TABLE_NAME}"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
partition_by=["name"] if partitioned else [],
|
|
)
|
|
|
|
instance.query(
|
|
f"create table {TABLE_NAME} (id Int32, name String) engine = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
|
|
num_insert_threads = 15
|
|
num_select_threads = 5
|
|
errors = ["" for _ in range(num_insert_threads)]
|
|
success = [0 for _ in range(num_insert_threads)]
|
|
|
|
def run_concurrent_queries():
|
|
def select(_):
|
|
instance.query(f"SELECT * FROM {TABLE_NAME}")
|
|
|
|
def insert(i):
|
|
try:
|
|
instance.query(
|
|
f"INSERT INTO {TABLE_NAME} SELECT number, toString(number) FROM numbers(50)",
|
|
)
|
|
success[i] += 1
|
|
except Exception as e:
|
|
errors[i] = str(e)
|
|
|
|
for _ in range(10):
|
|
insert(_)
|
|
|
|
select_pool = Pool(num_select_threads)
|
|
insert_pool = Pool(num_insert_threads)
|
|
sp = select_pool.map_async(select, range(num_select_threads))
|
|
ip = insert_pool.map_async(insert, range(num_insert_threads))
|
|
sp.wait()
|
|
ip.wait()
|
|
|
|
select(0)
|
|
|
|
num_rows = sum(success) * 50
|
|
assert num_rows == int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}",
|
|
)
|
|
)
|
|
|
|
for _ in range(3):
|
|
run_concurrent_queries()
|
|
if len([e for e in errors if e != ""]) > 0:
|
|
break
|
|
print("Did not catch commit conflict, will retry")
|
|
|
|
non_empty_errors = [e for e in errors if e != ""]
|
|
assert len(non_empty_errors) > 0
|
|
for e in non_empty_errors:
|
|
assert "commit conflict at version" in e
|
|
|
|
s3_objects = list(minio_client.list_objects(bucket, result_file, recursive=True))
|
|
file_names = []
|
|
for obj in s3_objects:
|
|
print(f"File: {obj.object_name}")
|
|
if obj.object_name.endswith("parquet") and not obj.object_name.endswith(
|
|
"snappy.parquet"
|
|
):
|
|
file_names.append(obj.object_name)
|
|
if partitioned:
|
|
assert len(file_names) == sum(success) * 50
|
|
else:
|
|
assert len(file_names) == sum(success)
|
|
|
|
|
|
def test_writes_spark_compatibility(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
cluster.instances["node_with_disabled_delta_kernel"]
|
|
table_name = randomize_table_name("test_writes")
|
|
result_file = f"/var/lib/clickhouse/user_files/{table_name}_data"
|
|
|
|
schema = pa.schema([("id", pa.int32()), ("name", pa.string())])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake(
|
|
f"file:///{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
mode="overwrite",
|
|
)
|
|
|
|
LocalUploader(instance).upload_directory(f"{result_file}/", f"{result_file}/")
|
|
files = (
|
|
instance.exec_in_container(["bash", "-c", f"ls {result_file}"])
|
|
.strip()
|
|
.split("\n")
|
|
)
|
|
assert len(files) == 1
|
|
assert "_delta_log" == files[0]
|
|
assert "" in instance.exec_in_container(
|
|
["bash", "-c", f"ls /{result_file}/_delta_log"]
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLakeLocal('{result_file}') SETTINGS output_format_parquet_compression_method = 'none'"
|
|
)
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number, toString(number) FROM numbers(10)"
|
|
)
|
|
|
|
LocalDownloader(instance).download_directory(f"{result_file}/", f"{result_file}/")
|
|
|
|
files = (
|
|
instance.exec_in_container(["bash", "-c", f"ls {result_file}"])
|
|
.strip()
|
|
.split("\n")
|
|
)
|
|
assert len(files) == 2
|
|
pfile = files[0] if files[0].endswith(".parquet") else files[1]
|
|
|
|
table = pq.read_table(f"{result_file}/{pfile}")
|
|
df = table.to_pandas()
|
|
assert (
|
|
"0 0 0\n1 1 1\n2 2 2\n3 3 3\n4 4 4\n5 5 5\n6 6 6\n7 7 7\n8 8 8\n9 9 9"
|
|
in str(df)
|
|
)
|
|
|
|
spark = started_cluster.spark_session
|
|
df = spark.read.format("delta").load(f"{result_file}").collect()
|
|
assert (
|
|
"[Row(id=0, name='0'), Row(id=1, name='1'), Row(id=2, name='2'), Row(id=3, name='3'), Row(id=4, name='4'), Row(id=5, name='5'), Row(id=6, name='6'), Row(id=7, name='7'), Row(id=8, name='8'), Row(id=9, name='9')]"
|
|
== str(df)
|
|
)
|
|
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number, toString(number) FROM numbers(10, 10)"
|
|
)
|
|
LocalDownloader(instance).download_directory(f"{result_file}/", f"{result_file}/")
|
|
files = (
|
|
instance.exec_in_container(["bash", "-c", f"ls {result_file}"])
|
|
.strip()
|
|
.split("\n")
|
|
)
|
|
assert len(files) == 3
|
|
|
|
df = spark.read.format("delta").load(f"{result_file}").collect()
|
|
assert (
|
|
"[Row(id=10, name='10'), Row(id=11, name='11'), Row(id=12, name='12'), Row(id=13, name='13'), Row(id=14, name='14'), Row(id=15, name='15'), Row(id=16, name='16'), Row(id=17, name='17'), Row(id=18, name='18'), Row(id=19, name='19'), Row(id=0, name='0'), Row(id=1, name='1'), Row(id=2, name='2'), Row(id=3, name='3'), Row(id=4, name='4'), Row(id=5, name='5'), Row(id=6, name='6'), Row(id=7, name='7'), Row(id=8, name='8'), Row(id=9, name='9')]"
|
|
== str(df)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("partitioned", [False, True])
|
|
@pytest.mark.parametrize("limit_enabled", [False, True])
|
|
def test_write_limits(started_cluster, partitioned, limit_enabled):
|
|
instance = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_write_limits")
|
|
result_file = f"/var/lib/clickhouse/user_files/{table_name}_data"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake(
|
|
f"file://{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
mode="overwrite",
|
|
partition_by=["id"] if partitioned else [],
|
|
)
|
|
LocalUploader(instance).upload_directory(f"{result_file}/", f"{result_file}/")
|
|
files = (
|
|
instance.exec_in_container(["bash", "-c", f"ls /{result_file}"])
|
|
.strip()
|
|
.split("\n")
|
|
)
|
|
assert len(files) == 1
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLakeLocal('{result_file}') SETTINGS output_format_parquet_compression_method = 'none'"
|
|
)
|
|
|
|
num_rows = 1000000
|
|
partitions_num = 5
|
|
limit_rows = 10 if limit_enabled else (num_rows + 1)
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number % {partitions_num}, randomString(10) FROM numbers({num_rows}) SETTINGS delta_lake_insert_max_rows_in_data_file = {limit_rows}, max_insert_block_size = 1000, min_chunk_bytes_for_parallel_parsing = 1000"
|
|
)
|
|
|
|
files = LocalDownloader(instance).download_directory(
|
|
f"{result_file}/", f"{result_file}/"
|
|
)
|
|
data_files = [file for file in files if file.endswith(".parquet")]
|
|
assert len(data_files) > 0, f"No data files: {files}"
|
|
|
|
if partitioned:
|
|
if limit_enabled:
|
|
assert len(data_files) > partitions_num, f"Data files: {data_files}"
|
|
else:
|
|
assert len(data_files) == partitions_num, f"Data files: {data_files}"
|
|
else:
|
|
if limit_enabled:
|
|
assert len(data_files) > 1, f"Data files: {data_files}"
|
|
else:
|
|
assert len(data_files) == 1, f"Data files: {data_files}"
|
|
|
|
assert num_rows == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
|
|
spark = started_cluster.spark_session
|
|
df = spark.read.format("delta").load(f"{result_file}")
|
|
assert df.count() == num_rows
|
|
|
|
|
|
def test_column_mapping_id(started_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_column_mapping_id")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("id", IntegerType(), True),
|
|
StructField(
|
|
"person",
|
|
StructType(
|
|
[
|
|
StructField("first_name", StringType(), True),
|
|
StructField("last_name", StringType(), True),
|
|
]
|
|
),
|
|
True,
|
|
),
|
|
]
|
|
)
|
|
data = [(1, ("Alice", "Smith")), (2, ("Bob", "Johnson"))]
|
|
df = spark.createDataFrame(data, schema=schema)
|
|
df.write.format("delta").option("delta.minReaderVersion", "2").option(
|
|
"delta.minWriterVersion", "5"
|
|
).option("delta.columnMapping.mode", "id").save(path)
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
assert (
|
|
"1\t('Alice','Smith')\n2\t('Bob','Johnson')"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("column_mapping", ["name", "id"])
|
|
def test_column_mapping_write_rejected(started_cluster, column_mapping):
|
|
# Writing to a column-mapped table must be rejected: ClickHouse writes logical
|
|
# Parquet field names, not the physical names/ids that a Delta reader expects,
|
|
# so committing an AddFile would produce an unreadable data file.
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name(f"test_column_mapping_write_{column_mapping}")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("id", IntegerType(), True),
|
|
StructField("name", StringType(), True),
|
|
]
|
|
)
|
|
data = [(1, "Alice"), (2, "Bob")]
|
|
df = spark.createDataFrame(data, schema=schema)
|
|
df.write.format("delta").option("delta.minReaderVersion", "2").option(
|
|
"delta.minWriterVersion", "5"
|
|
).option("delta.columnMapping.mode", column_mapping).save(path)
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
def list_objects():
|
|
return sorted(
|
|
obj.object_name
|
|
for obj in minio_client.list_objects(bucket, table_name, recursive=True)
|
|
)
|
|
|
|
objects_before = list_objects()
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}',
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
error = node.query_and_get_error(
|
|
f"INSERT INTO TABLE FUNCTION {delta_function} SELECT 3 AS id, 'Carol' AS name"
|
|
)
|
|
assert "NOT_IMPLEMENTED" in error
|
|
assert "column mapping" in error
|
|
|
|
# Nothing was committed: no new parquet data file and no new _delta_log entry.
|
|
assert objects_before == list_objects()
|
|
|
|
# The table is still readable and its data is unchanged.
|
|
assert (
|
|
"1\tAlice\n2\tBob"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("column_mapping", ["", "name"])
|
|
def test_subcolumns(started_cluster, column_mapping):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_struct")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"{table_name}"
|
|
|
|
data_file = "field_ids_struct_test/data/00000-1-7cad83a6-af90-42a9-8a10-114cbc862a42-0-00001.parquet"
|
|
|
|
def s3_function(path):
|
|
return f""" s3(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{path}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
func = s3_function(data_file)
|
|
assert (
|
|
"2025-06-04\t('100022','2025-06-04 18:40:56.000000','2025-06-09 21:19:00.364000')\t100022"
|
|
== node.query(f"select * from {func}").strip()
|
|
)
|
|
assert (
|
|
"col_x2D1\tNullable(Date32)\t\t\t\t\t\n"
|
|
"col_x2D2\tTuple(\\n col_x2D3 Nullable(String),\\n col_x2D4 Nullable(DateTime64(6, \\'UTC\\')),\\n col_x2D5 Nullable(DateTime64(6, \\'UTC\\')))\t\t\t\t\t\n"
|
|
"col_x2D6\tNullable(Int64)" == node.query(f"describe table {func}").strip()
|
|
)
|
|
|
|
df = spark.read.parquet(
|
|
os.path.join(
|
|
os.path.join(os.path.dirname(os.path.realpath(__file__))), data_file
|
|
)
|
|
)
|
|
write_delta_from_df(spark, df, f"/{path}", mode="overwrite")
|
|
default_upload_directory(started_cluster, "s3", f"/{path}", "")
|
|
|
|
s3_objects = list(minio_client.list_objects(bucket, table_name, recursive=True))
|
|
object_name = None
|
|
for obj in s3_objects:
|
|
print(f"File: {obj.object_name}")
|
|
if obj.object_name.endswith(".parquet"):
|
|
object_name = obj.object_name
|
|
|
|
func = s3_function(object_name)
|
|
assert (
|
|
"2025-06-04\t('100022','2025-06-04 18:40:56.000000000','2025-06-09 21:19:00.364000000')\t100022"
|
|
== node.query(f"select * from {func}").strip()
|
|
)
|
|
data_file_desc = node.query(f"describe table {func}").strip()
|
|
assert "col-" in data_file_desc
|
|
assert "col_" not in data_file_desc
|
|
|
|
data_file_schema = node.query(f"describe table {func}")
|
|
print(f"Data file schema: {data_file_schema}")
|
|
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
assert (
|
|
"2025-06-04\t('100022','2025-06-04 18:40:56.000000','2025-06-09 21:19:00.364000')\t100022"
|
|
== node.query(f"SELECT * FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
assert (
|
|
"col_x2D1\tNullable(Date32)\t\t\t\t\t\n"
|
|
"col_x2D2\tTuple(\\n col_x2D3 Nullable(String),\\n col_x2D4 Nullable(DateTime64(6)),\\n col_x2D5 Nullable(DateTime64(6)))\t\t\t\t\t\n"
|
|
"col_x2D6\tNullable(Int64)"
|
|
== node.query(f"describe table {delta_function}").strip()
|
|
)
|
|
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {table_name} (
|
|
col_x2D1 Nullable(Date32),
|
|
col_x2D2 Tuple(col_x2D3 Nullable(String), col_x2D4 Nullable(DateTime64(6)), col_x2D5 Nullable(DateTime64(6))),
|
|
col_x2D6 Nullable(Int64)) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
)
|
|
|
|
assert (
|
|
"2025-06-04\t('100022','2025-06-04 18:40:56.000000','2025-06-09 21:19:00.364000')\t100022"
|
|
== node.query(f"SELECT * FROM {table_name} ORDER BY all").strip()
|
|
)
|
|
assert (
|
|
"100022\t2025-06-04 18:40:56.000000"
|
|
== node.query(
|
|
f"SELECT col_x2D2.col_x2D3, col_x2D2.col_x2D4 FROM {table_name} ORDER BY all"
|
|
).strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("column_mapping", ["", "name"])
|
|
def test_subcolumns_2(started_cluster, column_mapping):
|
|
instance = started_cluster.instances["node1"]
|
|
cluster.instances["node_with_disabled_delta_kernel"]
|
|
table_name = randomize_table_name("test_write_column_order")
|
|
spark = started_cluster.spark_session
|
|
path = f"/var/lib/clickhouse/user_files/{table_name}"
|
|
|
|
if column_mapping == "name":
|
|
table_properties = "'delta.minReaderVersion' = '2', 'delta.minWriterVersion' = '5', 'delta.columnMapping.mode' = 'name'"
|
|
else:
|
|
table_properties = ""
|
|
|
|
create_query = f"""
|
|
CREATE TABLE {table_name}
|
|
(c1 STRUCT<_2: ARRAY<INT>> NOT NULL)
|
|
USING DELTA
|
|
LOCATION '{path}'
|
|
"""
|
|
if len(table_properties) > 0:
|
|
create_query += f" TBLPROPERTIES ({table_properties})"
|
|
|
|
spark.sql(create_query)
|
|
LocalUploader(instance).upload_directory(f"{path}/", f"{path}/")
|
|
|
|
table_function = f"deltaLakeLocal('{path}')"
|
|
assert 0 == int(instance.query(f"SELECT count() FROM {table_function}"))
|
|
assert (
|
|
"c1\tTuple(\\n _2 Array(Nullable(Int32)))"
|
|
== instance.query(f"DESCRIBE TABLE {table_function}").strip()
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} ENGINE = DeltaLakeLocal('{path}') SETTINGS output_format_parquet_compression_method = 'none'"
|
|
)
|
|
assert 0 == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
assert "" == instance.query(f"SELECT {table_name}.`c1._2` FROM {table_name}")
|
|
|
|
spark.sql(
|
|
f"""
|
|
INSERT INTO {table_name}
|
|
VALUES (named_struct('_2', array(1, NULL, 3))),
|
|
(named_struct('_2', array(4, 5)))
|
|
"""
|
|
)
|
|
LocalUploader(instance).upload_directory(f"{path}/", f"{path}/")
|
|
|
|
assert (
|
|
"([1,NULL,3])\n([4,5])" == instance.query(f"SELECT * FROM {table_name}").strip()
|
|
)
|
|
assert (
|
|
"[1,NULL,3]\n[4,5]"
|
|
== instance.query(f"SELECT {table_name}.`c1._2` FROM {table_name}").strip()
|
|
)
|
|
assert (
|
|
"3\n2"
|
|
== instance.query(
|
|
f"SELECT {table_name}.`c1._2`.size0 FROM {table_name}"
|
|
).strip()
|
|
)
|
|
assert (
|
|
"[0,1,0]\n[0,0]"
|
|
== instance.query(f"SELECT {table_name}.`c1._2`.null FROM {table_name}").strip()
|
|
)
|
|
|
|
|
|
def test_write_column_order(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_write_column_order")
|
|
result_file = f"/var/lib/clickhouse/user_files/{table_name}_data"
|
|
schema = pa.schema([("c1", pa.int32(), False), ("c0", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake(
|
|
f"file:///{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
mode="overwrite",
|
|
)
|
|
LocalUploader(instance).upload_directory(f"/{result_file}/", f"/{result_file}/")
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (c0 String, c1 Int32) ENGINE = DeltaLakeLocal('/{result_file}') SETTINGS output_format_parquet_compression_method = 'none'"
|
|
)
|
|
num_rows = 10
|
|
instance.query(
|
|
f"INSERT INTO {table_name} (c1, c0) SELECT number as c1, toString(number % 2) as c0 FROM numbers(10)"
|
|
)
|
|
|
|
assert num_rows == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
assert (
|
|
"0\t0\n1\t1\n0\t2\n1\t3\n0\t4\n1\t5\n0\t6\n1\t7\n0\t8\n1\t9"
|
|
== instance.query(f"SELECT c0, c1 FROM {table_name}").strip()
|
|
)
|
|
|
|
instance.query(
|
|
f"INSERT INTO {table_name} (c1, c0) SELECT c1, c0 FROM generateRandom('c1 Int32, c0 String', 16920040705558589162, 7706, 3) LIMIT {num_rows}"
|
|
)
|
|
|
|
assert num_rows * 2 == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
|
|
|
|
def test_write_schema_mismatch_raises_user_error(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_write_schema_mismatch")
|
|
|
|
# Case 1: a Nested column declared in the table definition flattens to
|
|
# subcolumns (c0.c1), which do not match the table-level write schema (c0).
|
|
nested_file = f"/var/lib/clickhouse/user_files/{table_name}_nested"
|
|
nested_type = pa.list_(pa.struct([("c1", pa.int32())]))
|
|
nested_schema = pa.schema([("c0", nested_type)])
|
|
write_deltalake(
|
|
f"file:///{nested_file}",
|
|
pa.Table.from_arrays([pa.array([], type=nested_type)], schema=nested_schema),
|
|
mode="overwrite",
|
|
)
|
|
LocalUploader(instance).upload_directory(f"/{nested_file}/", f"/{nested_file}/")
|
|
|
|
nested_table = randomize_table_name("nested")
|
|
instance.query(
|
|
f"CREATE TABLE {nested_table} (c0 Nested(c1 Int32)) "
|
|
f"ENGINE = DeltaLakeLocal('/{nested_file}') "
|
|
f"SETTINGS output_format_parquet_compression_method = 'none'"
|
|
)
|
|
error = instance.query_and_get_error(
|
|
f"INSERT INTO {nested_table} (c0.c1) SELECT [1, 2, 3]"
|
|
)
|
|
assert "INCOMPATIBLE_COLUMNS" in error
|
|
assert "do not match" in error
|
|
|
|
# Case 2: explicit structure inserts a subset of the table's columns.
|
|
subset_file = f"/var/lib/clickhouse/user_files/{table_name}_subset"
|
|
subset_schema = pa.schema([("c1", pa.int32()), ("c0", pa.int32())])
|
|
write_deltalake(
|
|
f"file:///{subset_file}",
|
|
pa.Table.from_arrays(
|
|
[pa.array([], type=pa.int32()), pa.array([], type=pa.int32())],
|
|
schema=subset_schema,
|
|
),
|
|
mode="overwrite",
|
|
)
|
|
LocalUploader(instance).upload_directory(f"/{subset_file}/", f"/{subset_file}/")
|
|
|
|
error = instance.query_and_get_error(
|
|
f"INSERT INTO TABLE FUNCTION deltaLakeLocal('/{subset_file}', 'Parquet', 'c0 Int32') (c0) "
|
|
f"VALUES (1)"
|
|
)
|
|
assert "INCOMPATIBLE_COLUMNS" in error
|
|
assert "do not match" in error
|
|
|
|
# Server stays alive: a well-formed write to the same table still works.
|
|
instance.query(
|
|
f"INSERT INTO TABLE FUNCTION deltaLakeLocal('/{subset_file}') (c1, c0) VALUES (1, 2)"
|
|
)
|
|
assert "1\t2" == instance.query(
|
|
f"SELECT c1, c0 FROM deltaLakeLocal('/{subset_file}')"
|
|
).strip()
|
|
|
|
|
|
@pytest.mark.parametrize("column_mapping", ["", "name"])
|
|
def test_type_from_storage_def(started_cluster, column_mapping):
|
|
instance = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_types_2")
|
|
spark = started_cluster.spark_session
|
|
path = f"/var/lib/clickhouse/user_files/{table_name}"
|
|
|
|
spark_schema = StructType(
|
|
[
|
|
StructField("c0", IntegerType(), nullable=False),
|
|
StructField("c1", TimestampType(), nullable=False),
|
|
StructField(
|
|
"c2",
|
|
StructType(
|
|
[
|
|
StructField("created_at", TimestampType(), nullable=True),
|
|
StructField("updated_at", TimestampType(), nullable=True),
|
|
]
|
|
),
|
|
nullable=False,
|
|
),
|
|
]
|
|
)
|
|
data = [
|
|
(
|
|
1,
|
|
datetime(2000, 10, 10, 0, 0, 0),
|
|
{
|
|
"created_at": datetime(2000, 11, 11, 0, 0, 0),
|
|
"updated_at": datetime(2000, 12, 12, 0, 0, 0),
|
|
},
|
|
)
|
|
]
|
|
|
|
df = spark.createDataFrame(data, schema=spark_schema)
|
|
if len(column_mapping) > 0:
|
|
df.write.format("delta").option("delta.minReaderVersion", "2").option(
|
|
"delta.minWriterVersion", "5"
|
|
).option("delta.columnMapping.mode", column_mapping).save(path)
|
|
else:
|
|
df.write.format("delta").save(path)
|
|
|
|
LocalUploader(instance).upload_directory(f"{path}/", f"{path}/")
|
|
|
|
instance.query(
|
|
f"""CREATE TABLE {table_name}
|
|
(c0 Int32, c1 DateTime, c2 Tuple(created_at DateTime, updated_at DateTime))
|
|
ENGINE = DeltaLakeLocal('{path}') SETTINGS output_format_parquet_compression_method = 'none'
|
|
"""
|
|
)
|
|
|
|
assert (
|
|
"('2000-11-11 00:00:00','2000-12-12 00:00:00')"
|
|
== instance.query(f"SELECT c2 FROM {table_name}").strip()
|
|
)
|
|
assert (
|
|
"2000-11-11 00:00:00"
|
|
== instance.query(f"SELECT c2.created_at FROM {table_name}").strip()
|
|
)
|
|
assert (
|
|
"2000-11-11 00:00:00"
|
|
== instance.query(f"SELECT c2.created_at FROM {table_name}").strip()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["0", "1"])
|
|
def test_system_table(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_multiple_log_files")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) == 2 # 1 metadata files + 1 data file
|
|
|
|
s3_objects = list(
|
|
minio_client.list_objects(bucket, f"{TABLE_NAME}/_delta_log/", recursive=True)
|
|
)
|
|
assert len(s3_objects) == 1
|
|
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}",
|
|
settings={"delta_lake_log_metadata": 1},
|
|
)
|
|
)
|
|
== 100
|
|
)
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 100, 200), f"/{TABLE_NAME}", mode="append"
|
|
)
|
|
files = upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
assert len(files) == 4 # 2 metadata files + 2 data files
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 200
|
|
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME}",
|
|
settings={"delta_lake_log_metadata": 1},
|
|
)
|
|
instance.query("SYSTEM FLUSH LOGS delta_lake_metadata_log")
|
|
|
|
assert (
|
|
int(
|
|
instance.query(
|
|
"SELECT count(DISTINCT file_path) FROM system.delta_lake_metadata_log"
|
|
)
|
|
)
|
|
== 2
|
|
)
|
|
contents = instance.query(
|
|
"SELECT content FROM system.delta_lake_metadata_log"
|
|
).split("\n")
|
|
for content in contents:
|
|
if len(content) == 0:
|
|
continue
|
|
assert "commitInfo" in content
|
|
assert ".parquet" in content
|
|
instance.query("TRUNCATE TABLE system.delta_lake_metadata_log")
|
|
|
|
|
|
def test_truncate(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
table_name = randomize_table_name("test_truncate")
|
|
result_file = f"{table_name}_data"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake_with_retry(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
)
|
|
|
|
instance.query(
|
|
f"CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')"
|
|
)
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number as name, toString(number) as id from numbers(10)"
|
|
)
|
|
|
|
assert 10 == int(instance.query(f"SELECT count() FROM {table_name}"))
|
|
|
|
s3_objects = list(minio_client.list_objects(bucket, result_file, recursive=True))
|
|
|
|
def count_files():
|
|
count = 0
|
|
for obj in s3_objects:
|
|
print(f"File: {obj.object_name}")
|
|
count = count + 1
|
|
return count
|
|
|
|
assert count_files() == 3
|
|
assert "not supported" in instance.query_and_get_error(
|
|
f"TRUNCATE TABLE {table_name}"
|
|
)
|
|
assert count_files() == 3
|
|
|
|
|
|
@pytest.mark.parametrize("on_cluster", [False, True])
|
|
def test_deletion_vector(started_cluster, on_cluster):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_dv")
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
path = f"/{table_name}"
|
|
|
|
suffix = "Cluster" if on_cluster else ""
|
|
cluster = "cluster," if on_cluster else ""
|
|
delta_function = f"""
|
|
deltaLake{suffix}({cluster}
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
|
|
(
|
|
DeltaTable.create(spark)
|
|
.tableName(table_name)
|
|
.location(path)
|
|
.addColumn("a", "INT", nullable=True)
|
|
.addColumn("b", "STRING", nullable=False)
|
|
.addColumn("c", "STRING", nullable=False)
|
|
.partitionedBy("a")
|
|
.property("delta.minReaderVersion", "2")
|
|
.property("delta.minWriterVersion", "5")
|
|
.property("delta.columnMapping.mode", "name")
|
|
.property("delta.enableDeletionVectors", "true")
|
|
.execute()
|
|
)
|
|
|
|
schema = StructType(
|
|
[
|
|
StructField("a", ShortType(), nullable=True),
|
|
StructField("b", StringType(), nullable=False),
|
|
StructField("c", StringType(), nullable=False),
|
|
]
|
|
)
|
|
|
|
data = [(1, "a", "a"), (1, "b", "a"), (1, "c", "a"), (1, "b", "c"), (2, "b", "a")]
|
|
df = spark.createDataFrame(data=data, schema=schema)
|
|
df.write.format("delta").partitionBy("a").mode("overwrite").save(path)
|
|
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
assert 5 == int(
|
|
node.query(f"SELECT count() FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
|
|
spark.sql(f"DELETE FROM {table_name} WHERE b = 'b'")
|
|
upload_directory(minio_client, bucket, path, "")
|
|
|
|
assert 2 == int(
|
|
node.query(f"SELECT count() FROM {delta_function} ORDER BY all").strip()
|
|
)
|
|
# check rows indexes size is 2
|
|
# (not 3 because third deleted row is inside a different partition and represents a single row inside it)
|
|
assert node.contains_in_log("Row indexes size 2 for file")
|
|
|
|
|
|
@pytest.mark.parametrize("cluster", [False, True])
|
|
def test_partition_columns_jumbled(started_cluster, cluster):
|
|
"""Test for bug https://github.com/ClickHouse/ClickHouse/issues/95526
|
|
|
|
Reproduces issue where partition column values become incorrect when inserting
|
|
from DeltaLake into ClickHouse with many columns and type conversions.
|
|
"""
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_partition_columns_jumbled")
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("id", pa.int32()),
|
|
("region", pa.string()),
|
|
("state", pa.string()),
|
|
]
|
|
)
|
|
|
|
data = [
|
|
pa.array([1, 2], type=pa.int32()),
|
|
pa.array(["west", "east"], type=pa.string()),
|
|
pa.array(["CA", "NY"], type=pa.string()),
|
|
]
|
|
|
|
storage_options = {
|
|
"AWS_ENDPOINT_URL": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}",
|
|
"AWS_ACCESS_KEY_ID": minio_access_key,
|
|
"AWS_SECRET_ACCESS_KEY": minio_secret_key,
|
|
"AWS_ALLOW_HTTP": "true",
|
|
"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
|
|
}
|
|
path = f"s3://root/{table_name}"
|
|
table = pa.Table.from_arrays(data, schema=schema)
|
|
|
|
write_deltalake_with_retry(
|
|
path, table, storage_options=storage_options, partition_by=["region", "state"]
|
|
)
|
|
|
|
if cluster:
|
|
delta_function = f"""
|
|
deltaLakeCluster(
|
|
cluster,
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}')
|
|
"""
|
|
else:
|
|
delta_function = f"""
|
|
deltaLake(
|
|
'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}' ,
|
|
'{minio_access_key}',
|
|
'{minio_secret_key}',
|
|
SETTINGS allow_experimental_delta_kernel_rs=0)
|
|
"""
|
|
|
|
dst_table = f"{table_name}_dst"
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {dst_table} (
|
|
id Int32,
|
|
region String,
|
|
state String
|
|
) ENGINE = MergeTree()
|
|
ORDER BY id
|
|
"""
|
|
)
|
|
|
|
node.query(
|
|
f"""
|
|
INSERT INTO {dst_table}
|
|
SELECT * FROM {delta_function}
|
|
"""
|
|
)
|
|
|
|
result_from_delta = node.query(
|
|
f"SELECT * FROM {delta_function} ORDER BY id",
|
|
settings={"allow_experimental_delta_kernel_rs": 1, "use_hive_partitioning": 0},
|
|
).strip()
|
|
|
|
result_from_table = node.query(f"SELECT * FROM {dst_table} ORDER BY id").strip()
|
|
|
|
assert (
|
|
result_from_delta == result_from_table
|
|
), f"Partition columns jumbled!\nFrom DeltaLake:\n{result_from_delta}\n\nFrom table:\n{result_from_table}"
|
|
|
|
expected = "1\twest\tCA\n2\teast\tNY"
|
|
assert (
|
|
result_from_table == expected
|
|
), f"Data doesn't match!\nExpected:\n{expected}\n\nGot:\n{result_from_table}"
|
|
|
|
|
|
def test_network_activity_with_system_tables(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
bucket = started_cluster.minio_bucket
|
|
table_name = randomize_table_name("test_network_activity_with_system_tables")
|
|
result_file = f"{table_name}_data"
|
|
|
|
schema = pa.schema([("id", pa.int32(), False), ("name", pa.string(), False)])
|
|
empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
|
|
write_deltalake(
|
|
f"s3://root/{result_file}",
|
|
pa.Table.from_arrays(empty_arrays, schema=schema),
|
|
storage_options=get_storage_options(started_cluster),
|
|
mode="overwrite",
|
|
)
|
|
|
|
instance.query(
|
|
f"""
|
|
CREATE TABLE {table_name} (id Int32, name String) ENGINE = DeltaLake('http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{result_file}/', 'minio', '{minio_secret_key}')
|
|
"""
|
|
)
|
|
|
|
instance.query(
|
|
f"INSERT INTO {table_name} SELECT number as name, toString(number) as id from numbers(10)"
|
|
)
|
|
|
|
query_id = f"{table_name}_query"
|
|
instance.query(
|
|
f"SELECT * FROM system.tables WHERE name = '{table_name}'", query_id=query_id
|
|
)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS text_log")
|
|
|
|
assert 0 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' AND message LIKE '%Initialized scan state%'"
|
|
)
|
|
)
|
|
|
|
|
|
def test_table_statistics(started_cluster):
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_table_statistics")
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
|
|
def get_parquet_files_size(table_name):
|
|
"""Calculate total size of parquet files in S3 for the Delta table."""
|
|
total_size = 0
|
|
s3_objects = minio_client.list_objects(bucket, table_name, recursive=True)
|
|
for obj in s3_objects:
|
|
# Only count parquet files, exclude _delta_log directory
|
|
if (
|
|
obj.object_name.endswith(".parquet")
|
|
and "/_delta_log/" not in obj.object_name
|
|
):
|
|
total_size += obj.size
|
|
return total_size
|
|
|
|
delta_path = f"/{TABLE_NAME}"
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, 0, 100),
|
|
delta_path,
|
|
partition_by="a",
|
|
mode="overwrite",
|
|
)
|
|
|
|
for i in range(1, 12):
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, i * 100, (i + 1) * 100),
|
|
delta_path,
|
|
partition_by="a",
|
|
mode="append",
|
|
)
|
|
|
|
default_upload_directory(
|
|
started_cluster,
|
|
"s3",
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
create_delta_table(
|
|
instance,
|
|
"s3",
|
|
TABLE_NAME,
|
|
started_cluster,
|
|
)
|
|
|
|
result = instance.query(
|
|
f"SELECT total_rows, total_bytes FROM system.tables WHERE name = '{TABLE_NAME}'"
|
|
)
|
|
|
|
total_rows, total_bytes = map(lambda x: int(x), result.strip().split("\t"))
|
|
expected_rows = 1200
|
|
expected_bytes = get_parquet_files_size(TABLE_NAME)
|
|
|
|
assert total_rows == expected_rows
|
|
assert total_bytes == expected_bytes
|
|
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, 1200, 1300),
|
|
delta_path,
|
|
partition_by="a",
|
|
mode="append",
|
|
)
|
|
|
|
default_upload_directory(
|
|
started_cluster,
|
|
"s3",
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
result = instance.query(
|
|
f"SELECT total_rows, total_bytes FROM system.tables WHERE name = '{TABLE_NAME}'"
|
|
)
|
|
|
|
total_rows, total_bytes = map(lambda x: int(x), result.strip().split("\t"))
|
|
expected_rows = 1300
|
|
expected_bytes = get_parquet_files_size(TABLE_NAME)
|
|
|
|
assert total_rows == expected_rows
|
|
assert total_bytes == expected_bytes
|
|
|
|
def check_with_condition(count, start_row, snapshot_version):
|
|
expected_rows = start_row + 100
|
|
query_id_1 = f"test_stats_partial_{TABLE_NAME}_{count}_query"
|
|
query_id_2 = f"test_stats_full_{TABLE_NAME}_query_2"
|
|
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, start_row, start_row + 100),
|
|
delta_path,
|
|
partition_by="a",
|
|
mode="append",
|
|
)
|
|
default_upload_directory(
|
|
started_cluster,
|
|
"s3",
|
|
delta_path,
|
|
"",
|
|
)
|
|
|
|
if count:
|
|
assert 100 == int(
|
|
instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} WHERE a >= {start_row}",
|
|
query_id=query_id_1,
|
|
)
|
|
)
|
|
else:
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} WHERE a >= 1200", query_id=query_id_1
|
|
)
|
|
|
|
if count:
|
|
assert expected_rows == int(
|
|
instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_2)
|
|
)
|
|
else:
|
|
instance.query(f"SELECT * FROM {TABLE_NAME}", query_id=query_id_2)
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
if count:
|
|
message = "Updated statistics for snapshot version " + str(snapshot_version)
|
|
else:
|
|
message = (
|
|
"Updated statistics from data files iterator for snapshot version "
|
|
+ str(snapshot_version)
|
|
)
|
|
log_result_1 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_1}' "
|
|
f"AND message LIKE '%{message}%'"
|
|
)
|
|
assert int(log_result_1) == 0
|
|
|
|
log_result_2 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_2}' "
|
|
f"AND message LIKE '%{message}%'"
|
|
)
|
|
assert int(log_result_2) == 1
|
|
|
|
check_with_condition(True, 1300, 13)
|
|
check_with_condition(False, 1400, 14)
|
|
|
|
# Test switching between snapshot versions
|
|
# Query with version 0 - should have 100 rows
|
|
query_id_v0 = f"test_switch_v0_{TABLE_NAME}"
|
|
result_v0 = instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} SETTINGS delta_lake_snapshot_version = 0",
|
|
query_id=query_id_v0,
|
|
)
|
|
assert int(result_v0.strip()) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
message_v0 = "Updated statistics for snapshot version 0"
|
|
log_result_v0 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_v0}' "
|
|
f"AND message LIKE '%{message_v0}%'"
|
|
)
|
|
assert int(log_result_v0) == 1
|
|
|
|
# Query with version 5 - should have 600 rows
|
|
query_id_v5 = f"test_switch_v5_{TABLE_NAME}"
|
|
result_v5 = instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} SETTINGS delta_lake_snapshot_version = 5",
|
|
query_id=query_id_v5,
|
|
)
|
|
assert int(result_v5.strip()) == 600
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
message_v5 = "Updated statistics for snapshot version 5"
|
|
log_result_v5 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_v5}' "
|
|
f"AND message LIKE '%{message_v5}%'"
|
|
)
|
|
assert int(log_result_v5) == 1
|
|
|
|
# Switch to version 10
|
|
query_id_v10 = f"test_switch_v10_{TABLE_NAME}"
|
|
result_v10 = instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME} SETTINGS delta_lake_snapshot_version = 10",
|
|
query_id=query_id_v10,
|
|
)
|
|
assert int(result_v10.strip()) == 1100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
message_v10 = "Updated statistics for snapshot version 10"
|
|
log_result_v10 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_v10}' "
|
|
f"AND message LIKE '%{message_v10}%'"
|
|
)
|
|
assert int(log_result_v10) == 1
|
|
|
|
# Query without version - should use latest with 1500 rows
|
|
query_id_latest = f"test_switch_latest_{TABLE_NAME}"
|
|
result_latest = instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_latest
|
|
)
|
|
assert int(result_latest.strip()) == 1500
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1"])
|
|
def test_system_reload_delta_kernel_tracing(started_cluster, use_delta_kernel):
|
|
"""Test SYSTEM RELOAD DELTA KERNEL TRACING command with different log levels."""
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_delta_kernel_tracing")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
# First query WITHOUT tracing enabled
|
|
query_id_before = f"{TABLE_NAME}_before_tracing"
|
|
result = instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_before)
|
|
assert int(result) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Verify that NO DeltaKernelTracing logs appear before enabling tracing
|
|
logs_before_count = int(instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE logger_name = 'DeltaKernelTracing' AND query_id = '{query_id_before}'"
|
|
).strip())
|
|
assert logs_before_count == 0, f"Expected 0 DeltaKernelTracing logs before enabling tracing, got {logs_before_count}"
|
|
|
|
# Now enable TRACE level tracing
|
|
instance.query("SYSTEM RELOAD DELTA KERNEL TRACING TRACE")
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Verify the reload was successful
|
|
reload_success = instance.query(
|
|
"SELECT count() FROM system.text_log WHERE message LIKE '%Delta kernel tracing level reloaded to TRACE%'"
|
|
)
|
|
assert int(reload_success), "Expected successful reload to TRACE level to be logged"
|
|
|
|
# Query the same table WITH tracing enabled at TRACE level
|
|
query_id_trace = f"{TABLE_NAME}_trace_query"
|
|
result = instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_trace)
|
|
assert int(result) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Now delta kernel logs should appear
|
|
logs_at_trace = int(instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE logger_name = 'DeltaKernelTracing' AND query_id = '{query_id_trace}'"
|
|
).strip())
|
|
assert logs_at_trace > 0, f"Expected DeltaKernelTracing logs at TRACE level, got {logs_at_trace}"
|
|
|
|
# For this query, we expect DEBUG level logs (not necessarily TRACE)
|
|
# At TRACE level, we should see DEBUG logs
|
|
debug_level_count = int(instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE logger_name = 'DeltaKernelTracing' AND query_id = '{query_id_trace}' AND level = 'Debug'"
|
|
).strip())
|
|
assert debug_level_count > 0, f"Expected Debug level logs at TRACE level, got {debug_level_count}"
|
|
|
|
# Now test switching to INFO level - should filter out DEBUG logs
|
|
instance.query("SYSTEM RELOAD DELTA KERNEL TRACING INFO")
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Verify the reload to INFO was successful
|
|
reload_info_success = instance.query(
|
|
"SELECT count() FROM system.text_log WHERE message LIKE '%Delta kernel tracing level reloaded to INFO%'"
|
|
)
|
|
assert int(reload_info_success), "Expected successful reload to INFO level to be logged"
|
|
|
|
query_id_info = f"{TABLE_NAME}_info_query"
|
|
result = instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_info)
|
|
assert int(result) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Verify that DEBUG level logs don't appear at INFO level (they should be filtered out)
|
|
debug_count = int(instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE logger_name = 'DeltaKernelTracing' AND query_id = '{query_id_info}' AND level = 'Debug'"
|
|
).strip())
|
|
|
|
assert debug_count == 0, f"Expected 0 Debug logs at INFO level, got {debug_count}"
|
|
|
|
# Test ERROR level - should not see any WARN, INFO, or DEBUG logs
|
|
instance.query("SYSTEM RELOAD DELTA KERNEL TRACING ERROR")
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# Verify the reload to ERROR was successful
|
|
reload_error_success = instance.query(
|
|
"SELECT count() FROM system.text_log WHERE message LIKE '%Delta kernel tracing level reloaded to ERROR%'"
|
|
)
|
|
assert int(reload_error_success), "Expected successful reload to ERROR level to be logged"
|
|
|
|
query_id_error = f"{TABLE_NAME}_error_query"
|
|
result = instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=query_id_error)
|
|
assert int(result) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
# At ERROR level, should not see any higher level logs (WARN/INFO/DEBUG)
|
|
higher_count = int(instance.query(
|
|
f"""
|
|
SELECT count()
|
|
FROM system.text_log
|
|
WHERE logger_name = 'DeltaKernelTracing'
|
|
AND query_id = '{query_id_error}'
|
|
AND level IN ('Warning', 'Information', 'Debug')
|
|
"""
|
|
).strip())
|
|
assert higher_count == 0, f"Expected 0 higher-level logs at ERROR level, got {higher_count}"
|
|
|
|
# Test invalid level
|
|
error = instance.query_and_get_error("SYSTEM RELOAD DELTA KERNEL TRACING INVALID")
|
|
assert "BAD_ARGUMENTS" in error or "Invalid delta kernel tracing level" in error
|
|
|
|
|
|
@pytest.mark.parametrize("use_delta_kernel", ["1"])
|
|
def test_early_return_limit(started_cluster, use_delta_kernel):
|
|
"""Test that scan callback stops early when using LIMIT"""
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_early_return_limit")
|
|
|
|
# Create 1000 partitions to test early return
|
|
write_delta_from_df(
|
|
spark,
|
|
generate_data(spark, 0, 1000),
|
|
f"/{TABLE_NAME}",
|
|
mode="overwrite",
|
|
partition_by="a",
|
|
)
|
|
|
|
files = default_upload_directory(started_cluster, "s3", f"/{TABLE_NAME}", "")
|
|
num_data_files = len([f for f in files if f.endswith(".parquet")])
|
|
assert num_data_files == 1000
|
|
|
|
# Explicit schema to avoid schema inference iterator
|
|
bucket = started_cluster.minio_bucket
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME} (a Int64, b String)
|
|
ENGINE=DeltaLake(s3, filename = '{TABLE_NAME}/', url = 'http://minio1:9001/{bucket}/')
|
|
"""
|
|
)
|
|
|
|
# Baseline: full scan
|
|
query_id_full = f"{TABLE_NAME}_full_scan"
|
|
instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} SETTINGS max_threads=1, max_streams_to_max_threads_ratio=1",
|
|
query_id=query_id_full,
|
|
)
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
full_scan_files = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['DeltaLakeScannedFiles'] FROM system.query_log "
|
|
f"WHERE query_id = '{query_id_full}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
# Test: LIMIT scan with early return
|
|
# Use small s3_list_object_keys_size (1) to force frequent queue pauses
|
|
# This makes callback check shutdown frequently and return false to stop iteration
|
|
query_id = f"{TABLE_NAME}_limit_query"
|
|
result = instance.query(
|
|
f"SELECT * FROM {TABLE_NAME} LIMIT 1 SETTINGS max_threads=1, max_streams_to_max_threads_ratio=1, s3_list_object_keys_size=1",
|
|
query_id=query_id,
|
|
)
|
|
assert len(result.strip().split("\n")) == 1
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
scanned_files = int(
|
|
instance.query(
|
|
f"SELECT ProfileEvents['DeltaLakeScannedFiles'] FROM system.query_log "
|
|
f"WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
)
|
|
)
|
|
|
|
assert scanned_files > 0
|
|
assert full_scan_files == num_data_files
|
|
|
|
# Check which shutdown path was hit by querying text_log
|
|
first_check_hits = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' AND message LIKE '%shutdown detected at first check%'"
|
|
).strip()
|
|
queue_check_hits = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' AND message LIKE '%shutdown detected after queue wait%'"
|
|
).strip()
|
|
|
|
first_check_hits = int(first_check_hits)
|
|
queue_check_hits = int(queue_check_hits)
|
|
|
|
print(f"First shutdown check hits: {first_check_hits}")
|
|
print(f"Queue shutdown check hits: {queue_check_hits}")
|
|
|
|
assert first_check_hits > 0 or queue_check_hits > 0
|
|
|
|
assert 1 == int(instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' AND message LIKE '%List batch size is 1/1, shutdown: true%'"
|
|
))
|
|
|
|
|
|
# Early return should scan significantly fewer files
|
|
# With s3_list_object_keys_size=1, queue pauses frequently forcing shutdown checks
|
|
# Should stop very early after consuming just a few files
|
|
assert scanned_files < full_scan_files, \
|
|
f"Early return should scan fewer files: {scanned_files} >= {full_scan_files}"
|
|
# 3 because:
|
|
# we have async reader creation with 2 existing readers at a moment of time,
|
|
# each calls next() and consumes 2 files from the scan.
|
|
# It takes 1 file for the query to stop because of LIMIT 1.
|
|
# But because scan is also asynchronous and continues once batch limit is not reached,
|
|
# we get +1 scanned file.
|
|
assert scanned_files == 3, \
|
|
f"Early return should scan 3 files with LIMIT 1, but scanned {scanned_files}"
|
|
|
|
|
|
def test_struct_dotted_field_names(started_cluster):
|
|
"""Regression test for struct fields whose logical (and physical) names contain dots.
|
|
|
|
In Delta Lake column mapping "name" mode the physical name equals the logical name,
|
|
so a field named `a.foo` has physical name `a.foo`. The map value for the child is
|
|
therefore "data.a.foo" (parent_physical + "." + child_physical). The old code used
|
|
`find_last_of('.')` to strip the parent prefix, which found the wrong dot and turned
|
|
every field into its last segment (e.g. "foo"). Duplicate segment names then caused
|
|
a `DUPLICATE_COLUMN` exception on `SELECT *` (Bug A) and silent NULLs on sub-field
|
|
access (Bug B).
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
table_name = randomize_table_name("test_struct_dotted_field_names")
|
|
path = f"/var/lib/clickhouse/user_files/{table_name}"
|
|
|
|
# ── Table 1: column mapping "name" mode ──────────────────────────────────────
|
|
spark.sql(
|
|
f"""
|
|
CREATE TABLE {table_name} (
|
|
id BIGINT,
|
|
data STRUCT<`a.foo`: STRING, `b.foo`: STRING>,
|
|
data_deep STRUCT<`a.b.c.bar`: STRING, `a.b.d.bar`: STRING, `a.b.c.d.e.bar`: STRING, `a.bar`: STRING>,
|
|
d2 STRUCT<inner: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>,
|
|
d3 STRUCT<l2: STRUCT<l3: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>>,
|
|
d4 STRUCT<l2: STRUCT<l3: STRUCT<l4: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>>>
|
|
)
|
|
USING DELTA
|
|
LOCATION '{path}'
|
|
TBLPROPERTIES (
|
|
'delta.minReaderVersion' = '2',
|
|
'delta.minWriterVersion' = '5',
|
|
'delta.columnMapping.mode' = 'name'
|
|
)
|
|
"""
|
|
)
|
|
spark.sql(
|
|
f"""
|
|
INSERT INTO {table_name}
|
|
VALUES (
|
|
1,
|
|
named_struct('a.foo', 'hello', 'b.foo', 'world'),
|
|
named_struct('a.b.c.bar', 'abcbar', 'a.b.d.bar', 'abdbar', 'a.b.c.d.e.bar', 'abcdebar', 'a.bar', 'abar'),
|
|
named_struct('inner', named_struct('a.foo', 'afoo2', 'b.foo', 'bfoo2')),
|
|
named_struct('l2', named_struct('l3', named_struct('a.foo', 'afoo3', 'b.foo', 'bfoo3'))),
|
|
named_struct('l2', named_struct('l3', named_struct('l4', named_struct('a.foo', 'afoo4', 'b.foo', 'bfoo4'))))
|
|
)
|
|
"""
|
|
)
|
|
LocalUploader(instance).upload_directory(f"{path}/", f"{path}/")
|
|
|
|
table_function = f"deltaLakeLocal('{path}')"
|
|
|
|
# Bug A: SELECT * must not raise DUPLICATE_COLUMN
|
|
result = instance.query(f"SELECT * FROM {table_function}").strip()
|
|
assert result == "1\t('hello','world')\t('abcbar','abdbar','abcdebar','abar')\t(('afoo2','bfoo2'))\t((('afoo3','bfoo3')))\t(((('afoo4','bfoo4'))))", \
|
|
f"Unexpected SELECT * result: {result!r}"
|
|
|
|
# Bug B: depth-1 sub-field access must return actual values, not NULL
|
|
result_a = instance.query(f"SELECT data.`a.foo` FROM {table_function}").strip()
|
|
assert result_a == "hello", f"Unexpected data.`a.foo` result: {result_a!r}"
|
|
|
|
result_b = instance.query(f"SELECT data.`b.foo` FROM {table_function}").strip()
|
|
assert result_b == "world", f"Unexpected data.`b.foo` result: {result_b!r}"
|
|
|
|
result_abc = instance.query(f"SELECT data_deep.`a.b.c.bar` FROM {table_function}").strip()
|
|
assert result_abc == "abcbar", f"Unexpected data_deep.`a.b.c.bar` result: {result_abc!r}"
|
|
|
|
result_abd = instance.query(f"SELECT data_deep.`a.b.d.bar` FROM {table_function}").strip()
|
|
assert result_abd == "abdbar", f"Unexpected data_deep.`a.b.d.bar` result: {result_abd!r}"
|
|
|
|
result_abcde = instance.query(f"SELECT data_deep.`a.b.c.d.e.bar` FROM {table_function}").strip()
|
|
assert result_abcde == "abcdebar", f"Unexpected data_deep.`a.b.c.d.e.bar` result: {result_abcde!r}"
|
|
|
|
result_abar = instance.query(f"SELECT data_deep.`a.bar` FROM {table_function}").strip()
|
|
assert result_abar == "abar", f"Unexpected data_deep.`a.bar` result: {result_abar!r}"
|
|
|
|
# Depth 2, 3, 4 sub-field access
|
|
result_d2 = instance.query(f"SELECT d2.inner.`a.foo` FROM {table_function}").strip()
|
|
assert result_d2 == "afoo2", f"Unexpected d2.inner.`a.foo` result: {result_d2!r}"
|
|
|
|
result_d3 = instance.query(f"SELECT d3.l2.l3.`a.foo` FROM {table_function}").strip()
|
|
assert result_d3 == "afoo3", f"Unexpected d3.l2.l3.`a.foo` result: {result_d3!r}"
|
|
|
|
result_d4 = instance.query(f"SELECT d4.l2.l3.l4.`a.foo` FROM {table_function}").strip()
|
|
assert result_d4 == "afoo4", f"Unexpected d4.l2.l3.l4.`a.foo` result: {result_d4!r}"
|
|
|
|
# ── Table 2: no column mapping (default) ─────────────────────────────────────
|
|
table_name2 = randomize_table_name("test_struct_dotted_field_names_no_cm")
|
|
path2 = f"/var/lib/clickhouse/user_files/{table_name2}"
|
|
|
|
spark.sql(
|
|
f"""
|
|
CREATE TABLE {table_name2} (
|
|
id BIGINT,
|
|
data STRUCT<`a.foo`: STRING, `b.foo`: STRING>,
|
|
data_deep STRUCT<`a.b.c.bar`: STRING, `a.b.d.bar`: STRING, `a.b.c.d.e.bar`: STRING, `a.bar`: STRING>,
|
|
d2 STRUCT<inner: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>,
|
|
d3 STRUCT<l2: STRUCT<l3: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>>,
|
|
d4 STRUCT<l2: STRUCT<l3: STRUCT<l4: STRUCT<`a.foo`: STRING, `b.foo`: STRING>>>>
|
|
)
|
|
USING DELTA
|
|
LOCATION '{path2}'
|
|
TBLPROPERTIES (
|
|
'delta.minReaderVersion' = '2',
|
|
'delta.minWriterVersion' = '5'
|
|
)
|
|
"""
|
|
)
|
|
spark.sql(
|
|
f"""
|
|
INSERT INTO {table_name2}
|
|
VALUES (
|
|
1,
|
|
named_struct('a.foo', 'hello', 'b.foo', 'world'),
|
|
named_struct('a.b.c.bar', 'abcbar', 'a.b.d.bar', 'abdbar', 'a.b.c.d.e.bar', 'abcdebar', 'a.bar', 'abar'),
|
|
named_struct('inner', named_struct('a.foo', 'afoo2', 'b.foo', 'bfoo2')),
|
|
named_struct('l2', named_struct('l3', named_struct('a.foo', 'afoo3', 'b.foo', 'bfoo3'))),
|
|
named_struct('l2', named_struct('l3', named_struct('l4', named_struct('a.foo', 'afoo4', 'b.foo', 'bfoo4'))))
|
|
)
|
|
"""
|
|
)
|
|
LocalUploader(instance).upload_directory(f"{path2}/", f"{path2}/")
|
|
|
|
table_function2 = f"deltaLakeLocal('{path2}')"
|
|
|
|
result2_d2 = instance.query(f"SELECT d2.inner.`a.foo` FROM {table_function2}").strip()
|
|
assert result2_d2 == "afoo2", f"Unexpected d2.inner.`a.foo` result: {result2_d2!r}"
|
|
|
|
result2_d3 = instance.query(f"SELECT d3.l2.l3.`a.foo` FROM {table_function2}").strip()
|
|
assert result2_d3 == "afoo3", f"Unexpected d3.l2.l3.`a.foo` result: {result2_d3!r}"
|
|
|
|
result2_d4 = instance.query(f"SELECT d4.l2.l3.l4.`a.foo` FROM {table_function2}").strip()
|
|
assert result2_d4 == "afoo4", f"Unexpected d4.l2.l3.l4.`a.foo` result: {result2_d4!r}"
|
|
|
|
|
|
def test_snapshot_consistency(started_cluster):
|
|
"""Test that snapshot version is correctly captured and used from metadata snapshot.
|
|
|
|
This test uses failpoints to pause a query, update the table concurrently, and verify
|
|
that the paused query uses the correct (original) snapshot version via log messages.
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_snapshot_consistency")
|
|
|
|
delta_path = f"/{TABLE_NAME}"
|
|
|
|
data = spark.range(100).selectExpr(
|
|
"id as a",
|
|
"id * 2 as b",
|
|
"id * 3 as c"
|
|
)
|
|
write_delta_from_df(spark, data, delta_path)
|
|
|
|
default_upload_directory(started_cluster, "s3", delta_path, "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
initial_count = int(instance.query(f"SELECT count() FROM {TABLE_NAME}").strip())
|
|
assert initial_count == 100, f"Expected 100 rows, got {initial_count}"
|
|
|
|
# Enable failpoint to pause at iterate()
|
|
instance.query("SYSTEM ENABLE FAILPOINT delta_lake_metadata_iterate_pause")
|
|
|
|
result_holder = {"result": None, "error": None, "query_id": None}
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import concurrent.futures
|
|
|
|
executor = ThreadPoolExecutor(max_workers=2)
|
|
|
|
# Wait for failpoint to be hit
|
|
wait_future = executor.submit(
|
|
lambda: instance.query("SYSTEM WAIT FAILPOINT delta_lake_metadata_iterate_pause PAUSE", timeout=30)
|
|
)
|
|
|
|
time.sleep(1)
|
|
|
|
# Start query that will be paused at iterate()
|
|
def run_query():
|
|
try:
|
|
# Use sum() instead of count() to force actual data reading and trigger iterate()
|
|
query_id = f"test_consistency_concurrent_{TABLE_NAME}"
|
|
result_holder["query_id"] = query_id
|
|
result = instance.query(
|
|
f"SELECT sum(a) FROM {TABLE_NAME} WHERE sleepEachRow(0.01) = 0",
|
|
settings={"max_threads": 1, "allow_experimental_delta_kernel_rs": 1},
|
|
query_id=query_id,
|
|
timeout=60
|
|
)
|
|
result_holder["result"] = int(result.strip())
|
|
except Exception as e:
|
|
result_holder["error"] = e
|
|
|
|
query_future = executor.submit(run_query)
|
|
|
|
# Wait for the query to hit the failpoint
|
|
concurrent.futures.wait([wait_future], timeout=15)
|
|
|
|
if wait_future.exception() is not None:
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_lake_metadata_iterate_pause")
|
|
raise Exception("Test setup failed - failpoint not hit")
|
|
|
|
# While query is paused, update the table (creates version 1 with 50 rows)
|
|
df = spark.read.format("delta").load(delta_path)
|
|
df_filtered = df.filter("a < 50")
|
|
df_filtered.write.format("delta").mode("overwrite").save(delta_path)
|
|
default_upload_directory(started_cluster, "s3", delta_path, "")
|
|
|
|
# Verify that a new query sees the updated data (version 1)
|
|
# Use sum() to trigger iterate() so we can check logs
|
|
query_id_v1 = f"test_consistency_version_1_{TABLE_NAME}"
|
|
second_query_result = instance.query(
|
|
f"SELECT sum(a) FROM {TABLE_NAME}",
|
|
query_id=query_id_v1,
|
|
settings={"allow_experimental_delta_kernel_rs": 1},
|
|
).strip()
|
|
# sum(0..49) = 1225
|
|
assert int(second_query_result) == 1225, f"Expected sum=1225 in updated table, got {second_query_result}"
|
|
|
|
# Resume the paused query
|
|
instance.query("SYSTEM NOTIFY FAILPOINT delta_lake_metadata_iterate_pause")
|
|
|
|
# Wait for query to complete
|
|
concurrent.futures.wait([query_future], timeout=60)
|
|
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_lake_metadata_iterate_pause")
|
|
|
|
if result_holder["error"]:
|
|
raise result_holder["error"]
|
|
|
|
first_query_result = result_holder["result"]
|
|
|
|
# The paused query should see the ORIGINAL snapshot (version 0, 100 rows)
|
|
# sum(0..99) = 4950
|
|
if first_query_result == 1225: # sum(0..49) - would indicate bug
|
|
raise Exception(f"Bug detected: First query saw updated snapshot (sum={first_query_result}, version 1) "
|
|
f"instead of its original snapshot (expected sum=4950, version 0)!")
|
|
|
|
assert first_query_result == 4950, f"Expected first query to see original data (sum=4950), got {first_query_result}"
|
|
|
|
# Now verify via logs that the correct snapshot version was used
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
|
|
query_id = result_holder["query_id"]
|
|
|
|
# Check that we used snapshot version 0 from metadata snapshot for iterate
|
|
log_check_iterate = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' "
|
|
f"AND message LIKE '%Using snapshot version 0 from storage metadata snapshot%'"
|
|
)
|
|
assert int(log_check_iterate) > 0, "Expected to find log message about using snapshot version 0 from metadata snapshot for iterate"
|
|
|
|
# Check that we used snapshot version 0 from metadata snapshot for prepareReadingFromFormat
|
|
log_check_prepare = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id}' "
|
|
f"AND message LIKE '%Using snapshot version 0 from storage metadata snapshot for prepareReadingFromFormat%'"
|
|
)
|
|
assert int(log_check_prepare) > 0, "Expected to find log message about using snapshot version 0 from metadata snapshot for prepareReadingFromFormat"
|
|
|
|
# Verify that the second query (after update) used snapshot version 1 from metadata snapshot
|
|
# Check that we used snapshot version 1 from metadata snapshot for iterate
|
|
log_check_iterate_v1 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_v1}' "
|
|
f"AND message LIKE '%Using snapshot version 1 from storage metadata snapshot%'"
|
|
)
|
|
assert int(log_check_iterate_v1) > 0, "Expected to find log message about using snapshot version 1 from metadata snapshot for iterate"
|
|
|
|
# Check that we used snapshot version 1 from metadata snapshot for prepareReadingFromFormat
|
|
log_check_prepare_v1 = instance.query(
|
|
f"SELECT count() FROM system.text_log WHERE query_id = '{query_id_v1}' "
|
|
f"AND message LIKE '%Using snapshot version 1 from storage metadata snapshot for prepareReadingFromFormat%'"
|
|
)
|
|
assert int(log_check_prepare_v1) > 0, "Expected to find log message about using snapshot version 1 from metadata snapshot for prepareReadingFromFormat"
|
|
|
|
|
|
def test_snapshot_initialized_once_per_query(started_cluster):
|
|
"""Test that DeltaLake table snapshot is initialized exactly once per SELECT query.
|
|
|
|
Verifies via the `DeltaLakeSnapshotInitializations` profile event that a single
|
|
SELECT query loads the snapshot from object storage only once, regardless of
|
|
how many files/threads are involved in the query. Two cases are tested:
|
|
- `SELECT count()` which can be served from metadata without reading data files
|
|
- `SELECT sum()` which reads actual data files
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_snapshot_init_once")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
def check_initializations_count(query, expected_result, query_id):
|
|
result = instance.query(query, query_id=query_id, settings={"max_threads": 4})
|
|
assert int(result.strip()) == expected_result
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
initializations = int(instance.query(
|
|
f"SELECT ProfileEvents['DeltaLakeSnapshotInitializations'] "
|
|
f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'"
|
|
).strip())
|
|
assert initializations == 1, (
|
|
f"Query '{query}': expected snapshot to be initialized exactly once, got {initializations}"
|
|
)
|
|
|
|
# FIXME
|
|
# count() currently produces 2 update() calls which reload table snapshot
|
|
# because of updateExternalDynamicMetadata
|
|
#check_initializations_count(
|
|
# f"SELECT count() FROM {TABLE_NAME}",
|
|
# expected_result=100,
|
|
# query_id=f"snapshot_init_count_{TABLE_NAME}",
|
|
#)
|
|
check_initializations_count(
|
|
f"SELECT sum(a) FROM {TABLE_NAME}",
|
|
expected_result=4950,
|
|
query_id=f"snapshot_init_sum_{TABLE_NAME}",
|
|
)
|
|
|
|
cluster_table_function = (
|
|
f"deltaLakeCluster(cluster, 'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/{bucket}/{TABLE_NAME}/', "
|
|
f"'minio', '{minio_secret_key}')"
|
|
)
|
|
# FIXME
|
|
# count() currently produces 2 update() calls which reload table snapshot
|
|
# because of updateExternalDynamicMetadata
|
|
#check_initializations_count(
|
|
# f"SELECT count() FROM {cluster_table_function}",
|
|
# expected_result=100,
|
|
# query_id=f"snapshot_init_cluster_count_{TABLE_NAME}",
|
|
#)
|
|
check_initializations_count(
|
|
f"SELECT sum(a) FROM {cluster_table_function}",
|
|
expected_result=4950,
|
|
query_id=f"snapshot_init_cluster_sum_{TABLE_NAME}",
|
|
)
|
|
|
|
@pytest.mark.parametrize("allow_experimental_analyzer", [0, 1])
|
|
def test_insert_select_from_cluster_with_partition_pruning(started_cluster, allow_experimental_analyzer):
|
|
node = started_cluster.instances["node1"]
|
|
table_name = randomize_table_name("test_insert_select_cluster_pruning")
|
|
|
|
schema = pa.schema(
|
|
[
|
|
("event_time", pa.date32()),
|
|
("account_id", pa.string()),
|
|
("impressions", pa.int64()),
|
|
]
|
|
)
|
|
storage_options = get_storage_options(started_cluster)
|
|
path = f"s3://root/{table_name}"
|
|
|
|
dates = [
|
|
datetime.strptime("2026-02-01", "%Y-%m-%d").date(),
|
|
datetime.strptime("2026-02-02", "%Y-%m-%d").date(),
|
|
datetime.strptime("2026-02-03", "%Y-%m-%d").date(),
|
|
]
|
|
|
|
for dt in dates:
|
|
data = pa.table(
|
|
{
|
|
"event_time": pa.array([dt] * 5, type=pa.date32()),
|
|
"account_id": pa.array([f"acc_{dt}_{i}" for i in range(5)], type=pa.string()),
|
|
"impressions": pa.array(list(range(5)), type=pa.int64()),
|
|
},
|
|
schema=schema,
|
|
)
|
|
write_deltalake_with_retry(
|
|
path,
|
|
data,
|
|
storage_options=storage_options,
|
|
partition_by=["event_time"],
|
|
mode="append",
|
|
)
|
|
|
|
table_function = (
|
|
f"deltaLakeCluster(cluster,"
|
|
f" 'http://{started_cluster.minio_ip}:{started_cluster.minio_port}/root/{table_name}',"
|
|
f" 'minio', '{minio_secret_key}',"
|
|
f" SETTINGS allow_experimental_delta_kernel_rs=1)"
|
|
)
|
|
|
|
total = int(node.query(f"SELECT count() FROM {table_function}"))
|
|
assert total == 15, f"Expected 15 total rows, got {total}"
|
|
|
|
zk_path = f"/clickhouse/tables/{{shard}}/{table_name}_dst"
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {table_name}_dst ON CLUSTER cluster
|
|
(event_time Nullable(Date), account_id Nullable(String), impressions Nullable(Int64))
|
|
ENGINE = ReplicatedMergeTree('{zk_path}', '{{replica}}') ORDER BY tuple()
|
|
"""
|
|
)
|
|
|
|
query_id = f"{table_name}-insert-{uuid.uuid4()}"
|
|
node.query(
|
|
f"""
|
|
INSERT INTO {table_name}_dst (event_time, account_id, impressions)
|
|
SELECT event_time, account_id, impressions
|
|
FROM {table_function}
|
|
WHERE (event_time >= '2026-02-01') AND (event_time < '2026-02-02')
|
|
""",
|
|
query_id=query_id,
|
|
settings={"allow_experimental_delta_kernel_rs": 1, "delta_lake_enable_engine_predicate": 0, "allow_experimental_analyzer" : allow_experimental_analyzer},
|
|
)
|
|
|
|
node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster'")
|
|
|
|
result = int(
|
|
node.query(
|
|
f"SELECT count() FROM {table_name}_dst WHERE event_time = '2026-02-01'"
|
|
)
|
|
)
|
|
assert result == 5
|
|
|
|
result = int(
|
|
node.query(
|
|
f"SELECT count() FROM {table_name}_dst WHERE event_time != '2026-02-01'"
|
|
)
|
|
)
|
|
assert result == 0
|
|
|
|
pruned = int(
|
|
node.query(
|
|
f"""
|
|
SELECT ProfileEvents['DeltaLakePartitionPrunedFiles']
|
|
FROM system.query_log
|
|
WHERE query_id = '{query_id}' AND type = 'QueryFinish' AND is_initial_query = 1
|
|
"""
|
|
)
|
|
)
|
|
assert pruned >= 2
|
|
node.query(f"DROP TABLE IF EXISTS {table_name}_dst ON CLUSTER cluster")
|
|
table_name2 = randomize_table_name("test_insert_select_cluster_pruning_virt")
|
|
zk_path2 = f"/clickhouse/tables/{{shard}}/{table_name2}_dst"
|
|
query_id = f"{table_name2}-insert-{uuid.uuid4()}"
|
|
|
|
node.query(
|
|
f"""
|
|
CREATE TABLE {table_name2}_dst ON CLUSTER cluster
|
|
(
|
|
event_time Nullable(Date),
|
|
account_id Nullable(String),
|
|
impressions Nullable(Int64),
|
|
file_path LowCardinality(String)
|
|
)
|
|
ENGINE = ReplicatedMergeTree('{zk_path2}', '{{replica}}') ORDER BY tuple()
|
|
"""
|
|
)
|
|
|
|
node.query(
|
|
f"""
|
|
INSERT INTO {table_name2}_dst (event_time, account_id, impressions, file_path)
|
|
SELECT event_time, account_id, impressions, _path
|
|
FROM {table_function}
|
|
WHERE _path LIKE '%2026-02-01%'
|
|
""",
|
|
query_id=query_id,
|
|
settings={
|
|
"allow_experimental_delta_kernel_rs": 1,
|
|
"delta_lake_enable_engine_predicate": 0,
|
|
"allow_experimental_analyzer": allow_experimental_analyzer,
|
|
},
|
|
)
|
|
|
|
result = int(
|
|
node.query(
|
|
f"SELECT count() FROM {table_name2}_dst WHERE file_path LIKE '%2026-02-01%'"
|
|
)
|
|
)
|
|
assert result == 5
|
|
|
|
result = int(
|
|
node.query(
|
|
f"SELECT count() FROM {table_name2}_dst WHERE NOT (file_path LIKE '%2026-02-01%')"
|
|
)
|
|
)
|
|
assert result == 0
|
|
node.query(f"DROP TABLE IF EXISTS {table_name2}_dst ON CLUSTER cluster")
|
|
node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster'")
|
|
filtered = int(
|
|
node.query(
|
|
f"""
|
|
SELECT ProfileEvents['ObjectStoragePredicateFilteredObjects']
|
|
FROM system.query_log
|
|
WHERE query_id = '{query_id}' AND type = 'QueryFinish' AND is_initial_query = 1
|
|
"""
|
|
)
|
|
)
|
|
assert filtered >= 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel",
|
|
["0", "1"],
|
|
)
|
|
def test_azure_url_encoded_partition_path(started_cluster, use_delta_kernel):
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
TABLE_NAME = randomize_table_name("test_azure_url_encoded")
|
|
|
|
container_client = started_cluster.blob_service_client.get_container_client(
|
|
started_cluster.azure_container_name
|
|
)
|
|
|
|
# Upload the parquet file at the on-disk (decoded-once) blob path.
|
|
# Partition value "@INTERNAL@" → directory "org=%40INTERNAL%40" (@ → %40).
|
|
partition_dir = "org=%40INTERNAL%40"
|
|
parquet_blob = f"{TABLE_NAME}/{partition_dir}/part-0.parquet"
|
|
|
|
schema = pa.schema([
|
|
pa.field("id", pa.int32()),
|
|
pa.field("org", pa.string()),
|
|
])
|
|
table_data = pa.table(
|
|
{
|
|
"id": pa.array([1, 2, 3], type=pa.int32()),
|
|
"org": pa.array(["@INTERNAL@"] * 3, type=pa.string()),
|
|
},
|
|
schema=schema,
|
|
)
|
|
buf = io.BytesIO()
|
|
pq.write_table(table_data, buf, compression=None)
|
|
parquet_bytes = buf.getvalue()
|
|
container_client.upload_blob(parquet_blob, parquet_bytes, overwrite=True)
|
|
|
|
# Build the Delta log manually. add.path is URI-encoded per the Delta protocol:
|
|
# '%' in the on-disk name (%40) is encoded to '%25', producing %2540.
|
|
add_path = "org=%2540INTERNAL%2540/part-0.parquet"
|
|
|
|
protocol = '{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}'
|
|
metadata = json.dumps(
|
|
{
|
|
"metaData": {
|
|
"id": str(uuid.uuid4()),
|
|
"format": {"provider": "parquet", "options": {}},
|
|
"schemaString": json.dumps(
|
|
{
|
|
"type": "struct",
|
|
"fields": [
|
|
{
|
|
"name": "id",
|
|
"type": "integer",
|
|
"nullable": True,
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"name": "org",
|
|
"type": "string",
|
|
"nullable": True,
|
|
"metadata": {},
|
|
},
|
|
],
|
|
}
|
|
),
|
|
"partitionColumns": ["org"],
|
|
"configuration": {},
|
|
"createdTime": 1600000000000,
|
|
}
|
|
}
|
|
)
|
|
add = json.dumps(
|
|
{
|
|
"add": {
|
|
"path": add_path,
|
|
"partitionValues": {"org": "@INTERNAL@"},
|
|
"size": len(parquet_bytes),
|
|
"modificationTime": 1600000000000,
|
|
"dataChange": True,
|
|
"stats": json.dumps({"numRecords": 3}),
|
|
}
|
|
}
|
|
)
|
|
|
|
log_content = "\n".join([protocol, metadata, add])
|
|
container_client.upload_blob(
|
|
f"{TABLE_NAME}/_delta_log/00000000000000000000.json",
|
|
log_content.encode(),
|
|
overwrite=True,
|
|
)
|
|
|
|
connection_string = started_cluster.env_variables["AZURITE_CONNECTION_STRING"]
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME}
|
|
ENGINE = DeltaLakeAzure('{connection_string}', '{started_cluster.azure_container_name}', '{TABLE_NAME}', Parquet)
|
|
"""
|
|
)
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 3
|
|
|
|
instance.query(f"DROP TABLE IF EXISTS {TABLE_NAME}")
|
|
|
|
@pytest.mark.parametrize(
|
|
"use_delta_kernel",
|
|
["0", "1"],
|
|
)
|
|
def test_azure_empty_blob_path(started_cluster, use_delta_kernel):
|
|
"""based off test_single_log_file_azure_connection_string but parameterized for delta_kernel"""
|
|
|
|
instance = get_node(started_cluster, use_delta_kernel)
|
|
spark = started_cluster.spark_session
|
|
TABLE_NAME = randomize_table_name("test_azure_empty_blob_path")
|
|
|
|
inserted_data = "SELECT number as a, toString(number + 1) as b FROM numbers(100)"
|
|
parquet_data_path = create_initial_data_file(
|
|
started_cluster, instance, inserted_data, TABLE_NAME, node_name=instance.name
|
|
)
|
|
|
|
delta_path = f"/{TABLE_NAME}"
|
|
write_delta_from_file(spark, parquet_data_path, delta_path)
|
|
|
|
files = default_upload_directory(
|
|
started_cluster,
|
|
"azure",
|
|
delta_path,
|
|
"/", # empty remote_path ends up inferring from local_path otherwise
|
|
)
|
|
|
|
assert len(files) == 2 # 1 metadata file + 1 data file
|
|
|
|
empty_blob_path = ""
|
|
connection_string = started_cluster.env_variables["AZURITE_CONNECTION_STRING"]
|
|
instance.query(
|
|
f"""
|
|
DROP TABLE IF EXISTS {TABLE_NAME};
|
|
CREATE TABLE {TABLE_NAME}
|
|
ENGINE=DeltaLakeAzure('{connection_string}', '{started_cluster.azure_container_name}', '{empty_blob_path}', Parquet)
|
|
"""
|
|
)
|
|
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
assert instance.query(f"SELECT * FROM {TABLE_NAME}") == instance.query(
|
|
inserted_data
|
|
)
|
|
|
|
|
|
def test_delta_kernel_rebuild_on_credentials_rotation(started_cluster):
|
|
"""Cache a DeltaLake snapshot, then change the credentials fingerprint between two
|
|
reads and assert the second read rebuilds the kernel engine instead of silently
|
|
reusing the stale one.
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_delta_kernel_rebuild_on_rotation")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
# First read: caches the kernel snapshot state under the current credentials fingerprint.
|
|
baseline_query_id = f"{TABLE_NAME}_baseline"
|
|
assert int(instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}",
|
|
query_id=baseline_query_id,
|
|
)) == 100
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
init_hits = int(instance.query(
|
|
"SELECT count() FROM system.text_log "
|
|
f"WHERE query_id = '{baseline_query_id}' "
|
|
" AND message ILIKE '%Initializing snapshot%'"
|
|
).strip())
|
|
assert init_hits >= 1, "First read should initialize the kernel snapshot state"
|
|
|
|
rebuild_hits_pre = int(instance.query(
|
|
"SELECT count() FROM system.text_log "
|
|
f"WHERE query_id = '{baseline_query_id}' "
|
|
" AND message ILIKE '%Rebuilding kernel snapshot state%'"
|
|
).strip())
|
|
assert rebuild_hits_pre == 0, "First read should NOT trigger the credentials-rotation rebuild"
|
|
|
|
# Second read: the failpoint perturbs the credentials fingerprint, simulating an STS
|
|
# rotation between consecutive reads of the same cached snapshot. The fingerprint
|
|
# mismatch must trigger the rebuild path; the row count must still come back correct.
|
|
rotated_query_id = f"{TABLE_NAME}_rotated"
|
|
instance.query("SYSTEM ENABLE FAILPOINT delta_kernel_force_credentials_fingerprint_drift")
|
|
try:
|
|
assert int(instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}",
|
|
query_id=rotated_query_id,
|
|
)) == 100
|
|
finally:
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_kernel_force_credentials_fingerprint_drift")
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
rebuild_hits = int(instance.query(
|
|
"SELECT count() FROM system.text_log "
|
|
f"WHERE query_id = '{rotated_query_id}' "
|
|
" AND message ILIKE '%Rebuilding kernel snapshot state (credentials rotated)%'"
|
|
).strip())
|
|
assert rebuild_hits >= 1, (
|
|
f"Expected the credentials-rotation rebuild log line to fire for query {rotated_query_id}, "
|
|
f"found {rebuild_hits} hits — the fingerprint check did not rebuild the kernel state."
|
|
)
|
|
|
|
|
|
def test_delta_kernel_retry_on_stale_token_via_catalog_callback(started_cluster):
|
|
"""Simulate delta-kernel reporting an `ExpiredToken` during snapshot init. The
|
|
`delta_kernel_force_stale_token_error` failpoint throws the kernel error exactly
|
|
once; `object_storage_force_refresh_callback_success` short-circuits the catalog
|
|
refresh callback to succeed without an actual catalog. The retry path must rebuild
|
|
the snapshot and return the correct row count.
|
|
"""
|
|
instance = started_cluster.instances["node1"]
|
|
spark = started_cluster.spark_session
|
|
minio_client = started_cluster.minio_client
|
|
bucket = started_cluster.minio_bucket
|
|
TABLE_NAME = randomize_table_name("test_delta_kernel_retry_on_stale_token")
|
|
|
|
write_delta_from_df(
|
|
spark, generate_data(spark, 0, 100), f"/{TABLE_NAME}", mode="overwrite"
|
|
)
|
|
upload_directory(minio_client, bucket, f"/{TABLE_NAME}", "")
|
|
create_delta_table(instance, "s3", TABLE_NAME, started_cluster)
|
|
|
|
# Prime the cache so the second read starts from a clean state.
|
|
assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
|
|
|
|
retry_query_id = f"{TABLE_NAME}_retry"
|
|
instance.query("SYSTEM ENABLE FAILPOINT object_storage_force_refresh_callback_success")
|
|
# Drift the fingerprint so initOrUpdateSnapshot enters its rebuild branch; the
|
|
# stale-token failpoint then trips inside the KernelSnapshotState ctor and the
|
|
# retry path takes over.
|
|
instance.query("SYSTEM ENABLE FAILPOINT delta_kernel_force_credentials_fingerprint_drift")
|
|
instance.query("SYSTEM ENABLE FAILPOINT delta_kernel_force_stale_token_error")
|
|
try:
|
|
assert int(instance.query(
|
|
f"SELECT count() FROM {TABLE_NAME}",
|
|
query_id=retry_query_id,
|
|
)) == 100
|
|
finally:
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_kernel_force_credentials_fingerprint_drift")
|
|
instance.query("SYSTEM DISABLE FAILPOINT object_storage_force_refresh_callback_success")
|
|
# delta_kernel_force_stale_token_error is FIU_ONETIME and self-disables, but
|
|
# disable defensively in case the retry path never reached it.
|
|
instance.query("SYSTEM DISABLE FAILPOINT delta_kernel_force_stale_token_error")
|
|
|
|
instance.query("SYSTEM FLUSH LOGS")
|
|
retry_hits = int(instance.query(
|
|
"SELECT count() FROM system.text_log "
|
|
f"WHERE query_id = '{retry_query_id}' "
|
|
" AND message ILIKE '%stale credentials during snapshot init; rebuilding%'"
|
|
" AND message ILIKE '%refreshed via callback: true%'"
|
|
).strip())
|
|
assert retry_hits >= 1, (
|
|
f"Expected the catalog-callback retry log line to fire for query {retry_query_id}, "
|
|
f"found {retry_hits} hits — the stale-token retry path was not exercised."
|
|
)
|