Skip to content

Antalya 26.6 - Read clustering columns from delta lake and use it for data skipping - #2395

Open
subkanthi wants to merge 17 commits into
antalya-26.6from
data_lake_clustering_read
Open

subkanthi wants to merge 17 commits into
antalya-26.6from
data_lake_clustering_read

Conversation

@subkanthi

@subkanthi subkanthi commented Sep 17, 2026 •

Copy link
Copy Markdown
Collaborator

closes: ##2394

Changelog category (leave one):

  • New Feature

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

Read clustering columns from delta-kernel-rs and use that for data skipping.

CI/CD Options

Exclude tests:

  • Fast test
  • Integration Tests
  • Stateless tests
  • Stateful tests
  • Unit tests
  • Performance tests
  • Aarch64 tests
  • All with ASAN
  • All with TSAN
  • All with MSAN
  • All with UBSAN
  • All with Coverage
  • All Regression
  • Disable CI Cache

Regression jobs to run:

  • Fast suites (mostly <1h)
  • Aggregate Functions (2h)
  • Alter (1.5h)
  • Benchmark (30m)
  • CAS (content-addressed storage; Antalya only)
  • ClickHouse Keeper (1h)
  • Iceberg (2h)
  • LDAP (1h)
  • OAuth (5m)
  • Parquet (1.5h)
  • RBAC (1.5h)
  • SSL Server (1h)
  • S3 (2h)
  • S3 Export (2h)
  • Swarms (30m)
  • Tiered Storage (2h)

@github-actions

github-actions Bot commented Sep 17, 2026 •

Copy link
Copy Markdown

Workflow [PR], commit [47cd546]

@subkanthi subkanthi changed the title Added integration test to validate reading of delta lake clustering c… Antalya 26.6 - Added integration test to validate reading of delta lake clustering c… Sep 18, 2026
@subkanthi subkanthi changed the title Antalya 26.6 - Added integration test to validate reading of delta lake clustering c… Antalya 26.6 - Read clustering columns from delta lake and use it for data skipping Sep 18, 2026
@subkanthi

Copy link
Copy Markdown
Collaborator Author

This PR is dependent on changes to delta-kernel-rs
created PR
ClickHouse/delta-kernel-rs#18

@subkanthi
subkanthi marked this pull request as ready for review October 7, 2026 15:20
@subkanthi

Copy link
Copy Markdown
Collaborator Author

Audit Report: Altinity/ClickHouse PR #2395

PR: Altinity/ClickHouse#2395 — Read clustering columns from delta lake and use it for data skipping

AI audit note: This review comment was generated by AI (claude-opus-4.6).


Scope

This PR adds support for reading Delta Lake liquid clustering columns from domain metadata (delta.clustering) and using them for a predicate-fallback data-skipping strategy. When the full filter DAG cannot be translated to a delta-kernel expression, the code restricts the filter to only clustering-column conjuncts and pushes that weaker predicate for file-level min/max skipping.

Files changed

  • src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp — core logic
  • src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h — new member + methods
  • src/Common/ProfileEvents.cpp — new profile event
  • src/Core/Settings.cpp — new setting delta_lake_clustering_predicate_fallback
  • src/Core/SettingsChangesHistory.cpp — setting history entries
  • .gitmodules / contrib/delta-kernel-rs — submodule pointer change
  • tests/integration/helpers/cluster.py — unrelated --ignore-pull-failures
  • tests/integration/test_storage_delta/test.py — 3 new integration tests

Call Graph

iterate()
  -> initOrUpdateSchemaIfChanged()             [lazy, mutex-held]
  -> initOrUpdateClusteringColumns()            [lazy, mutex-held]
  -> getKernelSnapshotState()
  -> Iterator ctor
       stores: clustering_columns (copy), enable_clustering_predicate_fallback
       -> scanDataFunc() [background thread]
            -> initScanState()
                 -> getEnginePredicate(full filter)
                 [on failure + fallback enabled + clustering cols non-empty]:
                    -> restrictFilterDAGToInputs(filter, clustering_inputs)
                    -> getEnginePredicate(restricted DAG)
                    -> ffi::scan(snapshot, engine, fallback_predicate)

getClusteringColumns() [public, mutex-held]
  -> initOrUpdateClusteringColumns()

initOrUpdateClusteringColumns() [mutex-held]
  -> getKernelSnapshotState()
  -> ffi::get_domain_metadata_internal("delta.clustering")
  -> JSON parse clusteringColumns array
  -> physical-to-logical name mapping via schema->physical_names_map

Transition Matrix

