[native] Adds the TPC-DS connector

Co-authored-by: Pramod Satya <pramod.satya@ibm.com>

presto-docs/src/main/sphinx/presto-cpp.rst

@@ -49,4 +49,6 @@ Only specific connectors are supported in the Presto C++ evaluation engine.
 
 * Iceberg connector supports both V1 and V2 tables, including tables with delete files.
 
-* TPCH connector, with ``tpch.naming=standard`` catalog property.
+* TPCH connector, with ``tpch.naming=standard`` catalog property.
+
+* TPCDS connector.

presto-native-execution/etc/catalog/tpcds.properties

@@ -0,0 +1 @@
+connector.name=tpcds

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

@@ -14,6 +14,7 @@ add_subdirectory(types)
 add_subdirectory(http)
 add_subdirectory(common)
 add_subdirectory(thrift)
+add_subdirectory(connectors)
 
 add_library(
   presto_server_lib
@@ -92,7 +93,7 @@ add_executable(presto_server PrestoMain.cpp)
 # "undefined reference to `vtable for velox::connector::tpch::TpchTableHandle`"
 # TODO: Fix these errors.
 target_link_libraries(presto_server presto_server_lib velox_hive_connector
-                      velox_tpch_connector)
+                      velox_tpch_connector presto_tpcds_connector)
 
 if(PRESTO_ENABLE_REMOTE_FUNCTIONS)
   add_library(presto_server_remote_function JsonSignatureParser.cpp

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

@@ -259,6 +259,8 @@ void PrestoServer::run() {
       std::make_unique<SystemPrestoToVeloxConnector>("system"));
   registerPrestoToVeloxConnector(
       std::make_unique<SystemPrestoToVeloxConnector>("$system@system"));
+  registerPrestoToVeloxConnector(
+      std::make_unique<TpcdsPrestoToVeloxConnector>("tpcds"));
 
   initializeVeloxMemory();
   initializeThreadPools();

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

@@ -0,0 +1,25 @@
+# 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.
+
+cmake_policy(SET CMP0079 NEW)
+
+add_library(presto_tpcds_connector OBJECT tpcds/TpcdsConnector.cpp)
+target_link_libraries(presto_tpcds_connector velox_connector tpcds_gen fmt::fmt)
+
+# Without this hack, there are multiple link errors similar to the one below
+# only on GCC. "undefined reference to `vtable for
+# velox::connector::tpcds::TpcdsTableHandle`. TODO: Fix this hack.
+target_link_libraries(velox_exec_test_lib presto_tpcds_connector)
+
+add_subdirectory(tpcds)

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

@@ -0,0 +1,50 @@
+# 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.
+
+cmake_minimum_required(VERSION 3.14)
+cmake_policy(SET CMP0079 NEW)
+
+project(TPCDS)
+
+if(CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+  add_compile_options(-Wno-deprecated-declarations -Wno-writable-strings
+                      -Wno-missing-field-initializers)
+endif()
+
+# This stringop-overflow warning seems to have lots of false positives and has
+# been the source of a lot of compiler bug reports (e.g.
+# https://gcc.gnu.org/bugzilla/show_bug.cgi?id=99578), which causes
+# parquet-amalgamation.cpp to fail to compile. For now, we disable this warning
+# on the affected compiler (GCC).
+if(CMAKE_CXX_COMPILER_ID MATCHES "GNU")
+  add_compile_options(-Wno-stringop-overflow -Wno-write-strings)
+endif()
+
+# Include directories
+include_directories(dsdgen/include)
+include_directories(dsdgen/include/dsdgen-c)
+include_directories(include)
+include_directories(..)
+
+# Add subdirectories
+add_subdirectory(dsdgen/dsdgen-c)
+
+add_library(append_info OBJECT include/append_info-c.cpp)
+target_link_libraries(append_info velox_vector_test_lib Folly::folly xsimd)
+target_link_libraries(dsdgen_c append_info)
+
+add_library(tpcds_gen TpcdsGen.cpp DSDGenIterator.cpp)
+target_include_directories(tpcds_gen PUBLIC dsdgen/include)
+target_link_libraries(tpcds_gen velox_memory velox_vector dsdgen_c append_info
+                      fmt::fmt)

presto-native-execution/presto_cpp/main/connectors/tpcds/DSDGenIterator.cpp

@@ -0,0 +1,101 @@
+/*
+ * 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 "DSDGenIterator.h"
+#include "address.h"
+#include "build_support.h"
+#include "config.h"
+#include "dist.h"
+#include "genrand.h"
+#include "iostream"
+#include "parallel.h"
+#include "params.h"
+#include "porting.h"
+#include "scaling.h"
+#include "tdefs.h"
+
+namespace facebook::velox::tpcds {
+
+void InitializeDSDgen(
+    double scale,
+    vector_size_t parallel,
+    vector_size_t child,
+    DSDGenContext& dsdGenContext) {
+  dsdGenContext.Reset();
+  resetCountCount();
+
+  std::string scaleStr = std::to_string(scale);
+  set_str("SCALE", scaleStr.c_str(), dsdGenContext);
+  std::string parallelStr = std::to_string(parallel);
+  set_str("PARALLEL", parallelStr.c_str(), dsdGenContext);
+  std::string childStr = std::to_string(child);
+  set_str("CHILD", childStr.c_str(), dsdGenContext);
+
+  init_rand(dsdGenContext); // no random numbers without this
+}
+
+std::string getQuery(int query) {
+  if (query <= 0 || query > TPCDS_QUERIES_COUNT) {
+    throw std::exception();
+  }
+  return TPCDS_QUERIES[query - 1];
+}
+
+DSDGenIterator::DSDGenIterator(
+    double scaleFactor,
+    vector_size_t parallel,
+    vector_size_t child) {
+  table_defs.resize(DBGEN_VERSION); // there are 24 TPC-DS tables
+  VELOX_CHECK_GE(scaleFactor, 0, "Tpcds scale factor must be non-negative");
+  dsdgenCtx_.scaleFactor = scaleFactor;
+  InitializeDSDgen(scaleFactor, parallel, child, dsdgenCtx_);
+}
+
+void DSDGenIterator::initializeTable(
+    std::vector<VectorPtr> children,
+    int table_id) {
+  auto tdef = getSimpleTdefsByNumber(table_id, dsdgenCtx_);
+  tpcds_table_def table_def;
+  table_def.name = tdef->name;
+  table_def.fl_child = tdef->flags & FL_CHILD ? 1 : 0;
+  table_def.fl_small = tdef->flags & FL_SMALL ? 1 : 0;
+  table_def.first_column = tdef->nFirstColumn;
+  table_def.children = children;
+  table_def.dsdGenContext = &dsdgenCtx_;
+  table_defs[table_id] = std::make_unique<tpcds_table_def>(table_def);
+}
+
+std::vector<std::unique_ptr<tpcds_table_def>>& DSDGenIterator::getTableDefs() {
+  return table_defs;
+};
+
+tpcds_builder_func DSDGenIterator::GetTDefFunctionByNumber(int table_id) {
+  auto table_funcs = getTdefFunctionsByNumber(table_id);
+  return table_funcs->builder;
+}
+
+void DSDGenIterator::initTableOffset(int32_t table_id, size_t offset) {
+  row_skip(table_id, offset, dsdgenCtx_);
+}
+void DSDGenIterator::genRow(int32_t table_id, size_t index) {
+  auto builder_func = GetTDefFunctionByNumber(table_id);
+  builder_func((void*)&table_defs, index, dsdgenCtx_);
+  row_stop(table_id, dsdgenCtx_);
+}
+
+int64_t DSDGenIterator::getRowCount(int32_t table) {
+  return get_rowcount(table, dsdgenCtx_);
+}
+
+} // namespace facebook::velox::tpcds

presto-native-execution/presto_cpp/main/connectors/tpcds/DSDGenIterator.h

@@ -0,0 +1,67 @@
+/*
+ * 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 <memory>
+
+#include <presto_cpp/main/connectors/tpcds/dsdgen/include/dsdgen-c/dist.h>
+#include <presto_cpp/main/connectors/tpcds/dsdgen/include/tpcds_constants.hpp>
+#include <presto_cpp/main/connectors/tpcds/include/append_info-c.hpp>
+
+namespace facebook::velox::tpcds {
+
+typedef int64_t ds_key_t;
+
+typedef int (*tpcds_builder_func)(void*, ds_key_t, DSDGenContext& dsdgenCtx);
+
+void InitializeDSDgen(
+    double scale,
+    vector_size_t parallel,
+    vector_size_t child,
+    DSDGenContext& dsdGenContext);
+
+std::string getQuery(int query);
+
+/// This class exposes a thread-safe and reproducible iterator over TPC-DS
+/// synthetically generated data, backed by DSDGEN.
+class DSDGenIterator {
+ public:
+  explicit DSDGenIterator(
+      double scaleFactor,
+      vector_size_t parallel,
+      vector_size_t child);
+
+  void initializeTable(std::vector<VectorPtr> children, int table);
+
+  std::vector<std::unique_ptr<tpcds_table_def>>& getTableDefs();
+
+  // Before generating records using the gen*() functions below, call the
+  // appropriate init*() function to correctly initialize the seed given the
+  // offset to be generated.
+  void initTableOffset(int32_t table_id, size_t offset);
+
+  // Generate different types of records.
+  void genRow(int32_t table_id, size_t index);
+
+  ds_key_t getRowCount(int32_t table_id);
+
+  tpcds_builder_func GetTDefFunctionByNumber(int table_id);
+
+ protected:
+  DSDGenContext dsdgenCtx_;
+  std::vector<std::unique_ptr<tpcds_table_def>> table_defs;
+};
+
+} // namespace facebook::velox::tpcds