.github/workflows/conbench_upload.yml

@@ -108,7 +108,7 @@ jobs:
           --run_id "GHA-${{ github.run_id }}-${{ github.run_attempt }}" \
           --pr_number "${{ steps.extract.outputs.pr_number }}" \
           --sha "${{ steps.download.outputs.contender_sha }}" \
-          --output_dir "/tmp/artifacts/benchmark-results/contender/"
+          --output_dir "/tmp/artifacts/contender/"
 
     - name: "Check the status of the upload"
       # Status functions like failure() only work in `if:`

CMake/VeloxUtils.cmake

@@ -77,9 +77,9 @@ function(velox_add_library TARGET)
       # Create the target if this is the first invocation.
       add_library(velox ${_type} ${ARGN})
       set_target_properties(velox PROPERTIES LIBRARY_OUTPUT_DIRECTORY
-                                             ${CMAKE_BINARY_DIR}/lib)
+                                             ${PROJECT_BINARY_DIR}/lib)
       set_target_properties(velox PROPERTIES ARCHIVE_OUTPUT_DIRECTORY
-                                             ${CMAKE_BINARY_DIR}/lib)
+                                             ${PROJECT_BINARY_DIR}/lib)
       install(TARGETS velox DESTINATION lib/velox)
     endif()
     # create alias for compatability

CMake/resolve_dependency_modules/simdjson.cmake

@@ -29,4 +29,8 @@ FetchContent_Declare(
   URL ${VELOX_SIMDJSON_SOURCE_URL}
   URL_HASH ${VELOX_SIMDJSON_BUILD_SHA256_CHECKSUM})
 
+if(${VELOX_SIMDJSON_SKIPUTF8VALIDATION})
+  set(SIMDJSON_SKIPUTF8VALIDATION ON)
+endif()
+
 FetchContent_MakeAvailable(simdjson)

CMakeLists.txt

@@ -149,6 +149,8 @@ option(
   VELOX_ENABLE_INT64_BUILD_PARTITION_BOUND
   "make buildPartitionBounds_ a vector int64 instead of int32 to avoid integer overflow when the hashtable has billions of records"
   OFF)
+option(VELOX_SIMDJSON_SKIPUTF8VALIDATION
+       "Skip simdjson utf8 validation in JSON parsing" OFF)
 
 # Explicitly force compilers to generate colored output. Compilers usually do
 # this by default if they detect the output is a terminal, but this assumption

README.md