State Trigger Next State
clustering_columns == nullopt initOrUpdateClusteringColumns clustering_columns == Names{...} or Names{}
Full predicate OK getEnginePredicate succeeds Use full predicate scan
Full predicate fails exception + fallback enabled + clustering cols present Try restricted DAG
Restricted DAG non-trivial getEnginePredicate on restricted DAG succeeds Use fallback predicate scan
Restricted DAG trivial / fallback fails fallthrough Use original no-predicate scan

Confirmed Defects

Medium: Clustering columns not invalidated on credential-rotation snapshot rebuild

  • Impact: After a stale-token recovery path resets kernel_snapshot_state (line 760), clustering_columns remains cached from the previous snapshot state. If the table's clustering columns changed between snapshots (schema evolution), stale clustering info could cause the fallback to push wrong columns or miss new ones. In practice this is low-probability because the snapshot version is pinned per-query, but the invalidation pattern is inconsistent with how schema is handled.
  • Anchor: TableSnapshot.cpp / initOrUpdateClusteringColumns and stale-token recovery path (line 760)
  • Trigger: Credential rotation causes kernel_snapshot_state.reset(), then initOrUpdateClusteringColumns returns early because clustering_columns.has_value() is true from the previous fetch.
  • Why defect: clustering_columns is semantically derived from the snapshot state, but is not invalidated when the snapshot state is rebuilt. The existing schema field has the same pattern (also not invalidated), so this is an inherited design issue — but the PR extends it to a new cached field without acknowledging the existing gap.
  • Fix direction: When kernel_snapshot_state is reset, also reset clustering_columns to std::nullopt.
  • Regression test direction: Test with a schema-evolving Delta table across credential rotation.

Code evidence (TableSnapshot.cpp, stale-token recovery):

// Line 760: kernel_snapshot_state is reset, but clustering_columns is NOT
kernel_snapshot_state.reset();
initOrUpdateSnapshot();
snapshot_stats = getSnapshotStatsImpl();
// clustering_columns still holds old value here

Code evidence (TableSnapshot.cpp, early-return guard):

void TableSnapshot::initOrUpdateClusteringColumns() const
{
    if (clustering_columns.has_value())  // <-- returns early with stale data
        return;
    // ...
}

Medium: restrictFilterDAGToInputs only handles AND — OR-rooted filters silently drop clustering predicates

  • Impact: If the user's WHERE clause is a = 42 OR cityHash64(b) != 0 (an OR), restrictFilterDAGToInputs does NOT replace non-computable children with true — it only does so for and nodes (ActionsDAG.cpp line 3848). The entire OR becomes non-computable, so has_function is false, the fallback path is skipped, and no file skipping occurs. This is correct but surprising (no data corruption), however the LOG_INFO message and the ProfileEvent would not fire, giving no observability that the fallback was attempted and rejected.
  • Anchor: TableSnapshot.cpp / initScanState fallback block; ActionsDAG.cpp:3848
  • Trigger: WHERE clause uses OR with a clustering column and an untranslatable expression.
  • Why defect: The fallback silently does nothing for OR predicates. A TRACE or DEBUG log for "restricted DAG was trivial, skipping fallback" would improve diagnosability.
  • Fix direction: Add a log message in the else branch of if (has_function) explaining the restricted DAG was trivial.
  • Regression test direction: Integration test with OR predicate verifying fallback is NOT applied but a diagnostic log exists.

Code evidence (ActionsDAG.cpp:3847-3856, only and is handled):

if (frame.node->type == ActionsDAG::ActionType::FUNCTION)
{
    const auto & name = frame.node->function_base->getName();

    /// Replace non-computable child in "and" with constant true.
    if (name == "and")
    {
        auto const_column = child->result_type->createColumnConst(0, 1);
        copy_map[child] = &actions.addColumn(std::move(const_column), child->result_type, child->result_name);
        it->second = true;
    }
    // NOTE: "or" is not handled — the non-computable child makes the whole OR non-computable
}

Code evidence (TableSnapshot.cpp, no logging on trivial DAG):

if (has_function)
{
    // ... fallback logic ...
}
// else: no log, no metric — silently falls through

