Merge branch 'main' of https://github.com/acvictor/velox into acvicto…

…r/updateFileHandleGen

.github/workflows/linux-build.yml

@@ -91,7 +91,7 @@ jobs:
       - name: Make Release Build
         env:
           MAKEFLAGS: 'NUM_THREADS=8 MAX_HIGH_MEM_JOBS=4 MAX_LINK_JOBS=4'
-          CUDA_ARCHITECTURES: 60
+          CUDA_ARCHITECTURES: 70
           CUDA_COMPILER: /usr/local/cuda-${CUDA_VERSION}/bin/nvcc
           # Without that, nvcc picks /usr/bin/c++ which is GCC 8
           CUDA_FLAGS: "-ccbin /opt/rh/gcc-toolset-9/root/usr/bin"

.github/workflows/scheduled.yml

@@ -264,14 +264,14 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: presto
-          path: velox/_build/debug/velox/expression/tests/velox_expression_fuzzer_test
+          path: velox/_build/debug/velox/expression/fuzzer/velox_expression_fuzzer_test
           retention-days: "${{ env.RETENTION }}"
 
       - name: Upload spark expression fuzzer
         uses: actions/upload-artifact@v4
         with:
           name: spark_expression_fuzzer
-          path: velox/_build/debug/velox/expression/tests/spark_expression_fuzzer_test
+          path: velox/_build/debug/velox/expression/fuzzer/spark_expression_fuzzer_test
           retention-days: "${{ env.RETENTION }}"
 
       - name: Upload spark aggregation fuzzer

Makefile

@@ -161,7 +161,7 @@ unittest: debug			#: Build with debugging and run unit tests
 # Build with debugging and run expression fuzzer test. Use a fixed seed to
 # ensure the tests are reproducible.
 fuzzertest: debug
-	$(BUILD_BASE_DIR)/debug/velox/expression/tests/velox_expression_fuzzer_test \
+	$(BUILD_BASE_DIR)/debug/velox/expression/fuzzer/velox_expression_fuzzer_test \
 			--seed $(FUZZER_SEED) \
 			--duration_sec $(FUZZER_DURATION_SEC) \
 			--repro_persist_path $(FUZZER_REPRO_PERSIST_PATH) \

README.md

