Merge branch 'antalya-26.6' into feature/antalya-26.6/CAS

Author: CarlosFelipeOR

.github/actions/create_workflow_report/create_workflow_report.py

@@ -691,6 +691,10 @@ def get_new_fails_this_pr(
 
     # Combine both types of fails and select only desired columns
     desired_columns = ["job_name", "test_name", "test_status", "results_link"]
+    if len(checks_fails) > 0 and "labels" in checks_fails.columns:
+        desired_columns.insert(desired_columns.index("results_link"), "labels")
+        if len(regression_fails) > 0:
+            regression_fails["labels"] = ""
     all_pr_fails = pd.concat([checks_fails, regression_fails], ignore_index=True)[
         desired_columns
     ]
@@ -976,6 +980,82 @@ def format_test_status(text: str) -> str:
     return f'<span style="font-weight: bold; color: {color}">{text}</span>'
 
 
+def _label_names_from_ext(ext: dict) -> list[str]:
+    names = []
+    for item in ext.get("labels") or []:
+        if isinstance(item, str):
+            name = item
+        elif isinstance(item, dict) and item.get("name"):
+            name = item["name"]
+        else:
+            continue
+        if name != "cidb":
+            names.append(name)
+    return names
+
+
+def fetch_workflow_result_json(
+    pr_number: int, branch: str, commit_sha: str
+) -> dict | None:
+    if pr_number == 0:
+        ref_param = f"REF={branch}"
+        workflow_name = "MasterCI"
+    else:
+        ref_param = f"PR={pr_number}"
+        workflow_name = "PR"
+
+    status_file = f"result_{workflow_name.lower()}.json"
+    s3_path = (
+        f"https://{S3_BUCKET}.s3.amazonaws.com/"
+        f"{ref_param.replace('=', 's/')}/{commit_sha}/{status_file}"
+    )
+    try:
+        response = requests.get(s3_path, timeout=30)
+        if response.status_code != 200:
+            return None
+        return response.json()
+    except Exception as e:
+        print(f"WARNING:Failed to fetch workflow result from {s3_path}: {e}")
+        return None
+
+
+def get_failure_labels_from_workflow(workflow_data: dict | None) -> dict:
+    if not workflow_data:
+        return {}
+    labels_map = {}
+    for job in workflow_data.get("results") or []:
+        job_name = job.get("name")
+        if not job_name:
+            continue
+        for leaf in job.get("results") or []:
+            test_name = leaf.get("name")
+            if not test_name:
+                continue
+            names = _label_names_from_ext(leaf.get("ext") or {})
+            if names:
+                labels_map[(job_name, test_name)] = ", ".join(names)
+    return labels_map
+
+
+def add_labels_to_checks_fails(
+    checks_fails: pd.DataFrame, workflow_data: dict | None
+) -> pd.DataFrame:
+    if checks_fails is None or len(checks_fails) == 0:
+        return checks_fails
+    labels_map = get_failure_labels_from_workflow(workflow_data)
+    df = checks_fails.copy()
+    df["labels"] = df.apply(
+        lambda row: labels_map.get((row["job_name"], row["test_name"]), ""),
+        axis=1,
+    )
+    cols = [c for c in df.columns if c != "labels"]
+    if "results_link" in cols:
+        cols.insert(cols.index("results_link"), "labels")
+    else:
+        cols.append("labels")
+    return df[cols]
+
+
 def format_results_as_html_table(results, *, branch_name: str = "") -> str:
     if not isinstance(results, pd.DataFrame):
         return results
@@ -1012,42 +1092,30 @@ def format_col_name(col_name: str) -> str:
         "PR Labels": lambda labels: format_pr_labels_with_verification(
             labels, branch_name=branch_name
         ),
+        "Labels": lambda labels: html.escape(str(labels), quote=True) if labels else "",
     }
 