Low: initOrUpdateClusteringColumns depends on schema but does not call initOrUpdateSchemaIfChanged

  • Impact: initOrUpdateClusteringColumns reads schema.has_value() and schema->physical_names_map (line 1148) without ensuring schema has been initialized. In the current call site (iterate), initOrUpdateSchemaIfChanged is always called first (line 985), so this is safe today. But getClusteringColumns (line 1081-1085) calls initOrUpdateClusteringColumns WITHOUT calling initOrUpdateSchemaIfChanged first. If getClusteringColumns is called before iterate on a fresh TableSnapshot where schema == nullopt, then initOrUpdateClusteringColumns will take the else branch (line 1160: logical_columns = std::move(physical_columns)) and return physical names instead of logical names, which would not match the filter DAG's column names.
  • Anchor: TableSnapshot.cpp:1081-1085 (getClusteringColumns) and TableSnapshot.cpp:1148 (initOrUpdateClusteringColumns)
  • Trigger: Call getClusteringColumns before iterate or getTableSchema on a fresh snapshot with physical name mapping.
  • Why defect: Missing precondition — the function depends on schema being populated but does not ensure it.
  • Fix direction: Call initOrUpdateSchemaIfChanged at the start of initOrUpdateClusteringColumns, or at the start of getClusteringColumns.
  • Regression test direction: Unit test calling getClusteringColumns on a fresh snapshot with physical name mapping before any other accessor.

Code evidence (TableSnapshot.cpp:1081-1085, missing schema init):

const DB::Names & TableSnapshot::getClusteringColumns() const
{
    std::lock_guard lock(mutex);
    initOrUpdateClusteringColumns();  // does NOT call initOrUpdateSchemaIfChanged()
    return *clustering_columns;
}

Code evidence (TableSnapshot.cpp:1148, conditional on schema):

DB::Names logical_columns;
if (schema.has_value() && !schema->physical_names_map.empty())
{
    // uses schema->physical_names_map for reverse mapping
}
else
{
    // falls through to physical names — wrong if schema just wasn't loaded yet
    logical_columns = std::move(physical_columns);
}

Low: Settings history has duplicate entry across two version blocks

  • Impact: delta_lake_clustering_predicate_fallback is registered in both the "26.7" and "26.6" blocks of SettingsChangesHistory.cpp. This is likely an error from rebasing. Having the same setting in two version blocks may cause confusion during upgrades, and the compatibility setting resolution picks the latest version entry, so the 26.7 entry wins and the 26.6 entry is dead code.
  • Anchor: src/Core/SettingsChangesHistory.cpp (both version blocks)
  • Trigger: N/A — cosmetic / maintenance issue.
  • Why defect: Redundant entry across version blocks; should be in exactly one.
  • Fix direction: Remove the entry from the version block where it does not belong (likely keep 26.7 only, or whichever is the target version).
  • Regression test direction: N/A.

Code evidence (SettingsChangesHistory.cpp, both entries):

// In the 26.7 block:
{"delta_lake_clustering_predicate_fallback", false, true,
 "New setting to enable clustering-aware predicate fallback ..."},

// In the 26.6 block:
{"delta_lake_clustering_predicate_fallback", false, true,
 "New setting to enable clustering-aware predicate fallback ..."},

Coverage Summary

  • Scope reviewed: TableSnapshot.cpp (Iterator ctor, initScanState fallback, initOrUpdateClusteringColumns, iterate, getClusteringColumns), TableSnapshot.h (new members), SettingsChangesHistory.cpp, Settings.cpp, ProfileEvents.cpp, cluster.py, test.py (3 new tests).
  • Categories passed: JSON parsing (well-guarded with Poco exception catch), FFI resource management (uses KernelPointerWrapper RAII wrappers correctly), predicate fallback logic (correct for AND-only filters), physical-to-logical name mapping (correct when schema is populated), thread safety of new members (Iterator copies values, no shared mutable state in hot path), integration tests (cover basic read, data skipping, and predicate fallback).
  • Categories failed: Stale cache invalidation (clustering_columns not reset on snapshot rebuild), missing schema precondition in getClusteringColumns, duplicate settings history entry.
  • Assumptions/limits: Static analysis only; delta-kernel-rs submodule changes (get_domain_metadata_internal FFI) were not inspected beyond the C++ call site. The restrictFilterDAGToInputs behavior was verified by reading its implementation.

@subkanthi

Copy link
Copy Markdown
Collaborator Author

AI audit note: This review comment was generated by AI (claude-4.6-opus).

Audit update for PR Altinity/ClickHouse#2395 (Delta Lake liquid-clustering predicate fallback):


Scope

Partition Files
Core feature (predicate fallback) TableSnapshot.cpp, TableSnapshot.h
Settings / profile events ProfileEvents.cpp, Settings.cpp, SettingsChangesHistory.cpp
Submodule change .gitmodules, contrib/delta-kernel-rs
Test infra tests/integration/helpers/cluster.py
Integration tests tests/integration/test_storage_delta/test.py

Call Graph

