Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions src/Planner/PlannerJoinTree.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -991,13 +991,16 @@ std::unique_ptr<ExpressionStep> createComputeAliasColumnsStep(
bool parallelReplicasEnabledForStorage(const StoragePtr & current_storage, const ContextPtr & context, const Settings & query_settings)
{
const auto * table_ptr = current_storage.get();
/// `getTargetTable` and `getUnderlyingMergeTreeStorageForParallelReplicas` return a `StoragePtr` by value.
/// A catalog lookup can be the only owner of that storage, so it has to stay alive for the checks below.
StoragePtr underlying_storage;

if (query_settings[Setting::parallel_replicas_allow_view_over_mergetree])
{
const auto * view = typeid_cast<const StorageView *>(current_storage.get());
if (view)
{
auto underlying_storage = view->getUnderlyingMergeTreeStorageForParallelReplicas(context);
underlying_storage = view->getUnderlyingMergeTreeStorageForParallelReplicas(context);
if (!underlying_storage)
return false;

Expand All @@ -1015,7 +1018,8 @@ bool parallelReplicasEnabledForStorage(const StoragePtr & current_storage, const
if (mv->isRefreshable())
return false;

table_ptr = mv->getTargetTable().get();
underlying_storage = mv->getTargetTable();
table_ptr = underlying_storage.get();
}

if (!table_ptr->isMergeTree())
Expand Down
2 changes: 2 additions & 0 deletions src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -411,6 +411,8 @@ std::shared_ptr<IObjectIterator> StorageObjectStorageSource::createFileIterator(
if (configuration->isDataLakeConfiguration()
&& (!storage_metadata || !storage_metadata->datalake_table_state.has_value()))
{
configuration->lazyInitializeIfNeeded(object_storage, local_context);

if (auto state = configuration->getTableStateSnapshot(local_context))
{
auto fixed_metadata = storage_metadata
Expand Down
215 changes: 215 additions & 0 deletions tests/integration/test_database_iceberg/test.py
Original file line number Diff line number Diff line change
Expand Up @@ -3088,3 +3088,218 @@ def test_catalog_listing_error_surfaces_in_system_tables(started_cluster):
assert "table_x" in result

node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}")


def test_alias_iceberg_catalog_cluster(started_cluster):
"""
An `Alias` over a catalog `Iceberg` table must apply the same filter as a direct
catalog read: local, with `object_storage_cluster`, and with parallel replicas.
`count` is the path that already succeeded before snapshot initialization on the
cluster file listing.
"""
node = started_cluster.instances["node1"]

test_ref = f"test_alias_repro_{uuid.uuid4().hex[:8]}"
namespace = f"{test_ref}_pnl"
table_name = "pnl_t"
catalog_table = f"{CATALOG_NAME}.`{namespace}.{table_name}`"
pnl_db = f"{test_ref}_db"
table_settings = {
"allow_database_iceberg": 1,
"allow_insert_into_iceberg": 1,
"write_full_path_in_iceberg_metadata": 1,
}
modes = {
"local": table_settings,
"osc": {**table_settings, "object_storage_cluster": "cluster_simple"},
"pr": {
**table_settings,
"enable_parallel_replicas": 1,
"cluster_for_parallel_replicas": "cluster_simple",
"parallel_replicas_for_cluster_engines": 1,
"max_parallel_replicas": 2,
"parallel_replicas_for_non_replicated_merge_tree": 1,
},
}
filtered = "WHERE TradeDate = '2026-10-06' AND Strategy = 's1'"
filtered_expected = "67\t2026-10-07 06:36:00.000000\n"
limit_expected = "2026-10-06 14:00:00.000000\ts0\n"

create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME)
node.query(
f"""
CREATE TABLE {catalog_table}
(
TimeSlice DateTime64(6, 'UTC'),
TradeDate Date32,
Strategy String,
NetPL Float64
)
ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/', '{minio_access_key}', '{minio_secret_key}')
PARTITION BY TradeDate
""",
settings={
"allow_experimental_database_iceberg": 1,
"write_full_path_in_iceberg_metadata": 1,
},
)
node.query(
f"""
INSERT INTO {catalog_table}
SELECT
toDateTime64('2026-10-06 14:00:00', 6, 'UTC') + number * 60,
toDate32('2026-10-06') + (number % 3),
's' || toString(number % 5),
number / 7
FROM numbers(1000)
""",
settings=table_settings,
)
node.query(f"CREATE DATABASE {pnl_db}")
node.query(f"CREATE TABLE {pnl_db}.pnl_t ENGINE = Alias({CATALOG_NAME}.`{namespace}.{table_name}`)")

try:
for settings in modes.values():
assert node.query(
f"SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM {catalog_table} {filtered}",
settings=settings,
) == filtered_expected
assert node.query(
f"SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM {pnl_db}.pnl_t {filtered}",
settings=settings,
) == filtered_expected
assert node.query(
f"SELECT toTimeZone(TimeSlice, 'UTC'), Strategy FROM {pnl_db}.pnl_t ORDER BY TimeSlice LIMIT 1",
settings=settings,
) == limit_expected
assert node.query(
f"SELECT count() FROM {pnl_db}.pnl_t",
settings=settings,
) == "1000\n"
finally:
node.query(f"DROP TABLE IF EXISTS {pnl_db}.pnl_t")
node.query(f"DROP DATABASE IF EXISTS {pnl_db}")
drop_clickhouse_iceberg_table(node, namespace, table_name, if_exists=True)


def _create_catalog_wrapper(node, engine, name, catalog_db, namespace, table_name, alias_name):
catalog_table = f"{catalog_db}.`{namespace}.{table_name}`"
qualified_name = f"{namespace}.{table_name}"
if engine == "Alias":
node.query(f"CREATE TABLE {name} ENGINE = Alias('{catalog_db}', '{qualified_name}')")
elif engine == "Buffer":
node.query(
f"CREATE TABLE {name} (id Int64) ENGINE = Buffer('{catalog_db}', '{qualified_name}', "
"1, 86400, 86400, 1000000, 1000000, 1000000000, 1000000000)"
)
elif engine == "Merge":
node.query(
f"CREATE TABLE {name} (id Int64) ENGINE = Merge('{catalog_db}', '^{namespace}[.]{table_name}$')"
)
elif engine == "MergeAlias":
node.query(f"CREATE TABLE {name} (id Int64) ENGINE = Merge(currentDatabase(), '^{alias_name}$')")
elif engine == "MaterializedView":
node.query(f"CREATE MATERIALIZED VIEW {name} TO {catalog_table} AS SELECT id FROM {alias_name}")
elif engine == "Distributed":
node.query(
f"CREATE TABLE {name} (id Int64) "
f"ENGINE = Distributed('cluster_simple', currentDatabase(), '{alias_name}')"
)
else:
raise AssertionError(f"Unexpected engine {engine}")


@pytest.mark.parametrize(
"engine",
["Alias", "Buffer", "Merge", "MergeAlias", "MaterializedView", "Loop", "Distributed"],
)
@pytest.mark.parametrize("cluster_mode", ["local", "query", "database"])
def test_datalake_catalog_wrapper_object_storage_cluster(started_cluster, engine, cluster_mode):
"""
A `DataLakeCatalog` lookup builds a fresh storage and leaves its configuration
uninitialized. Each engine below resolves that table again, then lists files.

`cluster_mode` chooses where `object_storage_cluster` comes from:
local keeps it unset, query sets it on the scan, database sets it on the catalog.
"""
if engine == "Loop" and cluster_mode != "local":
pytest.skip("loop has no cluster table function")

node = started_cluster.instances["node1"]
node2 = started_cluster.instances["node2"]

test_ref = f"test_wrapper_osc_{uuid.uuid4().hex[:8]}"
namespace = f"{test_ref}_ns"
table_name = f"{test_ref}_t"
catalog_table = f"{CATALOG_NAME}.`{namespace}.{table_name}`"
cluster_catalog_name = f"{test_ref}_catalog_cluster"
catalog_db = cluster_catalog_name if cluster_mode == "database" else CATALOG_NAME
alias_name = f"{test_ref}_alias"
wrapper_name = alias_name if engine == "Alias" else f"{test_ref}_wrapper"
table_settings = {
"allow_database_iceberg": 1,
"allow_insert_into_iceberg": 1,
"write_full_path_in_iceberg_metadata": 1,
}
settings = table_settings if cluster_mode != "query" else {
**table_settings,
"object_storage_cluster": "cluster_simple",
}
# A clustered `Buffer` destination reports `WithMergeableState`, so `Buffer` plans
# the in-memory branch as a second read of the same query.
distinct = "DISTINCT " if engine == "Buffer" and cluster_mode != "local" else ""
nodes = [node, node2] if engine == "Distributed" else [node]
created = {replica: [] for replica in nodes}

create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME)
create_clickhouse_iceberg_table(
started_cluster, node, namespace, table_name, "(id Int64)"
)
node.query(f"INSERT INTO {catalog_table} VALUES (1), (2), (3);", settings=table_settings)
if cluster_mode == "database":
create_clickhouse_iceberg_database(
started_cluster,
node,
cluster_catalog_name,
{"object_storage_cluster": "cluster_simple"},
)

try:
for replica in nodes:
if replica is node2:
catalog_settings = (
{"object_storage_cluster": "cluster_simple"} if cluster_mode == "database" else {}
)
create_clickhouse_iceberg_database(
started_cluster, replica, catalog_db, catalog_settings
)
if engine in ("Alias", "MergeAlias", "MaterializedView", "Distributed"):
replica.query(
f"CREATE TABLE {alias_name} ENGINE = Alias('{catalog_db}', '{namespace}.{table_name}')"
)
created[replica].append(alias_name)
if engine != "Alias" and engine != "Loop" and replica is node:
_create_catalog_wrapper(
replica, engine, wrapper_name, catalog_db, namespace, table_name, alias_name
)
created[replica].append(wrapper_name)

if engine == "Loop":
source = (
f"(SELECT id FROM loop('{catalog_db}', '{namespace}.{table_name}') "
"WHERE id > 0 LIMIT 3)"
)
else:
source = wrapper_name
assert node.query(
f"SELECT {distinct}id FROM {source} WHERE id > 0 ORDER BY id",
settings=settings,
) == "1\n2\n3\n"
finally:
for replica, names in created.items():
for name in reversed(names):
replica.query(f"DROP TABLE IF EXISTS {name}")
if engine == "Distributed":
node2.query(f"DROP DATABASE IF EXISTS {catalog_db}")
if cluster_mode == "database":
node.query(f"DROP DATABASE IF EXISTS {cluster_catalog_name}")
Loading