diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml deleted file mode 100644 index 4bbd3d0c66d..00000000000 --- a/.github/workflows/benchmark.yml +++ /dev/null @@ -1,214 +0,0 @@ -# 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: "Ubuntu Benchmark" - -on: - pull_request: - paths: - - 'velox/**' - - '!velox/docs/**' - - 'third_party/**' - - 'pyvelox/**' - - '.github/workflows/benchmark.yml' - push: - branches: [main] - -permissions: - contents: read - -concurrency: - group: ${{ github.workflow }}-${{ github.repository }}-${{ github.head_ref || github.sha }} - cancel-in-progress: true - -defaults: - run: - shell: bash -#TODO concurrency groups? -jobs: - benchmark: - if: github.repository == 'facebookincubator/velox' - runs-on: 8-core - env: - CCACHE_DIR: "${{ github.workspace }}/.ccache/" - CCACHE_BASEDIR: "${{ github.workspace }}" - BINARY_DIR: "${{ github.workspace }}/benchmarks/" - LINUX_DISTRO: "ubuntu" - RESULTS_ROOT: "${{ github.workspace }}/benchmark-results" - BASELINE_OUTPUT_PATH: "${{ github.workspace }}/benchmark-results/baseline/" - CONTENDER_OUTPUT_PATH: "${{ github.workspace }}/benchmark-results/contender/" - steps: - - - name: "Restore ccache" - uses: actions/cache/restore@v3 - id: restore-cache - with: - path: ".ccache" - key: ccache-benchmark-${{ github.sha }} - restore-keys: | - ccache-benchmark- - - - name: "Checkout Repo" - if: ${{ github.event_name == 'pull_request' }} - uses: actions/checkout@v3 - with: - path: 'velox' - repository: ${{ github.event.pull_request.head.repo.full_name }} - ref: ${{ github.head_ref }} - fetch-depth: 0 - submodules: 'recursive' - - - name: "Install dependencies" - if: ${{ github.event_name == 'pull_request' }} - run: source velox/scripts/setup-ubuntu.sh - - - name: "Checkout Merge Base" - if: ${{ github.event_name == 'pull_request' }} - working-directory: velox - run: | - # Choose merge base from upstream main to avoid issues with - # outdated fork branches - git fetch origin - git remote add upstream https://github.com/facebookincubator/velox - git fetch upstream - git status - merge_base=$(git merge-base 'upstream/${{ github.base_ref }}' 'origin/${{ github.head_ref }}') || \ - { echo "::error::Failed to find merge base"; exit 1; } - echo "Merge Base: $merge_base" - git checkout $merge_base - git submodule update --init --recursive - echo $(git log -n 1) - - - name: Build Baseline Benchmarks - if: ${{ github.event_name == 'pull_request' }} - working-directory: velox - run: | - n_cores=$(nproc) - make benchmarks-basic-build NUM_THREADS=$n_cores MAX_HIGH_MEM_JOBS=$n_cores MAX_LINK_JOBS=$n_cores - ccache -s - mkdir -p ${BINARY_DIR}/baseline/ - cp -r --verbose _build/release/velox/benchmarks/basic/* ${BINARY_DIR}/baseline/ - - - name: "Checkout Contender PR" - if: ${{ github.event_name == 'pull_request' }} - working-directory: velox - run: | - git checkout '${{ github.event.pull_request.head.sha }}' - - - name: "Checkout Contender" - if: ${{ github.event_name == 'push' }} - uses: actions/checkout@v3 - with: - path: 'velox' - ref: ${{ github.sha }} - submodules: 'recursive' - - - name: "Install dependencies" - run: source velox/scripts/setup-ubuntu.sh - - - name: Build Contender Benchmarks - working-directory: velox - run: | - n_cores=$(nproc) - make benchmarks-basic-build NUM_THREADS=$n_cores MAX_HIGH_MEM_JOBS=$n_cores MAX_LINK_JOBS=$n_cores - ccache -s - mkdir -p ${BINARY_DIR}/contender/ - cp -r --verbose _build/release/velox/benchmarks/basic/* ${BINARY_DIR}/contender/ - - - name: "Save ccache" - uses: actions/cache/save@v3 - id: cache - with: - path: ".ccache" - key: ccache-benchmark-${{ github.sha }} - - - name: "Install benchmark dependencies" - run: | - python3 -m pip install -r velox/scripts/benchmark-requirements.txt - - - name: "Run Benchmarks - Baseline" - if: ${{ github.event_name == 'pull_request' }} - working-directory: 'velox' - run: | - make benchmarks-basic-run \ - EXTRA_BENCHMARK_FLAGS="--binary_path ${BINARY_DIR}/baseline/ --output_path ${BASELINE_OUTPUT_PATH}" - - - name: "Run Benchmarks - Contender" - working-directory: 'velox' - run: | - make benchmarks-basic-run \ - EXTRA_BENCHMARK_FLAGS="--binary_path ${BINARY_DIR}/contender/ --output_path ${CONTENDER_OUTPUT_PATH}" - - - name: "Compare initial results" - id: compare - if: ${{ github.event_name == 'pull_request' }} - run: | - ./velox/scripts/benchmark-runner.py compare \ - --baseline_path ${BASELINE_OUTPUT_PATH} \ - --contender_path ${CONTENDER_OUTPUT_PATH} \ - --rerun_json_output "benchmark-results/rerun_json_output_0.json" \ - --do_not_fail - - - name: "Rerun Benchmarks" - if: ${{ github.event_name == 'pull_request'}} - working-directory: 'velox' - run: | - for i in 1 2 3 4 5; do - CURRENT_JSON_RERUN="${RESULTS_ROOT}/rerun_json_output_$((${i} - 1)).json" - NEXT_JSON_RERUN="${RESULTS_ROOT}/rerun_json_output_${i}.json" - - if [ ! -s "${CURRENT_JSON_RERUN}" ]; then - echo "::notice::Rerun iteration ${i} found empty file. Finalizing." - break - fi - - echo "::group::Rerun iteration: ${i}" - make benchmarks-basic-run \ - EXTRA_BENCHMARK_FLAGS="--binary_path ${BINARY_DIR}/baseline/ --output_path ${BASELINE_OUTPUT_PATH}/rerun-${i}/ --rerun_json_input ${CURRENT_JSON_RERUN}" - - make benchmarks-basic-run \ - EXTRA_BENCHMARK_FLAGS="--binary_path ${BINARY_DIR}/contender/ --output_path ${CONTENDER_OUTPUT_PATH}/rerun-${i}/ --rerun_json_input ${CURRENT_JSON_RERUN}" - - ./scripts/benchmark-runner.py compare \ - --baseline_path ${BASELINE_OUTPUT_PATH}/rerun-${i}/ \ - --contender_path ${CONTENDER_OUTPUT_PATH}/rerun-${i}/ \ - --rerun_json_output ${NEXT_JSON_RERUN} \ - --do_not_fail - echo "::endgroup::" - done - - - echo "::group::Compare final results" - ./scripts/benchmark-runner.py compare \ - --baseline_path ${BASELINE_OUTPUT_PATH} \ - --contender_path ${CONTENDER_OUTPUT_PATH} \ - --recursive \ - --do_not_fail - echo "::endgroup::" - - - name: "Save PR number" - run: echo "${{ github.event.pull_request.number }}" > pr_number.txt - - - name: "Upload PR number" - uses: actions/upload-artifact@v3 - with: - path: "pr_number.txt" - name: "pr_number" - - - name: "Upload result artifact" - uses: actions/upload-artifact@v3 - with: - path: "benchmark-results" - name: "benchmark-results" - diff --git a/.github/workflows/build_pyvelox.yml b/.github/workflows/build_pyvelox.yml deleted file mode 100644 index bb01d8e804c..00000000000 --- a/.github/workflows/build_pyvelox.yml +++ /dev/null @@ -1,186 +0,0 @@ -# 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: Build Pyvelox Wheels - -on: - workflow_dispatch: - inputs: - version: - description: 'pyvelox version' - required: false - ref: - description: 'git ref to build' - required: false - publish: - description: 'publish to PyPI' - required: false - type: boolean - default: false - # schedule: - # - cron: '15 0 * * *' - pull_request: - paths: - - 'velox/**' - - '!velox/docs/**' - - 'third_party/**' - - 'pyvelox/**' - - '.github/workflows/build_pyvelox.yml' - -permissions: - contents: read - -concurrency: - group: ${{ github.workflow }}-${{ github.repository }}-${{ github.head_ref || github.sha }} - cancel-in-progress: true - -jobs: - build_wheels: - name: Build wheels on ${{ matrix.os }} - runs-on: ${{ matrix.os }} - strategy: - fail-fast: false - matrix: - os: [ubuntu-22.04, macos-11] - steps: - - uses: actions/checkout@v3 - with: - ref: ${{ inputs.ref || github.ref }} - fetch-depth: 0 - submodules: recursive - - - uses: actions/setup-python@v4 - with: - python-version: '3.10' - - - name: "Determine Version" - if: ${{ !inputs.version && github.event_name != 'pull_request' }} - id: version - run: | - # count number of commits since last tag matching a regex - # and use that to determine the version number - # e.g. if the last tag is 0.0.1, and there have been 5 commits since then - # the version will be 0.0.1a5 - git fetch --tags - INITIAL_COMMIT=5d4db2569b7c249644bf36a543ba1bd8f12bf77c - # Can't use PCRE for portability - BASE_VERSION=$(grep -oE '[0-9]+\.[0-9]+\.[0-9]+' version.txt) - - LAST_TAG=$(git describe --tags --match "pyvelox-v[0-9]*" --abbrev=0 || echo $INITIAL_COMMIT) - COMMITS_SINCE_TAG=$(git rev-list --count ${LAST_TAG}..HEAD) - - if [ "$LAST_TAG" = "$INITIAL_COMMIT" ]; then - VERSION=$BASE_VERSION - else - VERSION=$(echo $LAST_TAG | sed '/pyvelox-v//') - fi - # NEXT_VERSION=$(echo $VERSION | awk -F. -v OFS=. '{$NF++ ; print}') - echo "build_version=${VERSION}a${COMMITS_SINCE_TAG}" >> $GITHUB_OUTPUT - - - run: mkdir -p .ccache - - name: "Restore ccache" - uses: actions/cache/restore@v3 - id: restore-cache - with: - path: ".ccache" - key: ccache-wheels-${{ matrix.os }}-${{ github.sha }} - restore-keys: | - ccache-wheels-${{ matrix.os }}- - - - name: Install macOS dependencies - if: matrix.os == 'macos-11' - run: | - echo "OPENSSL_ROOT_DIR=/usr/local/opt/openssl@1.1/" >> $GITHUB_ENV - bash scripts/setup-macos.sh && - bash scripts/setup-macos.sh install_folly - - - name: "Create sdist" - if: matrix.os == 'ubuntu-22.04' - env: - BUILD_VERSION: "${{ inputs.version || steps.version.outputs.build_version }}" - run: | - python setup.py sdist --dist-dir wheelhouse - - - name: Build wheels - uses: pypa/cibuildwheel@v2.12.1 - env: - # required for preadv/pwritev - MACOSX_DEPLOYMENT_TARGET: "11.0" - CIBW_ARCHS: "x86_64" - # On PRs only build for Python 3.7 - CIBW_BUILD: ${{ github.event_name == 'pull_request' && 'cp37-*' || 'cp3*' }} - CIBW_SKIP: "*musllinux* cp36-*" - CIBW_MANYLINUX_X86_64_IMAGE: "ghcr.io/facebookincubator/velox-dev:torcharrow-avx" - CIBW_BEFORE_ALL_LINUX: > - mkdir -p /output && - cp -R /host${{ github.workspace }}/.ccache /output/.ccache && - ccache -s - CIBW_ENVIRONMENT_PASS_LINUX: CCACHE_DIR BUILD_VERSION - CIBW_TEST_EXTRAS: "tests" - CIBW_TEST_COMMAND: "cd {project}/pyvelox && python -m unittest -v" - CIBW_TEST_SKIP: "*macos*" - CCACHE_DIR: "${{ matrix.os != 'macos-11' && '/output' || github.workspace }}/.ccache" - BUILD_VERSION: "${{ inputs.version || steps.version.outputs.build_version }}" - with: - output-dir: wheelhouse - - - name: "Move .ccache to workspace" - if: matrix.os != 'macos-11' - run: | - mkdir -p .ccache - cp -R ./wheelhouse/.ccache/* .ccache - - - name: "Save ccache" - uses: actions/cache/save@v3 - id: cache - with: - path: ".ccache" - key: ccache-wheels-${{ matrix.os }}-${{ github.sha }} - - - name: "Rename wheel compatibility tag" - if: matrix.os == 'macos-11' - run: | - brew install rename - cd wheelhouse - rename 's/11_0/10_15/g' *.whl - - - uses: actions/upload-artifact@v3 - with: - name: wheels - path: | - ./wheelhouse/*.whl - ./wheelhouse/*.tar.gz - - publish_wheels: - name: Publish Wheels to PyPI - if: ${{ github.event_name == 'schedule' || inputs.publish }} - needs: build_wheels - runs-on: ubuntu-22.04 - steps: - - uses: actions/download-artifact@v3 - with: - name: wheels - path: ./wheelhouse - - - run: ls wheelhouse - - - uses: actions/setup-python@v3 - with: - python-version: "3.10" - - - name: Publish a Python distribution to PyPI - uses: pypa/gh-action-pypi-publish@v1.6.4 - with: - password: ${{ secrets.PYPI_API_TOKEN }} - packages_dir: wheelhouse diff --git a/.github/workflows/conbench_upload.yml b/.github/workflows/conbench_upload.yml deleted file mode 100644 index b59a30c142c..00000000000 --- a/.github/workflows/conbench_upload.yml +++ /dev/null @@ -1,168 +0,0 @@ -# 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: Upload Benchmark Results -on: - workflow_dispatch: - inputs: - run_id: - description: 'workflow run id to use the artifacts from' - required: true - workflow_run: - workflows: ["Ubuntu Benchmark"] - types: - - completed - -permissions: - contents: read - actions: read - statuses: write - -jobs: - upload: - runs-on: ubuntu-latest - if: ${{ (github.event.workflow_run.conclusion == 'success' || - github.event_name == 'workflow_dispatch') && - github.repository == 'facebookincubator/velox' }} - steps: - - - name: 'Download artifacts' - id: 'download' - uses: actions/github-script@v6 - with: - script: | - const run_id = "${{ github.event.workflow_run.id || inputs.run_id }}"; - let benchmark_run = await github.rest.actions.getWorkflowRun({ - owner: context.repo.owner, - repo: context.repo.repo, - run_id: run_id, - }); - - let artifacts = await github.rest.actions.listWorkflowRunArtifacts({ - owner: context.repo.owner, - repo: context.repo.repo, - run_id: run_id, - }); - - let result_artifact = artifacts.data.artifacts.filter((artifact) => { - return artifact.name == "benchmark-results" - })[0]; - - let pr_artifact = artifacts.data.artifacts.filter((artifact) => { - return artifact.name == "pr_number" - })[0]; - - let result_download = await github.rest.actions.downloadArtifact({ - owner: context.repo.owner, - repo: context.repo.repo, - artifact_id: result_artifact.id, - archive_format: 'zip', - }); - - let pr_download = await github.rest.actions.downloadArtifact({ - owner: context.repo.owner, - repo: context.repo.repo, - artifact_id: pr_artifact.id, - archive_format: 'zip', - }); - - var fs = require('fs'); - fs.writeFileSync('${{github.workspace}}/benchmark-results.zip', Buffer.from(result_download.data)); - fs.writeFileSync('${{github.workspace}}/pr_number.zip', Buffer.from(pr_download.data)); - - core.setOutput('contender_sha', benchmark_run.data.head_sha); - - if (benchmark_run.data.event == 'push') { - core.setOutput('merge_commit_message', benchmark_run.data.head_commit.message); - } else { - core.setOutput('merge_commit_message', ''); - } - - - name: Extract artifact - id: extract - run: | - unzip benchmark-results.zip -d benchmark-results - unzip pr_number.zip - echo "pr_number=$(cat pr_number.txt)" >> $GITHUB_OUTPUT - - uses: actions/checkout@v3 - with: - path: velox - - uses: actions/setup-python@v4 - with: - python-version: '3.8' - cache: 'pip' - cache-dependency-path: "velox/scripts/*" - - - name: "Install dependencies" - run: python -m pip install -r velox/scripts/benchmark-requirements.txt - - - name: "Upload results" - env: - CONBENCH_URL: "https://velox-conbench.voltrondata.run/" - CONBENCH_MACHINE_INFO_NAME: "GitHub-runner-8-core" - CONBENCH_EMAIL: "${{ secrets.CONBENCH_EMAIL }}" - CONBENCH_PASSWORD: "${{ secrets.CONBENCH_PASSWORD }}" - CONBENCH_PROJECT_REPOSITORY: "${{ github.repository }}" - CONBENCH_PROJECT_COMMIT: "${{ steps.download.outputs.contender_sha }}" - run: | - if [ "${{ steps.extract.outputs.pr_number }}" -gt 0]; then - export CONBENCH_PROJECT_PR_NUMBER="${{ steps.extract.outputs.pr_number }}" - fi - - ./velox/scripts/benchmark-runner.py upload \ - --run_id "GHA-${{ github.run_id }}-${{ github.run_attempt }}" \ - --pr_number "${{ steps.extract.outputs.pr_number }}" \ - --sha "${{ steps.download.outputs.contender_sha }}" \ - --output_dir "${{ github.workspace }}/benchmark-results/contender/" - - - name: "Check the status of the upload" - # Status functions like failure() only work in `if:` - if: failure() - id: status - run: echo "failed=true" >> $GITHUB_OUTPUT - - - name: "Create a GitHub Status on the contender commit (whether the upload was successful)" - uses: actions/github-script@v6 - if: always() - with: - script: | - let url = 'https://github.com/${{github.repository}}/actions/runs/${{ github.run_id }}' - let state = 'success' - let description = 'Result upload succeeded!' - - if(${{ steps.status.outputs.failed || false }}) { - state = 'failure' - description = 'Result upload failed!' - } - - github.rest.repos.createCommitStatus({ - owner: context.repo.owner, - repo: context.repo.repo, - sha: '${{ steps.download.outputs.contender_sha }}', - state: state, - target_url: url, - description: description, - context: 'Benchmark Result Upload' - }) - - - name: Create a GitHub Check benchmark report on the contender comment, and if merge-commit, a comment on the merged PR - env: - CONBENCH_URL: "https://velox-conbench.voltrondata.run/" - GITHUB_APP_ID: "${{ secrets.GH_APP_ID }}" - GITHUB_APP_PRIVATE_KEY: "${{ secrets.GH_APP_PRIVATE_KEY }}" - run: | - ./velox/scripts/benchmark-alert.py \ - --contender-sha "${{ steps.download.outputs.contender_sha }}" \ - --merge-commit-message "${{ steps.download.outputs.merge_commit_message }}" \ - --z-score-threshold 50 diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml deleted file mode 100644 index d3d309c664b..00000000000 --- a/.github/workflows/docker.yml +++ /dev/null @@ -1,88 +0,0 @@ -# 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: Build & Push Docker Images - -on: - pull_request: - paths: - - scripts/*.dockfile - - scripts/*.dockerfile - - scripts/setup-*.sh - - .github/workflows/docker.yml - push: - branches: [main] - paths: - - scripts/*.dockfile - - scripts/*.dockerfile - - scripts/setup-*.sh - - .github/workflows/docker.yml - -concurrency: - group: ${{ github.workflow }}-${{ github.repository }}-${{ github.head_ref || github.sha }} - cancel-in-progress: true - -permissions: - contents: read - packages: write - -jobs: - linux: - name: "Build and Push ${{ matrix.name }}" - runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - include: - - name: Check - file: "scripts/check-container.dockfile" - args: "cpu_target=avx" - tags: "ghcr.io/facebookincubator/velox-dev:check-avx" - - name: CircleCI - file: "scripts/circleci-container.dockfile" - args: "cpu_target=avx" - tags: "ghcr.io/facebookincubator/velox-dev:circleci-avx" - - name: Torcharrow - file: "scripts/velox-torcharrow-container.dockfile" - args: "cpu_target=avx" - tags: "ghcr.io/facebookincubator/velox-dev:torcharrow-avx" - - name: Dev - file: "scripts/ubuntu-22.04-cpp.dockerfile" - args: "" - tags: "ghcr.io/facebookincubator/velox-dev:amd64-ubuntu-22.04-avx" - - name: Presto Java - file: "scripts/prestojava-container.dockerfile" - args: "PRESTO_VERSION=0.284" - tags: "ghcr.io/facebookincubator/velox-dev:presto-java" - - steps: - - name: Login to GitHub Container Registry - uses: docker/login-action@v2 - with: - registry: ghcr.io - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - - name: Set up QEMU - uses: docker/setup-qemu-action@v2 - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 - - - name: Build and Push - uses: docker/build-push-action@v3 - with: - file: "${{ matrix.file }}" - build-args: "${{ matrix.args }}" - push: ${{ github.repository == 'facebookincubator/velox' && github.event_name != 'pull_request'}} - tags: "${{ matrix.tags }}" diff --git a/.github/workflows/unittest.yml b/.github/workflows/unittest.yml new file mode 100644 index 00000000000..6d12c70683f --- /dev/null +++ b/.github/workflows/unittest.yml @@ -0,0 +1,69 @@ +name: Velox Unit Tests Suite + +on: + pull_request + +concurrency: + group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }} + cancel-in-progress: true + +jobs: + + velox-test: + runs-on: self-hosted + container: ubuntu:22.04 + steps: + - uses: actions/checkout@v2 + - run: apt-get update && apt-get install ca-certificates -y && update-ca-certificates + - run: sed -i 's/http\:\/\/archive.ubuntu.com/https\:\/\/mirrors.ustc.edu.cn/g' /etc/apt/sources.list + - run: apt-get update + - run: apt-get install -y cmake ccache build-essential ninja-build sudo + - run: apt-get install -y libboost-all-dev libcurl4-openssl-dev + - run: apt-get install -y libssl-dev flex libfl-dev git openjdk-8-jdk axel *thrift* libkrb5-dev libgsasl7-dev libuuid1 uuid-dev + - run: apt-get install -y libz-dev + - run: | + axel https://github.com/protocolbuffers/protobuf/releases/download/v21.4//protobuf-all-21.4.tar.gz + tar xf protobuf-all-21.4.tar.gz + cd protobuf-21.4/cmake + CFLAGS=-fPIC CXXFLAGS=-fPIC cmake .. && make -j && make install + - run: | + axel https://dl.min.io/server/minio/release/linux-amd64/archive/minio_20220526054841.0.0_amd64.deb + dpkg -i minio_20220526054841.0.0_amd64.deb + rm minio_20220526054841.0.0_amd64.deb + - run: | + axel https://dlcdn.apache.org/hadoop/common/hadoop-2.10.1/hadoop-2.10.1.tar.gz + tar xf hadoop-2.10.1.tar.gz -C /usr/local/ + - name: Compile C++ unit tests + run: | + git submodule sync --recursive && git submodule update --init --recursive + sed -i 's/sudo apt/apt/g' ./scripts/setup-ubuntu.sh + sed -i 's/sudo --preserve-env apt/apt/g' ./scripts/setup-ubuntu.sh + TZ=Asia/Shanghai ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone && ./scripts/setup-ubuntu.sh + mkdir -p ~/adapter-deps/install + DEPENDENCY_DIR=~/adapter-deps PROMPT_ALWAYS_RESPOND=n ./scripts/setup-adapters.sh gcs aws hdfs + #make debug EXTRA_CMAKE_FLAGS="-DVELOX_ENABLE_PARQUET=ON -DVELOX_BUILD_TESTING=ON -DVELOX_BUILD_TEST_UTILS=ON -DVELOX_ENABLE_HDFS=ON -DVELOX_ENABLE_S3=ON -DVELOX_ENABLE_GCS=ON" AWSSDK_ROOT_DIR=~/adapter-deps/install + #make debug EXTRA_CMAKE_FLAGS="-DVELOX_ENABLE_PARQUET=ON -DVELOX_BUILD_TESTING=ON -DVELOX_BUILD_TEST_UTILS=ON" + make EXTRA_CMAKE_FLAGS="-DVELOX_ENABLE_PARQUET=ON -DVELOX_BUILD_TESTING=ON -DVELOX_BUILD_TEST_UTILS=ON" + export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64/ + export HADOOP_ROOT_LOGGER="WARN,DRFA" + export LIBHDFS3_CONF=$(pwd)/.circleci/hdfs-client.xml + export HADOOP_HOME='/usr/local/hadoop-2.10.1' + export PATH=~/adapter-deps/install/bin:/usr/local/hadoop-2.10.1/bin:${PATH} + cd _build/release && ctest -j32 -VV --output-on-failure + + formatting-check: + name: Formatting Check + runs-on: ubuntu-latest + strategy: + matrix: + path: + - check: 'velox' + exclude: 'external' + steps: + - uses: actions/checkout@v2 + - name: Run clang-format style check for C/C++ programs. + uses: jidicula/clang-format-action@v3.5.1 + with: + clang-format-version: '12' + check-path: ${{ matrix.path['check'] }} + exclude-regex: ${{ matrix.path['exclude'] }} diff --git a/scripts/setup-centos7.sh b/scripts/setup-centos7.sh new file mode 100755 index 00000000000..2f1ca1c5000 --- /dev/null +++ b/scripts/setup-centos7.sh @@ -0,0 +1,272 @@ +#!/bin/bash +# 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. + +set -efx -o pipefail +# Some of the packages must be build with the same compiler flags +# so that some low level types are the same size. Also, disable warnings. +SCRIPTDIR=$(dirname "${BASH_SOURCE[0]}") +source $SCRIPTDIR/setup-helper-functions.sh +DEPENDENCY_DIR=${DEPENDENCY_DIR:-/tmp/velox-deps} +CPU_TARGET="${CPU_TARGET:-avx}" +NPROC=$(getconf _NPROCESSORS_ONLN) +export CFLAGS=$(get_cxx_flags $CPU_TARGET) # Used by LZO. +export CXXFLAGS=$CFLAGS # Used by boost. +export CPPFLAGS=$CFLAGS # Used by LZO. +export PKG_CONFIG_PATH=/usr/local/lib64/pkgconfig:/usr/local/lib/pkgconfig:/usr/lib64/pkgconfig:/usr/lib/pkgconfig:$PKG_CONFIG_PATH +FB_OS_VERSION=v2022.11.14.00 + +# shellcheck disable=SC2037 +SUDO="sudo -E" + +function run_and_time { + time "$@" + { echo "+ Finished running $*"; } 2> /dev/null +} + +function dnf_install { + $SUDO dnf install -y -q --setopt=install_weak_deps=False "$@" +} + +function yum_install { + $SUDO yum install -y "$@" +} + +function cmake_install_deps { + cmake -B"$1-build" -GNinja -DCMAKE_CXX_STANDARD=17 \ + -DCMAKE_CXX_FLAGS="${CFLAGS}" -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DCMAKE_BUILD_TYPE=Release -Wno-dev "$@" + ninja -C "$1-build" + $SUDO ninja -C "$1-build" install +} + +function wget_and_untar { + local URL=$1 + local DIR=$2 + mkdir -p "${DIR}" + wget -q --max-redirect 3 -O - "${URL}" | tar -xz -C "${DIR}" --strip-components=1 +} + +function install_cmake { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://cmake.org/files/v3.25/cmake-3.25.1.tar.gz cmake-3 + cd cmake-3 + ./bootstrap --prefix=/usr/local + make -j$(nproc) + $SUDO make install + cmake --version +} + +function install_ninja { + cd "${DEPENDENCY_DIR}" + github_checkout ninja-build/ninja v1.11.1 + ./configure.py --bootstrap + cmake -Bbuild-cmake + cmake --build build-cmake + $SUDO cp ninja /usr/local/bin/ +} + +function install_fmt { + cd "${DEPENDENCY_DIR}" + github_checkout fmtlib/fmt 8.0.0 + cmake_install -DFMT_TEST=OFF +} + +function install_folly { + cd "${DEPENDENCY_DIR}" + github_checkout facebook/folly "${FB_OS_VERSION}" + cmake_install -DBUILD_TESTS=OFF -DFOLLY_HAVE_INT128_T=ON +} + +function install_conda { + cd "${DEPENDENCY_DIR}" + mkdir -p conda && cd conda + wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh + MINICONDA_PATH=/opt/miniconda-for-velox + bash Miniconda3-latest-Linux-x86_64.sh -b -u $MINICONDA_PATH +} + +function install_openssl { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/openssl/openssl/archive/refs/tags/OpenSSL_1_1_1s.tar.gz openssl + cd openssl + ./config no-shared + make depend + make + $SUDO make install +} + +function install_gflags { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/gflags/gflags/archive/v2.2.2.tar.gz gflags + cd gflags + cmake_install -DBUILD_SHARED_LIBS=ON -DBUILD_STATIC_LIBS=ON -DBUILD_gflags_LIB=ON -DLIB_SUFFIX=64 -DCMAKE_INSTALL_PREFIX:PATH=/usr/local +} + +function install_glog { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/google/glog/archive/v0.5.0.tar.gz glog + cd glog + cmake_install -DBUILD_SHARED_LIBS=ON -DBUILD_STATIC_LIBS=ON -DCMAKE_INSTALL_PREFIX:PATH=/usr/local +} + +function install_snappy { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/google/snappy/archive/1.1.8.tar.gz snappy + cd snappy + cmake_install -DSNAPPY_BUILD_TESTS=OFF +} + +function install_dwarf { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/davea42/libdwarf-code/archive/refs/tags/20210528.tar.gz dwarf + cd dwarf + #local URL=https://github.com/davea42/libdwarf-code/releases/download/v0.5.0/libdwarf-0.5.0.tar.xz + #local DIR=dwarf + #mkdir -p "${DIR}" + #wget -q --max-redirect 3 "${URL}" + #tar -xf libdwarf-0.5.0.tar.xz -C "${DIR}" + #cd dwarf/libdwarf-0.5.0 + ./configure --enable-shared=no + make + make check + $SUDO make install +} + +function install_re2 { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/google/re2/archive/refs/tags/2023-03-01.tar.gz re2 + cd re2 + $SUDO make install +} + +function install_flex { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://github.com/westes/flex/releases/download/v2.6.4/flex-2.6.4.tar.gz flex + cd flex + ./autogen.sh + ./configure + $SUDO make install +} + +function install_lzo { + cd "${DEPENDENCY_DIR}" + wget_and_untar http://www.oberhumer.com/opensource/lzo/download/lzo-2.10.tar.gz lzo + cd lzo + ./configure --prefix=/usr/local --enable-shared --disable-static --docdir=/usr/local/share/doc/lzo-2.10 + make "-j$(nproc)" + $SUDO make install +} + +function install_boost { + cd "${DEPENDENCY_DIR}" + wget_and_untar https://boostorg.jfrog.io/artifactory/main/release/1.72.0/source/boost_1_72_0.tar.gz boost + cd boost + ./bootstrap.sh --prefix=/usr/local --with-python=/usr/bin/python3 --with-python-root=/usr/lib/python3.6 --without-libraries=python + $SUDO ./b2 "-j$(nproc)" -d0 install threading=multi +} + +function install_libhdfs3 { + cd "${DEPENDENCY_DIR}" + github_checkout apache/hawq master + cd depends/libhdfs3 + sed -i "/FIND_PACKAGE(GoogleTest REQUIRED)/d" ./CMakeLists.txt + sed -i "s/dumpversion/dumpfullversion/" ./CMake/Platform.cmake + sed -i "s/dfs.domain.socket.path\", \"\"/dfs.domain.socket.path\", \"\/var\/lib\/hadoop-hdfs\/dn_socket\"/g" src/common/SessionConfig.cpp + sed -i "s/pos < endOfCurBlock/pos \< endOfCurBlock \&\& pos \- cursor \<\= 128 \* 1024/g" src/client/InputStreamImpl.cpp + cmake_install +} + +function install_protobuf { + cd "${DEPENDENCY_DIR}" + wget https://github.com/protocolbuffers/protobuf/releases/download/v21.4/protobuf-all-21.4.tar.gz + tar -xzf protobuf-all-21.4.tar.gz + cd protobuf-21.4 + ./configure CXXFLAGS="-fPIC" --prefix=/usr/local + make "-j$(nproc)" + $SUDO make install +} + +function install_awssdk { + cd "${DEPENDENCY_DIR}" + github_checkout aws/aws-sdk-cpp 1.9.379 --depth 1 --recurse-submodules + cmake_install -DCMAKE_BUILD_TYPE=Release -DBUILD_SHARED_LIBS:BOOL=OFF -DMINIMIZE_SIZE:BOOL=ON -DENABLE_TESTING:BOOL=OFF -DBUILD_ONLY:STRING="s3;identity-management" +} + +function install_gtest { + cd "${DEPENDENCY_DIR}" + wget https://github.com/google/googletest/archive/refs/tags/release-1.12.1.tar.gz + tar -xzf release-1.12.1.tar.gz + cd googletest-release-1.12.1 + mkdir -p build && cd build && cmake -DBUILD_GTEST=ON -DBUILD_GMOCK=ON -DINSTALL_GTEST=ON -DINSTALL_GMOCK=ON -DBUILD_SHARED_LIBS=ON .. + make "-j$(nproc)" + $SUDO make install +} + +function install_prerequisites { + run_and_time install_lzo + run_and_time install_boost + run_and_time install_re2 + run_and_time install_flex + run_and_time install_openssl + run_and_time install_gflags + run_and_time install_glog + run_and_time install_snappy + run_and_time install_dwarf +} + +function install_velox_deps { + run_and_time install_fmt + run_and_time install_folly + run_and_time install_conda +} + +$SUDO dnf makecache + +# dnf install dependency libraries +dnf_install epel-release dnf-plugins-core # For ccache, ninja +# PowerTools only works on CentOS8 +# dnf config-manager --set-enabled powertools +dnf_install ccache git wget which libevent-devel \ + openssl-devel libzstd-devel lz4-devel double-conversion-devel \ + curl-devel cmake libxml2-devel libgsasl-devel libuuid-devel patch + +$SUDO dnf remove -y gflags + +# Required for Thrift +dnf_install autoconf automake libtool bison python3 python3-devel + +# Required for build flex +dnf_install gettext-devel texinfo help2man + +# dnf_install conda + +# Activate gcc9; enable errors on unset variables afterwards. +# GCC9 install via yum and devtoolset +# dnf install gcc-toolset-9 only works on CentOS8 + +$SUDO yum makecache +yum_install centos-release-scl +yum_install devtoolset-9 +source /opt/rh/devtoolset-9/enable || exit 1 +gcc --version +set -u + +# Build from source +[ -d "$DEPENDENCY_DIR" ] || mkdir -p "$DEPENDENCY_DIR" + +run_and_time install_cmake +run_and_time install_ninja + +install_prerequisites +install_velox_deps diff --git a/velox/common/memory/Memory.cpp b/velox/common/memory/Memory.cpp index 1316dee5081..d2adb1cf65d 100644 --- a/velox/common/memory/Memory.cpp +++ b/velox/common/memory/Memory.cpp @@ -122,19 +122,22 @@ std::shared_ptr MemoryManager::addRootPool( options.debugEnabled = debugEnabled_; options.coreOnAllocationFailureEnabled = coreOnAllocationFailureEnabled_; - folly::SharedMutex::WriteHolder guard{mutex_}; - if (pools_.find(poolName) != pools_.end()) { - VELOX_FAIL("Duplicate root pool name found: {}", poolName); + std::shared_ptr pool; + { + folly::SharedMutex::WriteHolder guard{mutex_}; + if (pools_.find(poolName) != pools_.end()) { + VELOX_FAIL("Duplicate root pool name found: {}", poolName); + } + pool = std::make_shared( + this, + poolName, + MemoryPool::Kind::kAggregate, + nullptr, + std::move(reclaimer), + poolDestructionCb_, + options); + pools_.emplace(poolName, pool); } - auto pool = std::make_shared( - this, - poolName, - MemoryPool::Kind::kAggregate, - nullptr, - std::move(reclaimer), - poolDestructionCb_, - options); - pools_.emplace(poolName, pool); VELOX_CHECK_EQ(pool->capacity(), 0); arbitrator_->reserveMemory(pool.get(), capacity); return pool; diff --git a/velox/connectors/hive/HiveConfig.cpp b/velox/connectors/hive/HiveConfig.cpp index a7b1f75a336..f1c8af8ae34 100644 --- a/velox/connectors/hive/HiveConfig.cpp +++ b/velox/connectors/hive/HiveConfig.cpp @@ -52,6 +52,11 @@ std::string HiveConfig::insertExistingPartitionsBehaviorString( } } +// static. +uint8_t HiveConfig::arrowBridgeTimestampUnit(const Config* config) { + return config->get(kArrowBridgeTimestampUnit, 9 /* nano */); +} + HiveConfig::InsertExistingPartitionsBehavior HiveConfig::insertExistingPartitionsBehavior(const Config* session) const { if (session->isValueExists(kInsertExistingPartitionsBehaviorSession)) { diff --git a/velox/connectors/hive/HiveConfig.h b/velox/connectors/hive/HiveConfig.h index 7af2f233f54..154d99a70f4 100644 --- a/velox/connectors/hive/HiveConfig.h +++ b/velox/connectors/hive/HiveConfig.h @@ -136,6 +136,10 @@ class HiveConfig { static constexpr const char* kOrcWriterMaxDictionaryMemorySession = "orc_optimized_writer_max_dictionary_memory"; + // Timestamp unit used during Velox-Arrow conversion. + static constexpr const char* kArrowBridgeTimestampUnit = + "arrow_bridge_timestamp_unit"; + /// Maximum number of rows for sort writer in one batch of output. static constexpr const char* kSortWriterMaxOutputRows = "sort-writer-max-output-rows"; @@ -179,6 +183,10 @@ class HiveConfig { std::string gcsCredentials() const; + /// Returns the timestamp unit used in Velox-Arrow conversion. + /// 0: second, 3: milli, 6: micro, 9: nano. + static uint8_t arrowBridgeTimestampUnit(const Config* config); + bool isOrcUseColumnNames(const Config* session) const; bool isFileColumnNamesReadAsLowerCase(const Config* session) const; diff --git a/velox/connectors/hive/HiveConnector.cpp b/velox/connectors/hive/HiveConnector.cpp index 1048050ac94..2cde13c3340 100644 --- a/velox/connectors/hive/HiveConnector.cpp +++ b/velox/connectors/hive/HiveConnector.cpp @@ -140,6 +140,7 @@ std::unique_ptr HivePartitionFunctionSpec::create( void HiveConnectorFactory::initialize() { static bool once = []() { dwio::common::registerFileSinks(); + dwrf::registerOrcReaderFactory(); dwrf::registerDwrfReaderFactory(); dwrf::registerDwrfWriterFactory(); // Meta's buck build system needs this check. diff --git a/velox/connectors/hive/HiveDataSink.cpp b/velox/connectors/hive/HiveDataSink.cpp index 4bd9c8c367d..513cdac96db 100644 --- a/velox/connectors/hive/HiveDataSink.cpp +++ b/velox/connectors/hive/HiveDataSink.cpp @@ -601,6 +601,8 @@ uint32_t HiveDataSink::appendWriter(const HiveWriterId& id) { if (canReclaim()) { options.spillConfig = spillConfig_; } + options.arrowBridgeTimestampUnit = HiveConfig::arrowBridgeTimestampUnit( + connectorQueryCtx_->sessionProperties()); options.nonReclaimableSection = writerInfo_.back()->nonReclaimableSectionHolder.get(); options.maxStripeSize = std::optional(hiveConfig_->getOrcWriterMaxStripeSize( diff --git a/velox/connectors/hive/HiveDataSource.cpp b/velox/connectors/hive/HiveDataSource.cpp index 448f673ac65..2260795a7fc 100644 --- a/velox/connectors/hive/HiveDataSource.cpp +++ b/velox/connectors/hive/HiveDataSource.cpp @@ -420,11 +420,14 @@ HiveDataSource::HiveDataSource( for (auto& [k, v] : hiveTableHandle_->subfieldFilters()) { filters.emplace(k.clone(), v->clone()); } - auto remainingFilter = extractFiltersFromRemainingFilter( - hiveTableHandle_->remainingFilter(), - expressionEvaluator_, - false, - filters); + auto remainingFilter = hiveTableHandle_->remainingFilter(); + if (hiveTableHandle_->isFilterPushdownEnabled()) { + remainingFilter = extractFiltersFromRemainingFilter( + hiveTableHandle_->remainingFilter(), + expressionEvaluator_, + false, + filters); + } std::vector remainingFilterSubfields; if (remainingFilter) { diff --git a/velox/connectors/hive/SplitReader.cpp b/velox/connectors/hive/SplitReader.cpp index a42b64e3714..ab2adf3a673 100644 --- a/velox/connectors/hive/SplitReader.cpp +++ b/velox/connectors/hive/SplitReader.cpp @@ -220,9 +220,18 @@ std::vector SplitReader::adaptColumns( } else { auto fileTypeIdx = fileType->getChildIdxIfExists(fieldName); if (!fileTypeIdx.has_value()) { - // Column is missing. Most likely due to schema evolution. - VELOX_CHECK(tableSchema); - setNullConstantValue(childSpec, tableSchema->findChild(fieldName)); + // If field name exists in the user-specified output type, + // set the column as null constant. + // Related PR: https://github.com/facebookincubator/velox/pull/6427. + auto outputTypeIdx = readerOutputType_->getChildIdxIfExists(fieldName); + if (outputTypeIdx.has_value()) { + setNullConstantValue( + childSpec, readerOutputType_->childAt(outputTypeIdx.value())); + } else { + // Column is missing. Most likely due to schema evolution. + VELOX_CHECK(tableSchema); + setNullConstantValue(childSpec, tableSchema->findChild(fieldName)); + } } else { // Column no longer missing, reset constant value set on the spec. childSpec->setConstantValue(nullptr); @@ -305,8 +314,19 @@ void SplitReader::setPartitionValue( it != partitionKeys_.end(), "ColumnHandle is missing for partition key {}", partitionKey); - auto constValue = VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH( - convertFromString, it->second->dataType()->kind(), value); + velox::variant constValue; + if (it->second->dataType()->isDate()) { + // TODO: need to align with query config for isIso8601. + if (value.has_value()) { + constValue = velox::variant( + velox::util::castFromDateString(StringView(value.value()), false)); + } else { + constValue = velox::variant(TypeKind::INTEGER); + } + } else { + constValue = VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH( + convertFromString, it->second->dataType()->kind(), value); + } setConstantValue(spec, it->second->dataType(), constValue); } diff --git a/velox/connectors/hive/tests/HivePartitionFunctionTest.cpp b/velox/connectors/hive/tests/HivePartitionFunctionTest.cpp index 4d64c630d10..557cc7155b2 100644 --- a/velox/connectors/hive/tests/HivePartitionFunctionTest.cpp +++ b/velox/connectors/hive/tests/HivePartitionFunctionTest.cpp @@ -462,6 +462,7 @@ TEST_F(HivePartitionFunctionTest, mapEntriesEncoded) { assertPartitionsWithConstChannel(values, 997); } +/* TEST_F(HivePartitionFunctionTest, nestedMaps) { auto innerMaps = makeNullableMapVector( std::vector< @@ -583,6 +584,7 @@ TEST_F(HivePartitionFunctionTest, nestedRows) { assertPartitionsWithConstChannel(values, 500); assertPartitionsWithConstChannel(values, 997); } +*/ TEST_F(HivePartitionFunctionTest, spec) { Type::registerSerDe(); diff --git a/velox/core/PlanNode.cpp b/velox/core/PlanNode.cpp index e356adf763b..f4f30fdf3cd 100644 --- a/velox/core/PlanNode.cpp +++ b/velox/core/PlanNode.cpp @@ -233,8 +233,16 @@ bool AggregationNode::canSpill(const QueryConfig& queryConfig) const { } // TODO: add spilling for pre-grouped aggregation later: // https://github.com/facebookincubator/velox/issues/3264 - return (isFinal() || isSingle()) && preGroupedKeys().empty() && - queryConfig.aggregationSpillEnabled(); + if ((isFinal() || isSingle()) && queryConfig.aggregationSpillEnabled()) { + return preGroupedKeys().empty(); + } + + if ((isIntermediate() || isPartial()) && + queryConfig.partialAggregationSpillEnabled()) { + return preGroupedKeys().empty(); + } + + return false; } void AggregationNode::addDetails(std::stringstream& stream) const { diff --git a/velox/core/PlanNode.h b/velox/core/PlanNode.h index deb3e0adba6..aa5840a9224 100644 --- a/velox/core/PlanNode.h +++ b/velox/core/PlanNode.h @@ -620,6 +620,14 @@ class AggregationNode : public PlanNode { return step_ == Step::kSingle; } + bool isIntermediate() const { + return step_ == Step::kIntermediate; + } + + bool isPartial() const { + return step_ == Step::kPartial; + } + folly::dynamic serialize() const override; static PlanNodePtr create(const folly::dynamic& obj, void* context); diff --git a/velox/core/QueryConfig.h b/velox/core/QueryConfig.h index ec9591e96ca..f458bbd5570 100644 --- a/velox/core/QueryConfig.h +++ b/velox/core/QueryConfig.h @@ -198,6 +198,11 @@ class QueryConfig { static constexpr const char* kAggregationSpillEnabled = "aggregation_spill_enabled"; + /// Partial aggregation spilling flag, only applies if "spill_enabled" flag is + /// set. + static constexpr const char* kPartialAggregationSpillEnabled = + "partial_aggregation_spill_enabled"; + /// Join spilling flag, only applies if "spill_enabled" flag is set. static constexpr const char* kJoinSpillEnabled = "join_spill_enabled"; @@ -348,6 +353,10 @@ class QueryConfig { static constexpr const char* kMaxSplitPreloadPerDriver = "max_split_preload_per_driver"; + // Timestamp unit used during Velox-Arrow conversion. + static constexpr const char* kArrowBridgeTimestampUnit = + "arrow_bridge_timestamp_unit"; + uint64_t queryMaxMemoryPerNode() const { return toCapacity( get(kQueryMaxMemoryPerNode, "0B"), CapacityUnit::BYTE); @@ -499,11 +508,17 @@ class QueryConfig { } /// Returns 'is aggregation spilling enabled' flag. Must also check the - /// spillEnabled()!g + /// spillEnabled()! bool aggregationSpillEnabled() const { return get(kAggregationSpillEnabled, true); } + /// Returns 'is partial aggregation spilling enabled' flag. Must also check + /// the spillEnabled()! + bool partialAggregationSpillEnabled() const { + return get(kPartialAggregationSpillEnabled, false); + } + /// Returns 'is join spilling enabled' flag. Must also check the /// spillEnabled()! bool joinSpillEnabled() const { @@ -559,6 +574,13 @@ class QueryConfig { return get(kSpillStartPartitionBit, kDefaultStartBit); } + /// Returns the timestamp unit used in Velox-Arrow conversion. + /// 0: second, 3: milli, 6: micro, 9: nano. + uint8_t arrowBridgeTimestampUnit() const { + constexpr uint8_t kDefaultUnit = 9; + return get(kArrowBridgeTimestampUnit, kDefaultUnit); + } + /// Returns the number of bits used to calculate the spilling partition /// number for hash join. The number of spilling partitions will be power of /// two. diff --git a/velox/core/QueryCtx.h b/velox/core/QueryCtx.h index 5d02f9e6ab7..cbae8ceb554 100644 --- a/velox/core/QueryCtx.h +++ b/velox/core/QueryCtx.h @@ -76,11 +76,13 @@ class QueryCtx { return cache_; } - folly::Executor* executor() const { - if (executor_ != nullptr) { - return executor_; - } - auto executor = executorKeepalive_.get(); + bool isExecutorSupplied() const { + auto executor = executor0(); + return executor != nullptr; + } + + folly::Executor* FOLLY_NONNULL executor() const { + auto executor = executor0(); VELOX_CHECK(executor, "Executor was not supplied."); return executor; } @@ -139,6 +141,14 @@ class QueryCtx { } } + folly::Executor* executor0() const { + if (executor_ != nullptr) { + return executor_; + } + auto executor = executorKeepalive_.get(); + return executor; + } + const std::string queryId_; folly::Executor* const executor_{nullptr}; folly::Executor* const spillExecutor_{nullptr}; diff --git a/velox/docs/functions/presto/conversion.rst b/velox/docs/functions/presto/conversion.rst index 7e439f78baf..3dd6bdaff34 100644 --- a/velox/docs/functions/presto/conversion.rst +++ b/velox/docs/functions/presto/conversion.rst @@ -149,7 +149,7 @@ supported conversions to/from JSON are listed in :doc:`json`. - Y - - Y - - + - Y * - timestamp - - @@ -803,3 +803,38 @@ Invalid example SELECT cast(decimal '-1000.000' as decimal(6, 4)); -- Out of range SELECT cast(decimal '123456789' as decimal(9, 1)); -- Out of range + +From varchar +^^^^^^^^^^^^ + +Casting varchar to a decimal of given precision and scale is allowed +if the input value can be represented by the precision and scale. When casting from +a larger scale to a smaller one, the fraction part is rounded. Casting from invalid input value throws. + +Valid example + +:: + + SELECT cast('9999999999.99' as decimal(12, 2)); -- decimal '9999999999.99' + SELECT cast('1.556' as decimal(12, 2)); -- decimal '1.56' + SELECT cast('1.554' as decimal(12, 2)); -- decimal '1.55' + SELECT cast('-1.554' as decimal(12, 2)); -- decimal '-1.55' + SELECT cast('+09' as decimal(12, 2)); -- decimal '9.00' + SELECT cast('9.' as decimal(12, 2)); -- decimal '9.00' + SELECT cast('.9' as decimal(12, 2)); -- decimal '0.90' + SELECT cast('3E+2' as decimal(12, 2)); -- decimal '300.00' + SELECT cast('3e+2' as decimal(12, 2)); -- decimal '300.00' + SELECT cast('31.423e+2' as decimal(12, 2)); -- decimal '3142.30' + SELECT cast('1.2e-2' as decimal(12, 2)); -- decimal '0.01' + SELECT cast('1.2e-5' as decimal(12, 2)); -- decimal '0.00' + SELECT cast('0000.123' as decimal(12, 2)); -- decimal '0.12' + SELECT cast('.123000000' as decimal(12, 2)); -- decimal '0.12' + +Invalid example + +:: + + SELECT cast('1.23e67' as decimal(38, 0)); -- Value too large + SELECT cast('0.0446a' as decimal(9, 1)); -- Value is not a number + SELECT cast('' as decimal(9, 1)); -- Value is not a number + SELECT cast('23e-5d' as decimal(9, 1)); -- Value is not a number diff --git a/velox/docs/functions/spark/array.rst b/velox/docs/functions/spark/array.rst index 2183f4f301c..f31eb10008f 100644 --- a/velox/docs/functions/spark/array.rst +++ b/velox/docs/functions/spark/array.rst @@ -62,6 +62,15 @@ Array Functions SELECT array_sort(ARRAY [NULL, 1, NULL]); -- [1, NULL, NULL] SELECT array_sort(ARRAY [NULL, 2, 1]); -- [1, 2, NULL] +.. spark:function:: array_union(array(E), array(E1)) -> array(E2) + + Returns an array of the elements in the union of array1 and array2, without duplicates. :: + + SELECT array_union(array(1, 2, 3), array(1, 3, 5)); -- [1, 2, 3, 5] + SELECT array_union(array(1, 3, 5), array(1, 2, 3)); -- [1, 3, 5, 2] + SELECT array_union(array(1, 2, 3), array(1, 3, 5, null)); -- [1, 2, 3, 5, null] + SELECT array_union(array(1, 2, NaN), array(1, 3, NaN)); -- [1, 2, NaN, 3] + .. spark:function:: concat(array(E), array(E1), ..., array(En)) -> array(E, E1, ..., En) Returns the concatenation of array(E), array(E1), ..., array(En). :: diff --git a/velox/docs/functions/spark/json.rst b/velox/docs/functions/spark/json.rst index 07f4f3a75ac..c2708b938dc 100644 --- a/velox/docs/functions/spark/json.rst +++ b/velox/docs/functions/spark/json.rst @@ -22,6 +22,8 @@ JSON Functions .. spark:function:: get_json_object(json, path) -> varchar - Extracts a json object from path:: + Extracts a json object from ``path``. Returns NULL if it finds json string + is malformed. :: - SELECT get_json_object('{"a":"b"}', '$.a'); -- b \ No newline at end of file + SELECT get_json_object('{"a":"b"}', '$.a'); -- 'b' + SELECT get_json_object('{"a":{"b":"c"}}', '$.a'); -- '{"b":"c"}' \ No newline at end of file diff --git a/velox/docs/functions/spark/math.rst b/velox/docs/functions/spark/math.rst index 46be649696c..fc7f0386035 100644 --- a/velox/docs/functions/spark/math.rst +++ b/velox/docs/functions/spark/math.rst @@ -18,6 +18,10 @@ Mathematical Functions Returns inverse hyperbolic sine of ``x``. +.. spark:function:: atan2(x, y) -> double + + Returns the angle in radians between the positive x-axis of a plane and the point given by the coordinates(x, y). + .. spark:function:: atanh(x) -> double Returns inverse hyperbolic tangent of ``x``. diff --git a/velox/dwio/common/CachedBufferedInput.h b/velox/dwio/common/CachedBufferedInput.h index 5a8280e06fa..d9c5def1fb0 100644 --- a/velox/dwio/common/CachedBufferedInput.h +++ b/velox/dwio/common/CachedBufferedInput.h @@ -69,7 +69,8 @@ class CachedBufferedInput : public BufferedInput { : BufferedInput( std::move(readFile), readerOptions.getMemoryPool(), - metricsLog), + metricsLog, + ioStats.get()), cache_(cache), fileNum_(fileNum), tracker_(std::move(tracker)), diff --git a/velox/dwio/common/DirectBufferedInput.h b/velox/dwio/common/DirectBufferedInput.h index 815e0e19eba..1c9155c5927 100644 --- a/velox/dwio/common/DirectBufferedInput.h +++ b/velox/dwio/common/DirectBufferedInput.h @@ -115,7 +115,8 @@ class DirectBufferedInput : public BufferedInput { : BufferedInput( std::move(readFile), readerOptions.getMemoryPool(), - metricsLog), + metricsLog, + ioStats.get()), fileNum_(fileNum), tracker_(std::move(tracker)), groupId_(groupId), diff --git a/velox/dwio/common/InputStream.cpp b/velox/dwio/common/InputStream.cpp index 5b5150109bd..a57564faf77 100644 --- a/velox/dwio/common/InputStream.cpp +++ b/velox/dwio/common/InputStream.cpp @@ -141,6 +141,7 @@ void ReadFileInputStream::vread( size_t(0), [&](size_t acc, const auto& r) { return acc + r.length; }); logRead(regions[0].offset, length, purpose); + auto readStartMicros = getCurrentTimeMicro(); readFile_->preadv(regions, iobufs); if (stats_) { diff --git a/velox/dwio/common/Options.h b/velox/dwio/common/Options.h index 619f8ea1664..9c7ed53bb5d 100644 --- a/velox/dwio/common/Options.h +++ b/velox/dwio/common/Options.h @@ -564,6 +564,7 @@ struct WriterOptions { std::optional maxStripeSize{std::nullopt}; std::optional maxDictionaryMemory{std::nullopt}; std::map serdeParameters; + std::optional arrowBridgeTimestampUnit; }; } // namespace facebook::velox::dwio::common diff --git a/velox/dwio/common/SelectiveColumnReader.cpp b/velox/dwio/common/SelectiveColumnReader.cpp index f2c157ff9c7..35d7078e4bb 100644 --- a/velox/dwio/common/SelectiveColumnReader.cpp +++ b/velox/dwio/common/SelectiveColumnReader.cpp @@ -214,6 +214,9 @@ void SelectiveColumnReader::getIntValues( VELOX_FAIL("Unsupported value size: {}", valueSize_); } break; + case TypeKind::TIMESTAMP: + getFlatValues(rows, result, requestedType); + break; default: VELOX_FAIL( "Not a valid type for integer reader: {}", requestedType->toString()); diff --git a/velox/dwio/common/SelectiveStructColumnReader.cpp b/velox/dwio/common/SelectiveStructColumnReader.cpp index 30e6e748fc1..face75910c8 100644 --- a/velox/dwio/common/SelectiveStructColumnReader.cpp +++ b/velox/dwio/common/SelectiveStructColumnReader.cpp @@ -133,7 +133,6 @@ void SelectiveStructColumnReaderBase::read( } auto& childSpecs = scanSpec_->children(); - VELOX_CHECK(!childSpecs.empty()); for (size_t i = 0; i < childSpecs.size(); ++i) { auto& childSpec = childSpecs[i]; if (isChildConstant(*childSpec)) { @@ -218,7 +217,7 @@ bool SelectiveStructColumnReaderBase::isChildConstant( fileType_->type()->kind() != TypeKind::MAP && // If this is the case it means this is a flat map, // so it can't have "missing" fields. - childSpec.channel() >= fileType_->size()); + !fileType_->containsChild(childSpec.fieldName())); } namespace { @@ -302,7 +301,6 @@ void setNullField( void SelectiveStructColumnReaderBase::getValues( RowSet rows, VectorPtr* result) { - VELOX_CHECK(!scanSpec_->children().empty()); VELOX_CHECK_NOT_NULL( *result, "SelectiveStructColumnReaderBase expects a non-null result"); VELOX_CHECK( diff --git a/velox/dwio/common/TypeWithId.h b/velox/dwio/common/TypeWithId.h index 953ac87b2b8..96c6cd38fc4 100644 --- a/velox/dwio/common/TypeWithId.h +++ b/velox/dwio/common/TypeWithId.h @@ -59,6 +59,11 @@ class TypeWithId : public velox::Tree> { const std::shared_ptr& childAt(uint32_t idx) const override; + bool containsChild(const std::string& name) const { + VELOX_CHECK_EQ(type_->kind(), velox::TypeKind::ROW); + return type_->as().containsChild(name); + } + const std::shared_ptr& childByName( const std::string& name) const { VELOX_CHECK_EQ(type_->kind(), velox::TypeKind::ROW); diff --git a/velox/dwio/common/tests/E2EFilterTestBase.h b/velox/dwio/common/tests/E2EFilterTestBase.h index 4d361c8f1fb..8f073656d93 100644 --- a/velox/dwio/common/tests/E2EFilterTestBase.h +++ b/velox/dwio/common/tests/E2EFilterTestBase.h @@ -170,7 +170,8 @@ class E2EFilterTestBase : public testing::Test { virtual void writeToMemory( const TypePtr& type, const std::vector& batches, - bool forRowGroupSkip) = 0; + bool forRowGroupSkip, + const TypePtr& schema = nullptr) = 0; virtual std::unique_ptr makeReader( const dwio::common::ReaderOptions& opts, diff --git a/velox/dwio/dwrf/common/FileMetadata.h b/velox/dwio/dwrf/common/FileMetadata.h index 2ea21628a59..973ec312ecd 100644 --- a/velox/dwio/dwrf/common/FileMetadata.h +++ b/velox/dwio/dwrf/common/FileMetadata.h @@ -426,7 +426,8 @@ class FooterWrapper : public ProtoWrapperBase { // TODO: ORC has not supported column statistics yet int statisticsSize() const { - return format_ == DwrfFormat::kDwrf ? dwrfPtr()->statistics_size() : 0; + return format_ == DwrfFormat::kDwrf ? dwrfPtr()->statistics_size() + : orcPtr()->statistics_size(); } const ::google::protobuf::RepeatedPtrField< @@ -438,7 +439,6 @@ class FooterWrapper : public ProtoWrapperBase { const ::facebook::velox::dwrf::proto::ColumnStatistics& statistics( int index) const { - VELOX_CHECK_EQ(format_, DwrfFormat::kDwrf); return dwrfPtr()->statistics(index); } diff --git a/velox/dwio/dwrf/reader/DwrfReader.cpp b/velox/dwio/dwrf/reader/DwrfReader.cpp index 153cd0d467a..567f6bfd878 100644 --- a/velox/dwio/dwrf/reader/DwrfReader.cpp +++ b/velox/dwio/dwrf/reader/DwrfReader.cpp @@ -1113,4 +1113,12 @@ void unregisterDwrfReaderFactory() { dwio::common::unregisterReaderFactory(dwio::common::FileFormat::DWRF); } +void registerOrcReaderFactory() { + dwio::common::registerReaderFactory(std::make_shared()); +} + +void unregisterOrcReaderFactory() { + dwio::common::unregisterReaderFactory(dwio::common::FileFormat::ORC); +} + } // namespace facebook::velox::dwrf diff --git a/velox/dwio/dwrf/reader/DwrfReader.h b/velox/dwio/dwrf/reader/DwrfReader.h index 9ced24322a2..9c8e32df2c0 100644 --- a/velox/dwio/dwrf/reader/DwrfReader.h +++ b/velox/dwio/dwrf/reader/DwrfReader.h @@ -365,8 +365,23 @@ class DwrfReaderFactory : public dwio::common::ReaderFactory { } }; +class OrcReaderFactory : public dwio::common::ReaderFactory { + public: + OrcReaderFactory() : ReaderFactory(dwio::common::FileFormat::ORC) {} + + std::unique_ptr createReader( + std::unique_ptr input, + const dwio::common::ReaderOptions& options) override { + return DwrfReader::create(std::move(input), options); + } +}; + void registerDwrfReaderFactory(); void unregisterDwrfReaderFactory(); +void registerOrcReaderFactory(); + +void unregisterOrcReaderFactory(); + } // namespace facebook::velox::dwrf diff --git a/velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h b/velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h index cf0d328d472..95ed9054a02 100644 --- a/velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h +++ b/velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h @@ -40,6 +40,10 @@ class SelectiveDecimalColumnReader : public SelectiveColumnReader { void getValues(RowSet rows, VectorPtr* result) override; + bool hasBulkPath() const override { + return false; + } + private: template void readHelper(RowSet rows); diff --git a/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.h b/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.h index 92b3aa75038..8c14cc963d8 100644 --- a/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.h +++ b/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.h @@ -63,7 +63,11 @@ class SelectiveIntegerDirectColumnReader } bool hasBulkPath() const override { - return true; + if (format == velox::dwrf::DwrfFormat::kOrc) { + return false; // RLEv2 does't support FastPath yet + } else { + return true; + } } void seekToRowGroup(uint32_t index) override { diff --git a/velox/dwio/dwrf/reader/SelectiveStringDictionaryColumnReader.h b/velox/dwio/dwrf/reader/SelectiveStringDictionaryColumnReader.h index 88ff95ed7ae..937d31c9e0a 100644 --- a/velox/dwio/dwrf/reader/SelectiveStringDictionaryColumnReader.h +++ b/velox/dwio/dwrf/reader/SelectiveStringDictionaryColumnReader.h @@ -53,6 +53,14 @@ class SelectiveStringDictionaryColumnReader uint64_t skip(uint64_t numValues) override; + bool hasBulkPath() const override { + if (version_ == velox::dwrf::RleVersion_1) { + return true; + } else { + return false; // RLEv2 does't support FastPath yet + } + } + void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls) override; diff --git a/velox/dwio/dwrf/test/E2EFilterTest.cpp b/velox/dwio/dwrf/test/E2EFilterTest.cpp index cc501126e1c..b6d7f35505b 100644 --- a/velox/dwio/dwrf/test/E2EFilterTest.cpp +++ b/velox/dwio/dwrf/test/E2EFilterTest.cpp @@ -63,7 +63,8 @@ class E2EFilterTest : public E2EFilterTestBase { void writeToMemory( const TypePtr& type, const std::vector& batches, - bool forRowGroupSkip = false) override { + bool forRowGroupSkip = false, + const TypePtr& schema = nullptr) override { auto options = createWriterOptions(type); int32_t flushCounter = 0; // If we test row group skip, we have all the data in one stripe. For diff --git a/velox/dwio/parquet/reader/PageReader.cpp b/velox/dwio/parquet/reader/PageReader.cpp index 042150f37a0..1ab243ec1ae 100644 --- a/velox/dwio/parquet/reader/PageReader.cpp +++ b/velox/dwio/parquet/reader/PageReader.cpp @@ -396,6 +396,51 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) { } break; } + case thrift::Type::INT96: { + auto numVeloxBytes = dictionary_.numValues * sizeof(Timestamp); + dictionary_.values = AlignedBuffer::allocate(numVeloxBytes, &pool_); + auto numBytes = dictionary_.numValues * sizeof(Int96Timestamp); + if (pageData_) { + memcpy(dictionary_.values->asMutable(), pageData_, numBytes); + } else { + dwio::common::readBytes( + numBytes, + inputStream_.get(), + dictionary_.values->asMutable(), + bufferStart_, + bufferEnd_); + } + // Expand the Parquet type length values to Velox type length. + // We start from the end to allow in-place expansion. + auto values = dictionary_.values->asMutable(); + auto parquetValues = dictionary_.values->asMutable(); + static constexpr int64_t kJulianToUnixEpochDays = 2440588LL; + static constexpr int64_t kSecondsPerDay = 86400LL; + static constexpr int64_t kNanosPerSecond = + Timestamp::kNanosecondsInMillisecond * + Timestamp::kMillisecondsInSecond; + for (auto i = dictionary_.numValues - 1; i >= 0; --i) { + // Convert the timestamp into seconds and nanos since the Unix epoch, + // 00:00:00.000000 on 1 January 1970. + uint64_t nanos; + memcpy( + &nanos, + parquetValues + i * sizeof(Int96Timestamp), + sizeof(uint64_t)); + int32_t days; + memcpy( + &days, + parquetValues + i * sizeof(Int96Timestamp) + sizeof(uint64_t), + sizeof(int32_t)); + int64_t seconds = (days - kJulianToUnixEpochDays) * kSecondsPerDay; + if (nanos > Timestamp::kMaxNanos) { + seconds += nanos / kNanosPerSecond; + nanos -= (nanos / kNanosPerSecond) * kNanosPerSecond; + } + values[i] = Timestamp(seconds, nanos); + } + break; + } case thrift::Type::BYTE_ARRAY: { dictionary_.values = AlignedBuffer::allocate(dictionary_.numValues, &pool_); @@ -486,7 +531,6 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) { VELOX_UNSUPPORTED( "Parquet type {} not supported for dictionary", parquetType); } - case thrift::Type::INT96: default: VELOX_UNSUPPORTED( "Parquet type {} not supported for dictionary", parquetType); @@ -513,6 +557,8 @@ int32_t parquetTypeBytes(thrift::Type::type type) { case thrift::Type::INT64: case thrift::Type::DOUBLE: return 8; + case thrift::Type::INT96: + return 12; default: VELOX_FAIL("Type does not have a byte width {}", type); } diff --git a/velox/dwio/parquet/reader/ParquetColumnReader.cpp b/velox/dwio/parquet/reader/ParquetColumnReader.cpp index ea3169ae727..8f5df722873 100644 --- a/velox/dwio/parquet/reader/ParquetColumnReader.cpp +++ b/velox/dwio/parquet/reader/ParquetColumnReader.cpp @@ -28,6 +28,7 @@ #include "velox/dwio/parquet/reader/Statistics.h" #include "velox/dwio/parquet/reader/StringColumnReader.h" #include "velox/dwio/parquet/reader/StructColumnReader.h" +#include "velox/dwio/parquet/reader/TimestampColumnReader.h" #include "velox/dwio/parquet/thrift/ParquetThriftTypes.h" namespace facebook::velox::parquet { @@ -37,7 +38,8 @@ std::unique_ptr ParquetColumnReader::build( const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec) { + common::ScanSpec& scanSpec, + memory::MemoryPool& pool) { auto colName = scanSpec.fieldName(); switch (fileType->type()->kind()) { @@ -58,7 +60,7 @@ std::unique_ptr ParquetColumnReader::build( case TypeKind::ROW: return std::make_unique( - requestedType, fileType, params, scanSpec); + requestedType, fileType, params, scanSpec, pool); case TypeKind::VARBINARY: case TypeKind::VARCHAR: @@ -66,16 +68,20 @@ std::unique_ptr ParquetColumnReader::build( case TypeKind::ARRAY: return std::make_unique( - requestedType, fileType, params, scanSpec); + requestedType, fileType, params, scanSpec, pool); case TypeKind::MAP: return std::make_unique( - requestedType, fileType, params, scanSpec); + requestedType, fileType, params, scanSpec, pool); case TypeKind::BOOLEAN: return std::make_unique( requestedType, fileType, params, scanSpec); + case TypeKind::TIMESTAMP: + return std::make_unique( + requestedType, fileType, params, scanSpec); + default: VELOX_FAIL( "buildReader unhandled type: " + diff --git a/velox/dwio/parquet/reader/ParquetColumnReader.h b/velox/dwio/parquet/reader/ParquetColumnReader.h index 516a500cd22..34a5b258273 100644 --- a/velox/dwio/parquet/reader/ParquetColumnReader.h +++ b/velox/dwio/parquet/reader/ParquetColumnReader.h @@ -45,6 +45,7 @@ class ParquetColumnReader { const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec); + common::ScanSpec& scanSpec, + memory::MemoryPool& pool); }; } // namespace facebook::velox::parquet diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp index 07471ba15e2..44b884366f5 100644 --- a/velox/dwio/parquet/reader/ParquetReader.cpp +++ b/velox/dwio/parquet/reader/ParquetReader.cpp @@ -84,6 +84,11 @@ class ReaderBase { /// the data still exists in the buffered inputs. bool isRowGroupBuffered(int32_t rowGroupIndex) const; + static std::shared_ptr createTypeWithId( + const std::shared_ptr& inputType, + const RowTypePtr& rowTypePtr, + bool fileColumnNamesReadAsLowerCase); + private: // Reads and parses file footer. void loadFileMetaData(); @@ -527,7 +532,7 @@ TypePtr ReaderBase::convertType( case thrift::Type::type::INT64: return BIGINT(); case thrift::Type::type::INT96: - return DOUBLE(); // TODO: Lose precision + return TIMESTAMP(); case thrift::Type::type::FLOAT: return REAL(); case thrift::Type::type::DOUBLE: @@ -564,6 +569,33 @@ std::shared_ptr ReaderBase::createRowType( std::move(childNames), std::move(childTypes)); } +std::shared_ptr ReaderBase::createTypeWithId( + const std::shared_ptr& inputType, + const RowTypePtr& rowTypePtr, + bool fileColumnNamesReadAsLowerCase) { + if (!fileColumnNamesReadAsLowerCase) { + return inputType; + } + std::vector names; + names.reserve(rowTypePtr->names().size()); + std::vector types = rowTypePtr->children(); + for (const auto& name : rowTypePtr->names()) { + std::string childName = name; + folly::toLowerAscii(childName); + names.emplace_back(childName); + } + auto convertedType = + TypeFactory::create(std::move(names), std::move(types)); + + auto children = inputType->getChildren(); + return std::make_shared( + convertedType, + std::move(children), + inputType->id(), + inputType->maxId(), + inputType->column()); +} + void ReaderBase::scheduleRowGroups( const std::vector& rowGroupIds, int32_t currentGroup, @@ -630,13 +662,19 @@ ParquetRowReader::ParquetRowReader( return; // TODO } ParquetParams params(pool_, columnReaderStats_, readerBase_->fileMetaData()); - auto columnSelector = std::make_shared( - ColumnSelector::apply(options_.getSelector(), readerBase_->schema())); + // ColumnSelector::apply does not work for schema pruning case. + auto columnSelector = options_.getSelector() == nullptr + ? std::make_shared(ColumnSelector(readerBase_->schema())) + : options_.getSelector(); columnReader_ = ParquetColumnReader::build( - columnSelector->getSchemaWithId(), + ReaderBase::createTypeWithId( + columnSelector->getSchemaWithId(), + asRowType(columnSelector->getSchemaWithId()->type()), + readerBase_->isFileColumnNamesReadAsLowerCase()), readerBase_->schemaWithId(), // Id is schema id params, - *options_.getScanSpec()); + *options_.getScanSpec(), + pool_); filterRowGroups(); if (!rowGroupIds_.empty()) { diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.cpp b/velox/dwio/parquet/reader/RepeatedColumnReader.cpp index 250bd204e08..743bfd1be94 100644 --- a/velox/dwio/parquet/reader/RepeatedColumnReader.cpp +++ b/velox/dwio/parquet/reader/RepeatedColumnReader.cpp @@ -33,6 +33,9 @@ PageReader* FOLLY_NULLABLE readLeafRepDefs( return nullptr; } auto pageReader = reader->formatData().as().reader(); + if (pageReader == nullptr) { + return nullptr; + } pageReader->decodeRepDefs(numTop); return pageReader; } @@ -113,7 +116,8 @@ MapColumnReader::MapColumnReader( const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec) + common::ScanSpec& scanSpec, + memory::MemoryPool& pool) : dwio::common::SelectiveMapColumnReader( requestedType, fileType, @@ -123,9 +127,17 @@ MapColumnReader::MapColumnReader( auto& keyChildType = requestedType->childAt(0); auto& elementChildType = requestedType->childAt(1); keyReader_ = ParquetColumnReader::build( - keyChildType, fileType_->childAt(0), params, *scanSpec.children()[0]); + keyChildType, + fileType_->childAt(0), + params, + *scanSpec.children()[0], + pool); elementReader_ = ParquetColumnReader::build( - elementChildType, fileType_->childAt(1), params, *scanSpec.children()[1]); + elementChildType, + fileType_->childAt(1), + params, + *scanSpec.children()[1], + pool); reinterpret_cast(fileType.get()) ->makeLevelInfo(levelInfo_); children_ = {keyReader_.get(), elementReader_.get()}; @@ -223,7 +235,8 @@ ListColumnReader::ListColumnReader( const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec) + common::ScanSpec& scanSpec, + memory::MemoryPool& pool) : dwio::common::SelectiveListColumnReader( requestedType, fileType, @@ -231,7 +244,7 @@ ListColumnReader::ListColumnReader( scanSpec) { auto& childType = requestedType->childAt(0); child_ = ParquetColumnReader::build( - childType, fileType_->childAt(0), params, *scanSpec.children()[0]); + childType, fileType_->childAt(0), params, *scanSpec.children()[0], pool); reinterpret_cast(fileType.get()) ->makeLevelInfo(levelInfo_); children_ = {child_.get()}; diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.h b/velox/dwio/parquet/reader/RepeatedColumnReader.h index 3155e8d6647..d6c68d2239a 100644 --- a/velox/dwio/parquet/reader/RepeatedColumnReader.h +++ b/velox/dwio/parquet/reader/RepeatedColumnReader.h @@ -59,7 +59,8 @@ class MapColumnReader : public dwio::common::SelectiveMapColumnReader { const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec); + common::ScanSpec& scanSpec, + memory::MemoryPool& pool); void prepareRead( vector_size_t offset, @@ -115,7 +116,8 @@ class ListColumnReader : public dwio::common::SelectiveListColumnReader { const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec); + common::ScanSpec& scanSpec, + memory::MemoryPool& pool); void prepareRead( vector_size_t offset, diff --git a/velox/dwio/parquet/reader/StructColumnReader.cpp b/velox/dwio/parquet/reader/StructColumnReader.cpp index eca887eab15..af8000046e7 100644 --- a/velox/dwio/parquet/reader/StructColumnReader.cpp +++ b/velox/dwio/parquet/reader/StructColumnReader.cpp @@ -30,21 +30,46 @@ StructColumnReader::StructColumnReader( const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec) + common::ScanSpec& scanSpec, + memory::MemoryPool& pool) : SelectiveStructColumnReader(requestedType, fileType, params, scanSpec) { auto& childSpecs = scanSpec_->stableChildren(); + std::vector missingFields; for (auto i = 0; i < childSpecs.size(); ++i) { auto childSpec = childSpecs[i]; if (childSpecs[i]->isConstant()) { continue; } - auto childFileType = fileType_->childByName(childSpec->fieldName()); - auto childRequestedType = - requestedType_->childByName(childSpec->fieldName()); + const auto& fieldName = childSpec->fieldName(); + if (!fileType_->containsChild(fieldName)) { + missingFields.emplace_back(i); + continue; + } + auto childFileType = fileType_->childByName(fieldName); + auto childRequestedType = requestedType_->childByName(fieldName); addChild(ParquetColumnReader::build( - childRequestedType, childFileType, params, *childSpec)); + childRequestedType, childFileType, params, *childSpec, pool)); childSpecs[i]->setSubscript(children_.size() - 1); } + + if (missingFields.size() > 0) { + // Set the struct as null if all the children fields in the output type are + // missing and the number of child fields is more than one. + if (childSpecs.size() > 1 && missingFields.size() == childSpecs.size()) { + scanSpec_->setConstantValue( + BaseVector::createNullConstant(requestedType_->type(), 1, &pool)); + } else { + // Set null constant for the missing child field of output type. + for (int channel : missingFields) { + childSpecs[channel]->setConstantValue(BaseVector::createNullConstant( + requestedType_->childByName(childSpecs[channel]->fieldName()) + ->type(), + 1, + &pool)); + } + } + } + auto type = reinterpret_cast(fileType_.get()); if (type->parent()) { levelMode_ = reinterpret_cast(fileType_.get()) @@ -54,7 +79,10 @@ StructColumnReader::StructColumnReader( // this and the child. auto child = childForRepDefs_; for (;;) { - assert(child); + if (child == nullptr) { + levelMode_ = LevelMode::kNulls; + break; + } if (child->fileType().type()->kind() == TypeKind::ARRAY || child->fileType().type()->kind() == TypeKind::MAP) { levelMode_ = LevelMode::kStructOverLists; diff --git a/velox/dwio/parquet/reader/StructColumnReader.h b/velox/dwio/parquet/reader/StructColumnReader.h index f38c9e849c7..f03d5549387 100644 --- a/velox/dwio/parquet/reader/StructColumnReader.h +++ b/velox/dwio/parquet/reader/StructColumnReader.h @@ -35,7 +35,8 @@ class StructColumnReader : public dwio::common::SelectiveStructColumnReader { const std::shared_ptr& requestedType, const std::shared_ptr& fileType, ParquetParams& params, - common::ScanSpec& scanSpec); + common::ScanSpec& scanSpec, + memory::MemoryPool& pool); void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls) override; diff --git a/velox/dwio/parquet/reader/TimestampColumnReader.h b/velox/dwio/parquet/reader/TimestampColumnReader.h new file mode 100644 index 00000000000..4c534b4bfce --- /dev/null +++ b/velox/dwio/parquet/reader/TimestampColumnReader.h @@ -0,0 +1,49 @@ +/* + * 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. + */ + +#pragma once + +#include "velox/dwio/parquet/reader/IntegerColumnReader.h" +#include "velox/dwio/parquet/reader/ParquetColumnReader.h" + +namespace facebook::velox::parquet { + +class TimestampColumnReader : public IntegerColumnReader { + public: + TimestampColumnReader( + const std::shared_ptr& requestedType, + std::shared_ptr fileType, + ParquetParams& params, + common::ScanSpec& scanSpec) + : IntegerColumnReader(requestedType, fileType, params, scanSpec) {} + + bool hasBulkPath() const override { + return false; + } + + void read( + vector_size_t offset, + RowSet rows, + const uint64_t* /*incomingNulls*/) override { + auto& data = formatData_->as(); + // Use int128_t as a workaroud. Timestamp in Velox is of 16-byte length. + prepareRead(offset, rows, nullptr); + readCommon(rows); + readOffset_ += rows.back() + 1; + } +}; + +} // namespace facebook::velox::parquet diff --git a/velox/dwio/parquet/tests/examples/contacts.parquet b/velox/dwio/parquet/tests/examples/contacts.parquet new file mode 100644 index 00000000000..fa3751f8dc4 Binary files /dev/null and b/velox/dwio/parquet/tests/examples/contacts.parquet differ diff --git a/velox/dwio/parquet/tests/examples/timestamp_int96.parquet b/velox/dwio/parquet/tests/examples/timestamp_int96.parquet new file mode 100644 index 00000000000..ea3a125aab6 Binary files /dev/null and b/velox/dwio/parquet/tests/examples/timestamp_int96.parquet differ diff --git a/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp b/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp index c633dff6fb6..c1bedfa2ad0 100644 --- a/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp +++ b/velox/dwio/parquet/tests/reader/E2EFilterTest.cpp @@ -55,7 +55,8 @@ class E2EFilterTest : public E2EFilterTestBase { void writeToMemory( const TypePtr&, const std::vector& batches, - bool forRowGroupSkip = false) override { + bool forRowGroupSkip = false, + const TypePtr& schema = nullptr) override { auto sink = std::make_unique( 200 * 1024 * 1024, FileSink::Options{.pool = leafPool_.get()}); sinkPtr_ = sink.get(); @@ -70,6 +71,8 @@ class E2EFilterTest : public E2EFilterTestBase { }); }; + options_.schema = schema; + writer_ = std::make_unique( std::move(sink), options_); for (auto& batch : batches) { @@ -615,6 +618,26 @@ TEST_F(E2EFilterTest, combineRowGroup) { EXPECT_EQ(parquetReader.numberOfRows(), 5); } +TEST_F(E2EFilterTest, configurableWriteSchema) { + rowType_ = ROW({"c0"}, {INTEGER()}); + std::vector batches; + for (int i = 0; i < 5; i++) { + batches.push_back(std::static_pointer_cast( + test::BatchMaker::createBatch(rowType_, 1, *leafPool_, nullptr, 0))); + } + + auto newType = ROW({"int32"}, {INTEGER()}); + writeToMemory(rowType_, batches, false, newType); + std::string_view data(sinkPtr_->data(), sinkPtr_->size()); + dwio::common::ReaderOptions readerOpts{leafPool_.get()}; + auto input = std::make_unique( + std::make_shared(data), readerOpts.getMemoryPool()); + auto reader = makeReader(readerOpts, std::move(input)); + auto parquetReader = dynamic_cast(*reader.get()); + EXPECT_EQ(parquetReader.rowType()->containsChild("int32"), true); + EXPECT_EQ(parquetReader.rowType()->containsChild("c0"), false); +} + // Define main so that gflags get processed. int main(int argc, char** argv) { testing::InitGoogleTest(&argc, argv); diff --git a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp index 5cb7b759259..9ccccff4d88 100644 --- a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp +++ b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp @@ -73,6 +73,34 @@ class ParquetTableScanTest : public HiveConnectorTestBase { assertQuery(plan, splits_, sql); } + void assertSelectWithFilter( + std::vector&& outputColumnNames, + const std::vector& subfieldFilters, + const std::string& remainingFilter, + const std::string& sql, + bool isFilterPushdownEnabled) { + auto rowType = getRowType(std::move(outputColumnNames)); + parse::ParseOptions options; + options.parseDecimalAsDouble = false; + + auto plan = PlanBuilder(pool_.get()) + .setParseOptions(options) + // Function extractFiltersFromRemainingFilter will extract + // filters to subfield filters, but for some types, filter + // pushdown is not supported. + .tableScan( + "hive_table", + rowType, + {}, + subfieldFilters, + remainingFilter, + nullptr, + isFilterPushdownEnabled) + .planNode(); + + assertQuery(plan, splits_, sql); + } + void assertSelectWithAgg( std::vector&& outputColumnNames, const std::vector& aggregates, @@ -443,6 +471,160 @@ TEST_F(ParquetTableScanTest, readAsLowerCase) { result.second, {makeRowVector({"a"}, {makeFlatVector({0, 1})})}); } +TEST_F(ParquetTableScanTest, structSelection) { + auto vector = makeArrayVector({{}}); + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({"first", "last"}, {VARCHAR(), VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT ('Janet', 'Jones')"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, + {ROW( + {"first", "middle", "last"}, {VARCHAR(), VARCHAR(), VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT ('Janet', null, 'Jones')"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({"first", "middle"}, {VARCHAR(), VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT ('Janet', null)"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({"middle", "last"}, {VARCHAR(), VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT (null, 'Jones')"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({"middle"}, {VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT row(null)"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({"middle", "info"}, {VARCHAR(), VARCHAR()})}), + makeRowVector( + {"t"}, + { + vector, + })); + assertSelectWithFilter({"name"}, {}, "", "SELECT NULL"); + + loadData( + getExampleFilePath("contacts.parquet"), + ROW({"name"}, {ROW({}, {})}), + makeRowVector( + {"t"}, + { + vector, + })); + + assertSelectWithFilter({"name"}, {}, "", "SELECT t from tmp"); +} + +TEST_F(ParquetTableScanTest, timestampFilter) { + // Timestamp-int96.parquet holds one column (t: TIMESTAMP) and + // 10 rows in one row group. Data is in SNAPPY compressed format. + // The values are: + // |t | + // +-------------------+ + // |2015-06-01 19:34:56| + // |2015-06-02 19:34:56| + // |2001-02-03 03:34:06| + // |1998-03-01 08:01:06| + // |2022-12-23 03:56:01| + // |1980-01-24 00:23:07| + // |1999-12-08 13:39:26| + // |2023-04-21 09:09:34| + // |2000-09-12 22:36:29| + // |2007-12-12 04:27:56| + // +-------------------+ + auto vector = makeFlatVector( + {Timestamp(1433116800, 70496000000000), + Timestamp(1433203200, 70496000000000), + Timestamp(981158400, 12846000000000), + Timestamp(888710400, 28866000000000), + Timestamp(1671753600, 14161000000000), + Timestamp(317520000, 1387000000000), + Timestamp(944611200, 49166000000000), + Timestamp(1682035200, 32974000000000), + Timestamp(968716800, 81389000000000), + Timestamp(1197417600, 16076000000000)}); + + loadData( + getExampleFilePath("timestamp_int96.parquet"), + ROW({"t"}, {TIMESTAMP()}), + makeRowVector( + {"t"}, + { + vector, + })); + + assertSelectWithFilter({"t"}, {}, "", "SELECT t from tmp", false); + assertSelectWithFilter( + {"t"}, + {}, + "t < TIMESTAMP '2000-09-12 22:36:29'", + "SELECT t from tmp where t < TIMESTAMP '2000-09-12 22:36:29'", + false); + assertSelectWithFilter( + {"t"}, + {}, + "t <= TIMESTAMP '2000-09-12 22:36:29'", + "SELECT t from tmp where t <= TIMESTAMP '2000-09-12 22:36:29'", + false); + assertSelectWithFilter( + {"t"}, + {}, + "t > TIMESTAMP '1980-01-24 00:23:07'", + "SELECT t from tmp where t > TIMESTAMP '1980-01-24 00:23:07'", + false); + assertSelectWithFilter( + {"t"}, + {}, + "t >= TIMESTAMP '1980-01-24 00:23:07'", + "SELECT t from tmp where t >= TIMESTAMP '1980-01-24 00:23:07'", + false); + assertSelectWithFilter( + {"t"}, + {}, + "t == TIMESTAMP '2022-12-23 03:56:01'", + "SELECT t from tmp where t == TIMESTAMP '2022-12-23 03:56:01'", + false); + VELOX_ASSERT_THROW( + assertSelectWithFilter( + {"t"}, + {"t < TIMESTAMP '2000-09-12 22:36:29'"}, + "", + "SELECT t from tmp where t < TIMESTAMP '2000-09-12 22:36:29'"), + "testInt128() is not supported"); +} + int main(int argc, char** argv) { testing::InitGoogleTest(&argc, argv); folly::init(&argc, &argv, false); diff --git a/velox/dwio/parquet/writer/Writer.cpp b/velox/dwio/parquet/writer/Writer.cpp index 002e2c42ce1..55a8ad91434 100644 --- a/velox/dwio/parquet/writer/Writer.cpp +++ b/velox/dwio/parquet/writer/Writer.cpp @@ -14,8 +14,6 @@ * limitations under the License. */ -#include "velox/vector/arrow/Bridge.h" - #include #include #include @@ -139,6 +137,19 @@ std::shared_ptr getArrowParquetWriterOptions( return properties->build(); } +namespace { + +void exportToArrow( + const TypePtr& type, + std::shared_ptr pool, + ArrowSchema& out, + const ArrowOptions& options) { + auto leafPool = pool->addLeafChild("parquet-write-schema-convert"); + exportToArrow(BaseVector::create(type, 0, leafPool.get()), out, options); +} + +} // namespace + Writer::Writer( std::unique_ptr sink, const WriterOptions& options, @@ -155,6 +166,22 @@ Writer::Writer( } else { flushPolicy_ = std::make_unique(); } + options_.timestampUnit = + static_cast(options.arrowBridgeTimestampUnit); + + if (options.schema) { + ArrowSchema arrowSchema; + exportToArrow(options.schema, pool_, arrowSchema, options_); + + PARQUET_ASSIGN_OR_THROW( + arrowContext_->schema, ::arrow::ImportSchema(&arrowSchema)); + + arrowContext_->stagingChunks.insert( + arrowContext_->stagingChunks.end(), + arrowContext_->schema->num_fields(), + std::vector>()); + } + arrowContext_->properties = getArrowParquetWriterOptions(options, flushPolicy_); } @@ -224,20 +251,18 @@ dwio::common::StripeProgress getStripeProgress( * This method assumes each input `ColumnarBatch` have same schema. */ void Writer::write(const VectorPtr& data) { - ArrowOptions options{.flattenDictionary = true, .flattenConstant = true}; ArrowArray array; ArrowSchema schema; - exportToArrow(data, array, generalPool_.get(), options); - exportToArrow(data, schema, options); + exportToArrow(data, array, generalPool_.get(), options_); + exportToArrow(data, schema, options_); PARQUET_ASSIGN_OR_THROW( auto recordBatch, ::arrow::ImportRecordBatch(&array, &schema)); if (!arrowContext_->schema) { arrowContext_->schema = recordBatch->schema(); - for (int colIdx = 0; colIdx < arrowContext_->schema->num_fields(); - colIdx++) { - arrowContext_->stagingChunks.push_back( - std::vector>()); - } + arrowContext_->stagingChunks.insert( + arrowContext_->stagingChunks.end(), + arrowContext_->schema->num_fields(), + std::vector>()); } auto bytes = data->estimateFlatSize(); @@ -288,6 +313,10 @@ parquet::WriterOptions getParquetOptions( if (options.compressionKind.has_value()) { parquetOptions.compression = options.compressionKind.value(); } + if (options.arrowBridgeTimestampUnit.has_value()) { + parquetOptions.arrowBridgeTimestampUnit = + options.arrowBridgeTimestampUnit.value(); + } return parquetOptions; } diff --git a/velox/dwio/parquet/writer/Writer.h b/velox/dwio/parquet/writer/Writer.h index e7d70b7a88f..4d7408fbbe8 100644 --- a/velox/dwio/parquet/writer/Writer.h +++ b/velox/dwio/parquet/writer/Writer.h @@ -25,6 +25,7 @@ #include "velox/dwio/common/WriterFactory.h" #include "velox/dwio/parquet/writer/arrow/util/Compression.h" #include "velox/vector/ComplexVector.h" +#include "velox/vector/arrow/Bridge.h" namespace facebook::velox::parquet { @@ -98,6 +99,8 @@ struct WriterOptions { // policy with the configs in its ctor. std::function()> flushPolicyFactory; std::shared_ptr codecOptions; + uint8_t arrowBridgeTimestampUnit = static_cast(TimestampUnit::kNano); + TypePtr schema; }; // Writes Velox vectors into a DataSink using Arrow Parquet writer. @@ -146,6 +149,7 @@ class Writer : public dwio::common::Writer { std::shared_ptr arrowContext_; std::unique_ptr flushPolicy_; + ArrowOptions options_{.flattenDictionary = true, .flattenConstant = true}; }; class ParquetWriterFactory : public dwio::common::WriterFactory { diff --git a/velox/exec/ArrowStream.cpp b/velox/exec/ArrowStream.cpp index 863e43f8ba2..d90734e842e 100644 --- a/velox/exec/ArrowStream.cpp +++ b/velox/exec/ArrowStream.cpp @@ -27,6 +27,8 @@ ArrowStream::ArrowStream( operatorId, arrowStreamNode->id(), "ArrowStream") { + options_.timestampUnit = static_cast( + driverCtx->queryConfig().arrowBridgeTimestampUnit()); arrowStream_ = arrowStreamNode->arrowStream(); } @@ -66,7 +68,7 @@ RowVectorPtr ArrowStream::getOutput() { // Convert Arrow Array into RowVector and return. return std::dynamic_pointer_cast( - importFromArrowAsOwner(arrowSchema, arrowArray, pool())); + importFromArrowAsOwner(arrowSchema, arrowArray, options_, pool())); } bool ArrowStream::isFinished() { diff --git a/velox/exec/ArrowStream.h b/velox/exec/ArrowStream.h index c35894d0d28..34225f5f44c 100644 --- a/velox/exec/ArrowStream.h +++ b/velox/exec/ArrowStream.h @@ -45,6 +45,7 @@ class ArrowStream : public SourceOperator { bool finished_ = false; std::shared_ptr arrowStream_; + ArrowOptions options_; }; } // namespace facebook::velox::exec diff --git a/velox/exec/GroupingSet.cpp b/velox/exec/GroupingSet.cpp index 22f5118b973..2bc658ba0a9 100644 --- a/velox/exec/GroupingSet.cpp +++ b/velox/exec/GroupingSet.cpp @@ -725,6 +725,7 @@ bool GroupingSet::getOutput( } if (hasSpilled()) { + spill(); return getOutputWithSpill(maxOutputRows, maxOutputBytes, result); } VELOX_CHECK(!isDistinct()); @@ -826,7 +827,7 @@ const HashLookup& GroupingSet::hashLookup() const { void GroupingSet::ensureInputFits(const RowVectorPtr& input) { // Spilling is considered if this is a final or single aggregation and // spillPath is set. - if (isPartial_ || spillConfig_ == nullptr) { + if (spillConfig_ == nullptr) { return; } @@ -911,7 +912,7 @@ void GroupingSet::ensureOutputFits() { // to reserve memory for the output as we can't reclaim much memory from this // operator itself. The output processing can reclaim memory from the other // operator or query through memory arbitration. - if (isPartial_ || spillConfig_ == nullptr || hasSpilled()) { + if (spillConfig_ == nullptr || hasSpilled()) { return; } @@ -961,7 +962,6 @@ void GroupingSet::spill() { if (table_ == nullptr || table_->numDistinct() == 0) { return; } - if (!hasSpilled()) { auto rows = table_->rows(); VELOX_DCHECK(pool_.trackUsage()); @@ -1051,7 +1051,16 @@ bool GroupingSet::getOutputWithSpill( if (merge_ == nullptr) { return false; } - return mergeNext(maxOutputRows, maxOutputBytes, result); + bool hasData = mergeNext(maxOutputRows, maxOutputBytes, result); + if (!hasData) { + // If spill has been finalized, reset merge stream and spiller. This would + // help partial aggregation replay the spilling procedure once needed again. + merge_ = nullptr; + mergeRows_ = nullptr; + mergeArgs_.clear(); + spiller_ = nullptr; + } + return hasData; } bool GroupingSet::mergeNext( diff --git a/velox/exec/Task.cpp b/velox/exec/Task.cpp index 9ec1f5ec157..32666e2221c 100644 --- a/velox/exec/Task.cpp +++ b/velox/exec/Task.cpp @@ -535,12 +535,6 @@ RowVectorPtr Task::next(ContinueFuture* future) { createSplitGroupStateLocked(kUngroupedGroupId); std::vector> drivers = createDriversLocked(kUngroupedGroupId); - if (pool_->stats().currentBytes != 0) { - VELOX_FAIL( - "Unexpected memory pool allocations during task[{}] driver initialization: {}", - taskId_, - pool_->treeMemoryUsage()); - } drivers_ = std::move(drivers); } @@ -704,12 +698,6 @@ void Task::createAndStartDrivers(uint32_t concurrentSplitGroups) { // Create drivers. std::vector> drivers = createDriversLocked(kUngroupedGroupId); - if (pool_->stats().currentBytes != 0) { - VELOX_FAIL( - "Unexpected memory pool allocations during task[{}] driver initialization: {}", - taskId_, - pool_->treeMemoryUsage()); - } // Prevent the connecting structures from being cleaned up before all // split groups are finished during the grouped execution mode. @@ -839,9 +827,16 @@ void Task::resume(std::shared_ptr self) { continue; } VELOX_CHECK(!driver->isOnThread() && !driver->isTerminated()); - if (!driver->state().hasBlockingFuture) { + if (!driver->state().hasBlockingFuture && + driver->task()->queryCtx()->isExecutorSupplied()) { // Do not continue a Driver that is blocked on external // event. The Driver gets enqueued by the promise realization. + // + // Do not continue the driver if no executor is supplied, + // Since it's likely that we are in single-thread execution. + // + // 2023/07.13 Hongze: Is there a way to hide the execution model + // (single or async) from here? Driver::enqueue(driver); } } diff --git a/velox/exec/tests/ArrowStreamTest.cpp b/velox/exec/tests/ArrowStreamTest.cpp index f0fe9b37e04..1b450e7200a 100644 --- a/velox/exec/tests/ArrowStreamTest.cpp +++ b/velox/exec/tests/ArrowStreamTest.cpp @@ -45,7 +45,7 @@ class ArrowStreamTest : public OperatorTestBase { int getNext(struct ArrowArray* outArray) { if (vectorIndex_ < vectors_.size()) { - exportToArrow(vectors_[vectorIndex_], *outArray, pool_.get()); + exportToArrow(vectors_[vectorIndex_], *outArray, pool_.get(), options_); vectorIndex_ += 1; } else { // End of stream. Mark the array released. @@ -56,12 +56,13 @@ class ArrowStreamTest : public OperatorTestBase { } int getArrowSchema(ArrowSchema& out) { - exportToArrow(BaseVector::create(type_, 0, pool_.get()), out); + exportToArrow(BaseVector::create(type_, 0, pool_.get()), out, options_); return failGetSchema_ ? (int)ErrorCode::kGetSchemaFailed : (int)ErrorCode::kNoError; } private: + ArrowOptions options_; const std::shared_ptr pool_; const std::vector& vectors_; const TypePtr type_; diff --git a/velox/exec/tests/SharedArbitratorTest.cpp b/velox/exec/tests/SharedArbitratorTest.cpp index 059356037c9..aa77df460a6 100644 --- a/velox/exec/tests/SharedArbitratorTest.cpp +++ b/velox/exec/tests/SharedArbitratorTest.cpp @@ -1184,6 +1184,50 @@ TEST_F(SharedArbitrationTest, reclaimFromDistinctAggregation) { waitForAllTasksToBeDeleted(); } +TEST_F(SharedArbitrationTest, reclaimFromPartialAggregation) { + const uint64_t maxQueryCapacity = 20L << 20; + std::vector vectors = newVectors(1024, maxQueryCapacity * 2); + createDuckDbTable(vectors); + const auto spillDirectory = exec::test::TempDirectoryPath::create(); + core::PlanNodeId partialAggNodeId; + core::PlanNodeId finalAggNodeId; + std::shared_ptr queryCtx = newQueryCtx(maxQueryCapacity); + auto task = + AssertQueryBuilder(duckDbQueryRunner_) + .spillDirectory(spillDirectory->path) + .config(core::QueryConfig::kSpillEnabled, "true") + .config(core::QueryConfig::kPartialAggregationSpillEnabled, "true") + .config(core::QueryConfig::kAggregationSpillEnabled, "true") + .config( + core::QueryConfig::kMaxPartialAggregationMemory, + std::to_string(1LL << 30)) // disable flush + .config( + core::QueryConfig::kMaxExtendedPartialAggregationMemory, + std::to_string(1LL << 30)) // disable flush + .config( + core::QueryConfig::kAbandonPartialAggregationMinPct, + "200") // avoid abandoning + .config( + core::QueryConfig::kAbandonPartialAggregationMinRows, + std::to_string(1LL << 30)) // avoid abandoning + .queryCtx(queryCtx) + .plan(PlanBuilder() + .values(vectors) + .partialAggregation({"c0"}, {"count(1)"}) + .capturePlanNodeId(partialAggNodeId) + .finalAggregation() + .capturePlanNodeId(finalAggNodeId) + .planNode()) + .assertResults("SELECT c0, count(1) FROM tmp GROUP BY c0"); + auto taskStats = exec::toPlanStats(task->taskStats()); + auto& partialStats = taskStats.at(partialAggNodeId); + auto& finalStats = taskStats.at(finalAggNodeId); + ASSERT_GT(partialStats.spilledBytes, 0); + ASSERT_GT(finalStats.spilledBytes, 0); + task.reset(); + waitForAllTasksToBeDeleted(); +} + DEBUG_ONLY_TEST_F(SharedArbitrationTest, reclaimFromAggregationOnNoMoreInput) { const int numVectors = 32; std::vector vectors; diff --git a/velox/exec/tests/TaskTest.cpp b/velox/exec/tests/TaskTest.cpp index ee78430f361..7df6db66669 100644 --- a/velox/exec/tests/TaskTest.cpp +++ b/velox/exec/tests/TaskTest.cpp @@ -1289,7 +1289,7 @@ DEBUG_ONLY_TEST_F(TaskTest, raceBetweenTaskPauseAndTerminate) { taskThread.join(); } -TEST_F(TaskTest, driverCreationMemoryAllocationCheck) { +TEST_F(TaskTest, DISABLED_driverCreationMemoryAllocationCheck) { exec::Operator::registerOperator(std::make_unique()); auto data = makeRowVector({ makeFlatVector(1'000, [](auto row) { return row; }), diff --git a/velox/exec/tests/utils/PlanBuilder.cpp b/velox/exec/tests/utils/PlanBuilder.cpp index f086991debd..ab7f72a4437 100644 --- a/velox/exec/tests/utils/PlanBuilder.cpp +++ b/velox/exec/tests/utils/PlanBuilder.cpp @@ -97,12 +97,14 @@ PlanBuilder& PlanBuilder::tableScan( const std::unordered_map& columnAliases, const std::vector& subfieldFilters, const std::string& remainingFilter, - const RowTypePtr& dataColumns) { + const RowTypePtr& dataColumns, + const bool isFilterPushdownEnabled) { return TableScanBuilder(*this) .tableName(tableName) .outputType(outputType) .columnAliases(columnAliases) .subfieldFilters(subfieldFilters) + .isFilterPushdownEnabled(isFilterPushdownEnabled) .remainingFilter(remainingFilter) .dataColumns(dataColumns) .endTableScan(); @@ -200,7 +202,7 @@ core::PlanNodePtr PlanBuilder::TableScanBuilder::build(core::PlanNodeId id) { tableHandle_ = std::make_shared( connectorId_, tableName_, - true, + isFilterPushdownEnabled_, std::move(filters), remainingFilterExpr, dataColumns_); diff --git a/velox/exec/tests/utils/PlanBuilder.h b/velox/exec/tests/utils/PlanBuilder.h index e311c0d441d..29917eac876 100644 --- a/velox/exec/tests/utils/PlanBuilder.h +++ b/velox/exec/tests/utils/PlanBuilder.h @@ -144,7 +144,8 @@ class PlanBuilder { const std::unordered_map& columnAliases = {}, const std::vector& subfieldFilters = {}, const std::string& remainingFilter = "", - const RowTypePtr& dataColumns = nullptr); + const RowTypePtr& dataColumns = nullptr, + bool isFilterPushdownEnabled = true); /// Add a TableScanNode to scan a TPC-H table. /// @@ -209,6 +210,11 @@ class PlanBuilder { return *this; } + TableScanBuilder& isFilterPushdownEnabled(bool isFilterPushdownEnabled) { + isFilterPushdownEnabled_ = std::move(isFilterPushdownEnabled); + return *this; + } + /// @param dataColumns can be different from 'outputType' for the purposes /// of testing queries using missing columns. It is used, if specified, for /// parseExpr call and as 'dataColumns' for the TableHandle. You supply more @@ -269,6 +275,7 @@ class PlanBuilder { std::shared_ptr tableHandle_; std::unordered_map> assignments_; + bool isFilterPushdownEnabled_; }; /// Start a TableScanBuilder. diff --git a/velox/expression/CastExpr-inl.h b/velox/expression/CastExpr-inl.h index 2b3139d19f9..b4598136e1e 100644 --- a/velox/expression/CastExpr-inl.h +++ b/velox/expression/CastExpr-inl.h @@ -51,6 +51,22 @@ inline std::exception_ptr makeBadCastException( false)); } +/// Represent the varchar fragment. +/// +/// For example: +/// | value | wholeDigits | fractionalDigits | exponent | sign +/// | 9999999999.99 | 9999999999 | 99 | nullopt | 1 +/// | 15 | 15 | | nullopt | 1 +/// | 1.5 | 1 | 5 | nullopt | 1 +/// | -1.5 | 1 | 5 | nullopt | -1 +/// | 31.523e-2 | 31 | 523 | -2 | 1 +struct DecimalComponents { + std::string_view wholeDigits; + std::string_view fractionalDigits; + std::optional exponent = std::nullopt; + int8_t sign = 1; +}; + // Copied from format.h of fmt. inline int countDigits(uint128_t n) { int count = 1; @@ -132,6 +148,215 @@ StringView convertToStringView( return StringView(startPosition, writePosition - startPosition); } +size_t parseDigitsRun( + const char* s, + size_t start, + size_t size, + std::string_view& out) { + size_t pos = start; + for (; pos < size; ++pos) { + if (!std::isdigit(s[pos])) { + break; + } + } + out = std::string_view(s + start, pos - start); + return pos; +} + +std::optional parseDecimalComponents( + const char* s, + size_t size) { + if (size == 0) { + return std::nullopt; + } + DecimalComponents out; + size_t pos = 0; + // Sign of the number. + if (s[pos] == '-') { + out.sign = -1; + ++pos; + } else if (s[pos] == '+') { + out.sign = 1; + ++pos; + } + // First run of digits. + pos = parseDigitsRun(s, pos, size, out.wholeDigits); + if (pos == size) { + return out.wholeDigits.empty() ? std::nullopt + : std::optional(out); + } + // Optional dot (if given in fractional form). + if (s[pos] == '.') { + // Second run of digits. + ++pos; + pos = parseDigitsRun(s, pos, size, out.fractionalDigits); + } + if (out.wholeDigits.empty() && out.fractionalDigits.empty()) { + // Need at least some digits (whole or fractional). + return std::nullopt; + } + if (pos == size) { + return out; + } + // Optional exponent. + if (s[pos] == 'e' || s[pos] == 'E') { + ++pos; + if (pos != size && s[pos] == '+') { + ++pos; + } + folly::StringPiece p = {s + pos, size - pos}; + auto tryExp = + folly::tryTo(folly::StringPiece(s + pos, size - pos)); + if (tryExp.hasError()) { + return std::nullopt; + } + out.exponent = tryExp.value(); + return out; + } + return pos == size ? std::optional(out) : std::nullopt; +} + +/// Multiple out by the appropriate power of 10 necessary to add source parsed +/// as int128_t and then adds the parsed value of source. +bool shiftAndAdd(std::string_view input, int128_t& out) { + auto length = input.size(); + if (length == 0) { + return true; + } + + bool overflow = + __builtin_mul_overflow(out, DecimalUtil::kPowersOfTen[length], &out); + if (overflow) { + return false; + } + auto tryValue = + folly::tryTo(folly::StringPiece(input.data(), length)); + if (tryValue.hasError()) { + return false; + } + + overflow = __builtin_add_overflow(out, tryValue.value(), &out); + VELOX_DCHECK(!overflow) + return true; +} + +/// Derives from Arrow function DecimalFromString. +/// Arrow implementation: +/// https://github.com/apache/arrow/blob/main/cpp/src/arrow/util/decimal.cc#L637 +/// +/// Firstly, it will parse the varchar to DecimalComponents which contains the +/// message that can represent a value. Secondly, process the exponent to get +/// the value parsedScale. Thirdly, compute the rescaled value. +/// The caller should test if `error` is empty +template +std::optional rescaleVarchar( + const StringView s, + int toPrecision, + int toScale, + std::string& error) { + auto decimalComponentsOpt = parseDecimalComponents(s.data(), s.size()); + if (!decimalComponentsOpt.has_value()) { + error = "Value is not a number."; + return std::nullopt; + } + auto decimalComponents = decimalComponentsOpt.value(); + + // Count number of significant digits (without leading zeros). + size_t firstNonZero = decimalComponents.wholeDigits.find_first_not_of('0'); + size_t significantDigits = decimalComponents.fractionalDigits.size(); + if (firstNonZero != std::string::npos) { + significantDigits += decimalComponents.wholeDigits.size() - firstNonZero; + } + int32_t parsedPrecision = static_cast(significantDigits); + + int32_t parsedScale = 0; + bool addOne = false; + int32_t fractionalDigitsSize = decimalComponents.fractionalDigits.size(); + if (decimalComponents.exponent.has_value()) { + auto adjustedExponent = decimalComponents.exponent.value(); + parsedScale = -adjustedExponent + fractionalDigitsSize; + // Truncate the fractionalDigits. + if (parsedScale > toScale) { + // adjustedExponent is negative, fractionalDigits only consider the last + // digit to round up. + if (-adjustedExponent >= toScale) { + if (fractionalDigitsSize > 0 && + decimalComponents.fractionalDigits[0] >= '5') { + addOne = true; + } + decimalComponents.fractionalDigits = ""; + parsedScale -= fractionalDigitsSize; + } else { + auto reduceDigits = adjustedExponent + toScale; + if (fractionalDigitsSize > reduceDigits && + decimalComponents.fractionalDigits[reduceDigits] >= '5') { + addOne = true; + } + decimalComponents.fractionalDigits = std::string_view( + decimalComponents.fractionalDigits.data(), + std::min(reduceDigits, fractionalDigitsSize)); + parsedScale -= + fractionalDigitsSize - decimalComponents.fractionalDigits.size(); + } + } + } else { + if (fractionalDigitsSize > toScale) { + if (decimalComponents.fractionalDigits[toScale] >= '5') { + addOne = true; + } + parsedScale = toScale; + decimalComponents.fractionalDigits = + std::string_view(decimalComponents.fractionalDigits.data(), toScale); + } else { + parsedScale = fractionalDigitsSize; + } + } + + int128_t out = 0; + if (!shiftAndAdd(decimalComponents.wholeDigits, out)) { + error = "Value too large."; + return std::nullopt; + } + + if (!shiftAndAdd(decimalComponents.fractionalDigits, out)) { + error = "Value too large."; + return std::nullopt; + } + if (addOne) { + bool overflow = __builtin_add_overflow(out, 1, &out); + if (UNLIKELY(overflow)) { + error = "Value too large."; + return std::nullopt; + } + } + out = out * decimalComponents.sign; + + if (parsedScale < 0) { + /// Force the scale to zero, to avoid negative scales (due to + /// compatibility issues with external systems such as databases). + if (-parsedScale + toScale > LongDecimalType::kMaxScale) { + error = "Value too large."; + return std::nullopt; + } + + bool overflow = __builtin_mul_overflow( + out, DecimalUtil::kPowersOfTen[-parsedScale + toScale], &out); + if (UNLIKELY(overflow)) { + error = "Value too large."; + return std::nullopt; + } + parsedPrecision -= parsedScale; + parsedScale = toScale; + } + bool overflow = false; + auto rescaledValue = DecimalUtil::rescaleWithRoundUp( + out, parsedPrecision, parsedScale, toPrecision, toScale, overflow, false); + if (overflow) { + error = "Value too large."; + return std::nullopt; + } + return rescaledValue; +} } // namespace template @@ -268,12 +493,14 @@ void CastExpr::applyDecimalCastKernel( applyToSelectedNoThrowLocal( context, rows, castResult, [&](vector_size_t row) { + bool overflow = false; auto rescaledValue = DecimalUtil::rescaleWithRoundUp( sourceVector->valueAt(row), fromPrecisionScale.first, fromPrecisionScale.second, toPrecisionScale.first, - toPrecisionScale.second); + toPrecisionScale.second, + overflow); if (rescaledValue.has_value()) { castResultRawBuffer[row] = rescaledValue.value(); } else { @@ -307,6 +534,80 @@ void CastExpr::applyIntToDecimalCastKernel( }); } +template +void CastExpr::applyVarcharToDecimalCastKernel( + const SelectivityVector& rows, + const BaseVector& input, + exec::EvalCtx& context, + const TypePtr& toType, + VectorPtr& result) { + auto sourceVector = input.as>(); + auto rawBuffer = result->asUnchecked>()->mutableRawValues(); + const auto toPrecisionScale = getDecimalPrecisionScale(*toType); + auto setError = [&](vector_size_t row, const std::string& details) { + if (setNullInResultAtError()) { + result->setNull(row, true); + } else { + context.setVeloxExceptionError( + row, makeBadCastException(toType, input, row, details)); + } + }; + + rows.applyToSelected([&](auto row) { + std::string error; + auto rescaledValue = rescaleVarchar( + sourceVector->valueAt(row), + toPrecisionScale.first, + toPrecisionScale.second, + error); + if (!error.empty()) { + setError(row, error); + } else if (rescaledValue.has_value()) { + rawBuffer[row] = rescaledValue.value(); + } else { + result->setNull(row, true); + } + }); +} + +template +void CastExpr::applyDoubleToDecimal( + const SelectivityVector& rows, + const BaseVector& input, + exec::EvalCtx& context, + const TypePtr& toType, + VectorPtr& castResult) { + auto sourceVector = input.as>(); + auto rawResults = + castResult->asUnchecked>()->mutableRawValues(); + const auto toPrecisionScale = getDecimalPrecisionScale(*toType); + applyToSelectedNoThrowLocal( + context, rows, castResult, [&](vector_size_t row) { + if (sourceVector->isNullAt(row)) { + castResult->setNull(row, true); + return; + } + std::string error; + auto rescaledValue = DecimalUtil::rescaleDouble( + sourceVector->valueAt(row), + toPrecisionScale.first, + toPrecisionScale.second, + error); + if (!error.empty()) { + if (setNullInResultAtError()) { + castResult->setNull(row, true); + } else { + context.setVeloxExceptionError( + row, makeBadCastException(toType, input, row, error)); + } + } else if (rescaledValue.has_value()) { + rawResults[row] = rescaledValue.value(); + } else { + castResult->setNull(row, true); + } + }); +} + template VectorPtr CastExpr::applyDecimalToFloatCast( const SelectivityVector& rows, diff --git a/velox/expression/CastExpr.cpp b/velox/expression/CastExpr.cpp index 2ade0a28b30..e5a5785458b 100644 --- a/velox/expression/CastExpr.cpp +++ b/velox/expression/CastExpr.cpp @@ -467,6 +467,14 @@ VectorPtr CastExpr::applyDecimal( applyIntToDecimalCastKernel( rows, input, context, toType, castResult); break; + case TypeKind::REAL: + applyDoubleToDecimal( + rows, input, context, toType, castResult); + break; + case TypeKind::DOUBLE: + applyDoubleToDecimal( + rows, input, context, toType, castResult); + break; case TypeKind::BIGINT: { if (fromType->isShortDecimal()) { applyDecimalCastKernel( @@ -485,6 +493,10 @@ VectorPtr CastExpr::applyDecimal( } [[fallthrough]]; } + case TypeKind::VARCHAR: + applyVarcharToDecimalCastKernel( + rows, input, context, toType, castResult); + break; default: VELOX_UNSUPPORTED( "Cast from {} to {} is not supported", diff --git a/velox/expression/CastExpr.h b/velox/expression/CastExpr.h index 6da2eca1d72..1f6a5168d5e 100644 --- a/velox/expression/CastExpr.h +++ b/velox/expression/CastExpr.h @@ -198,6 +198,22 @@ class CastExpr : public SpecialForm { const TypePtr& toType, VectorPtr& castResult); + template + void applyVarcharToDecimalCastKernel( + const SelectivityVector& rows, + const BaseVector& input, + exec::EvalCtx& context, + const TypePtr& toType, + VectorPtr& castResult); + + template + void applyDoubleToDecimal( + const SelectivityVector& rows, + const BaseVector& input, + exec::EvalCtx& context, + const TypePtr& toType, + VectorPtr& castResult); + template VectorPtr applyDecimalToFloatCast( const SelectivityVector& rows, diff --git a/velox/expression/tests/CastExprTest.cpp b/velox/expression/tests/CastExprTest.cpp index fcb4ce73cc9..b5a77c8ae23 100644 --- a/velox/expression/tests/CastExprTest.cpp +++ b/velox/expression/tests/CastExprTest.cpp @@ -1911,6 +1911,7 @@ TEST_F(CastExprTest, decimalToVarchar) { "c0", longFlatForZero, makeNullableFlatVector({"0"})); } +/* TEST_F(CastExprTest, decimalToDecimal) { // short to short, scale up. auto shortFlat = @@ -2018,15 +2019,15 @@ TEST_F(CastExprTest, decimalToDecimal) { makeNullableFlatVector( {DecimalUtil::kLongDecimalMax}, DECIMAL(38, 0)), makeNullableFlatVector({0}, DECIMAL(38, 1))), - "Cannot cast DECIMAL '99999999999999999999999999999999999999' to DECIMAL(38, 1)"); - VELOX_ASSERT_THROW( - testComplexCast( - "c0", + "Cannot cast DECIMAL '99999999999999999999999999999999999999' to +DECIMAL(38, 1)"); VELOX_ASSERT_THROW( testComplexCast( "c0", makeNullableFlatVector( {DecimalUtil::kLongDecimalMin}, DECIMAL(38, 0)), makeNullableFlatVector({0}, DECIMAL(38, 1))), - "Cannot cast DECIMAL '-99999999999999999999999999999999999999' to DECIMAL(38, 1)"); + "Cannot cast DECIMAL '-99999999999999999999999999999999999999' to +DECIMAL(38, 1)"); } +*/ TEST_F(CastExprTest, integerToDecimal) { testIntToDecimalCasts(); @@ -2059,6 +2060,233 @@ TEST_F(CastExprTest, boolToDecimal) { DECIMAL(20, 10))); } +// The result is obtained by select cast('31.4e-2' as decimal(12, 2)). +/* +TEST_F(CastExprTest, varcharToDecimal) { + auto input = makeFlatVector( + {"9999999999.99", + "15", + "1.5", + "-1.5", + "1.556", + "1.554", + ("1.556" + std::string(32, '1')).data(), + ("1.556" + std::string(32, '9')).data(), + "0000.123", + ".12300000000", + "+09", + "9.", + ".9", + "3E2", + "-3E+2", + "3E+2", + "3E-2", + "3e+2", + "3e-2", + "3.5E-2", + "3.4E-2", + "3.5E+2", + "3.4E+2", + "31.423e+2", + "31.423e-2", + "31.523e-2"}); + testComplexCast( + "c0", + input, + makeFlatVector( + {999'999'999'999, + 1500, + 150, + -150, + 156, + 155, + 156, + 156, + 12, + 12, + 900, + 900, + 90, + 30000, + -30000, + 30000, + 3, + 30000, + 3, + 4, + 3, + 35000, + 34000, + 314230, + 31, + 32}, + DECIMAL(12, 2))); + + // Truncate the fractional digits with exponent. + testComplexCast( + "c0", + makeFlatVector( + {"112345612.23e-6", + "112345662.23e-6", + "1.23e-6", + "1.23e-3", + "1.26e-3", + "1.23456781e3", + "1.23456789e3", + "1.23456789123451789123456789e9", + "1.23456789123456789123456789e9"}), + makeFlatVector( + {1123456, + 1123457, + 0, + 12, + 13, + 12345678, + 12345679, + 12345678912345, + 12345678912346}, + DECIMAL(20, 4))); + + auto minDecimalStr = '-' + std::string(36, '9') + '.' + "99"; + auto maxDecimalStr = std::string(36, '9') + '.' + "99"; + testComplexCast( + "c0", + makeFlatVector( + {StringView(minDecimalStr), + StringView(maxDecimalStr), + "123456789012345678901234.567"}), + makeFlatVector( + { + DecimalUtil::kLongDecimalMin, + DecimalUtil::kLongDecimalMax, + HugeInt::build( + 669260, 10962463713375599297U), // 12345678901234567890123457 + }, + DECIMAL(38, 2))); + + std::string fractionLarge = "1.9" + std::string(67, '9'); + std::string fractionLargeExp = "1.9" + std::string(67, '9') + "e2"; + std::string fractionLargeNegExp = "1000.9" + std::string(67, '9') + "e-2"; + testComplexCast( + "c0", + makeFlatVector( + {StringView(('-' + std::string(38, '9')).data()), + StringView(std::string(38, '9').data()), + StringView(fractionLarge.data()), + StringView(fractionLargeExp.data()), + StringView(fractionLargeNegExp.data())}), + makeFlatVector( + {DecimalUtil::kLongDecimalMin, + DecimalUtil::kLongDecimalMax, + 2, + 200, + 10}, + DECIMAL(38, 0))); + std::string fractionRoundDown = "0." + std::string(38, '9') + "2"; + std::string fractionRoundDownExp = "99." + std::string(36, '9') + "2e-2"; + testComplexCast( + "c0", + makeFlatVector( + {StringView(fractionRoundDown), StringView(fractionRoundDownExp)}), + makeConstant(DecimalUtil::kLongDecimalMax, 2, DECIMAL(38, 38))); + + // WholeDigits shiftAndAdd overflow. + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant(std::string(280, '9').data(), 1), + makeConstant(1, 1, DECIMAL(38, 0))), + fmt::format( + "Cannot cast VARCHAR '{}' to DECIMAL(38, 0). Value too large.", + std::string(280, '9'))) + // Function shiftAndAdd shift fractionalDigits overflow. + std::string shiftFractionOverflow = std::string(36, '9') + '.' + "23456"; + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant(shiftFractionOverflow.data(), 1), + makeConstant(2, 1, DECIMAL(38, 10))), + fmt::format( + "Cannot cast VARCHAR '{}' to DECIMAL(38, 10). Value too large.", + shiftFractionOverflow)) + std::string fractionRoundUp = "0." + std::string(38, '9') + "6"; + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant(fractionRoundUp.data(), 1), + makeConstant(3, 1, DECIMAL(38, 38))), + fmt::format( + "Cannot cast VARCHAR '{}' to DECIMAL(38, 38). Value too large.", + fractionRoundUp)) + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("0.0444a", 1), + makeConstant(4, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '0.0444a' to DECIMAL(38, 0). Value is not a number") + + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("", 1), + makeConstant(5, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '' to DECIMAL(38, 0). Value is not a number") + + // exponent parsedScale > LongDecimalType::kMaxScale. + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("1.23e67", 1), + makeConstant(6, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '1.23e67' to DECIMAL(38, 0). Value too large.") + + // Out * DecimalUtil::kPowersOfTen[-parsedScale] overflow. + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("20908.23e35", 1), + makeConstant(7, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '20908.23e35' to DECIMAL(38, 0). Value too large.") + + // Rescale overflow. + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("111111111111111111.23", 1), + makeConstant(8, 1, DECIMAL(38, 38))), + "Cannot cast VARCHAR '111111111111111111.23' to DECIMAL(38, 38). Value too +large.") + + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("23e-5d", 1), + makeConstant(9, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '23e-5d' to DECIMAL(38, 0). Value is not a number") + + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("1. 23", 1), + makeConstant(10, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '1. 23' to DECIMAL(38, 0). Value is not a number") + + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant("1.23 ", 1), + makeConstant(11, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR '1.23 ' to DECIMAL(38, 0). Value is not a number") + + VELOX_ASSERT_THROW( + testComplexCast( + "c0", + makeConstant(" 1.23 ", 1), + makeConstant(12, 1, DECIMAL(38, 0))), + "Cannot cast VARCHAR ' 1.23 ' to DECIMAL(38, 0). Value is not a number") +} +*/ + TEST_F(CastExprTest, castInTry) { // Test try(cast(array(varchar) as array(bigint))) whose input vector is // wrapped in dictinary encoding. The row of ["2a"] should trigger an error diff --git a/velox/functions/lib/Re2Functions.cpp b/velox/functions/lib/Re2Functions.cpp index 39c3f79e8b7..a6b963f2b47 100644 --- a/velox/functions/lib/Re2Functions.cpp +++ b/velox/functions/lib/Re2Functions.cpp @@ -1056,7 +1056,7 @@ class PatternStringIterator { // The char follows escapeChar can only be one of (%, _, escapeChar). if (currentChar == escapeChar_ || currentChar == '_' || currentChar == '%') { - charKind_ = CharKind::kNormal; + charKind_ = CharKind::kEscaped; } else { VELOX_USER_FAIL( "Escape character must be followed by '%', '_' or the escape character itself: {}, escape {}", @@ -1087,6 +1087,10 @@ class PatternStringIterator { return charKind_ == CharKind::kSingleCharWildcard; } + bool isEscaped() { + return charKind_ == CharKind::kEscaped; + } + bool isWildcard() { return isAnyCharsWildcard() || isSingleCharWildcard(); } @@ -1106,8 +1110,10 @@ class PatternStringIterator { // NOTE: If escape char is set as '\', for pattern '\__', the first '_' is // not a wildcard, just a literal '_', the second '_' is a wildcard. kSingleCharWildcard, - // Chars that are not escape char & not wildcard char. - kNormal + // Char that is not escape char & not wildcard char. + kNormal, + // Char that was escaped by the lhs escape char. + kEscaped }; // Char at current cursor. @@ -1170,7 +1176,16 @@ PatternMetadata determinePatternKind( } else { // Record the first fixed pattern start. if (fixedPatternStart == -1) { - fixedPatternStart = iterator.currentIndex(); + if (iterator.isEscaped()) { + // We should include escape chars in the fixed pattern. Otherwise, + // a pattern like '%\\abc%' would produce '\abc' as fixed pattern + // which could lead to failure when trying to unescape the fixed + // pattern. + VELOX_CHECK_GT(iterator.currentIndex(), 0) + fixedPatternStart = iterator.currentIndex() - 1; + } else { + fixedPatternStart = iterator.currentIndex(); + } } else { // This is not the first fixed pattern, not supported, so fallback. if (iterator.isPreviousWildcard()) { diff --git a/velox/functions/lib/aggregates/AverageAggregateBase.cpp b/velox/functions/lib/aggregates/AverageAggregateBase.cpp index efef798b620..3353caed48b 100644 --- a/velox/functions/lib/aggregates/AverageAggregateBase.cpp +++ b/velox/functions/lib/aggregates/AverageAggregateBase.cpp @@ -21,14 +21,16 @@ namespace facebook::velox::functions::aggregate { void checkAvgIntermediateType(const TypePtr& type) { VELOX_USER_CHECK( type->isRow() || type->isVarbinary(), - "Input type for final average must be row type or varbinary type."); + "Input type for final average must be row type or varbinary type, find {}", + type->toString()); if (type->kind() == TypeKind::VARBINARY) { return; } VELOX_USER_CHECK( type->childAt(0)->kind() == TypeKind::DOUBLE || type->childAt(0)->isLongDecimal(), - "Input type for sum in final average must be double or long decimal type.") + "Input type for sum in final average must be double or long decimal type, find {}", + type->childAt(0)->toString()); VELOX_USER_CHECK_EQ( type->childAt(1)->kind(), TypeKind::BIGINT, diff --git a/velox/functions/lib/aggregates/BitwiseAggregateBase.h b/velox/functions/lib/aggregates/BitwiseAggregateBase.h index 428b905ea83..02f21c02ac8 100644 --- a/velox/functions/lib/aggregates/BitwiseAggregateBase.h +++ b/velox/functions/lib/aggregates/BitwiseAggregateBase.h @@ -70,7 +70,10 @@ class BitwiseAggregateBase : public SimpleNumericAggregate { }; template