iterate()                          [locks mutex]
  ├─ initOrUpdateSchemaIfChanged()
  │    └─ getKernelSnapshotState() → initOrUpdateSnapshot()
  ├─ initOrUpdateClusteringColumns()          ← NEW
  │    ├─ getKernelSnapshotState()
  │    ├─ ffi::get_domain_metadata_internal()  ← NEW FFI call
  │    ├─ Poco::JSON parse
  │    └─ physical→logical column-name remap
  └─ new Iterator(…, *clustering_columns, …)
       └─ [background thread] scanDataFunc()
            └─ initScanState()
                 ├─ getEnginePredicate(filter)
                 ├─ ffi::scan(predicate)
                 └─ [if exception + fallback enabled + clustering_cols non-empty]  ← NEW
                      ├─ restrictFilterDAGToInputs(clustering_inputs)
                      ├─ has_function check
                      ├─ getEnginePredicate(restricted_dag)
                      └─ ffi::scan(fallback_predicate)

getClusteringColumns()             [locks mutex]           ← NEW public API
  └─ initOrUpdateClusteringColumns()

getSnapshotStats() stale-token retry:
  └─ clustering_columns.reset()                            ← NEW invalidation

Confirmed defects:

Low: Trivial AND(true, true) passes has_function gate, causing wasted ffi::scan

  • Impact: When the original predicate fails to translate and no filter conjunct references a clustering column, restrictFilterDAGToInputs produces AND(true, true). The AND node passes the has_function check, so getEnginePredicate + ffi::scan are called for a predicate that is semantically equivalent to "no filter". The result is a second scan object that provides zero data-skipping benefit. Not a correctness issue — the scan result is identical — but it adds unnecessary latency (one extra FFI round-trip).
  • Anchor: TableSnapshot.cpp / initScanState() — the has_function loop (~line 220).
  • Trigger: WHERE cityHash64(b) != 0 AND length(c) > 5 on a table clustered by column a (no clustering column in the filter, but top-level AND with 2+ non-computable branches).
  • Why defect: Wasted FFI work on every such query; the trivial-predicate scan is functionally identical to the already-created no-predicate scan.
  • Fix direction: After restrictFilterDAGToInputs, also check that at least one input node in the restricted DAG matches a clustering column (i.e. the restricted DAG actually references a clustering column, not just constants).
  • Regression test direction: Add a case where the filter has no clustering-column conjuncts and verify DeltaLakeClusteringPredicateFallbacks is not incremented.

Coverage summary:

  • Scope reviewed: All changed C++ code (TableSnapshot.cpp/.h, Settings.cpp, SettingsChangesHistory.cpp, ProfileEvents.cpp), integration tests, docker infra change.
  • Categories executed & passed:
    • Thread safety / data-race analysis — pass (Iterator owns copies of all mutable data; clustering_columns is guarded by mutex with TSA annotations; no shared mutable access from the background scan thread).
    • Memory lifetime / use-after-free — pass (filter is cloned; clustering_columns is copied by value into Iterator; restricted_dag and fallback_predicate are locals with clear lifetimes).
    • Iterator / reference invalidation — pass (no container mutation concurrent with iteration).
    • Exception safety / partial-update — pass (initOrUpdateClusteringColumns either sets the optional or throws; clustering_columns stays nullopt on failure, safe to retry).
    • Cache invalidation consistency — pass (stale-token retry in getSnapshotStats resets clustering_columns; normal credential-rotation in initOrUpdateSnapshot does not need to reset because the snapshot version is pinned and domain metadata is version-stable).
    • Integer overflow / signedness — pass (loop indices are size_t; static_cast<unsigned int>(i) for Poco API is safe within realistic array sizes).
    • Lock-order / deadlock — pass (single mutex; no nested lock acquisition).
    • Rollback / partial-update — pass (scan replacement in fallback is a simple move-assignment; no intermediate state inconsistency).
    • Settings history — pass (added to both "unreleased" and "26.6" sections, consistent with existing entries in the Altinity fork).
    • Logical name consistency — pass (clustering_columns are mapped to logical names; filter DAG uses logical names; restrictFilterDAGToInputs and getEnginePredicate both operate in the logical-name space).
    • JSON parsing robustness — pass (getArray null checks present; empty sub-arrays skipped via continue; Poco exceptions caught and re-thrown as DB::Exception).
    • FFI ownership — pass (result from get_domain_metadata_internal wrapped in unique_ptr<std::string> matching KernelUtils::allocateString contract).
  • Categories not applicable:
    • Security / auth bypass — N/A (feature is read-only data skipping; no auth changes).
    • Protocol / API parity — N/A (no new external API; internal-only change).
  • Assumptions / limits: Static analysis only; delta-kernel-rs submodule update (Altinity fork, commit 625c2fa) was not audited at the Rust level. The behavior of ffi::get_domain_metadata_internal and restrictFilterDAGToInputs was assessed by API contract, not by reading their full implementations.

