K-Means UDTF & Intel oneDAL initial integration

CMakeLists.txt

@@ -304,6 +304,20 @@ if(ENABLE_MLPACK)
   add_definitions("-DHAVE_MLPACK")
 endif()
 
+option(ENABLE_ONEDAL "Use Intel oneDAL" OFF)
+if(ENABLE_ONEDAL)
+  set(USE_DPCPP no)
+  set(USE_NEW_IFACES no)
+  set(TARGET_LINK static)
+  find_package(oneDAL REQUIRED)
+  list(APPEND ONEDAL_LIBRARIES "libonedal_core.a")
+  list(APPEND ONEDAL_LIBRARIES "libonedal_thread.a")
+  list(APPEND LATEST_TBB_LIBRARIES "$ENV{TBBROOT}/lib/intel64/gcc4.8/libtbb.so.12")
+  list(APPEND LATEST_TBB_LIBRARIES "$ENV{TBBROOT}/lib/intel64/gcc4.8/libtbbmalloc.so.2")
+  include_directories(${oneDAL_INCLUDE_DIRS})
+  add_definitions("-DHAVE_ONEDAL")
+endif()
+
 if(MSVC)
   include_directories(include_directories("${LIBS_PATH}/include/pdcurses"))
 else()
@@ -730,6 +744,11 @@ endif()
 
 list(APPEND MAPD_LIBRARIES ${TBB_LIBS})
 
+if(ENABLE_ONEDAL)
+  list(APPEND MAPD_LIBRARIES ${ONEDAL_LIBRARIES})
+endif()
+
+
 if(ENABLE_CANONICAL_RAFT)
   list(APPEND MAPD_LIBRARIES raft_canonical)
 endif()
@@ -812,11 +831,11 @@ add_custom_target(rerun_cmake ALL
   )
 add_dependencies(omnisci_server rerun_cmake)
 
-target_link_libraries(omnisci_server mapd_thrift thrift_handler ${MAPD_LIBRARIES} ${Boost_LIBRARIES} ${CMAKE_DL_LIBS} ${CUDA_LIBRARIES} ${PROFILER_LIBS} ${ZLIB_LIBRARIES} ${LOCALE_LINK_FLAG})
+target_link_libraries(omnisci_server mapd_thrift thrift_handler ${MAPD_LIBRARIES} ${Boost_LIBRARIES} ${CMAKE_DL_LIBS} ${CUDA_LIBRARIES} ${PROFILER_LIBS} ${ZLIB_LIBRARIES} ${LOCALE_LINK_FLAG} ${LATEST_TBB_LIBRARIES})
 
 target_link_libraries(initdb mapd_thrift thrift_handler ${MAPD_LIBRARIES} ${Boost_LIBRARIES} ${CMAKE_DL_LIBS}
     ${CUDA_LIBRARIES} ${PROFILER_LIBS} ${ZLIB_LIBRARIES} ${BLOSC_LIBRARIES}
-    ${LOCALE_LINK_FLAG})
+    ${LOCALE_LINK_FLAG} ${LATEST_TBB_LIBRARIES})
 
 macro(set_dpkg_arch arch_in arch_out)
   if("${arch_in}" STREQUAL "x86_64")

QueryEngine/CMakeLists.txt

@@ -237,7 +237,7 @@ add_custom_target(QueryEngineFunctionsTargets
         ${CMAKE_CURRENT_BINARY_DIR}/GeosRuntime.bc
     )
 
-set(TABLE_FUNCTION_HEADERS ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/TableFunctions.hpp ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/TableFunctionsTesting.hpp)
+set(TABLE_FUNCTION_HEADERS ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/TableFunctions.hpp ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/TableFunctionsTesting.hpp ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/MLFunctions.hpp)
 if(ENABLE_MLPACK)
   list(APPEND TABLE_FUNCTION_HEADERS ${CMAKE_CURRENT_SOURCE_DIR}/TableFunctions/MLFunctions.hpp)
 endif()

QueryEngine/OmniSciTypes.h

