sunchao commented on code in PR #5974: URL: https://github.com/apache/datafusion-comet/pull/5974#discussion_r4039239743
########## dev/local-ci.sh: ########## @@ -0,0 +1,421 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# Run the Spark SQL or Iceberg CI workflow locally. Mirrors +# .github/workflows/spark_sql_test_reusable.yml and +# .github/workflows/iceberg_spark_test_reusable.yml. +# +# Versions, matrix rows and the shard count are read from ci.yml and dev/ci/ at +# run time, so a version bump needs no change here. Written for bash 3.2. + +set -euo pipefail + +REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +SHARDS_PY="$REPO/dev/ci/check-iceberg-shards.py" +# Rebuildable trees, so keep them out of $HOME. Override to survive a reboot. +SANDBOX="${COMET_LOCAL_CI_HOME:-/tmp/comet-local-ci}" +case "$(uname -s)" in + # bsdtar reads Spark's dot-prefixed .crc fixtures as AppleDouble metadata and + # exits nonzero after extracting them correctly. + Darwin) LIB=libcomet.dylib TAR_FLAGS=--no-mac-metadata ;; + *) LIB=libcomet.so TAR_FLAGS= ;; +esac + +# Yellow starting, green finished, red failed. +say() { printf '\n\033[1;33m[local-ci] %s\033[0m\n' "$*" >&2; } +ok() { printf '\n\033[1;32m[local-ci] %s\033[0m\n' "$*" >&2; } +fail() { printf '\n\033[1;31m[local-ci] %s\033[0m\n' "$*" >&2; } +die() { + fail "$*" + exit 1 +} +FAILED="" + +# Seconds as 1h 05m 12s. +hms() { + if [ "$1" -ge 3600 ]; then + printf '%dh %02dm %02ds' $(($1 / 3600)) $(($1 % 3600 / 60)) $(($1 % 60)) + elif [ "$1" -ge 60 ]; then + printf '%dm %02ds' $(($1 / 60)) $(($1 % 60)) + else + printf '%ds' "$1" + fi +} + +usage() { + cat >&2 <<'EOF' +Usage: dev/local-ci.sh <spark|iceberg> [version] [target...] + + dev/local-ci.sh spark every Spark SQL matrix row + dev/local-ci.sh spark sql_core-1 one row, or all/core/hive + dev/local-ci.sh iceberg every Iceberg target + dev/local-ci.sh iceberg shard-2 one shard, or extensions/runtime + +The version defaults to the one the merge queue gates on. Name an older one to +reproduce a nightly failure: dev/local-ci.sh spark 3.5 sql_core-1 + + --print-config show what is read from ci.yml and dev/ci, then exit + + SKIP_PREPARE=1 run the tests only. Skips the Comet install too, so do + not use it after changing Comet + COMET_LOCAL_CI_HOME where the sources live (default /tmp/comet-local-ci) +EOF + exit 2 +} + +# Read every value from dev/ci/local-ci-config.py, the single parser. It +# validates shapes and fails loudly, so nothing below has to re-guard a parse. +# shellcheck disable=SC2153 # VERSION/FULL/JAVA/ROWS/... all come from here +load_config() { + eval "$(cd "$REPO" && python3 dev/ci/local-ci-config.py --shell "$@")" || Review Comment: ### Correctness [P2] Check the parser status before evaluating its output Could the command substitution be assigned and checked separately before calling `eval`? `eval "$(...)"` returns the status of `eval`, so an empty result from a failed parser succeeds and bypasses `die`. On the explicitly supported macOS Bash 3.2, I ran the unmodified `/bin/bash dev/local-ci.sh spark nonexistent` and `spark 9.9`. Both report the parser error, hit `DEFAULTED: unbound variable`, print the green runtime message and exit 0 without running a test. The parser itself returns 1. Checking the assignment first makes both cases exit 1. This needs a negative-path check because callers use the exit status to decide whether local validation passed. ########## dev/ci/local-ci-config.py: ########## @@ -0,0 +1,257 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Everything dev/local-ci.sh needs to know about the CI configuration. +# +# The shell used to parse ci.yml and dev/ci/ with its own awk and sed, and +# check-ci-config.py parsed them again in Python so preflight could compare the +# two. That is worse than one parser twice over: it is two implementations to +# keep in step, and when they share a blind spot -- both stripped only single +# quotes, so `spark-full: "4.1.3"` came out with the quotes attached -- they +# agree with each other and the comparison says nothing. +# +# So there is one parser, here. The shell evals `--shell`, and preflight +# imports `config()` directly and checks the values rather than a second +# rendering of them. +# +# Usage: +# local-ci-config.py --shell spark 4.1 eval-able assignments for one job +# local-ci-config.py --print every value, for eyeballing + +import argparse +import json +import re +import shlex +import subprocess +import sys +from pathlib import Path + +CI_YML = Path(".github/workflows/ci.yml") +SPARK_YML = Path(".github/workflows/spark_sql_test_reusable.yml") +ICEBERG_YML = Path(".github/workflows/iceberg_spark_test_reusable.yml") +SHARDS_PY = Path("dev/ci/check-iceberg-shards.py") +MODULES_PY = Path("dev/ci/spark-sql-modules.py") +POLICY_PY = Path("dev/ci/compute-changes.py") + +# Values are versions or a JDK major. Anything else means a quoting or +# indentation change upstream has fooled the parse, and the shell would go on +# to build a URL or a Gradle task name out of it. +VERSION = re.compile(r"^\d+\.\d+$") +FULL_VERSION = re.compile(r"^\d+\.\d+\.\d+$") +MAJOR = re.compile(r"^\d+$") + + +class ConfigError(Exception): + pass + + +def _check(value, pattern, what): + if not pattern.match(value or ""): + raise ConfigError(f"{what} is {value!r}, which is not shaped like a version") + return value + + +def _job_inputs(): + """The `with:` inputs of every ci.yml job, keyed by job name.""" + jobs, job, in_with = {}, None, False + for line in CI_YML.read_text(encoding="utf-8").splitlines(): + header = re.match(r"^ ([A-Za-z0-9_-]+):\s*$", line) + if header: + job, in_with = header.group(1), False + elif job and re.match(r"^ with:\s*$", line): + in_with = True + elif in_with: + entry = re.match(r"^ ([a-z][a-z0-9-]*):\s*(\S.*?)\s*$", line) + if entry: + # YAML accepts either quote style. + jobs.setdefault(job, {})[entry.group(1)] = entry.group(2).strip("'\"") + elif re.match(r"^ [a-z]", line): + in_with = False + return jobs + + +def _load(name, path): + import importlib.util + + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _queue_version(prefix, policy): + """The version the merge queue gates on: the newest Comet fully supports. + + Only purely version-shaped keys, since `spark_4_1_hive` is queue-tier too. + """ + found = [ + match.group(1).replace("_", ".") + for key, events in policy.items() + for match in [re.fullmatch(rf"{prefix}_(\d+(?:_\d+)*)", key)] + if match and "queue" in events + ] + if not found: + raise ConfigError(f"no queue-tier {prefix} version in {POLICY_PY}") + return max(found, key=lambda v: [int(part) for part in v.split(".")]) + + +def _dedicated_gate(): + """The one Spark version whose suites the workflow process-isolates. + + Returns (version, suites). Every other version legitimately gets nothing, + so the guard is on the line being present and parseable. + """ + text = SPARK_YML.read_text(encoding="utf-8") + match = re.search(r"DEDICATED_JVM_SBT_TESTS:.*?spark-short == '([^']*)' && '([^']*)'", text) + if not match: + raise ConfigError(f"no parseable DEDICATED_JVM_SBT_TESTS in {SPARK_YML}") + return _check(match.group(1), VERSION, "dedicated-jvm gate"), match.group(2) + + +def _iceberg_scala(): + """ci.yml leaves `scala` unset, so the reusable workflow default applies.""" + text = ICEBERG_YML.read_text(encoding="utf-8") + match = re.search(r"^ scala:.*?^ default:\s*'?([0-9.]+)'?", text, re.S | re.M) + if not match: + raise ConfigError(f"no scala default in {ICEBERG_YML}") + return _check(match.group(1), VERSION, "iceberg scala") + + +def _iceberg_shards(): + shards = _load("iceberg_shards", SHARDS_PY).SHARD_COUNT + if not isinstance(shards, int) or shards < 1: + raise ConfigError(f"SHARD_COUNT in {SHARDS_PY} is {shards!r}") + return shards + + +def spark_rows(selectors=()): + """The Spark SQL matrix rows a selector names, in workflow order.""" + modules = _load("spark_sql_modules", MODULES_PY) + rows = modules.select("all") + names = [row["name"] for row in rows] + picked = [] + for want in selectors or ["all"]: + if want in ("all", "core", "hive"): + picked += [r for r in rows if want == "all" or r["group"] == want] Review Comment: ### Correctness [P2] Deduplicate selected rows before launching them concurrently Could selection preserve each row only once, or reject overlaps before preparation? `spark core sql_core-1` currently selects `sql_core-1` twice. `run_spark_rows` gives both copies the same tree and log path, and the second `clone_tree` removes that directory while the first process is still using it. With bounded test doubles I observed the first row's directory inode change during execution, and another run failed to open `build/sbt` after the deletion. Both invocations also write the same log. The disjoint `core hive` control launches seven distinct rows without replacement. Deduplicating by row name preserves the promised per-row isolation and avoids repeating the expensive suite. ########## dev/local-ci.sh: ########## @@ -0,0 +1,421 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# Run the Spark SQL or Iceberg CI workflow locally. Mirrors +# .github/workflows/spark_sql_test_reusable.yml and +# .github/workflows/iceberg_spark_test_reusable.yml. +# +# Versions, matrix rows and the shard count are read from ci.yml and dev/ci/ at +# run time, so a version bump needs no change here. Written for bash 3.2. + +set -euo pipefail + +REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +SHARDS_PY="$REPO/dev/ci/check-iceberg-shards.py" +# Rebuildable trees, so keep them out of $HOME. Override to survive a reboot. +SANDBOX="${COMET_LOCAL_CI_HOME:-/tmp/comet-local-ci}" +case "$(uname -s)" in + # bsdtar reads Spark's dot-prefixed .crc fixtures as AppleDouble metadata and + # exits nonzero after extracting them correctly. + Darwin) LIB=libcomet.dylib TAR_FLAGS=--no-mac-metadata ;; + *) LIB=libcomet.so TAR_FLAGS= ;; +esac + +# Yellow starting, green finished, red failed. +say() { printf '\n\033[1;33m[local-ci] %s\033[0m\n' "$*" >&2; } +ok() { printf '\n\033[1;32m[local-ci] %s\033[0m\n' "$*" >&2; } +fail() { printf '\n\033[1;31m[local-ci] %s\033[0m\n' "$*" >&2; } +die() { + fail "$*" + exit 1 +} +FAILED="" + +# Seconds as 1h 05m 12s. +hms() { + if [ "$1" -ge 3600 ]; then + printf '%dh %02dm %02ds' $(($1 / 3600)) $(($1 % 3600 / 60)) $(($1 % 60)) + elif [ "$1" -ge 60 ]; then + printf '%dm %02ds' $(($1 / 60)) $(($1 % 60)) + else + printf '%ds' "$1" + fi +} + +usage() { + cat >&2 <<'EOF' +Usage: dev/local-ci.sh <spark|iceberg> [version] [target...] + + dev/local-ci.sh spark every Spark SQL matrix row + dev/local-ci.sh spark sql_core-1 one row, or all/core/hive + dev/local-ci.sh iceberg every Iceberg target + dev/local-ci.sh iceberg shard-2 one shard, or extensions/runtime + +The version defaults to the one the merge queue gates on. Name an older one to +reproduce a nightly failure: dev/local-ci.sh spark 3.5 sql_core-1 + + --print-config show what is read from ci.yml and dev/ci, then exit + + SKIP_PREPARE=1 run the tests only. Skips the Comet install too, so do + not use it after changing Comet + COMET_LOCAL_CI_HOME where the sources live (default /tmp/comet-local-ci) +EOF + exit 2 +} + +# Read every value from dev/ci/local-ci-config.py, the single parser. It +# validates shapes and fails loudly, so nothing below has to re-guard a parse. +# shellcheck disable=SC2153 # VERSION/FULL/JAVA/ROWS/... all come from here +load_config() { + eval "$(cd "$REPO" && python3 dev/ci/local-ci-config.py --shell "$@")" || + die "could not read the CI configuration" + if [ -n "$DEFAULTED" ]; then + say "$1 $VERSION (the version the merge queue gates on)" + fi +} + +# --- sandbox --------------------------------------------------------------—-- + +setup_jdk() { + if [ -x /usr/libexec/java_home ]; then + JAVA_HOME="$(/usr/libexec/java_home -v "$1")" || die "JDK $1 not installed" + export JAVA_HOME + fi + [ -n "${JAVA_HOME:-}" ] || die "export JAVA_HOME pointing at a JDK $1" + say "JDK $1: $JAVA_HOME" +} + +# The ci profile is release without LTO. Stage it where -Prelease looks. +build_native() { + say "cargo build --profile ci" + (cd "$REPO/native" && cargo build --profile ci) + mkdir -p "$REPO/native/target/release" + cp "$REPO/native/target/ci/$LIB" "$REPO/native/target/release/$LIB" +} + +# Spark's build reaches for git only through build/spark-build-info, which has no +# `set -e`, so the tag archive works and skips the git objects a clone carries. +# Extract beside the target and move, so a failed download leaves nothing that +# the next run mistakes for a complete tree. +fetch_archive() { + if [ -d "$2" ]; then return 0; fi + say "downloading $1" + rm -rf "$2.part" + mkdir -p "$2.part" + # Checked, not left to `set -e`, which bash disables inside a function called + # from a conditional. + # shellcheck disable=SC2086 + if ! curl -fsSL "$1" | tar -xz $TAR_FLAGS -C "$2.part" --strip-components=1; then + rm -rf "$2.part" + die "could not download $1" + fi + mv "$2.part" "$2" +} + +# Iceberg does need a repository: its build takes the project version from the +# latest apache-iceberg-* tag via com.palantir.git-version. +clone_tag() { + if [ -d "$3/.git" ]; then return 0; fi + say "cloning $1 at $2" + git clone --depth 1 --branch "$2" "$1" "$3" +} + +# `git apply` works outside a repository, so this covers both trees. The applied +# diff is recorded so that editing dev/diffs/ and re-running reverts the old one +# first; otherwise the tree wedges and only deleting it helps. +apply_diff() { + marker="$1/.local-ci-applied.diff" + if [ -f "$marker" ] && cmp -s "$marker" "$2"; then return 0; fi + # A tree patched before the record existed: adopt it rather than fail. + if [ ! -f "$marker" ] && (cd "$1" && git apply --check --reverse "$2") 2>/dev/null; then + cp "$2" "$marker" + return 0 + fi + if [ -f "$marker" ]; then + say "reverting the previously applied diff" + (cd "$1" && git apply -R "$marker") || die "cannot revert $marker. Delete $1 and re-run." + rm -f "$marker" + fi + say "applying $(basename "$2")" + (cd "$1" && git apply "$2") || die "$(basename "$2") does not apply. Delete $1 and re-run." + cp "$2" "$marker" +} + +install_comet() { + say "mvnw install -Prelease -DskipTests $*" + (cd "$REPO" && ./mvnw -B install -Prelease -DskipTests "$@") +} + +# Ask Maven rather than assuming ~/.m2/repository: settings.xml and +# -Dmaven.repo.local can relocate it, and guessing wrong makes the purges below +# silent no-ops for the people who need them most. +MAVEN_REPO="" +maven_repo() { + if [ -z "$MAVEN_REPO" ]; then + MAVEN_REPO="$(cd "$REPO" && ./mvnw -q -N help:evaluate \ + -Dexpression=settings.localRepository -DforceStdout 2>/dev/null | tail -1)" + case "$MAVEN_REPO" in /*) ;; *) die "could not resolve the local Maven repository" ;; esac + fi + printf '%s\n' "$MAVEN_REPO" +} + +# Comet's install leaves POMs whose JARs it never fetched. Coursier then calls +# the artifact found-locally and refuses to fall back to Maven Central, so sbt +# dies on a JAR it can see a POM for. Both workflows drop the Parquet tree for +# that reason; the wider sweep is the same problem one level out. +purge_parquet() { + dir="$(maven_repo)/org/apache/parquet" + [ -d "$dir" ] || return 0 + say "removing $dir so sbt and gradle refetch it" + rm -rf "$dir" +} + +# setup-spark-builder greps for an explicit <packaging>jar|bundle</packaging> and +# so misses a POM declaring none, which Maven defaults to jar. org.antlr:antlr4 +# is one of those. A <packaging>pom</packaging> parent has no JAR by design. +purge_partial_poms() { + repo="$(maven_repo)" + [ -d "$repo" ] || return 0 + say "dropping POM-only entries across all of $repo" + find "$repo" -name '*.pom' | while read -r pom; do + [ -f "${pom%.pom}.jar" ] && continue + packaging="$(sed -n 's:.*<packaging>\(.*\)</packaging>.*:\1:p' "$pom" | head -1)" + case "${packaging:-jar}" in jar | bundle) ;; *) continue ;; esac + rm -f "$pom" "$pom.sha1" "${pom%.pom}.pom.lastUpdated" \ + "$(dirname "$pom")/_remote.repositories" + done +} + +# Purging invalidates what sbt already resolved: plugins such as sbt-antlr4 build +# their classpath from the cached update report, not the filesystem, so a +# refetched artifact stays invisible until the report is rebuilt. +drop_cached_resolution() { + [ -d "$1" ] || return 0 + say "clearing cached sbt resolution so it re-reads the Maven repository" + find "$1" -type d -name update -path '*/target/*' -prune -exec rm -rf {} + +} + +# Copy a prepared tree so a shard can have one to itself, the way each CI +# matrix row gets its own runner and its own extracted apache-spark/. On APFS +# and btrfs this is copy-on-write, so a 4 GB tree costs kilobytes until the +# shards start writing their own reports. +clone_tree() { + rm -rf "$2" + cp -Rc "$1" "$2" 2>/dev/null || + cp -R --reflink=auto "$1" "$2" 2>/dev/null || + cp -R "$1" "$2" +} + +# --- runners ----------------------------------------------------------------- + +row_label() { printf 'spark-sql-%s / spark-%s-jdk%s\n' "$1" "$FULL" "$JAVA"; } + +# spark_row <name> <args1> <args2> <heap> <metaspace> <tree> +# One row, in the tree it is given. A subshell so HEAP_SIZE cannot leak. +spark_row() { + ( + cd "$6" + printf -- '-J-Xms1g\n-J-Xmx4g\n-J-XX:MaxMetaspaceSize=1g\n' > .sbtopts + export LC_ALL=C.UTF-8 NOLINT_ON_COMPILE=true + # shellcheck disable=SC2030 + export ENABLE_COMET=true ENABLE_COMET_ONHEAP=true + export SBT_OPTS="-Xss4m -XX:+UseG1GC -XX:+UseStringDeduplication -XX:MaxMetaspaceSize=384m -XX:G1HeapRegionSize=2m -XX:InitiatingHeapOccupancyPercent=35 -XX:+ParallelRefProcEnabled -XX:+ExitOnOutOfMemoryError" + [ -n "$4" ] && export HEAP_SIZE="$4" + [ -n "$5" ] && export METASPACE_SIZE="$5" + # What the workflow exports. SERIAL_SBT_TESTS suppresses Spark's own test + # grouping and parallelExecution; the cap keeps one forked test JVM. Rows run + # beside each other in their own trees instead, the way CI does it. + export SERIAL_SBT_TESTS=1 + set -- -Dsbt.log.noformat=true -mem 1024 \ + "set Global / concurrentRestrictions := Seq(Tags.limit(Tags.ForkedTestGroup, 1))" \ + ${2:+"$2"} ${3:+"$3"} + build/sbt "$@" + ) +} + +# Every selected row at once, each in its own copy of the tree. That is exactly +# what CI does: seven matrix rows, seven runners, seven extracted apache-spark/ +# trees. Because each row gets a tree to itself, the per-row settings stay +# identical to CI's -- no shared sbt server, target/ or metastore tmpdir. +run_spark_rows() { + logs="$SANDBOX/logs-spark-$FULL" + mkdir -p "$logs" + running="" + while IFS=$'\037' read -r name args1 args2 heap metaspace; do + [ -n "$name" ] || continue + tree="$dest-$name" + clone_tree "$dest" "$tree" Review Comment: ### Correctness [P2] Reap launched rows when a later tree copy fails Could every exit path clean up or await the rows already launched? This loop starts a row before preparing the next tree. If a later copy fails, `set -e` exits before `await_rows`, and the EXIT trap only prints timing. In a focused probe using this control flow, an injected second-copy failure made the parent exit 73 while the first bounded child was still running. I then awaited that child separately. With real sbt rows, a disk-full copy failure can leave expensive JVMs and report writers running after the command has failed, and a retry removes their directories. Tracking the children in an exit cleanup path, or preparing all selected trees before launching any row, would close this failure path. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