@@ -131,6 +131,12 @@ subsequent Velox builds can use the installed packages.
 *You can reuse `DEPENDENCY_INSTALL` and `INSTALL_PREFIX` for Velox clients such as Prestissimo
 by specifying a common shared directory.`*
 
+The build parallelism for dependencies can be controlled by the `BUILD_THREADS` environment
+variable and overrides the default number of parallel processes used for compiling and linking.
+The default value is the number of cores on your machine.
+This is useful if your machine has lots of cores but no matching memory to process all
+compile and link processes in parallel resulting in OOM kills by the kernel.
+
 ### Setting up on macOS
 
 On a macOS machine (either Intel or Apple silicon) you can setup and then build like so:

scripts/setup-centos9.sh

@@ -30,7 +30,7 @@ set -efx -o pipefail
 # so that some low level types are the same size. Also, disable warnings.
 SCRIPTDIR=$(dirname "${BASH_SOURCE[0]}")
 source $SCRIPTDIR/setup-helper-functions.sh
-NPROC=$(getconf _NPROCESSORS_ONLN)
+NPROC=${BUILD_THREADS:-$(getconf _NPROCESSORS_ONLN)}
 export CXXFLAGS=$(get_cxx_flags) # Used by boost.
 export CFLAGS=${CXXFLAGS//"-std=c++17"/} # Used by LZO.
 CMAKE_BUILD_TYPE="${BUILD_TYPE:-Release}"

scripts/setup-helper-functions.sh

@@ -18,6 +18,7 @@
 
 DEPENDENCY_DIR=${DEPENDENCY_DIR:-$(pwd)/deps-download}
 OS_CXXFLAGS=""
+NPROC=${BUILD_THREADS:-$(getconf _NPROCESSORS_ONLN)}
 
 function run_and_time {
   time "$@" || (echo "Failed to run $* ." ; exit 1 )
@@ -214,7 +215,7 @@ function cmake_install {
     -DBUILD_TESTING=OFF \
     "$@"
   # Exit if the build fails.
-  cmake --build "${BINARY_DIR}" || { echo 'build failed' ; exit 1; }
+  cmake --build "${BINARY_DIR}" "-j ${NPROC}" || { echo 'build failed' ; exit 1; }
   ${SUDO} cmake --install "${BINARY_DIR}"
 }
 

scripts/setup-macos.sh

@@ -36,7 +36,7 @@ PYTHON_VENV=${PYTHON_VENV:-"${SCRIPTDIR}/../.venv"}
 # by tagging the brew packages to be system packages.
 # This is used during package builds.
 export OS_CXXFLAGS=" -isystem $(brew --prefix)/include "
-NPROC=$(getconf _NPROCESSORS_ONLN)
+NPROC=${BUILD_THREADS:-$(getconf _NPROCESSORS_ONLN)}
 
 BUILD_DUCKDB="${BUILD_DUCKDB:-true}"
 VELOX_BUILD_SHARED=${VELOX_BUILD_SHARED:-"OFF"} #Build folly shared for use in libvelox.so.

scripts/setup-ubuntu.sh

@@ -34,7 +34,7 @@ source $SCRIPTDIR/setup-helper-functions.sh
 # are the same size.
 COMPILER_FLAGS=$(get_cxx_flags)
 export COMPILER_FLAGS
-NPROC=$(getconf _NPROCESSORS_ONLN)
+NPROC=${BUILD_THREADS:-$(getconf _NPROCESSORS_ONLN)}
 BUILD_DUCKDB="${BUILD_DUCKDB:-true}"
 export CMAKE_BUILD_TYPE=Release
 VELOX_BUILD_SHARED=${VELOX_BUILD_SHARED:-"OFF"} #Build folly shared for use in libvelox.so.

setup.py

@@ -136,7 +136,7 @@ def build_extension(self, ext):
             f"-DCMAKE_BUILD_TYPE={cfg}",
             f"-DCMAKE_INSTALL_PREFIX={extdir}",
             "-DCMAKE_VERBOSE_MAKEFILE=ON",
-            "-DVELOX_BUILD_MINIMAL=ON",
+            "-DVELOX_BUILD_MINIMAL_WITH_DWIO=ON",
             "-DVELOX_BUILD_PYTHON_PACKAGE=ON",
             f"-DPYTHON_EXECUTABLE={exec_path} ",
         ]

velox/common/caching/tests/AsyncDataCacheTest.cpp

@@ -718,7 +718,7 @@ TEST_P(AsyncDataCacheTest, pin) {
 
 TEST_P(AsyncDataCacheTest, replace) {
   constexpr int64_t kMaxBytes = 64 << 20;
-  FLAGS_velox_exception_user_stacktrace_enabled = false;
+  config::globalConfig.exceptionUserStacktraceEnabled = false;
   initializeCache(kMaxBytes);
   // Load 10x the max size, inject an error every 21 batches.
   loadLoop(0, kMaxBytes * 10, 21);
@@ -736,7 +736,7 @@ TEST_P(AsyncDataCacheTest, replace) {
 
 TEST_P(AsyncDataCacheTest, evictAccounting) {
   constexpr int64_t kMaxBytes = 64 << 20;
-  FLAGS_velox_exception_user_stacktrace_enabled = false;
+  config::globalConfig.exceptionUserStacktraceEnabled = false;
   initializeCache(kMaxBytes);
   auto pool = manager_->addLeafPool("test");
 
@@ -760,7 +760,7 @@ TEST_P(AsyncDataCacheTest, evictAccounting) {
 TEST_P(AsyncDataCacheTest, largeEvict) {
   constexpr int64_t kMaxBytes = 256 << 20;
   constexpr int32_t kNumThreads = 24;
-  FLAGS_velox_exception_user_stacktrace_enabled = false;
+  config::globalConfig.exceptionUserStacktraceEnabled = false;
   initializeCache(kMaxBytes);
   // Load 10x the max size, inject an allocation of 1/8 the capacity every 4
   // batches.
@@ -839,7 +839,7 @@ TEST_P(AsyncDataCacheTest, DISABLED_ssd) {
   constexpr uint64_t kRamBytes = 32 << 20;
   constexpr uint64_t kSsdBytes = 512UL << 20;
 #endif
-  FLAGS_velox_exception_user_stacktrace_enabled = false;
+  config::globalConfig.exceptionUserStacktraceEnabled = false;
   initializeCache(kRamBytes, kSsdBytes);
   cache_->setVerifyHook(
       [&](const AsyncDataCacheEntry& entry) { checkContents(entry); });

velox/common/config/GlobalConfig.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <cstdint>
+#include <string>
 
 namespace facebook::velox::config {
 
@@ -53,6 +54,29 @@ struct GlobalConfiguration {
   /// Min time interval in milliseconds between stack traces captured in
   /// system type of VeloxException; off when set to 0 (the default).
   int32_t exceptionSystemStacktraceRateLimitMs{0};
+  /// Whether to overwrite queryCtx and force the use of simplified expression
+  /// evaluation path.
+  bool forceEvalSimplified{false};
+  /// This is an experimental flag only to be used for debugging purposes. If
+  /// set to true, serializes the input vector data and all the SQL expressions
+  /// in the ExprSet that is currently executing, whenever a fatal signal is
+  /// encountered. Enabling this flag makes the signal handler async signal
+  /// unsafe, so it should only be used for debugging purposes. The vector and
+  /// SQLs are serialized to files in directories specified by either
+  /// 'saveInputOnExpressionAnyFailurePath' or
+  /// 'saveInputOnExpressionSystemFailurePath'
+  bool experimentalSaveInputOnFatalSignal{false};
+  /// Used to enable saving input vector and expression SQL on disk in case
+  /// of any (user or system) error during expression evaluation. The value
+  /// specifies a path to a directory where the vectors will be saved. That
+  /// directory must exist and be writable.
+  std::string saveInputOnExpressionAnyFailurePath;
+  /// Used to enable saving input vector and expression SQL on disk in case
+  /// of a system error during expression evaluation. The value specifies a path
+  /// to a directory where the vectors will be saved. That directory must exist
+  /// and be writable. This flag is ignored if
+  /// saveInputOnExpressionAnyFailurePath flag is set.
+  std::string saveInputOnExpressionSystemFailurePath;
 };
 
 extern GlobalConfiguration globalConfig;

velox/common/file/tests/FaultyFileSystem.h

@@ -47,8 +47,8 @@ class FaultyFileSystem : public FileSystem {
   // Extracts the delegated real file path by removing the faulty file system
   // scheme prefix.
   inline std::string_view extractPath(std::string_view path) const override {
-    VELOX_CHECK_EQ(path.find(scheme()), 0, "");
-    const auto filePath = path.substr(scheme().length());
+    const auto filePath =
+        (path.find(scheme()) == 0) ? path.substr(scheme().length()) : path;
     return getFileSystem(filePath, config_)->extractPath(filePath);
   }
 

velox/common/future/VeloxPromise.h

@@ -42,7 +42,7 @@ class VeloxPromise : public folly::Promise<T> {
     }
   }
 
-  explicit VeloxPromise(VeloxPromise<T>&& other)
+  explicit VeloxPromise(VeloxPromise<T>&& other) noexcept
       : folly::Promise<T>(std::move(other)),
         context_(std::move(other.context_)) {}
 

velox/connectors/Connector.h

@@ -43,6 +43,10 @@ namespace facebook::velox::config {
 class ConfigBase;
 }
 
+namespace facebook::velox::core {
+class ITypedExpr;
+}
+
 namespace facebook::velox::connector {
 
 class DataSource;
@@ -287,6 +291,73 @@ class DataSource {
   }
 };
 
+class IndexSource {
+ public:
+  virtual ~IndexSource() = default;
+
+  /// Represents a lookup request for a given input.
+  struct LookupRequest {
+    /// Contains the input column vectors used by lookup join and range
+    /// conditions.
+    RowVectorPtr input;
+
+    explicit LookupRequest(RowVectorPtr input) : input(std::move(input)) {}
+  };
+
+  /// Represents the lookup result for a subset of input produced by the
+  /// 'LookupResultIterator'.
+  struct LookupResult {
+    /// Specifies the indices of input row in the lookup request that have
+    /// matches in 'output'. It contains the input indices in the order
+    /// of the input rows in the lookup request. Any gap in the indices means
+    /// the input rows that has no matches in output.
+    ///
+    /// Example:
+    ///   LookupRequest: input = [0, 1, 2, 3, 4]
+    ///   LookupResult:  inputHits = [0, 0, 2, 2, 3, 4, 4, 4]
+    ///                  output    = [0, 1, 2, 3, 4, 5, 6, 7]
+    ///
+    ///   Here is match results for each input row:
+    ///   input row #0: match with output rows #0 and #1.
+    ///   input row #1: no matches
+    ///   input row #2: match with output rows #2 and #3.
+    ///   input row #3: match with output row #4.
+    ///   input row #4: match with output rows #5, #6 and #7.
+    ///
+    /// 'LookupResultIterator' must also produce the output result in order of
+    /// input rows.
+    BufferPtr inputHits;
+
+    /// Contains the lookup result rows.
+    RowVectorPtr output;
+
+    LookupResult(BufferPtr _inputHits, RowVectorPtr _output)
+        : inputHits(std::move(_inputHits)), output(std::move(_output)) {
+      VELOX_CHECK_EQ(inputHits->size() / sizeof(int32_t), output->size());
+    }
+  };
+
+  /// The lookup result iterator used to fetch the lookup result in batch for a
+  /// given lookup request.
+  class LookupResultIterator {
+   public:
+    virtual ~LookupResultIterator() = default;
+
+    /// Invoked to fetch update to 'size' number of output rows. Returns nullptr
+    /// if all the lookup results have been fetched. Returns std::nullopt and
+    /// sets the 'future' if started asynchronous work and needs to wait for it
+    /// to complete to continue processing. The caller will wait for the
+    /// 'future' to complete before calling 'next' again.
+    virtual std::optional<std::shared_ptr<LookupResult>> next(
+        vector_size_t size,
+        velox::ContinueFuture& future) = 0;
+  };
+
+  virtual LookupResultIterator lookup(const LookupRequest& request) = 0;
+
+  virtual std::unordered_map<std::string, RuntimeCounter> runtimeStats() = 0;
+};
+
 /// Collection of context data for use in a DataSource, IndexSource or DataSink.
 /// One instance of this per DataSource and DataSink. This may be passed between
 /// threads but methods must be invoked sequentially. Serializing use is the
@@ -479,6 +550,57 @@ class Connector {
     return false;
   }
 
+  /// Creates index source for index join lookup.
+  /// @param inputType The list of probe-side columns that either used in
+  /// equi-clauses or join conditions.
+  /// @param numJoinKeys The number of key columns used in join equi-clauses.
+  /// The first 'numJoinKeys' columns in 'inputType' form a prefix of the
+  /// index, and the rest of the columns in inputType are expected to be used in
+  /// 'joinCondition'.
+  /// @param joinConditions The join conditions. It expects inputs columns from
+  /// the 'tail' of 'inputType' and from 'columnHandles'.
+  /// @param outputType The lookup output type from index source.
+  /// @param tableHandle The index table handle.
+  /// @param columnHandles The column handles which maps from column name
+  /// used in 'outputType' and 'joinConditions' to the corresponding column
+  /// handles in the index table.
+  /// @param connectorQueryCtx The query context.
+  ///
+  /// Here is an example that how the lookup join operator uses index source:
+  ///
+  /// SELECT t.sid, t.day_ts, u.event_value
+  /// FROM t LEFT JOIN u
+  /// ON t.sid = u.sid
+  ///  AND contains(t.event_list, u.event_type)
+  ///  AND t.ds BETWEEN '2024-01-01' AND '2024-01-07'
+  ///
+  /// Here,
+  /// - 'inputType' is ROW{t.sid, t.event_list}
+  /// - 'numJoinKeys' is 1 since only t.sid is used in join equi-clauses.
+  /// - 'joinConditions' is list of one expression: contains(t.event_list,
+  ///    u.event_type)
+  /// - 'outputType' is ROW{u.event_value}
+  /// - 'tableHandle' specifies the metadata of the index table.
+  /// - 'columnHandles' is a map from 'u.event_type' (in 'joinConditions') and
+  ///   'u.event_value' (in 'outputType') to the actual column names in the
+  ///   index table.
+  /// - 'connectorQueryCtx' provide the connector query execution context.
+  ///
+  virtual std::unique_ptr<IndexSource> createIndexSource(
+      const RowTypePtr& inputType,
+      size_t numJoinKeys,
+      const std::vector<std::shared_ptr<const core::ITypedExpr>>&
+          joinConditions,
+      const RowTypePtr& outputType,
+      const std::shared_ptr<ConnectorTableHandle>& tableHandle,
+      const std::unordered_map<
+          std::string,
+          std::shared_ptr<connector::ColumnHandle>>& columnHandles,
+      ConnectorQueryCtx* connectorQueryCtx) {
+    VELOX_UNSUPPORTED(
+        "Connector {} does not support index source", connectorId());
+  }
+
   virtual std::unique_ptr<DataSink> createDataSink(
       RowTypePtr inputType,
       std::shared_ptr<ConnectorInsertTableHandle> connectorInsertTableHandle,

velox/connectors/hive/HiveConfig.h

@@ -165,9 +165,9 @@ class HiveConfig {
       "hive.reader.timestamp_unit";
 
   static constexpr const char* kReadStatsBasedFilterReorderDisabled =
-      "hive.reader.stats-based-filter-reorder-disabaled";
+      "hive.reader.stats-based-filter-reorder-disabled";
   static constexpr const char* kReadStatsBasedFilterReorderDisabledSession =
-      "hive.reader.stats_based_filter_reorder_disabaled";
+      "hive.reader.stats_based_filter_reorder_disabled";
 
   static constexpr const char* kLocalDataPath = "hive_local_data_path";
   static constexpr const char* kLocalFileFormat = "hive_local_file_format";

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -288,7 +288,7 @@ void checkColumnNameLowerCase(const std::shared_ptr<const Type>& type) {
 }
 
 void checkColumnNameLowerCase(
-    const SubfieldFilters& filters,
+    const common::SubfieldFilters& filters,
     const std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>&
         infoColumns) {
   for (const auto& filterIt : filters) {
@@ -349,7 +349,7 @@ std::shared_ptr<common::ScanSpec> makeScanSpec(
     const RowTypePtr& rowType,
     const folly::F14FastMap<std::string, std::vector<const common::Subfield*>>&
         outputSubfields,
-    const SubfieldFilters& filters,
+    const common::SubfieldFilters& filters,
     const RowTypePtr& dataColumns,
     const std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>&
         partitionKeys,
@@ -837,7 +837,7 @@ core::TypedExprPtr extractFiltersFromRemainingFilter(
     const core::TypedExprPtr& expr,
     core::ExpressionEvaluator* evaluator,
     bool negated,
-    SubfieldFilters& filters,
+    common::SubfieldFilters& filters,
     double& sampleRate) {
   auto* call = dynamic_cast<const core::CallTypedExpr*>(expr.get());
   if (call == nullptr) {