@@ -155,6 +155,8 @@ struct GeoMultiPolygon {
 
 template <typename T>
 struct Column {
+  using value_type = T;
+
   T* ptr_;        // row data
   int64_t size_;  // row count
 
@@ -216,6 +218,8 @@ struct Column {
 */
 template <typename T>
 struct ColumnList {
+  using value_type = T;
+
   int8_t** ptrs_;     // ptrs to columns data
   int64_t num_cols_;  // the length of columns list
   int64_t size_;      // the size of columns

QueryEngine/TableFunctions/TableFunctions.hpp

@@ -17,6 +17,11 @@
 #include "../../QueryEngine/OmniSciTypes.h"
 #include "../../Shared/funcannotations.h"
 
+#ifdef HAVE_ONEDAL
+#include <type_traits>
+#include "daal.h"
+#endif
+
 // clang-format off
 /*
   UDTF: row_copier(Column<double>, RowMultiplier) -> Column<double>
@@ -248,4 +253,72 @@ EXTENSION_NOINLINE int32_t column_list_row_sum__cpu_(const ColumnList<int32_t>&
   return output_num_rows;
 }
 
+#ifdef HAVE_ONEDAL
+
+// clang-format off
+/*
+  UDTF: k_means(Cursor<Column<int>, ColumnList<float>>, int, int, RowMultiplier) -> Column<int>, Column<int>
+ */
+// clang-format on
+
+EXTENSION_NOINLINE int32_t k_means(const Column<int>& input_ids,
+                                   const ColumnList<float>& input,
+                                   const int num_clusters,
+                                   const int num_iterations,
+                                   const int output_multiplier,
+                                   Column<int>& output_ids,
+                                   Column<int>& output_cluster) {
+  using namespace daal::algorithms;
+  using namespace daal::data_management;
+
+  // Float data type
+  using float_type =
+      std::remove_cv_t<std::remove_reference_t<decltype(input)>>::value_type;
+
+  // Assignments data type
+  using assignments_type =
+      std::remove_cv_t<std::remove_reference_t<decltype(output_cluster)>>::value_type;
+
+  // Data dimensions
+  const size_t num_rows = input_ids.size();
+  const size_t num_columns = input.numCols();
+
+  // Prepare input data as structure of arrays (SOA) as columnar format (zero-copy)
+  const auto dataTable = SOANumericTable::create(num_columns, num_rows);
+  for (size_t i = 0; i < num_columns; ++i) {
+    dataTable->setArray<float_type>(input[i].ptr_, i);
+  }
+
+  // Initialization phase of K-Means
+  kmeans::init::Batch<float_type, kmeans::init::randomDense> init(num_clusters);
+  init.input.set(kmeans::init::data, dataTable);
+  init.compute();
+  const NumericTablePtr centroids = init.getResult()->get(kmeans::init::centroids);
+
+  // Prepare output data as homogeneous numeric table to allow zero-copy for assignments
+  const auto assignmentsTable =
+      HomogenNumericTable<assignments_type>::create(output_cluster.ptr_, 1, num_rows);
+  const kmeans::ResultPtr result(new kmeans::Result);
+  result->set(kmeans::assignments, assignmentsTable);
+  result->set(kmeans::objectiveFunction,
+              HomogenNumericTable<float>::create(1, 1, NumericTable::doAllocate));
+  result->set(kmeans::nIterations,
+              HomogenNumericTable<int>::create(1, 1, NumericTable::doAllocate));
+
+  // Clustering phase of K-Means
+  kmeans::Batch<> algorithm(num_clusters, num_iterations);
+  algorithm.input.set(kmeans::data, dataTable);
+  algorithm.input.set(kmeans::inputCentroids, centroids);
+  algorithm.parameter().resultsToEvaluate = kmeans::computeAssignments;
+  algorithm.setResult(result);
+  algorithm.compute();
+
+  // Copying from input_ids to output_ids
+  output_ids = input_ids;
+
+  return num_rows;
+}
+
+#endif
+
 #include "MLFunctions.hpp"

Tests/CMakeLists.txt

@@ -115,7 +115,7 @@ endif()
 # Tests + Microbenchmarks
 add_executable(TableUpdateDeleteBenchmark TableUpdateDeleteBenchmark.cpp)
 
-set(EXECUTE_TEST_LIBS gtest mapd_thrift QueryRunner ${MAPD_LIBRARIES} ${CMAKE_DL_LIBS} ${CUDA_LIBRARIES} ${Boost_LIBRARIES} ${ZLIB_LIBRARIES} ${PROFILER_LIBS})
+set(EXECUTE_TEST_LIBS gtest mapd_thrift QueryRunner ${MAPD_LIBRARIES} ${CMAKE_DL_LIBS} ${CUDA_LIBRARIES} ${Boost_LIBRARIES} ${ZLIB_LIBRARIES} ${PROFILER_LIBS} ${LATEST_TBB_LIBRARIES})
 set(THRIFT_HANDLER_TEST_LIBRARIES thrift_handler ${EXECUTE_TEST_LIBS})
 
 # Replace Licensing library with TestLicensing library

scripts/mapd-deps-prebuilt.sh

@@ -104,6 +104,10 @@ if [ "$ID" == "ubuntu" ] ; then
       python-yaml \
       libxerces-c-dev \
       swig
+  sudo wget https://apt.repos.intel.com/intel-gpg-keys/GPG-PUB-KEY-INTEL-SW-PRODUCTS.PUB -O - | sudo apt-key add -
+  sudo add-apt-repository "deb https://apt.repos.intel.com/oneapi all main"
+  sudo $PACKAGER update
+  sudo $PACKAGER install intel-oneapi-dal-devel
 
 # Set up gcc-8 as default gcc
 sudo update-alternatives \
@@ -148,6 +152,8 @@ VK_LAYER_PATH=\$PREFIX/etc/vulkan/explicit_layer.d
 CMAKE_PREFIX_PATH=\$PREFIX:\$CMAKE_PREFIX_PATH
 
 export LD_LIBRARY_PATH PATH VULKAN_SDK VK_LAYER_PATH CMAKE_PREFIX_PATH
+
+source /opt/intel/oneapi/setvars.sh > /dev/null
 EOF
 
   PROFPATH=/etc/profile.d/xx-mapd-deps.sh

scripts/mapd-deps.sh.in

@@ -27,3 +27,5 @@ export PATH=$PREFIX/maven/bin:$PATH
 
 export VULKAN_SDK=$PREFIX
 export VK_LAYER_PATH=$PREFIX/etc/vulkan/explicit_layer.d
+
+source /opt/intel/oneapi/setvars.sh  > /dev/null