Run TPC-H with cuDF (facebookincubator#24)

* Enable cudf in TPC-H queries.
* Draft work on strings conversion.
* Improved DATE / logical type support.
* Fix for join build to allow multiple batches.
* Add environment variables VELOX_CUDF_DISABLED and VELOX_CUDF_DEBUG.
* Apply tuning parameters to TPC-H queries to use larger batches when cuDF is enabled.
* Add env vars for disabling cudf / enabling debug.
* num_drivers=1, num_repeats=1
* Don't reset hashObject_ and only set finished_ to true if noMoreInput_ is true.
* Apply tuning parameters to cuDF engine.

---------

Co-authored-by: Karthikeyan Natarajan <[email protected]>

benchmark.sh

@@ -25,8 +25,29 @@ set -euo pipefail
 # Run a GPU build and test
 pushd "$(dirname ${0})"
 
-#CUDA_ARCHITECTURES="native" EXTRA_CMAKE_FLAGS="-DVELOX_ENABLE_ARROW=ON -DVELOX_ENABLE_PARQUET=ON -DVELOX_ENABLE_BENCHMARKS=ON -DVELOX_ENABLE_BENCHMARKS_BASIC=ON" make gpu
+mkdir -p benchmark_results
 
-./_build/release/velox/benchmarks/tpch/velox_tpch_benchmark --data_path=velox-tpch-sf10-data --data_format=parquet --run_query_verbose=5 --num_repeats=6
+queries=${1:-$(seq 1 20)}
+devices=${2:-"cpu gpu"}
+
+
+for query_number in ${queries}; do
+    printf -v query_number '%02d' "${query_number}"
+    for device in ${devices}; do
+        case "${device}" in
+            "cpu")
+                num_drivers=40
+                export VELOX_CUDF_DISABLED=1;;
+            "gpu")
+                num_drivers=1
+                export VELOX_CUDF_DISABLED=0;;
+        esac
+        echo "Running query ${query_number} on ${device} with ${num_drivers} drivers."
+        # The benchmarks segfault after reporting results, so we disable errors
+        set +e
+        ./_build/release/velox/benchmarks/tpch/velox_tpch_benchmark --data_path=velox-tpch-sf10-data --data_format=parquet --run_query_verbose=${query_number} --num_repeats=1 --num_drivers ${num_drivers} 2>&1 | tee benchmark_results/q${query_number}_${device}_${num_drivers}_drivers
+        set -e
+    done
+done
 
 popd

velox/benchmarks/tpch/CMakeLists.txt

@@ -17,6 +17,7 @@ add_library(velox_tpch_benchmark_lib TpchBenchmark.cpp)
 target_link_libraries(
   velox_tpch_benchmark_lib
   velox_aggregates
+  velox_cudf_exec
   velox_exec
   velox_exec_test_lib
   velox_dwio_common

velox/benchmarks/tpch/TpchBenchmark.cpp

@@ -35,6 +35,7 @@
 #include "velox/exec/Split.h"
 #include "velox/exec/tests/utils/HiveConnectorTestBase.h"
 #include "velox/exec/tests/utils/TpchQueryBuilder.h"
+#include "velox/experimental/cudf/exec/ToCudf.h"
 #include "velox/functions/prestosql/aggregates/RegisterAggregateFunctions.h"
 #include "velox/functions/prestosql/registration/RegistrationFunctions.h"
 #include "velox/parse/TypeResolver.h"
@@ -274,9 +275,13 @@ class TpchBenchmark {
             connector::hive::HiveConnectorFactory::kHiveConnectorName)
             ->newConnector(kHiveConnectorId, properties, ioExecutor_.get());
     connector::registerConnector(hiveConnector);
+
+    // Enable cuDF operators
+    cudf_velox::registerCudf();
   }
 
   void shutdown() {
+    cudf_velox::unregisterCudf();
     cache_->shutdown();
   }
 
@@ -290,6 +295,14 @@ class TpchBenchmark {
         params.planNode = tpchPlan.plan;
         params.queryConfigs[core::QueryConfig::kMaxSplitPreloadPerDriver] =
             std::to_string(FLAGS_split_preload_per_driver);
+        if (cudf_velox::cudfIsRegistered()) {
+          params.queryConfigs[core::QueryConfig::kPreferredOutputBatchBytes] =
+              "536870912"; // 512 MB
+          params.queryConfigs[core::QueryConfig::kPreferredOutputBatchRows] =
+              "500000";
+          params.queryConfigs[core::QueryConfig::kMaxOutputBatchRows] =
+              "500000";
+        }
         const int numSplitsPerFile = FLAGS_num_splits_per_file;
 
         bool noMoreSplits = false;

velox/experimental/cudf/exec/CMakeLists.txt

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_library(velox_cudf_exec CudfHashJoin.cpp ToCudf.cpp VeloxCudfInterop.cpp)
+add_library(velox_cudf_exec CudfHashJoin.cpp ToCudf.cpp Utilities.cpp
+                            VeloxCudfInterop.cpp)
 
 set_target_properties(
   velox_cudf_exec