@@ -72,11 +72,8 @@ Blog posts are available [here](https://velox-lib.io/blog).
 
 ### Get the Velox Source
 ```
-git clone --recursive https://github.com/facebookincubator/velox.git
+git clone https://github.com/facebookincubator/velox.git
 cd velox
-# if you are updating an existing checkout
-git submodule sync --recursive
-git submodule update --init --recursive
 ```
 Once Velox is checked out, the first step is to install the dependencies.
 Details on the dependencies and how Velox manages some of them for you
@@ -90,7 +87,7 @@ dependencies for a given platform.
 On an Intel MacOS machine you can setup and then build like so:
 
 ```shell
-$ ./scripts/setup-macos.sh 
+$ ./scripts/setup-macos.sh
 $ make
 ```
 
@@ -117,7 +114,7 @@ $ CPU_TARGET="aarch64" make
 Once you have checked out Velox, you can setup and build like so:
 
 ```shell
-$ ./scripts/setup-ubuntu.sh 
+$ ./scripts/setup-ubuntu.sh
 $ make
 ```
 
@@ -135,7 +132,7 @@ Note that,
   * f16c
 * Velox tries to use the following (or equivalent) instruction sets where available:
   * On Intel CPUs
-    * avx  
+    * avx
     * avx2
     * sse
   * On ARM
@@ -167,7 +164,7 @@ contribute to the project.
 ## Community
 
 The main communication channel with the Velox OSS community is through the
-[the Velox-OSS Slack workspace](http://velox-oss.slack.com). 
+[the Velox-OSS Slack workspace](http://velox-oss.slack.com).
 Please reach out to **[email protected]** to get access to Velox Slack Channel.
 
 

velox/common/base/CMakeLists.txt

@@ -22,6 +22,7 @@ add_library(
   BitUtil.cpp
   Counters.cpp
   Fs.cpp
+  PeriodicStatsReporter.cpp
   RandomUtil.cpp
   RawVector.cpp
   RuntimeMetrics.cpp

velox/common/base/Counters.cpp

@@ -48,6 +48,24 @@ void registerVeloxMetrics() {
 
   /// ================== Memory Arbitration Counters =================
 
+  // The number of arbitration requests.
+  DEFINE_METRIC(
+      kMetricArbitratorRequestsCount, facebook::velox::StatType::COUNT);
+
+  // The number of times a query level memory pool is aborted as a result of a
+  // memory arbitration process. The memory pool aborted will eventually result
+  // in a cancelling of the original query.
+  DEFINE_METRIC(
+      kMetricArbitratorAbortedCount, facebook::velox::StatType::COUNT);
+
+  // The number of times a memory arbitration request failed. This may occur
+  // either because the requester was terminated during the processing of its
+  // request, the arbitration request would surpass the maximum allowed capacity
+  // for the requester, or the arbitration process couldn't release the
+  // requested amount of memory.
+  DEFINE_METRIC(
+      kMetricArbitratorFailuresCount, facebook::velox::StatType::COUNT);
+
   // Tracks the memory reclaim count on an operator.
   DEFINE_METRIC(kMetricMemoryReclaimCount, facebook::velox::StatType::COUNT);
 
@@ -82,10 +100,6 @@ void registerVeloxMetrics() {
   DEFINE_METRIC(
       kMetricMemoryNonReclaimableCount, facebook::velox::StatType::COUNT);
 
-  // The number of arbitration requests.
-  DEFINE_METRIC(
-      kMetricArbitratorRequestsCount, facebook::velox::StatType::COUNT);
-
   // The number of arbitration that reclaims the used memory from the query
   // which initiates the memory arbitration request itself. It ensures the
   // memory arbitration request won't exceed its per-query memory capacity
@@ -103,20 +117,6 @@ void registerVeloxMetrics() {
       kMetricArbitratorGlobalArbitrationCount,
       facebook::velox::StatType::COUNT);
 
-  // The number of times a query level memory pool is aborted as a result of a
-  // memory arbitration process. The memory pool aborted will eventually result
-  // in a cancelling the original query.
-  DEFINE_METRIC(
-      kMetricArbitratorAbortedCount, facebook::velox::StatType::COUNT);
-
-  // The number of times a memory arbitration request failed. This may occur
-  // either because the requester was terminated during the processing of its
-  // request, the arbitration request would surpass the maximum allowed capacity
-  // for the requester, or the arbitration process couldn't release the
-  // requested amount of memory.
-  DEFINE_METRIC(
-      kMetricArbitratorFailuresCount, facebook::velox::StatType::COUNT);
-
   // The distribution of the amount of time an arbitration request stays queued
   // in range of [0, 600s] with 20 buckets. It is configured to report the
   // latency at P50, P90, P99, and P100 percentiles.

velox/common/base/Counters.h

@@ -70,21 +70,12 @@ constexpr folly::StringPiece kMetricMemoryPoolReservationLeakBytes{
 constexpr folly::StringPiece kMetricMemoryAllocatorDoubleFreeCount{
     "velox.memory_allocator_double_free_count"};
 
-constexpr folly::StringPiece kMetricArbitratorRequestsCount{
-    "velox.arbitrator_requests_count"};
-
 constexpr folly::StringPiece kMetricArbitratorLocalArbitrationCount{
     "velox.arbitrator_local_arbitration_count"};
 
 constexpr folly::StringPiece kMetricArbitratorGlobalArbitrationCount{
     "velox.arbitrator_global_arbitration_count"};
 
-constexpr folly::StringPiece kMetricArbitratorAbortedCount{
-    "velox.arbitrator_aborted_count"};
-
-constexpr folly::StringPiece kMetricArbitratorFailuresCount{
-    "velox.arbitrator_failures_count"};
-
 constexpr folly::StringPiece kMetricArbitratorQueueTimeMs{
     "velox.arbitrator_queue_time_ms"};
 
@@ -128,4 +119,13 @@ constexpr folly::StringPiece kMetricSpillWriteTimeMs{
 
 constexpr folly::StringPiece kMetricFileWriterEarlyFlushedRawBytes{
     "velox.file_writer_early_flushed_raw_bytes"};
+
+constexpr folly::StringPiece kMetricArbitratorRequestsCount{
+    "velox.arbitrator_requests_count"};
+
+constexpr folly::StringPiece kMetricArbitratorAbortedCount{
+    "velox.arbitrator_aborted_count"};
+
+constexpr folly::StringPiece kMetricArbitratorFailuresCount{
+    "velox.arbitrator_failures_count"};
 } // namespace facebook::velox

velox/common/base/PeriodicStatsReporter.cpp

@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+
+#include "velox/common/base/PeriodicStatsReporter.h"
+#include "velox/common/base/Counters.h"
+#include "velox/common/base/StatsReporter.h"
+#include "velox/common/memory/Memory.h"
+
+namespace facebook::velox {
+
+namespace {
+#define REPORT_IF_NOT_ZERO(name, counter)   \
+  if ((counter) != 0) {                     \
+    RECORD_METRIC_VALUE((name), (counter)); \
+  }
+
+std::mutex& instanceMutex() {
+  static std::mutex instanceMu;
+  return instanceMu;
+}
+
+// Global instance. Must be called while holding a lock over instanceMutex().
+std::unique_ptr<PeriodicStatsReporter>& instance() {
+  static std::unique_ptr<PeriodicStatsReporter> reporter;
+  return reporter;
+}
+} // namespace
+
+void startPeriodicStatsReporter(const PeriodicStatsReporter::Options& options) {
+  std::lock_guard<std::mutex> l(instanceMutex());
+  auto& instanceRef = instance();
+  VELOX_CHECK_NULL(
+      instanceRef, "The periodic stats reporter has already started.");
+  instanceRef = std::make_unique<PeriodicStatsReporter>(options);
+  instanceRef->start();
+}
+
+void stopPeriodicStatsReporter() {
+  std::lock_guard<std::mutex> l(instanceMutex());
+  auto& instanceRef = instance();
+  VELOX_CHECK_NOT_NULL(instanceRef, "No periodic stats reporter to stop.");
+  instanceRef->stop();
+  instanceRef.reset();
+}
+
+PeriodicStatsReporter::PeriodicStatsReporter(const Options& options)
+    : arbitrator_(options.arbitrator), options_(options) {}
+
+void PeriodicStatsReporter::start() {
+  LOG(INFO) << "Starting PeriodicStatsReporter with options "
+            << options_.toString();
+  addTask(
+      "report_arbitrator_stats",
+      [this]() { reportArbitratorStats(); },
+      options_.arbitratorStatsIntervalMs);
+}
+
+void PeriodicStatsReporter::stop() {
+  LOG(INFO) << "Stopping PeriodicStatsReporter";
+  scheduler_.stop();
+}
+
+void PeriodicStatsReporter::reportArbitratorStats() {
+  if (arbitrator_ == nullptr) {
+    return;
+  }
+
+  const auto stats = arbitrator_->stats();
+  RECORD_METRIC_VALUE(
+      kMetricArbitratorFreeCapacityBytes,
+      stats.freeCapacityBytes + stats.freeReservedCapacityBytes);
+  RECORD_METRIC_VALUE(
+      kMetricArbitratorFreeReservedCapacityBytes,
+      stats.freeReservedCapacityBytes);
+}
+
+} // namespace facebook::velox

velox/common/base/PeriodicStatsReporter.h

@@ -0,0 +1,95 @@
+/*
+ * 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 <folly/experimental/ThreadedRepeatingFunctionRunner.h>
+#include "velox/common/memory/MemoryArbitrator.h"
+
+namespace folly {
+class CPUThreadPoolExecutor;
+}
+
+namespace facebook::velox {
+
+namespace memory {
+class MemoryAllocator;
+}
+
+namespace cache {
+class AsyncDataCache;
+}
+
+/// Manages a background daemon thread to report stats through 'StatsReporter'.
+class PeriodicStatsReporter {
+ public:
+  struct Options {
+    Options() {}
+
+    const memory::MemoryArbitrator* arbitrator{nullptr};
+
+    uint64_t arbitratorStatsIntervalMs{60'000};
+
+    std::string toString() const {
+      return fmt::format(
+          "arbitratorStatsIntervalMs:{}", arbitratorStatsIntervalMs);
+    }
+  };
+
+  PeriodicStatsReporter(const Options& options = Options());
+
+  /// Invoked to start the report daemon in background.
+  void start();
+
+  /// Invoked to stop the report daemon in background.
+  void stop();
+
+ private:
+  // Add a task to run periodically.
+  template <typename TFunc>
+  void addTask(const std::string& taskName, TFunc&& func, size_t intervalMs) {
+    scheduler_.add(
+        taskName,
+        [taskName,
+         intervalMs,
+         func = std::forward<TFunc>(func)]() mutable noexcept {
+          try {
+            func();
+          } catch (const std::exception& e) {
+            LOG(ERROR) << "Error running periodic task " << taskName << ": "
+                       << e.what();
+          }
+          return std::chrono::milliseconds(intervalMs);
+        });
+  }
+
+  void reportArbitratorStats();
+
+  const velox::memory::MemoryArbitrator* const arbitrator_{nullptr};
+  const Options options_;
+
+  folly::ThreadedRepeatingFunctionRunner scheduler_;
+};
+
+/// Initializes and starts the process-wide periodic stats reporter. Before
+/// 'stopPeriodicStatsReporter()' is called, this method can only be called once
+/// process-wide, and additional calls to this method will throw.
+void startPeriodicStatsReporter(const PeriodicStatsReporter::Options& options);
+
+/// Stops the process-wide periodic stats reporter.
+void stopPeriodicStatsReporter();
+
+} // namespace facebook::velox