diff --git a/.github/scripts/cmake-flags.sh b/.github/scripts/cmake-flags.sh index 0f78df25036..8f299caf74f 100644 --- a/.github/scripts/cmake-flags.sh +++ b/.github/scripts/cmake-flags.sh @@ -25,6 +25,8 @@ # * dep-graph — Dep-graph generator action; same as adapters minus # WAVE/CUDF (they gate experimental code, which the # selective-build planner short-circuits to full). +# * gpu-spark-hosted — Minimal cuDF/UCX production-object compile on a +# standard GitHub-hosted runner. # * ubuntu-debug — Ubuntu debug with system dependencies. # * ubuntu-bundled-deps — Ubuntu debug with velox's BUNDLED dependency # resolution; mirrors ubuntu-debug minus @@ -43,6 +45,14 @@ # EXTRA_CMAKE_FLAGS=("${CMAKE_FLAGS[@]}" -DVELOX_ENABLE_X=ON) case "${BUILD_PROFILE:?BUILD_PROFILE must be set before sourcing cmake-flags.sh}" in +gpu-spark-hosted) + CMAKE_FLAGS=( + -DCMAKE_EXPORT_COMPILE_COMMANDS=ON + -DVELOX_MONO_LIBRARY=OFF + -DVELOX_ENABLE_CUDF=ON + ) + ;; + adapters | dep-graph) # adapters and dep-graph are sister profiles: dep-graph regenerates # the planner's source-of-truth graph for what adapters builds, so diff --git a/.github/workflows/README.md b/.github/workflows/README.md index a3ed557051f..9b3d1ad3a3a 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -18,12 +18,13 @@ For current build times and performance trends, see the [CI performance metrics] | Workflow | File | Triggers | Purpose | |----------|------|----------|---------| +| GPU Spark Hosted Compile | `gpu-spark-hosted-compile.yml` | push to dev, PRs, manual | Compile cuDF and UCX production objects on a standard GitHub-hosted runner without requiring a GPU | | Linux Build using GCC | `linux-build.yml` + `linux-build-base.yml` | push to main, PRs | Main build & test (3 configs); selective by default, full on push / sticky-approval | | Detect Force-Full Trigger | `detect-force-full.yml` | PR approving review | Step 1 of the approval-path chain: trivial trigger for the workflow_run hand-off (needed because fork PR reviews get a read-only token) | | Rerun Linux Build on Force-Full Trigger | `rerun-on-force-full.yml` | workflow_run (after Detect Force-Full Trigger) | Step 2 of the approval-path chain: BASE-context, dedups, calls the `rerun-linux-build` composite action | | Force-Full Build on /full-build Comment | `force-full-on-comment.yml` | PR comment containing `/full-build` | Direct (no workflow_run hop): dedups, calls the `rerun-linux-build` composite action | | Selective Build Comment | `selective-build-comment.yml` | workflow_run (after Linux Build) | Post selective build plan as PR comment | -| macOS Build | `macos.yml` | push, PRs | Compilation check (debug + release) | +| macOS Build | `macos.yml` | push to dev, PRs | Core API compilation check (debug) | | Breeze Linux Build | `breeze.yml` | push to main, PRs | Tracing module with sanitizers | | Fuzzer Jobs | `scheduled.yml` | PRs, push to main, daily cron, manual | Randomized correctness testing | | Run Checks | `preliminary_checks.yml` | PRs | Formatting, linting, PR title | @@ -40,6 +41,22 @@ For current build times and performance trends, see the [CI performance metrics] ## Core Build & Test +### GPU Spark Hosted Compile (`gpu-spark-hosted-compile.yml`) + +Uses the existing adapters CI image and a minimal cuDF CMake profile on a +standard four-core GitHub-hosted runner. It configures the complete cuDF build +graph without unrelated cloud adapters, then asks Ninja to compile only +Velox-owned production objects under +`velox/experimental/cudf` and `velox/experimental/ucx-exchange`. Building +objects directly avoids compiling and linking all of libcudf while still +checking C++, CUDA, generated-header, and public API compatibility. +The current adapters image does not include UCX, so the workflow installs it +through `scripts/setup-centos-adapters.sh` and caches the resulting prefix. + +This lane does not link the resulting libraries or execute GPU tests. Those +remain responsibilities of the full adapters and `cudf-tests` jobs on larger +and GPU-backed runners. + ### Linux Build using GCC (`linux-build.yml` + `linux-build-base.yml`) The main CI workflow for Velox. Triggered on pushes to `main` and on pull request `opened`/`synchronize`/`reopened` events. The entry point `linux-build.yml` first calls the `selective-build-plan.yml` reusable workflow to decide the build mode (see [Selective Build Plan](#selective-build-plan-selective-build-planyml)), then delegates to `linux-build-base.yml` which can build and test up to four configurations in parallel on 32-core Ubuntu runners: @@ -65,7 +82,16 @@ Status jobs handle cancelled runs gracefully — when a job is cancelled (e.g., ### macOS Build (`macos.yml`) -Builds Velox on macOS 15 (ARM64/Apple Silicon) with both debug and release configurations. Triggered on pushes to any branch and on pull requests when relevant files change. Uses the Ninja build generator and ccache for faster builds. Tests are currently disabled on macOS — the workflow focuses on ensuring compilation compatibility with Apple's toolchain rather than full test coverage. Dependencies are installed via the `setup-macos.sh` script. +Builds the `velox_core_test` target in the debug configuration on macOS 15 +(ARM64/Apple Silicon) for pushes and pull requests targeting `dev`. This target +compiles and links the shared core API surface and its production dependencies +without spending hosted-runner time on every unrelated test binary. Pure cuDF +and UCX changes are excluded because those modules are not enabled by the macOS +CMake profile; changes to shared Velox code still receive the cross-platform +compile check. Uses the Ninja build generator and ccache for faster builds. +Tests are currently disabled on macOS — the workflow focuses on ensuring +compilation compatibility with Apple's toolchain rather than full test +coverage. Dependencies are installed via the `setup-macos.sh` script. ### Breeze Linux Build (`breeze.yml`) diff --git a/.github/workflows/gpu-spark-hosted-compile.yml b/.github/workflows/gpu-spark-hosted-compile.yml new file mode 100644 index 00000000000..b832e4d86cc --- /dev/null +++ b/.github/workflows/gpu-spark-hosted-compile.yml @@ -0,0 +1,186 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed 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. + +name: GPU Spark Hosted Compile + +on: + workflow_dispatch: {} + push: + branches: + - dev + paths: + - velox/experimental/cudf/** + - velox/experimental/ucx-exchange/** + - velox/functions/sparksql/** + - velox/common/** + - velox/core/** + - velox/exec/** + - velox/vector/** + - CMakeLists.txt + - CMake/** + - Makefile + - scripts/setup-centos-adapters.sh + - scripts/setup-centos9.sh + - scripts/setup-common.sh + - scripts/setup-helper-functions.sh + - scripts/setup-versions.sh + - .github/scripts/cmake-flags.sh + - .github/workflows/gpu-spark-hosted-compile.yml + pull_request: + branches: + - dev + paths: + - velox/experimental/cudf/** + - velox/experimental/ucx-exchange/** + - velox/functions/sparksql/** + - velox/common/** + - velox/core/** + - velox/exec/** + - velox/vector/** + - CMakeLists.txt + - CMake/** + - Makefile + - scripts/setup-centos-adapters.sh + - scripts/setup-centos9.sh + - scripts/setup-common.sh + - scripts/setup-helper-functions.sh + - scripts/setup-versions.sh + - .github/scripts/cmake-flags.sh + - .github/workflows/gpu-spark-hosted-compile.yml + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.repository }}-${{ github.head_ref || github.sha }} + cancel-in-progress: true + +jobs: + compile: + name: cuDF and UCX object compile + runs-on: ubuntu-22.04 + timeout-minutes: 90 + env: + BUILD_IMAGE: ghcr.io/facebookincubator/velox-dev@sha256:480989697666235f5d5c4ea7cdb26145b9722fc6db6e0cc2fe0bcfa369ffa199 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Free disk space + run: | + df -h / + sudo rm -rf \ + /usr/share/dotnet \ + /usr/local/lib/android \ + /opt/ghc \ + /usr/local/.ghcup \ + /opt/hostedtoolcache/CodeQL + df -h / + + - name: Restore UCX + id: ucx-cache + uses: actions/cache/restore@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4 + with: + path: hosted-deps/ucx + key: hosted-ucx-${{ runner.os }}-1.20.1-v1 + + - name: Restore compiler cache + uses: actions/cache@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4 + with: + path: ccache + key: hosted-gpu-spark-${{ runner.os }}-${{ hashFiles('CMake/resolve_dependency_modules/cudf.cmake', '.github/scripts/cmake-flags.sh') }}-${{ github.sha }} + restore-keys: | + hosted-gpu-spark-${{ runner.os }}- + + - name: Pull build image + run: | + docker pull "$BUILD_IMAGE" + + - name: Build UCX dependency + if: steps.ucx-cache.outputs.cache-hit != 'true' + run: | + mkdir -p hosted-deps/ucx + docker run --rm \ + --entrypoint /bin/bash \ + --env BUILD_THREADS=4 \ + --env DEPENDENCY_DIR=/tmp/velox-hosted-deps \ + --env INSTALL_PREFIX=/workspace/hosted-deps/ucx \ + --env UCX_VERSION=1.20.1 \ + --mount "type=bind,source=$GITHUB_WORKSPACE,target=/workspace" \ + --workdir /workspace \ + "$BUILD_IMAGE" -lc ' + set -euo pipefail + mkdir -p "$DEPENDENCY_DIR" + source scripts/setup-centos-adapters.sh + install_ucx + ' + + - name: Save UCX + if: steps.ucx-cache.outputs.cache-hit != 'true' + uses: actions/cache/save@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4 + with: + path: hosted-deps/ucx + key: hosted-ucx-${{ runner.os }}-1.20.1-v1 + + - name: Configure and compile GPU Spark objects + run: | + mkdir -p ccache + docker run --rm \ + --entrypoint /bin/bash \ + --env CCACHE_DIR=/workspace/ccache \ + --env CCACHE_MAXSIZE=2G \ + --env CCACHE_NOHASHDIR=true \ + --env CMAKE_PREFIX_PATH=/workspace/hosted-deps/ucx \ + --env CPATH=/workspace/hosted-deps/ucx/include \ + --env LD_LIBRARY_PATH=/workspace/hosted-deps/ucx/lib:/workspace/hosted-deps/ucx/lib64 \ + --env LIBRARY_PATH=/workspace/hosted-deps/ucx/lib:/workspace/hosted-deps/ucx/lib64 \ + --env PKG_CONFIG_PATH=/workspace/hosted-deps/ucx/lib/pkgconfig:/workspace/hosted-deps/ucx/lib64/pkgconfig \ + --env VELOX_DEPENDENCY_SOURCE=SYSTEM \ + --env cudf_SOURCE=BUNDLED \ + --mount "type=bind,source=$GITHUB_WORKSPACE,target=/workspace" \ + --workdir /workspace \ + "$BUILD_IMAGE" -lc ' + set -euo pipefail + git config --global --add safe.directory /workspace + source /opt/rh/gcc-toolset-14/enable + dnf -y install rdma-core-devel + + BUILD_PROFILE=gpu-spark-hosted source .github/scripts/cmake-flags.sh + export CUDA_ARCHITECTURES=70 + export CUDA_COMPILER=/usr/local/cuda-12.9/bin/nvcc + export CUDA_FLAGS="-ccbin /opt/rh/gcc-toolset-14/root/usr/bin" + + CC=gcc CXX=g++ make cmake \ + BUILD_DIR=hosted-gpu-spark \ + BUILD_TYPE=Release \ + VELOX_BUILD_TESTING=OFF \ + MAX_HIGH_MEM_JOBS=2 \ + MAX_LINK_JOBS=1 \ + EXTRA_CMAKE_FLAGS="${CMAKE_FLAGS[*]}" + + mapfile -t objects < <( + ninja -C _build/hosted-gpu-spark -t targets all | + sed -nE "s#^(velox/experimental/(cudf|ucx-exchange)/[^:]+\\.o):.*#\\1#p" + ) + if (( ${#objects[@]} == 0 )); then + echo "::error::CMake generated no cuDF or UCX object targets." + exit 1 + fi + + printf "Compiling %d Velox GPU object targets.\n" "${#objects[@]}" + printf "%s\n" "${objects[@]}" + ninja -C _build/hosted-gpu-spark -j 4 "${objects[@]}" + ccache -s + ' diff --git a/.github/workflows/macos.yml b/.github/workflows/macos.yml index 52020d70d65..53a04471044 100644 --- a/.github/workflows/macos.yml +++ b/.github/workflows/macos.yml @@ -17,9 +17,29 @@ name: macOS Build on: workflow_dispatch: {} push: + branches: + - dev paths: - velox/** - '!velox/docs/**' + - '!velox/experimental/cudf/**' + - '!velox/experimental/ucx-exchange/**' + - CMakeLists.txt + - CMake/** + - scripts/setup-macos.sh + - scripts/setup-common.sh + - scripts/setup-versions.sh + - scripts/setup-helper-functions.sh + - .github/workflows/macos.yml + + pull_request: + branches: + - dev + paths: + - velox/** + - '!velox/docs/**' + - '!velox/experimental/cudf/**' + - '!velox/experimental/ucx-exchange/**' - CMakeLists.txt - CMake/** - scripts/setup-macos.sh @@ -37,13 +57,13 @@ concurrency: jobs: macos-build: - if: ${{ github.repository == 'facebookincubator/velox' }} name: macos-15-${{ matrix.type }} + timeout-minutes: 120 strategy: fail-fast: false matrix: # macos-15 = arm64 Mac and cmake 4.0 with 7GB RAM - type: [debug, release] + type: [debug] runs-on: macos-15 env: CCACHE_DIR: ${{ github.workspace }}/ccache @@ -104,12 +124,13 @@ jobs: "${CMAKE_FLAGS[@]}" \ -DCMAKE_BUILD_TYPE=$BUILD_TYPE - - name: Build + - name: Build core API surface run: | - cmake --build _build/$BUILD_TYPE -j $NJOBS + cmake --build _build/$BUILD_TYPE --target velox_core_test -j $NJOBS ccache -s - uses: apache/infrastructure-actions/stash/save@3354c1565d4b0e335b78a76aedd82153a9e144d4 + if: ${{ !cancelled() }} with: path: ${{ env.CCACHE_DIR }} key: ccache-macos-1-macos-15-${{ matrix.type }} diff --git a/velox/core/tests/PlanNodeTest.cpp b/velox/core/tests/PlanNodeTest.cpp index e81ae9949cb..f61b6a5fafd 100644 --- a/velox/core/tests/PlanNodeTest.cpp +++ b/velox/core/tests/PlanNodeTest.cpp @@ -594,290 +594,242 @@ TEST_F(PlanNodeTest, aggregationNodeNoGroupsSpanBatches) { } } -TEST_F(PlanNodeTest, exchangeNodeTransportTypeSerialization) { - // Round-trip with kUcx transport. - { - auto node = std::make_shared( - "exchange1", rowType_, "presto", ExchangeNode::TransportType::kUcx); - auto serialized = node->serialize(); - ASSERT_EQ(serialized["transportType"].asString(), "UCX"); - auto deserialized = std::dynamic_pointer_cast( - ExchangeNode::create(serialized, nullptr)); - ASSERT_EQ(deserialized->transportType(), ExchangeNode::TransportType::kUcx); - } - - // Round-trip with kHttp transport. - { - auto node = std::make_shared( - "exchange2", rowType_, "presto", ExchangeNode::TransportType::kHttp); - auto serialized = node->serialize(); - ASSERT_EQ(serialized["transportType"].asString(), "HTTP"); - auto deserialized = std::dynamic_pointer_cast( - ExchangeNode::create(serialized, nullptr)); - ASSERT_EQ( - deserialized->transportType(), ExchangeNode::TransportType::kHttp); - } - - // Backward compatibility: missing transportType field defaults to kHttp. - { - auto node = std::make_shared( - "exchange3", rowType_, "presto", ExchangeNode::TransportType::kUcx); - auto serialized = node->serialize(); - serialized.erase("transportType"); - auto deserialized = std::dynamic_pointer_cast( - ExchangeNode::create(serialized, nullptr)); - ASSERT_EQ( - deserialized->transportType(), ExchangeNode::TransportType::kHttp); - } -} - -TEST_F(PlanNodeTest, partitionedOutputNodeTransportTypeSerialization) { +TEST_F(PlanNodeTest, partitionedOutputNodeTransportKindSerialization) { auto source = std::make_shared("source", rowData_); const auto serdeKind = "presto"; // Round-trip with kUcx transport. { auto node = PartitionedOutputNode::single( - "po1", - rowType_, - serdeKind, - source, - PartitionedOutputNode::TransportType::kUcx); + "po1", rowType_, serdeKind, std::string{TransportKind::kUcx}, source); auto serialized = node->serialize(); - ASSERT_EQ(serialized["transportType"].asString(), "UCX"); + ASSERT_EQ(serialized["transportKind"].asString(), "UCX"); auto deserialized = std::dynamic_pointer_cast( PartitionedOutputNode::create(serialized, nullptr)); - ASSERT_EQ( - deserialized->transportType(), - PartitionedOutputNode::TransportType::kUcx); + ASSERT_EQ(deserialized->transportKind(), std::string{TransportKind::kUcx}); } - // Round-trip with kHttp transport. + // Round-trip with in-memory transport. { auto node = PartitionedOutputNode::single( "po2", rowType_, serdeKind, - source, - PartitionedOutputNode::TransportType::kHttp); + std::string{TransportKind::kInMemory}, + source); auto serialized = node->serialize(); - ASSERT_EQ(serialized["transportType"].asString(), "HTTP"); + ASSERT_EQ(serialized["transportKind"].asString(), "in-memory"); auto deserialized = std::dynamic_pointer_cast( PartitionedOutputNode::create(serialized, nullptr)); ASSERT_EQ( - deserialized->transportType(), - PartitionedOutputNode::TransportType::kHttp); + deserialized->transportKind(), std::string{TransportKind::kInMemory}); } - // Backward compatibility: missing transportType field defaults to kHttp. + // Backward compatibility: missing transportKind defaults to in-memory. { auto node = PartitionedOutputNode::single( - "po3", - rowType_, - serdeKind, - source, - PartitionedOutputNode::TransportType::kUcx); + "po3", rowType_, serdeKind, std::string{TransportKind::kUcx}, source); auto serialized = node->serialize(); - serialized.erase("transportType"); + serialized.erase("transportKind"); auto deserialized = std::dynamic_pointer_cast( PartitionedOutputNode::create(serialized, nullptr)); ASSERT_EQ( - deserialized->transportType(), - PartitionedOutputNode::TransportType::kHttp); - } - // The FixedPointNode constructor and the state declarations validate their - // inputs up front, so a malformed plan fails at construction rather than at - // execution. - TEST_F(PlanNodeTest, fixedPointValidation) { - auto vecSchema = ROW("x", BIGINT()); - auto htSchema = ROW({"k", "v"}, BIGINT()); - - // A single body that reads the output entry -- the minimal valid body, - // reused as the valid baseline that each case mutates one field of. - auto body = - std::make_shared("b", "n", vecSchema, /*delta=*/true); - auto vectorN = [&] { - return std::make_shared( - "n", vecSchema, /*initialPlan=*/nullptr, /*append=*/true); - }; - - // maxIterations must be positive. - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{vectorN()}, - std::vector{body}, - ConvergenceConfig{ - .maxIterations = 0, .errorWhenMaxIterationReached = false}, - "n"), - "maxIterations must be positive"); - - // A hash table needs at least one key column, and every key must be in the - // schema -- checked when the declaration is built. - VELOX_ASSERT_USER_THROW( - std::make_shared( - "h", htSchema, std::vector{}), - "at least one key column"); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "h", htSchema, std::vector{"missing"}), - "key column is not in the schema"); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "h", htSchema, std::vector{"k", "k"}), - "key columns must be unique"); - - // StateHashJoin needs a non-null probe source and at least one probe key. - VELOX_ASSERT_USER_THROW( - std::make_shared( - "j", "h", std::vector{"k"}, htSchema, nullptr), - "non-null probe source"); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "j", "h", std::vector{}, htSchema, body), - "at least one probe key"); - - // State declaration names must be unique across all kinds. - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{ - vectorN(), - std::make_shared("n", vecSchema)}, - std::vector{body}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "n"), - "duplicate state declaration name"); - - // A StateSource must reference a declared vector entry. - auto typoBody = std::make_shared( - "b", "typo", vecSchema, /*delta=*/true); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{vectorN()}, - std::vector{typoBody}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "n"), - "StateSource references no declared vector state entry"); - - // errorWhenMaxIterationReached requires a convergence plan (a null plan - // never converges, so it would always fail). - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{vectorN()}, - std::vector{body}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = true}, - "n"), - "errorWhenMaxIterationReached requires a convergence plan"); - - // A convergence plan must emit exactly one BOOLEAN column. - auto nonBoolConvergence = - std::make_shared("c", "n", vecSchema, /*delta=*/false); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{vectorN()}, - std::vector{body}, - ConvergenceConfig{.plan = nonBoolConvergence, .maxIterations = 5}, - "n"), - "convergence plan output column must be BOOLEAN"); - - auto twoColSchema = ROW({"a", "b"}, BOOLEAN()); - auto twoColConvergence = std::make_shared( - "c", "flags", twoColSchema, /*delta=*/false); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{ - vectorN(), - std::make_shared( - "flags", twoColSchema)}, - std::vector{body}, - ConvergenceConfig{.plan = twoColConvergence, .maxIterations = 5}, - "n"), - "exactly one output column"); - - // A StateHashJoin output arity must equal probe columns plus the hash - // table's payload columns. - auto badArityJoin = std::make_shared( - "j", - "h", - std::vector{"x"}, - vecSchema, - std::make_shared("b", "n", vecSchema, /*delta=*/true)); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{ - vectorN(), - std::make_shared( - "h", htSchema, std::vector{"k"})}, - std::vector{badArityJoin}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "n"), - "output arity must equal probe columns plus hash table payload columns"); - - // A StateHashJoin's leading probe key column types must match the hash - // table's build key types (keys-first on both sides). - auto varcharProbe = ROW("k", VARCHAR()); - auto badKeyTypeJoin = std::make_shared( - "j", - "h", - std::vector{"k"}, - ROW({"k", "v"}, {VARCHAR(), BIGINT()}), - std::make_shared( - "s", "probe", varcharProbe, /*delta=*/true)); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{ - std::make_shared("probe", varcharProbe), - std::make_shared( - "h", htSchema, std::vector{"k"})}, - std::vector{badKeyTypeJoin}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "probe"), - "probe key column type at channel 0 must match the hash table build key"); - - // A null body plan is rejected with a clean error rather than crashing. - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{vectorN()}, - std::vector{nullptr}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "n"), - "plan 0 must not be null"); - - // Hash table key columns must be the leading schema columns (keys-first). - VELOX_ASSERT_USER_THROW( - std::make_shared( - "h", ROW({"v", "k"}, BIGINT()), std::vector{"k"}), - "leading schema columns in order"); - - // An initial plan must not read state -- it runs in Phase 1 before state - // exists. - auto stateReadingInitial = - std::make_shared("s", "n", vecSchema, /*delta=*/true); - VELOX_ASSERT_USER_THROW( - std::make_shared( - "fp", - std::vector{ - std::make_shared( - "n", vecSchema, stateReadingInitial, /*append=*/true)}, - std::vector{body}, - ConvergenceConfig{ - .maxIterations = 5, .errorWhenMaxIterationReached = false}, - "n"), - "initial plan must not read state"); + deserialized->transportKind(), std::string{TransportKind::kInMemory}); } +} + +// The FixedPointNode constructor and the state declarations validate their +// inputs up front, so a malformed plan fails at construction rather than at +// execution. +TEST_F(PlanNodeTest, fixedPointValidation) { + auto vecSchema = ROW("x", BIGINT()); + auto htSchema = ROW({"k", "v"}, BIGINT()); + + // A single body that reads the output entry -- the minimal valid body, + // reused as the valid baseline that each case mutates one field of. + auto body = + std::make_shared("b", "n", vecSchema, /*delta=*/true); + auto vectorN = [&] { + return std::make_shared( + "n", vecSchema, /*initialPlan=*/nullptr, /*append=*/true); + }; + + // maxIterations must be positive. + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{vectorN()}, + std::vector{body}, + ConvergenceConfig{ + .maxIterations = 0, .errorWhenMaxIterationReached = false}, + "n"), + "maxIterations must be positive"); + + // A hash table needs at least one key column, and every key must be in the + // schema -- checked when the declaration is built. + VELOX_ASSERT_USER_THROW( + std::make_shared( + "h", htSchema, std::vector{}), + "at least one key column"); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "h", htSchema, std::vector{"missing"}), + "key column is not in the schema"); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "h", htSchema, std::vector{"k", "k"}), + "key columns must be unique"); + + // StateHashJoin needs a non-null probe source and at least one probe key. + VELOX_ASSERT_USER_THROW( + std::make_shared( + "j", "h", std::vector{"k"}, htSchema, nullptr), + "non-null probe source"); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "j", "h", std::vector{}, htSchema, body), + "at least one probe key"); + + // State declaration names must be unique across all kinds. + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{ + vectorN(), + std::make_shared("n", vecSchema)}, + std::vector{body}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "n"), + "duplicate state declaration name"); + + // A StateSource must reference a declared vector entry. + auto typoBody = + std::make_shared("b", "typo", vecSchema, /*delta=*/true); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{vectorN()}, + std::vector{typoBody}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "n"), + "StateSource references no declared vector state entry"); + + // errorWhenMaxIterationReached requires a convergence plan (a null plan + // never converges, so it would always fail). + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{vectorN()}, + std::vector{body}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = true}, + "n"), + "errorWhenMaxIterationReached requires a convergence plan"); + + // A convergence plan must emit exactly one BOOLEAN column. + auto nonBoolConvergence = + std::make_shared("c", "n", vecSchema, /*delta=*/false); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{vectorN()}, + std::vector{body}, + ConvergenceConfig{.plan = nonBoolConvergence, .maxIterations = 5}, + "n"), + "convergence plan output column must be BOOLEAN"); + + auto twoColSchema = ROW({"a", "b"}, BOOLEAN()); + auto twoColConvergence = std::make_shared( + "c", "flags", twoColSchema, /*delta=*/false); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{ + vectorN(), + std::make_shared("flags", twoColSchema)}, + std::vector{body}, + ConvergenceConfig{.plan = twoColConvergence, .maxIterations = 5}, + "n"), + "exactly one output column"); + + // A StateHashJoin output arity must equal probe columns plus the hash + // table's payload columns. + auto badArityJoin = std::make_shared( + "j", + "h", + std::vector{"x"}, + vecSchema, + std::make_shared("b", "n", vecSchema, /*delta=*/true)); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{ + vectorN(), + std::make_shared( + "h", htSchema, std::vector{"k"})}, + std::vector{badArityJoin}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "n"), + "output arity must equal probe columns plus hash table payload columns"); + + // A StateHashJoin's leading probe key column types must match the hash + // table's build key types (keys-first on both sides). + auto varcharProbe = ROW("k", VARCHAR()); + auto badKeyTypeJoin = std::make_shared( + "j", + "h", + std::vector{"k"}, + ROW({"k", "v"}, {VARCHAR(), BIGINT()}), + std::make_shared( + "s", "probe", varcharProbe, /*delta=*/true)); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{ + std::make_shared("probe", varcharProbe), + std::make_shared( + "h", htSchema, std::vector{"k"})}, + std::vector{badKeyTypeJoin}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "probe"), + "probe key column type at channel 0 must match the hash table build key"); + + // A null body plan is rejected with a clean error rather than crashing. + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{vectorN()}, + std::vector{nullptr}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "n"), + "plan 0 must not be null"); + + // Hash table key columns must be the leading schema columns (keys-first). + VELOX_ASSERT_USER_THROW( + std::make_shared( + "h", ROW({"v", "k"}, BIGINT()), std::vector{"k"}), + "leading schema columns in order"); + + // An initial plan must not read state -- it runs in Phase 1 before state + // exists. + auto stateReadingInitial = + std::make_shared("s", "n", vecSchema, /*delta=*/true); + VELOX_ASSERT_USER_THROW( + std::make_shared( + "fp", + std::vector{ + std::make_shared( + "n", vecSchema, stateReadingInitial, /*append=*/true)}, + std::vector{body}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "n"), + "initial plan must not read state"); +} } // namespace diff --git a/velox/exec/CMakeLists.txt b/velox/exec/CMakeLists.txt index a8fa7eeaa08..289a12908c3 100644 --- a/velox/exec/CMakeLists.txt +++ b/velox/exec/CMakeLists.txt @@ -11,16 +11,6 @@ # 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. -if(VELOX_ENABLE_CUDF) - # Task owns output-queue initialization. Without this definition the cuDF - # UcxPartitionedOutput operator runs without a registered output queue. - set_source_files_properties( - Task.cpp - TARGET_DIRECTORY velox - PROPERTIES COMPILE_DEFINITIONS VELOX_ENABLE_CUDF - ) -endif() - velox_add_library( velox_exec AddressableNonNullValueList.cpp @@ -267,7 +257,7 @@ if(VELOX_ENABLE_CUDF) # Task.cpp includes the cuDF-backed queue manager under the same guard. # Propagate cuDF's CCCL include directories (including ) to # this target without exposing them to velox_exec consumers. - velox_link_libraries(velox_exec PRIVATE cudf::cudf) + set_property(TARGET velox_exec APPEND PROPERTY LINK_LIBRARIES cudf::cudf) endif() if(VELOX_ENABLE_GEO) diff --git a/velox/experimental/cudf/tests/TaskOutputManagerTest.cpp b/velox/experimental/cudf/tests/TaskOutputManagerTest.cpp index 45a332aff8f..1f16aa02181 100644 --- a/velox/experimental/cudf/tests/TaskOutputManagerTest.cpp +++ b/velox/experimental/cudf/tests/TaskOutputManagerTest.cpp @@ -58,7 +58,7 @@ class TaskOutputManagerTest : public HiveConnectorTestBase { std::shared_ptr startTask( const std::string& taskId, - core::PartitionedOutputNode::TransportType transport, + std::string_view transportKind, core::QueryConfig queryConfig = core::QueryConfig{{}}) { auto output = std::dynamic_pointer_cast( PlanBuilder() @@ -66,7 +66,7 @@ class TaskOutputManagerTest : public HiveConnectorTestBase { .partitionedOutputBroadcast() .planNode()); auto plan = core::PartitionedOutputNode::Builder(*output) - .transportType(transport) + .transportKind(std::string{transportKind}) .build(); auto task = Task::create( taskId, @@ -93,8 +93,7 @@ TEST_F(TaskOutputManagerTest, selectionAndCancellationCleanup) { auto cleanup = folly::makeGuard([&]() { queueManager->removeTask(ucxTaskId); }); - auto ucxTask = - startTask(ucxTaskId, core::PartitionedOutputNode::TransportType::kUcx); + auto ucxTask = startTask(ucxTaskId, core::TransportKind::kUcx); ASSERT_TRUE(queueManager->stats(ucxTaskId).has_value()); std::atomic_bool cancellationDelivered{false}; @@ -111,16 +110,15 @@ TEST_F(TaskOutputManagerTest, selectionAndCancellationCleanup) { queueManager->removeTask(ucxTaskId); cleanup.dismiss(); - const std::string httpTaskId = "http-output-manager-lifecycle"; - auto httpTask = - startTask(httpTaskId, core::PartitionedOutputNode::TransportType::kHttp); - EXPECT_FALSE(queueManager->stats(httpTaskId).has_value()); - httpTask->requestAbort().wait(); + const std::string inMemoryTaskId = "in-memory-output-manager-lifecycle"; + auto inMemoryTask = startTask(inMemoryTaskId, core::TransportKind::kInMemory); + EXPECT_FALSE(queueManager->stats(inMemoryTaskId).has_value()); + inMemoryTask->requestAbort().wait(); const std::string disabledTaskId = "disabled-ucx-output-manager-lifecycle"; auto disabledTask = startTask( disabledTaskId, - core::PartitionedOutputNode::TransportType::kUcx, + core::TransportKind::kUcx, core::QueryConfig( std::unordered_map{ {cudf_velox::CudfConfig::kCudfEnabled, "false"}})); diff --git a/velox/experimental/ucx-exchange/CMakeLists.txt b/velox/experimental/ucx-exchange/CMakeLists.txt index 351aa7692c2..fd699cf0294 100644 --- a/velox/experimental/ucx-exchange/CMakeLists.txt +++ b/velox/experimental/ucx-exchange/CMakeLists.txt @@ -49,13 +49,27 @@ target_compile_options(velox_ucx_exchange PRIVATE -Wno-error=deprecated-declarat target_include_directories(velox_ucx_exchange PRIVATE ${CMAKE_CURRENT_BINARY_DIR}) -find_package(ucxx CONFIG REQUIRED) -find_package(ucx REQUIRED) +if(NOT TARGET ucxx::ucxx) + find_package(ucxx CONFIG REQUIRED) +endif() +if(NOT TARGET ucx::ucp) + find_package(ucx REQUIRED) +endif() +if(NOT TARGET nvtx3::nvtx3-cpp) + find_package(nvtx3 REQUIRED) +endif() target_link_libraries( velox_ucx_exchange PUBLIC ucxx::ucxx ucx::ucp - PRIVATE cudf::cudf CUDA::cudart CUDA::cuda_driver Folly::folly velox_core ucx::ucs + PRIVATE + cudf::cudf + nvtx3::nvtx3-cpp + CUDA::cudart + CUDA::cuda_driver + Folly::folly + velox_core + ucx::ucs ) if(VELOX_BUILD_TESTING) diff --git a/velox/experimental/ucx-exchange/Communicator.cpp b/velox/experimental/ucx-exchange/Communicator.cpp index 1147cf405ee..e8dec01cbcb 100644 --- a/velox/experimental/ucx-exchange/Communicator.cpp +++ b/velox/experimental/ucx-exchange/Communicator.cpp @@ -20,11 +20,11 @@ #include #include #include +#include "UcxExchangeModules.h" #include "velox/common/base/Exceptions.h" #include "velox/experimental/cudf/CudfConfig.h" #include "velox/experimental/ucx-exchange/CommElement.h" #include "velox/experimental/ucx-exchange/EndpointRef.h" -#include "velox/experimental/ucx-exchange/UcxExchangeModules.h" #include "velox/experimental/ucx-exchange/UcxExchangeServer.h" #include "velox/experimental/ucx-exchange/UcxExchangeSource.h"