-    html = results.to_html(
+    return results.to_html(
         index=False,
         formatters=formatters,
         escape=False,
         border=0,
         classes=["test-results-table"],
     )
-    return html
 
 
 def backfill_skipped_statuses(
-    job_statuses: pd.DataFrame, pr_number: int, branch: str, commit_sha: str
+    job_statuses: pd.DataFrame,
+    workflow_result: dict | None,
 ):
     """
     Fill in the job statuses for skipped jobs.
     """
-
-    if pr_number == 0:
-        ref_param = f"REF={branch}"
-        workflow_name = "MasterCI"
-    else:
-        ref_param = f"PR={pr_number}"
-        workflow_name = "PR"
-
-    status_file = f"result_{workflow_name.lower()}.json"
-    s3_path = f"https://{S3_BUCKET}.s3.amazonaws.com/{ref_param.replace('=', 's/')}/{commit_sha}/{status_file}"
-    response = requests.get(s3_path)
-
-    if response.status_code != 200:
+    if workflow_result is None:
         return job_statuses
 
-    status_data = response.json()
     skipped_jobs = []
-    for job in status_data["results"]:
+    for job in workflow_result["results"]:
         if job["status"] == "skipped" and len(job["links"]) > 0:
             skipped_jobs.append(
                 {
@@ -1192,10 +1260,15 @@ def create_workflow_report(
         settings={"use_numpy": True},
     )
 
+    workflow_result = fetch_workflow_result_json(pr_number, branch_name, commit_sha)
+
     results_dfs = {
         "prs_in_release": [],
         "job_statuses": get_commit_statuses(commit_sha),
-        "checks_fails": get_checks_fails(db_client, commit_sha, branch_name),
+        "checks_fails": add_labels_to_checks_fails(
+            get_checks_fails(db_client, commit_sha, branch_name),
+            workflow_result,
+        ),
         "checks_known_fails": [],
         "pr_new_fails": [],
         "checks_errors": get_checks_errors(db_client, commit_sha, branch_name),
@@ -1257,7 +1330,8 @@ def create_workflow_report(
             pr_info = {}
 
     results_dfs["job_statuses"] = backfill_skipped_statuses(
-        results_dfs["job_statuses"], pr_number, branch_name, commit_sha
+        results_dfs["job_statuses"],
+        workflow_result,
     )
 
     high_cve_count = 0

.github/actions/docker_setup/action.yml

@@ -20,7 +20,8 @@ runs:
         sudo cat <<EOT > /etc/docker/daemon.json
         {
           "ipv6": true,
-          "fixed-cidr-v6": "${{ env.ipv6_subnet }}"
+          "fixed-cidr-v6": "${{ env.ipv6_subnet }}",
+          "dns": ["1.1.1.1", "1.0.0.1"]
         }
         EOT
         sudo chown root:root /etc/docker/daemon.json

.github/actions/runner_setup/action.yml

@@ -3,6 +3,13 @@ description: Setup environment
 runs:
   using: "composite"
   steps:
+    - name: Configure docker proxy and dataset mirror host
+      shell: bash
+      run: |
+          proxy_host=dockerhub-proxy.dockerhub-proxy-zone
+          if ! timeout 5 getent ahostsv4 "$proxy_host" >/dev/null; then
+            echo "65.108.242.32 $proxy_host" | sudo tee -a /etc/hosts
+          fi
     - name: Setup zram
       shell: bash
       run: |

ci/jobs/stress_job.py

@@ -1,6 +1,7 @@
 import csv
 import logging
 import os
+import socket
 import sys
 from pathlib import Path
 from typing import List, Tuple
@@ -169,12 +170,24 @@ def get_run_command(
     else:
         run_script = "/repo/tests/docker_scripts/stress_runner.sh"
 
+    # Nested docker does not inherit the runner /etc/hosts.
+    # --network=host would expose minio and azurite. Skip --add-host when the
+    # name is unknown (CI Tests e2e / local) rather than failing the job.
+    proxy_host = "dockerhub-proxy.dockerhub-proxy-zone"
+    add_host = ""
+    try:
+        proxy_ip = socket.getaddrinfo(proxy_host, None)[0][4][0]
+        add_host = f"--add-host={proxy_host}:{proxy_ip} "
+    except OSError:
+        logging.info("Could not resolve %s; not passing --add-host", proxy_host)
+
     cmd = (
         "docker run --cap-add=SYS_PTRACE "
         # For dmesg and sysctl
         "--privileged "
         # azurite-rs (in-process Azure Blob Storage emulator) needs many fds under parallel load
         "--ulimit nofile=1048576:1048576 "
+        f"{add_host}"
         # a static link, don't use S3_URL or S3_DOWNLOAD
         "-e S3_URL='https://s3.amazonaws.com/clickhouse-datasets' "
         "--tmpfs /tmp/clickhouse:mode=1777 "

ci/praktika/runner.py

@@ -5,6 +5,7 @@
 import os
 import re
 import shlex
+import socket
 import sys
 import traceback
 from pathlib import Path
@@ -446,6 +447,18 @@ def _run(
             )
             from_root = "root" in docker_settings
             settings = [s for s in docker_settings if s.startswith("-")]
+            # NOTE (strtgbb): FT/integration docker is not --network=host, so
+            # the runner /etc/hosts mapping for the web-disk proxy is invisible
+            # inside the container. Resolve on the host and pass --add-host.
+            # Skip if the name is not configured on this runner (e.g. CI Tests).
+            _proxy_host = "dockerhub-proxy.dockerhub-proxy-zone"
+            try:
+                _proxy_ip = socket.getaddrinfo(_proxy_host, None)[0][4][0]
+                settings.append(f"--add-host={_proxy_host}:{_proxy_ip}")
+            except OSError as e:
+                print(
+                    f"NOTE: skipping --add-host for {_proxy_host}: {e}"
+                )
             if ":" in job.run_in_docker:
                 docker_name, docker_tag = job.run_in_docker.split(":")
                 print(

docs/en/antalya/part_export.md

@@ -53,10 +53,10 @@ Source and destination tables must support positional schema conversion. The fol
 - **Column types** may differ, as long as the source type is safely castable to the destination type. Set `export_merge_tree_part_allow_lossy_cast = 1` to also permit lossy casts.
 - **`Tuple` element names** may differ if either the source or destination declares the tuple without named elements: an unnamed `Tuple` (e.g. `Tuple(Int32, Int32)`) is matched against the destination by element position and type only, not by name. For example, exporting from `t Tuple(Int32, Int32)` to `t Tuple(x Int32, y Int32)` is allowed as long as element types match positionally.
 
-The following must match between source and destination:
+The following requirements apply to the source and destination:
 
 1. **Column count** - source and destination must have the same number of columns by default. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode.
-2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must be representable as an Iceberg partition spec and must match the destination partition fields and transforms.
+2. **`PARTITION BY` expressions** - the whole part must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the part's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](/docs/en/antalya/partition_export.md#source-partition-key-compatibility).
 3. **The position of every column backing the partition key** - it is not enough for the `PARTITION BY` expressions to be textually identical: every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. If such a column contains a named `Tuple`, its element names must also be declared in the same order (an unnamed `Tuple` on either side is exempt from this, per the allowance above). This comparison is recursive through nested tuples and through container types such as `Array` and `Map`.
 
   For example, `CREATE TABLE src (a Int32, b Int32) ... PARTITION BY a` and `CREATE TABLE dst (b Int32, a Int32) ... PARTITION BY a` both have the expression `PARTITION BY a`, but `a` is at position 0 in `src` and position 1 in `dst`. The export is rejected with a `BAD_ARGUMENTS` exception whose message includes `Cannot export to <destination>: partition key column 'a' is at position 0 in the source table, but the destination's column at that position is named 'b'`.

docs/en/antalya/partition_export.md

@@ -24,6 +24,13 @@ The manifest file produced by the commit contains a summary field `clickhouse.ex
 
 The Iceberg manifest files contain statistics about the data. Exporting a merge tree partition is a non ephemeral long running task, in which nodes can be turned off and turned on. This means the stats of individual files need to be persisted somewhere in order to produce the final manifest. This is implemented through sidecars. Each data file exported will contain a "sibling" sidecar file named `<data_file_name>_clickhouse_export_part_sidecar.avro`. ClickHouse does not clean up these files, and they can be safely deleted once the data is comitted.
 
+#### Source partition key compatibility
+
+The source partition must not be split in the destination. This is validated at schedule time through two mechanisms:
+
+1. Structural match: in case the source and destination are identical, the destination expression is a subset of the source expression or the destination expression can be entirely computed using only constants and the exact values guaranteed (pinned) by the source.
+2. Dynamic proof: the destination expression is monotonic over the source partition min/max range.
+
 ### On plain object storage exports:
 
 Each MergeTree part will become a separate file with the following name convention: `<table_directory>/<partitioning>/<data_part_name>_<merge_tree_part_checksum>.<format>`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `<table_directory>/commit_<partition_id>_<transaction_id>`.
@@ -45,10 +52,10 @@ TO TABLE [destination_database.]destination_table
 
 ## Requirements
 
-`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following must match:
+`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following requirements apply:
 
 1. **Column count** - source and destination must have the same number of columns by default. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode.
-2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must match the destination partition fields and transforms.
+2. **`PARTITION BY` expressions** - the whole source partition must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the partition's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](#source-partition-key-compatibility).
 3. **Partition key column positions and layouts** - every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. Named `Tuple` elements within such a column must also be declared in the same order, including tuples nested inside `Array` or `Map`. This applies even if both tables' `PARTITION BY` expressions are textually identical. See [`EXPORT PART` requirements](/docs/en/antalya/part_export.md#requirements) for a worked example and the corresponding exception message.
 
 ## Settings

src/Core/FormatFactorySettings.h

@@ -199,7 +199,13 @@ When reading Parquet files, parse JSON columns as ClickHouse JSON Column.
 Schedule prefetches more aggressively if memory usage is below than threshold. Potentially useful e.g. if there are many small bloom filters to read over network.
 )", 0) \
     DECLARE(UInt64, input_format_parquet_memory_high_watermark, 4ul << 30, R"(
-Approximate memory limit for Parquet reader v3. Limits how many row groups or columns can be read in parallel. When reading multiple files in one query, the limit is on total memory usage across those files.
+Approximate memory limit for the Parquet reader. Limits how many row groups or columns can be read in parallel. When reading multiple files in one query, the limit is on total memory usage across those files.
+)", 0) \
+    DECLARE(Double, input_format_parquet_prefetch_memory_fraction, 0.6, R"(
+Advanced tuning knob for the Parquet reader scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting.
+)", 0) \
+    DECLARE(Double, input_format_parquet_decode_thread_fraction, 0.375, R"(
+Advanced tuning knob for the Parquet reader scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1].
 )", 0) \
     DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"(
 Skip pages using min/max values from column index.

src/Core/SettingsChangesHistory.cpp

@@ -49,6 +49,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."},
             {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."},
             {"reserve_memory", 0, 0, "New setting to reserve memory for specific workload before starting a query."},
+            {"input_format_parquet_prefetch_memory_fraction", 0.6, 0.6, "New setting to tune the Parquet reader split of the column-data memory budget between compressed read-ahead and decode."},
+            {"input_format_parquet_decode_thread_fraction", 0.375, 0.375, "New setting to tune the Parquet reader share of the parsing thread pool given to column decoding."},
             {"output_format_image_width", 1024, 1024, "New setting controlling the width of the output image for image output formats such as PNG."},
             {"output_format_image_height", 1024, 1024, "New setting controlling the height of the output image for image output formats such as PNG."},
             {"output_format_image_terminal_mode", "", "", "New setting controlling whether image output formats such as PNG are rendered directly to the terminal using an inline image protocol."},