Additional observations (not defects):

  1. --ignore-pull-failures in cluster.py: This is a docker-compose v1 flag. Docker Compose v2 added it later for backward compatibility, but very old v2 releases may not support it. If the CI images use a sufficiently recent Docker Compose v2, this is fine. Otherwise it could cause a noisy (but non-fatal) warning.

  2. getClusteringColumns() public API: Currently only called indirectly via iterate(). If called independently before initOrUpdateSchemaIfChanged() (e.g., from a new caller), the physical→logical name mapping would fall back to identity (the else branch at line ~1164), which is correct for most tables but could diverge for tables with column mapping enabled. Not a current bug since iterate() calls initOrUpdateSchemaIfChanged() first.

@subkanthi subkanthi closed this Oct 8, 2026
@subkanthi subkanthi reopened this Oct 8, 2026
@subkanthi

subkanthi commented Oct 8, 2026 •

Copy link
Copy Markdown
Collaborator Author

TESTING

docker run --rm -i \
  --user 0:0 \
  -v "$PWD/tmp/liquid_delta:/delta" \
  -v "$PWD/tmp/ivy:/tmp/ivy" \
  --entrypoint /opt/spark/bin/spark-sql \
  apache/spark:3.5.3 \
  --master 'local[2]' \
  --packages io.delta:delta-spark_2.12:3.3.0 \
  --conf spark.driver.extraJavaOptions='-Divy.cache.dir=/tmp/ivy -Divy.home=/tmp/ivy' \
  --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
  --conf spark.sql.warehouse.dir=/tmp/warehouse \
  --conf spark.sql.files.maxRecordsPerFile=500 \
  --conf spark.databricks.delta.optimize.maxFileSize=32768 <<'SQL'
CREATE TABLE delta.`/delta/table` (a INT, b STRING)
USING DELTA
CLUSTER BY (a)
TBLPROPERTIES (
  'delta.checkpointPolicy' = 'v2',
  'delta.checkpointInterval' = '1'
);

INSERT INTO delta.`/delta/table` SELECT id, CAST(id AS STRING) FROM range(0, 1000);
INSERT INTO delta.`/delta/table` SELECT id, CAST(id AS STRING) FROM range(1000, 2000);
INSERT INTO delta.`/delta/table` SELECT id, CAST(id AS STRING) FROM range(2000, 3000);
INSERT INTO delta.`/delta/table` SELECT id, CAST(id AS STRING) FROM range(3000, 4000);
INSERT INTO delta.`/delta/table` SELECT id, CAST(id AS STRING) FROM range(4000, 5000);

OPTIMIZE delta.`/delta/table` FULL;
DESCRIBE DETAIL delta.`/delta/table`;
SHOW TBLPROPERTIES delta.`/delta/table`;
SELECT count(*), min(a), max(a), sum(a) FROM delta.`/delta/table`;
SQL
mkdir -p /root/Documents/ClickHouse/programs/server/user_files
mv /root/Documents/ClickHouse/tmp/liquid_delta/table \
   /root/Documents/ClickHouse/programs/server/user_files/liquid_delta_table
Ubuntu-2404-noble-amd64-base :) SELECT count(), min(a), max(a), sum(a)
FROM deltaLakeLocal('/root/Documents/ClickHouse/programs/server/user_files/liquid_delta_table');

SELECT
    count(),
    min(a),
    max(a),
    sum(a)
FROM deltaLakeLocal('/root/Documents/ClickHouse/programs/server/user_files/liquid_delta_table')

Query id: 0a665eca-7f52-42b5-8e56-53bbdb4305bb

   ┌─count()─┬─min(a)─┬─max(a)─┬───sum(a)─┐
1. │    5000 │      0 │   4999 │ 12497500 │ -- 12.50 million
   └─────────┴────────┴────────┴──────────┘
[Ubuntu-2404-noble-amd64-base] 2026.10.08 19:06:07.807155 [ 3215907 ] {dca3e07d-9491-4aba-9eaa-9023cb26d1f3} <Trace> DeltaLakeMetadata: Clustering columns: a

@subkanthi subkanthi closed this Oct 9, 2026
@subkanthi subkanthi reopened this Oct 9, 2026

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants