[IGNORE] Squashed commit of cudf exchange presto setup

presto-native-execution/entrypoint.sh

@@ -11,4 +11,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-GLOG_logtostderr=1 presto_server --etc-dir=/opt/presto-server/etc
+if [[ "$PROFILE" == "ON" ]]; then
+  mkdir /presto_profiles
+
+  if [[ -z $PROFILE_ARGS ]]; then
+    PROFILE_ARGS="-t nvtx,cuda,osrt 
+                  --cuda-memory-usage=true 
+                  --cuda-um-cpu-page-faults=true 
+                  --cuda-um-gpu-page-faults=true 
+                  --cudabacktrace=true"
+  fi
+  PROFILE_CMD="nsys launch $PROFILE_ARGS"
+fi
+
+GLOG_logtostderr=1 $PROFILE_CMD presto_server --etc-dir=/opt/presto-server/etc

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -79,6 +79,7 @@
 #ifdef PRESTO_ENABLE_CUDF
 #include "velox/experimental/cudf/CudfConfig.h"
 #include "velox/experimental/cudf/exec/ToCudf.h"
+#include "velox/experimental/cudf-exchange/Communicator.h"
 #endif
 
 #ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
@@ -166,7 +167,8 @@ bool isSharedLibrary(const fs::path& path) {
   return pathExt == kLinuxSharedLibExt || pathExt == kMacOSSharedLibExt;
 }
 
-void registerVeloxCudf() {
+std::shared_ptr<std::thread> registerVeloxCudf() {
+  std::shared_ptr<std::thread> serverThread = nullptr;
 #ifdef PRESTO_ENABLE_CUDF
   // Disable by default.
   velox::cudf_velox::CudfConfig::getInstance().enabled = false;
@@ -179,20 +181,38 @@ void registerVeloxCudf() {
         systemConfig->values());
     if (velox::cudf_velox::CudfConfig::getInstance().enabled) {
       velox::cudf_velox::registerCudf();
+      if (velox::cudf_velox::CudfConfig::getInstance().exchange) {
+        PRESTO_STARTUP_LOG(INFO) << "cuDF exchange server started";
+        auto server = facebook::velox::cudf_exchange::Communicator::initAndGet(
+            SystemConfig::instance()->cudfServerPort(),
+            SystemConfig::instance()->discoveryUri().value());
+        if (server) {
+          serverThread = std::make_shared<std::thread>(
+              &velox::cudf_exchange::Communicator::run, server.get());
+        }
+      }
       PRESTO_STARTUP_LOG(INFO) << "cuDF is registered.";
     }
   }
 #endif
+  return serverThread;
 }
 
-void unregisterVeloxCudf() {
+void unregisterVeloxCudf(std::shared_ptr<std::thread> serverThread) {
 #ifdef PRESTO_ENABLE_CUDF
   auto systemConfig = SystemConfig::instance();
   if (systemConfig->values().contains(
           velox::cudf_velox::CudfConfig::kCudfEnabled) &&
       velox::cudf_velox::CudfConfig::getInstance().enabled) {
     velox::cudf_velox::unregisterCudf();
     PRESTO_SHUTDOWN_LOG(INFO) << "cuDF is unregistered.";
+    if (serverThread) {
+      auto server = facebook::velox::cudf_exchange::Communicator::getInstance();
+      server->stop();
+      server.reset();      
+      PRESTO_SHUTDOWN_LOG(INFO) << "Joining UCX Communicator thread for shutdown.";
+      serverThread->join();
+    }
   }
 #endif
 }
@@ -352,7 +372,7 @@ void PrestoServer::run() {
 
   // We need to register cuDF before the connectors so that the cuDF connector
   // factories can be used.
-  registerVeloxCudf();
+  auto communicatorThread = registerVeloxCudf();
 
   // Register Presto connector factories and connectors
   registerConnectors();
@@ -746,7 +766,7 @@ void PrestoServer::run() {
   unregisterFileReadersAndWriters();
   unregisterFileSystems();
   unregisterConnectors();
-  unregisterVeloxCudf();
+  unregisterVeloxCudf(communicatorThread);
 
   PRESTO_SHUTDOWN_LOG(INFO)
       << "Joining Driver CPU Executor '" << driverCpuExecutor_->getName()

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -169,6 +169,7 @@ SystemConfig::SystemConfig() {
           NONE_PROP(kHttpsKeyPath),
           NONE_PROP(kHttpsClientCertAndKeyPath),
           NONE_PROP(kHttpsClientCaFile),
+          NONE_PROP(kCudfServerPort),
           NUM_PROP(kExchangeHttpClientNumIoThreadsHwMultiplier, 1.0),
           NUM_PROP(kExchangeHttpClientNumCpuThreadsHwMultiplier, 1.0),
           NUM_PROP(kConnectorNumCpuThreadsHwMultiplier, 0.0),
@@ -306,6 +307,10 @@ int SystemConfig::httpServerHttpPort() const {
   return requiredProperty<int>(kHttpServerHttpPort);
 }
 
+int SystemConfig::cudfServerPort() const {
+  return requiredProperty<int>(kCudfServerPort);
+}
+
 bool SystemConfig::httpServerReusePort() const {
   return optionalProperty<bool>(kHttpServerReusePort).value();
 }

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -708,6 +708,8 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kHttpClientConnectionReuseCounterEnabled{
       "http-client.connection-reuse-counter-enabled"};
 
+  static constexpr std::string_view kCudfServerPort{"cudf.exchange.server.port"};
+
   static constexpr std::string_view kExchangeMaxErrorDuration{
       "exchange.max-error-duration"};
 
@@ -898,6 +900,8 @@ class SystemConfig : public ConfigBase {
 
   int httpServerHttpPort() const;
 
+  int cudfServerPort() const;
+
   bool httpServerReusePort() const;
 
   bool httpServerBindToNodeInternalAddressOnlyEnabled() const;

presto-native-execution/presto_cpp/main/connectors/CMakeLists.txt

@@ -26,7 +26,8 @@ if(PRESTO_ENABLE_ARROW_FLIGHT_CONNECTOR)
 endif()
 
 if(PRESTO_ENABLE_CUDF)
-  target_link_libraries(presto_connectors velox_cudf_hive_connector cudf::cudf)
+   target_link_libraries(presto_connectors velox_cudf_hive_connector
+                        ucxx::ucxx cudf::cudf)
 endif()
 
 target_link_libraries(

presto-native-execution/presto_cpp/main/types/PrestoToVeloxQueryPlan.cpp

@@ -2245,7 +2245,8 @@ core::PlanNodePtr VeloxQueryPlanConverterBase::toVeloxQueryPlan(
       node->id,
       toRowType(node->outputVariables, typeParser_),
       VectorSerde::Kind::kPresto,
-      toVeloxQueryPlan(node->source, tableWriteInfo, taskId));
+      toVeloxQueryPlan(node->source, tableWriteInfo, taskId),
+      true);
 }
 
 core::PlanNodePtr VeloxInteractiveQueryPlanConverter::toVeloxQueryPlan(

presto-native-execution/scripts/dockerfiles/centos-dependency.dockerfile

@@ -20,6 +20,8 @@ ENV PROMPT_ALWAYS_RESPOND=y
 ENV CC=/opt/rh/gcc-toolset-12/root/bin/gcc
 ENV CXX=/opt/rh/gcc-toolset-12/root/bin/g++
 ENV ARM_BUILD_TARGET=${ARM_BUILD_TARGET}
+ENV CUDA_VERSION="12.9"
+ENV UCX_VERSION="1.19.0"
 
 RUN mkdir -p /scripts /velox/scripts
 COPY scripts /scripts
@@ -35,7 +37,8 @@ RUN bash -c "mkdir build && \
                  source ../velox/scripts/setup-centos-adapters.sh && \
                  install_adapters && \
                  install_clang15 && \
-                 install_cuda 12.8) && \
+                 install_cuda && \
+                 install_ucx) && \
     rm -rf build"
 
 # put CUDA binaries on the PATH

presto-native-execution/scripts/dockerfiles/prestissimo-runtime.dockerfile

@@ -31,8 +31,12 @@ RUN --mount=type=cache,target=/root/.ccache,sharing=locked \
     EXTRA_CMAKE_FLAGS=${EXTRA_CMAKE_FLAGS} \
     NUM_THREADS=${NUM_THREADS} make --directory="/prestissimo/" cmake-and-build BUILD_TYPE=${BUILD_TYPE} BUILD_DIR=${BUILD_DIR} BUILD_BASE_DIR=${BUILD_BASE_DIR} && \
     ccache -sz -v'
-RUN !(LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/local/lib:/usr/local/lib64 ldd /prestissimo/${BUILD_BASE_DIR}/${BUILD_DIR}/presto_cpp/main/presto_server  | grep "not found") && \
-    LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/local/lib:/usr/local/lib64 ldd /prestissimo/${BUILD_BASE_DIR}/${BUILD_DIR}/presto_cpp/main/presto_server | awk 'NF == 4 { system("cp " $3 " /runtime-libraries") }'
+RUN !(LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/local/lib:/usr/local/lib64:/usr/local/cuda/compat ldd /prestissimo/${BUILD_BASE_DIR}/${BUILD_DIR}/presto_cpp/main/presto_server  | grep "not found") && \
+    LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/local/lib:/usr/local/lib64:/usr/local/cuda/compat ldd /prestissimo/${BUILD_BASE_DIR}/${BUILD_DIR}/presto_cpp/main/presto_server | awk 'NF == 4 { system("cp " $3 " /runtime-libraries") }'
+
+RUN cp -rf /usr/local/cuda/targets/*/lib /runtime-libraries/cuda
+RUN cp /usr/local/cuda/compat/libnvidia-ptxjitcompiler.*.*.* /runtime-libraries/cuda
+RUN cp -rf /usr/local/lib/ucx /runtime-libraries/ucx
 
 #/////////////////////////////////////////////
 #          prestissimo-runtime
@@ -45,8 +49,18 @@ ENV BUILD_DIR=""
 
 COPY --chmod=0775 --from=prestissimo-image /prestissimo/${BUILD_BASE_DIR}/${BUILD_DIR}/presto_cpp/main/presto_server /usr/bin/
 COPY --chmod=0775 --from=prestissimo-image /runtime-libraries/* /usr/lib64/prestissimo-libs/
+COPY --chmod=0775 --from=prestissimo-image /runtime-libraries/ucx /usr/lib64/prestissimo-libs/ucx
+COPY --chmod=0775 --from=prestissimo-image /runtime-libraries/cuda /usr/lib64/prestissimo-libs/cuda
 COPY --chmod=0755 ./etc /opt/presto-server/etc
 COPY --chmod=0775 ./entrypoint.sh /opt/entrypoint.sh
 RUN echo "/usr/lib64/prestissimo-libs" > /etc/ld.so.conf.d/prestissimo.conf && ldconfig
+RUN echo "/usr/lib64/prestissimo-libs/cuda" >> /etc/ld.so.conf.d/cuda.conf && ldconfig
+RUN echo "/usr/lib64/prestissimo-libs/ucx" >> /etc/ld.so.conf.d/prestissimo.conf && ldconfig
+
+RUN dnf install -y librdmacm libibverbs
+
+RUN rpm --import https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64/7fa2af80.pub && \
+    dnf config-manager --add-repo "https://developer.download.nvidia.com/devtools/repos/rhel$(source /etc/os-release; echo ${VERSION_ID%%.*})/$(rpm --eval '%{_arch}' | sed s/aarch/arm/)/" && \
+    dnf install -y nsight-systems-cli
 
 ENTRYPOINT ["/opt/entrypoint.sh"]