Merge pull request #50 from sirius-db/type_decimal

Support basic usage of decimal type.

Author: Yifei-yang7

src/cuda/expression_executor/gpu_dispatch_materialize.cu

@@ -41,6 +41,36 @@ struct MaterializeNumeric
   }
 };
 
+//----------Decimals----------//
+template <typename T>
+struct MaterializeDecimal
+{
+  static std::unique_ptr<cudf::column> Do(const T* input_data,
+                                          const uint64_t* row_ids,
+                                          uint64_t row_id_count,
+                                          const cudf::data_type& cudf_type,
+                                          rmm::device_async_resource_ref mr,
+                                          rmm::cuda_stream_view stream = rmm::cuda_stream_default)
+  {
+    // Define the thrust execution policy
+    rmm::mr::thrust_allocator<uint8_t> thrust_allocator(stream, mr);
+    auto exec = thrust::cuda::par(thrust_allocator).on(stream);
+
+    // Allocate output buffer
+    rmm::device_uvector<T> output_data(row_id_count, stream, mr);
+
+    // Gather the input data
+    thrust::gather(exec, row_ids, row_ids + row_id_count, input_data, output_data.begin());
+
+    // Construct and return a cudf::column
+    return std::make_unique<cudf::column>(cudf_type,
+                                          static_cast<cudf::size_type>(row_id_count),
+                                          output_data.release(),
+                                          rmm::device_buffer(0, stream, mr),
+                                          0);
+  }
+};
+
 //----------Strings----------//
 struct MaterializeString
 {
@@ -214,6 +244,34 @@ std::unique_ptr<cudf::column> GpuDispatcher::DispatchMaterialize(const GPUColumn
                                    input->row_id_count,
                                    mr,
                                    stream);
+    case GPUColumnTypeId::DECIMAL: {
+      switch (input->data_wrapper.getColumnTypeSize()) {
+        case sizeof(int32_t): {
+          // cudf decimal type uses negative scale, same for below
+          cudf::data_type cudf_type(cudf::type_id::DECIMAL32,
+                                    -input->data_wrapper.type.GetDecimalTypeInfo()->scale_);
+          return MaterializeDecimal<int32_t>::Do(reinterpret_cast<const int32_t*>(input_data),
+                                                 input->row_ids,
+                                                 input->row_id_count,
+                                                 cudf_type,
+                                                 mr,
+                                                 stream);
+        }
+        case sizeof(int64_t): {
+          cudf::data_type cudf_type(cudf::type_id::DECIMAL64,
+                                    -input->data_wrapper.type.GetDecimalTypeInfo()->scale_);
+          return MaterializeDecimal<int64_t>::Do(reinterpret_cast<const int64_t*>(input_data),
+                                                 input->row_ids,
+                                                 input->row_id_count,
+                                                 cudf_type,
+                                                 mr,
+                                                 stream);
+        }
+        default:
+          throw NotImplementedException("Unsupported sirius DECIMAL column type size in `Dispatch[Materialize]`: %zu",
+                                        input->data_wrapper.getColumnTypeSize());
+      }
+    }
     default:
       throw InternalException("Unsupported sirius column type in `Dispatch[Materialize]`: %d",
                               static_cast<int>(input->data_wrapper.type.id()));

src/cuda/operator/arbitrary_expression.cu

@@ -131,7 +131,7 @@ __global__ void table_scan_expression(uint8_t **col, uint64_t** offset, uint8_t
         for (int expr = 0; expr < num_expr; expr++) {
             if (threadIdx.x + ITEM * B < num_tile_items) {
 
-                if (data_type[expr] == INT32 || data_type[expr] == DATE) {
+                if (data_type[expr] == INT32 || data_type[expr] == DATE || data_type[expr] == DECIMAL32) {
                     uint64_t item_idx = tile_offset + threadIdx.x + ITEM * B;
                     int item = (reinterpret_cast<int*>(col[expr]))[item_idx];
 
@@ -141,15 +141,15 @@ __global__ void table_scan_expression(uint8_t **col, uint64_t** offset, uint8_t
 
                     selection_flags[ITEM] = device_comparison<int>(item, constant, compare_mode[expr]);
 
-                } else if (data_type[expr] == INT64) {
+                } else if (data_type[expr] == INT64 || data_type[expr] == DECIMAL64) {
                     uint64_t item_idx = tile_offset + threadIdx.x + ITEM * B;
-                    uint64_t item = (reinterpret_cast<uint64_t*>(col[expr]))[item_idx];
+                    int64_t item = (reinterpret_cast<int64_t*>(col[expr]))[item_idx];
 
                     uint64_t start_constant_offset = constant_offset[expr]; 
-                    uint64_t constant;
-                    memcpy(&constant, constant_compare + start_constant_offset, sizeof(uint64_t));
+                    int64_t constant;
+                    memcpy(&constant, constant_compare + start_constant_offset, sizeof(int64_t));
 
-                    selection_flags[ITEM] = device_comparison<uint64_t>(item, constant, compare_mode[expr]);
+                    selection_flags[ITEM] = device_comparison<int64_t>(item, constant, compare_mode[expr]);
 
                 } else if (data_type[expr] == FLOAT32) {
                     uint64_t item_idx = tile_offset + threadIdx.x + ITEM * B;

src/cuda/operator/materialize.cu

@@ -60,6 +60,8 @@ template
 __global__ void materialize_expression<double, BLOCK_THREADS, ITEMS_PER_THREAD>(const double *a, double* result, uint64_t *row_ids, uint64_t N);
 template
 __global__ void materialize_expression<uint8_t, BLOCK_THREADS, ITEMS_PER_THREAD>(const uint8_t *a, uint8_t* result, uint64_t *row_ids, uint64_t N);
+template
+__global__ void materialize_expression<int64_t, BLOCK_THREADS, ITEMS_PER_THREAD>(const int64_t *a, int64_t* result, uint64_t *row_ids, uint64_t N);
 
 template <typename T>
 void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_len) {
@@ -189,5 +191,7 @@ template
 void materializeExpression<double>(double *a, double*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
 template
 void materializeExpression<uint8_t>(uint8_t *a, uint8_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
+template
+void materializeExpression<int64_t>(int64_t *a, int64_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
 
 } // namespace duckdb

src/expression_executor/gpu_expression_executor.cpp

@@ -198,7 +198,7 @@ void GpuExpressionExecutor::Execute(const GPUIntermediateRelation& input_relatio
     if (result->type().id() != cudf_return_type.id())
     {
       result =
-        cudf::cast(result->view(), cudf_return_type, cudf::get_default_stream(), resource_ref);
+        cudf::cast(result->view(), cudf_return_type, execution_stream, resource_ref);
     }
 
     // Transfer to output relation (zero copy)

src/expression_executor/specializations/gpu_execute_function.cpp

@@ -149,7 +149,7 @@ struct NumericBinaryFunctionDispatcher
       : executor(exec)
   {}
 
-  // Left scalar binary operator
+  // Left scalar binary operator for numeric types
   template <typename T>
   std::unique_ptr<cudf::column> DoLeftScalarBinaryOp(const T& left_value,
                                                      const cudf::column_view& right,
@@ -165,7 +165,24 @@ struct NumericBinaryFunctionDispatcher
                                   executor.resource_ref);
   }
 
-  // Right scalar binary operator
+  // Left scalar binary operator for decimal types
+  template <typename T>
+  std::unique_ptr<cudf::column> DoLeftScalarBinaryOp(typename T::rep left_value,
+                                                     numeric::scale_type scale,
+                                                     const cudf::column_view& right,
+                                                     const cudf::data_type& return_type)
+  {
+    auto left_decimal_scalar = cudf::fixed_point_scalar<T>(
+      left_value, scale, true, executor.execution_stream, executor.resource_ref);
+    return cudf::binary_operation(left_decimal_scalar,
+                                  right,
+                                  BinOp,
+                                  return_type,
+                                  executor.execution_stream,
+                                  executor.resource_ref);
+  }
+
+  // Right scalar binary operator for numeric types
   template <typename T>
   std::unique_ptr<cudf::column> DoRightScalarBinaryOp(const cudf::column_view& left,
                                                       const T& right_value,
@@ -181,6 +198,23 @@ struct NumericBinaryFunctionDispatcher
                                   executor.resource_ref);
   }
 
+  // Right scalar binary operator for decimal types
+  template <typename T>
+  std::unique_ptr<cudf::column> DoRightScalarBinaryOp(const cudf::column_view& left,
+                                                      typename T::rep right_value,
+                                                      numeric::scale_type scale,
+                                                      const cudf::data_type& return_type)
+  {
+    auto right_decimal_scalar = cudf::fixed_point_scalar<T>(
+      right_value, scale, true, executor.execution_stream, executor.resource_ref);
+    return cudf::binary_operation(left,
+                                  right_decimal_scalar,
+                                  BinOp,
+                                  return_type,
+                                  executor.execution_stream,
+                                  executor.resource_ref);
+  }
+
   // Dispatch operator
   std::unique_ptr<cudf::column> operator()(const BoundFunctionExpression& expr,
                                            GpuExpressionState* state)
@@ -205,6 +239,17 @@ struct NumericBinaryFunctionDispatcher
           return DoLeftScalarBinaryOp(left_value.GetValue<float_t>(), right->view(), return_type);
         case cudf::type_id::FLOAT64:
           return DoLeftScalarBinaryOp(left_value.GetValue<double_t>(), right->view(), return_type);
+        case cudf::type_id::DECIMAL32:
+          // cudf decimal type uses negative scale, same for below
+          return DoLeftScalarBinaryOp<numeric::decimal32>(
+            left_value.GetValueUnsafe<int32_t>(),
+            numeric::scale_type{-duckdb::DecimalType::GetScale(left_value.type())},
+            right->view(), return_type);
+        case cudf::type_id::DECIMAL64:
+          return DoLeftScalarBinaryOp<numeric::decimal64>(
+            left_value.GetValueUnsafe<int64_t>(),
+            numeric::scale_type{-duckdb::DecimalType::GetScale(left_value.type())},
+            right->view(), return_type);
         case cudf::type_id::BOOL8:
           throw NotImplementedException("Execute[Function]: Boolean types not supported for "
                                         "numeric binary operations!");
@@ -229,6 +274,17 @@ struct NumericBinaryFunctionDispatcher
           return DoRightScalarBinaryOp(left->view(), right_value.GetValue<float_t>(), return_type);
         case cudf::type_id::FLOAT64:
           return DoRightScalarBinaryOp(left->view(), right_value.GetValue<double_t>(), return_type);
+        case cudf::type_id::DECIMAL32:
+          // cudf decimal type uses negative scale, same for below
+          return DoRightScalarBinaryOp<numeric::decimal32>(
+            left->view(), right_value.GetValueUnsafe<int32_t>(),
+            numeric::scale_type{-duckdb::DecimalType::GetScale(right_value.type())},
+            return_type);
+        case cudf::type_id::DECIMAL64:
+          return DoRightScalarBinaryOp<numeric::decimal64>(
+            left->view(), right_value.GetValueUnsafe<int64_t>(),
+            numeric::scale_type{-duckdb::DecimalType::GetScale(right_value.type())},
+            return_type);
         case cudf::type_id::BOOL8:
           throw NotImplementedException("Execute[Function]: Boolean types not supported for "
                                         "numeric binary operations!");

src/gpu_buffer_manager.cpp

@@ -18,6 +18,9 @@ GPUBufferManager::customCudaMalloc<int>(size_t size, int gpu, bool caching);
 template uint64_t*
 GPUBufferManager::customCudaMalloc<uint64_t>(size_t size, int gpu, bool caching);
 
+template int64_t*
+GPUBufferManager::customCudaMalloc<int64_t>(size_t size, int gpu, bool caching);
+
 template uint8_t*
 GPUBufferManager::customCudaMalloc<uint8_t>(size_t size, int gpu, bool caching);
 

src/gpu_columns.cpp

@@ -1,9 +1,25 @@
 #include "gpu_columns.hpp"
 #include "gpu_buffer_manager.hpp"
 #include "log/logging.hpp"
+#include "duckdb/common/types/decimal.hpp"
 
 namespace duckdb {
 
+size_t GPUDecimalTypeInfo::GetDecimalTypeSize() const {
+    if (width_ <= Decimal::MAX_WIDTH_INT16) {
+        return sizeof(int16_t);
+    } else if (width_ <= Decimal::MAX_WIDTH_INT32) {
+        return sizeof(int32_t);
+    } else if (width_ <= Decimal::MAX_WIDTH_INT64) {
+        return sizeof(int64_t);
+    } else if (width_ <= Decimal::MAX_WIDTH_INT128) {
+        return sizeof(__int128_t);
+    } else {
+        throw InternalException("Decimal has a width of %d which is bigger than the maximum supported width of %d",
+                                width_, DecimalType::MaxWidth());
+    }
+}
+
 DataWrapper::DataWrapper(GPUColumnType _type, uint8_t* _data, size_t _size) : data(_data), size(_size) {
     type = _type;
     num_bytes = size * getColumnTypeSize();
@@ -14,7 +30,7 @@ DataWrapper::DataWrapper(GPUColumnType _type, uint8_t* _data, uint64_t* _offset,
     data(_data), size(_size), type(_type), offset(_offset), num_bytes(_num_bytes), is_string_data(_is_string_data) {};
 
 size_t 
-DataWrapper::getColumnTypeSize() {
+DataWrapper::getColumnTypeSize() const {
     switch (type.id()) {
         case GPUColumnTypeId::INT32:
         case GPUColumnTypeId::DATE:
@@ -31,6 +47,13 @@ DataWrapper::getColumnTypeSize() {
             return sizeof(uint8_t);
         case GPUColumnTypeId::VARCHAR:
             return 128;
+        case GPUColumnTypeId::DECIMAL: {
+            GPUDecimalTypeInfo* decimal_type_info = type.GetDecimalTypeInfo();
+            if (decimal_type_info == nullptr) {
+                throw InternalException("`decimal_type_info` not set for DECIMAL type in `getColumnTypeSize`");
+            }
+            return decimal_type_info->GetDecimalTypeSize();
+        }
         default:
             throw duckdb::InternalException("Unsupported sirius column type in `getColumnTypeSize()`: %d",
                                             static_cast<int>(type.id()));
@@ -114,6 +137,23 @@ GPUColumn::convertToCudfColumn() {
             std::move(children)
         );
         return str_col;
+    } else if (data_wrapper.type.id() == GPUColumnTypeId::DECIMAL) {
+        cudf::data_type cudf_type;
+        switch (data_wrapper.getColumnTypeSize()) {
+            case sizeof(int32_t): {
+                // cudf decimal type uses negative scale, same for below
+                cudf_type = cudf::data_type(cudf::type_id::DECIMAL32, -data_wrapper.type.GetDecimalTypeInfo()->scale_);
+                break;
+            }
+            case sizeof(int64_t): {
+                cudf_type = cudf::data_type(cudf::type_id::DECIMAL64, -data_wrapper.type.GetDecimalTypeInfo()->scale_);
+                break;
+            }
+            default:
+                throw duckdb::InternalException("Unsupported sirius DECIMAL column type size in `convertToCudfColumn()`: %zu",
+                                                data_wrapper.getColumnTypeSize());
+        }
+        return cudf::column_view(cudf_type, size, reinterpret_cast<void*>(data_wrapper.data), nullptr, 0);
     }
     throw duckdb::InternalException("Unsupported sirius column type in `convertToCudfColumn()`: %d", data_wrapper.type.id());
 }
@@ -179,6 +219,27 @@ GPUColumn::setFromCudfColumn(cudf::column& cudf_column, bool _is_unique, int32_t
         data_wrapper.type = GPUColumnType(GPUColumnTypeId::BOOLEAN);
         data_wrapper.num_bytes = col_size * data_wrapper.getColumnTypeSize();
         data_wrapper.offset = nullptr;
+    } else if (col_type == cudf::data_type(cudf::type_id::TIMESTAMP_DAYS)) {
+        data_wrapper.is_string_data = false;
+        data_wrapper.type = GPUColumnType(GPUColumnTypeId::DATE);
+        data_wrapper.num_bytes = col_size * data_wrapper.getColumnTypeSize();
+        data_wrapper.offset = nullptr;
+    } else if (col_type.id() == cudf::type_id::DECIMAL32) {
+        data_wrapper.is_string_data = false;
+        data_wrapper.type = GPUColumnType(GPUColumnTypeId::DECIMAL);
+        // cudf decimal type uses negative scale, same for below
+        data_wrapper.type.SetDecimalTypeInfo(Decimal::MAX_WIDTH_INT32, -col_type.scale());
+        data_wrapper.num_bytes = col_size * data_wrapper.getColumnTypeSize();
+        data_wrapper.offset = nullptr;
+    } else if (col_type.id() == cudf::type_id::DECIMAL64) {
+        data_wrapper.is_string_data = false;
+        data_wrapper.type = GPUColumnType(GPUColumnTypeId::DECIMAL);
+        data_wrapper.type.SetDecimalTypeInfo(Decimal::MAX_WIDTH_INT64, -col_type.scale());
+        data_wrapper.num_bytes = col_size * data_wrapper.getColumnTypeSize();
+        data_wrapper.offset = nullptr;
+    } else {
+        throw NotImplementedException("Unsupported cudf data type in `setFromCudfColumn`: %d",
+                                      static_cast<int>(col_type.id()));
     }
 
     if (_row_ids != nullptr) {
@@ -224,6 +285,24 @@ GPUColumn::setFromCudfScalar(cudf::scalar& cudf_scalar, GPUBufferManager* gpuBuf
         callCudaMemcpyDeviceToDevice<uint8_t>(data_wrapper.data, reinterpret_cast<uint8_t*>(typed_scalar.data()), sizeof(uint8_t), 0);
         data_wrapper.type = GPUColumnType(GPUColumnTypeId::BOOLEAN);
         data_wrapper.num_bytes = sizeof(uint8_t);
+    } else if (scalar_type.id() == cudf::type_id::DECIMAL32){
+        auto& typed_scalar = static_cast<cudf::fixed_point_scalar<numeric::decimal32>&>(cudf_scalar);
+        data_wrapper.data = gpuBufferManager->customCudaMalloc<uint8_t>(sizeof(int32_t), 0, 0);
+        callCudaMemcpyDeviceToDevice<uint8_t>(data_wrapper.data, reinterpret_cast<uint8_t*>(typed_scalar.data()), sizeof(int32_t), 0);
+        data_wrapper.type = GPUColumnType(GPUColumnTypeId::DECIMAL);
+         // cudf decimal type uses negative scale, same for below
+        data_wrapper.type.SetDecimalTypeInfo(Decimal::MAX_WIDTH_INT32, -typed_scalar.type().scale());
+        data_wrapper.num_bytes = sizeof(int32_t);
+    } else if (scalar_type.id() == cudf::type_id::DECIMAL64){
+        auto& typed_scalar = static_cast<cudf::fixed_point_scalar<numeric::decimal64>&>(cudf_scalar);
+        data_wrapper.data = gpuBufferManager->customCudaMalloc<uint8_t>(sizeof(int64_t), 0, 0);
+        callCudaMemcpyDeviceToDevice<uint8_t>(data_wrapper.data, reinterpret_cast<uint8_t*>(typed_scalar.data()), sizeof(int64_t), 0);
+        data_wrapper.type = GPUColumnType(GPUColumnTypeId::DECIMAL);
+        data_wrapper.type.SetDecimalTypeInfo(Decimal::MAX_WIDTH_INT64, -typed_scalar.type().scale());
+        data_wrapper.num_bytes = sizeof(int64_t);
+    } else {
+        throw NotImplementedException("Unsupported cudf data type in `setFromCudfScalar`: %d",
+                                      static_cast<int>(scalar_type.id()));
     }
 
     data_wrapper.size = 1;