Merge remote-tracking branch 'origin' into type_date

Author: Yifei-yang7

docker/Dockerfile

@@ -1,29 +1,32 @@
-FROM nvidia/cuda:12.5.1-cudnn-devel-ubuntu22.04
+FROM nvidia/cuda:12.6.1-devel-ubuntu22.04
+ENV DEBIAN_FRONTEND=noninteractive
+WORKDIR /root
 
-ENV TZ=Asia/Dubai
-RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
+# Common dependencies
+RUN apt-get update && apt-get install -y git g++ cmake ninja-build libssl-dev wget vim unzip
 
-RUN apt update -y
-RUN apt-get update -y
-RUN apt install -y g++ make wget unzip vim htop git dstat
-RUN apt install -y software-properties-common
-RUN add-apt-repository -y ppa:ubuntu-toolchain-r/test
-RUN apt update
-RUN apt-get install -y build-essential
-RUN apt-get install -y libc-dev
-RUN update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-11 11
-RUN update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-11 11
+# CMake 3.30.4
+RUN wget https://github.com/Kitware/CMake/releases/download/v3.30.4/cmake-3.30.4-linux-x86_64.sh && \
+    chmod +x cmake-3.30.4-linux-x86_64.sh && \
+    ./cmake-3.30.4-linux-x86_64.sh --skip-license --prefix=/usr/local && \
+    rm cmake-3.30.4-linux-x86_64.sh
 
-# install cmake
-RUN wget https://github.com/Kitware/CMake/releases/download/v3.28.0/cmake-3.28.0-linux-x86_64.sh
-RUN mkdir /opt/cmake
-RUN sh cmake-3.28.0-linux-x86_64.sh --skip-license --prefix=/opt/cmake/
-RUN ln -s /opt/cmake/bin/cmake /usr/local/bin/cmake
-RUN apt install -y ninja-build
+# Install Miniconda
+ENV CONDA_DIR=/opt/conda
+RUN wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh && \
+    bash Miniconda3-latest-Linux-x86_64.sh -b -p $CONDA_DIR && \
+    rm Miniconda3-latest-Linux-x86_64.sh
+ENV PATH=$CONDA_DIR/bin:$PATH
 
-# Install nsight compute
-RUN apt-get install -y nsight-compute-2024.1.1
+# Create conda env with libcudf
+RUN conda update -n base -c defaults conda && \
+    conda create -n libcudf-env -y && \
+    conda run -n libcudf-env conda install -y -c rapidsai -c conda-forge -c nvidia rapidsai::libcudf
+ENV USE_CUDF=1
+ENV LIBCUDF_ENV_PREFIX=$CONDA_DIR/envs/libcudf-env
 
-RUN mkdir -p /working_dir
-WORKDIR /working_dir
-RUN git config --global --add safe.directory "*"
+# Activates libcudf env at runtime beginning
+RUN echo 'echo "[INFO] Initializing conda environment... please wait (conda activation can take a few seconds)"' >> ~/.bashrc && \
+    echo "source /opt/conda/etc/profile.d/conda.sh" >> ~/.bashrc && \
+    echo "conda activate libcudf-env" >> ~/.bashrc && \
+    echo 'echo "[INFO] Finished initializing conda environment"' >> ~/.bashrc

docs/README.md

@@ -17,7 +17,22 @@ Sirius is a GPU acceleration layer for SQL analytics. It plugs into existing eng
 - CUDA >= 11.2
 - CMake >= 3.30.4 (follow this [instruction](https://medium.com/@yulin_li/how-to-update-cmake-on-ubuntu-9602521deecb) to upgrade CMake)
 
-## Installing dependencies
+## Dependencies (Option 1): Use Docker Image
+To use the docker image with dependencies fully installed:
+```
+sudo docker run --gpus all -it yifeiyang7/sirius_dependencies:latest bash
+```
+
+Make sure both `nvidia-driver` and `nvidia-container-toolkit` are installed, `nvidia-driver` can be installed by
+```
+sudo apt install nvidia-driver-535
+```
+and `nvidia-container-toolkit` can be installed following the [instructions](https://docs.nvidia.com/datacenter/cloud-native/container-toolkit/latest/install-guide.html). Then restart docker by
+```
+sudo systemctl restart docker
+```
+
+## Dependencies (Option 2): Install Manually
 
 ### Install duckdb dependencies
 ```
@@ -46,7 +61,8 @@ export USE_CUDF=1
 export LIBCUDF_ENV_PREFIX={PATH to libcudf-env}
 ```
 
-### Clone the Sirius repository
+## Building Sirius
+To clone the Sirius repository:
 ```
 git clone --recurse-submodules https://github.com/sirius-db/sirius.git
 cd sirius
@@ -60,14 +76,20 @@ cd $SIRIUS_HOME_PATH
 ```
 The `--recurse-submodules` will ensure DuckDB is pulled which is required to build the extension.
 
-## Building Sirius
 To build Sirius:
 ```
 make -j {nproc}
 ```
 
-## Generating TPC-H dataset
-Unzip `dbgen.zip` and run `./dbgen -s {SF}`.
+## Generating and Loading TPC-H dataset
+To generate the TPC-H dataset
+```
+unzip dbgen.zip
+cd dbgen
+./dbgen -s 1 && mkdir s1 && mv *.tbl s1  # this generates dataset of SF1
+cd ..
+```
+
 To load the TPC-H dataset to duckdb:
 ```
 ./build/release/duckdb {DATABASE_NAME}.duckdb
@@ -79,7 +101,7 @@ To run Sirius, simply start the shell with `./build/release/duckdb {DATABASE_NAM
 From the duckdb shell, initialize the Sirius buffer manager with `call gpu_buffer_init`. This API accepts 2 parameters, the GPU caching region size and the GPU processing region size. The GPU caching region is a memory region where the raw data is stored in GPUs, whereas the GPU processing region is where intermediate results are stored in GPUs (hash tables, join results .etc).
 For example, to set the caching region as 1 GB and the processing region as 2 GB, we can run the following command:
 ```
-call gpu_buffer_init("1 GB", "2 GB")
+call gpu_buffer_init("1 GB", "2 GB");
 ```
 
 After setting up Sirius, we can execute SQL queries using the `call gpu_processing`:

src/cuda/cudf/cudf_aggregate.cu

@@ -25,11 +25,15 @@ static std::unique_ptr<cudf::reduce_aggregation> make_reduce_aggregation()
 
 void cudf_aggregate(vector<shared_ptr<GPUColumn>>& column, uint64_t num_aggregates, AggregationType* agg_mode) 
 {
+    GPUBufferManager *gpuBufferManager = &(GPUBufferManager::GetInstance());
+    cudf::set_current_device_resource(gpuBufferManager->mr);
     if (column[0]->column_length == 0) {
         SIRIUS_LOG_DEBUG("Input size is 0");
         for (int agg_idx = 0; agg_idx < num_aggregates; agg_idx++) {
             if (agg_mode[agg_idx] == AggregationType::COUNT_STAR || agg_mode[agg_idx] == AggregationType::COUNT) {
-                column[agg_idx] = make_shared_ptr<GPUColumn>(0, ColumnType::INT64, column[agg_idx]->data_wrapper.data);
+                uint64_t* temp = gpuBufferManager->customCudaMalloc<uint64_t>(1, 0, 0);
+                cudaMemset(temp, 0, sizeof(uint64_t));
+                column[agg_idx] = make_shared_ptr<GPUColumn>(1, ColumnType::INT64, reinterpret_cast<uint8_t*>(temp));
             } else {
                 column[agg_idx] = make_shared_ptr<GPUColumn>(0, column[agg_idx]->data_wrapper.type, column[agg_idx]->data_wrapper.data);
             }
@@ -42,9 +46,6 @@ void cudf_aggregate(vector<shared_ptr<GPUColumn>>& column, uint64_t num_aggregat
     SETUP_TIMING();
     START_TIMER();
 
-    GPUBufferManager *gpuBufferManager = &(GPUBufferManager::GetInstance());
-    cudf::set_current_device_resource(gpuBufferManager->mr);
-
     uint64_t size = 0;
     for (int agg = 0; agg < num_aggregates; agg++) {
         if (column[agg]->data_wrapper.data != nullptr || (column[agg]->data_wrapper.data == nullptr && agg_mode[agg] == AggregationType::COUNT_STAR && column[agg]->column_length > 0)) {

src/cuda/utils.cu

@@ -26,11 +26,63 @@ __device__ __forceinline__ uint64_t convert_endianess(uint64_t value) {
   return (bottom_swapped << 32) | top_swapped;
 }
 
+__global__ void convert_int64_to_int128(uint8_t *input, uint8_t *output, size_t count) {
+  size_t idx = blockIdx.x * blockDim.x + threadIdx.x;
+  if (idx < count) {
+    // Store the converted value as 16 bytes in output
+    uint64_t* input_ptr = reinterpret_cast<uint64_t*>(input + idx * 8);
+    uint64_t* output_ptr = reinterpret_cast<uint64_t*>(output + idx * 16);
+    // for (int i = 0; i < 8; ++i) {
+      // output[idx * 16 + i] = input[idx * 8 + i];
+      // output[idx * 16 + i + 8] = 0;
+      output_ptr[0] = input_ptr[0];
+      output_ptr[1] = 0; // Set the upper 64 bits to zero
+    // }
+  }
+}
+
+__global__ void convert_int32_to_int128(uint8_t *input, uint8_t *output, size_t count) {
+  size_t idx = blockIdx.x * blockDim.x + threadIdx.x;
+  if (idx < count) {
+    // Store the converted value as 16 bytes in output
+    int32_t* input_ptr = reinterpret_cast<int32_t*>(input + idx * 4);
+    int32_t* output_ptr = reinterpret_cast<int32_t*>(output + idx * 16);
+    // for (int i = 0; i < 8; ++i) {
+      output_ptr[0] = input_ptr[0];
+      output_ptr[1] = 0; // Set the upper 64 bits to zero
+      output_ptr[2] = 0; // Set the upper 64 bits to zero
+      output_ptr[3] = 0; // Set the upper 64 bits to zero
+    // }
+  }
+}
+
 void warmup_gpu() {
   // Perform the warmup
   cudaFree(0);
   warmup_kernel<<<WARMUP_BLOCKS, WARMUP_THREADS_PER_BLOCK>>>();
   cudaDeviceSynchronize();
 }
 
+void convertInt64ToInt128(uint8_t *input, uint8_t *output, size_t count) {
+  if (count == 0) return;
+
+  // Launch the kernel to convert the data
+  size_t threads_per_block = 256;
+  size_t blocks = (count + threads_per_block - 1) / threads_per_block;
+
+  convert_int64_to_int128<<<blocks, threads_per_block>>>(input, output, count);
+  cudaDeviceSynchronize();
+}
+
+void convertInt32ToInt128(uint8_t *input, uint8_t *output, size_t count) {
+  if (count == 0) return;
+
+  // Launch the kernel to convert the data
+  size_t threads_per_block = 256;
+  size_t blocks = (count + threads_per_block - 1) / threads_per_block;
+
+  convert_int32_to_int128<<<blocks, threads_per_block>>>(input, output, count);
+  cudaDeviceSynchronize();
+}
+
 } // namespace duckdb

src/include/gpu_columns.hpp

@@ -22,7 +22,8 @@ enum class ColumnType {
     FLOAT64,
     BOOLEAN,
     DATE,
-    VARCHAR
+    VARCHAR,
+    INT128
 };
 
 inline ColumnType convertLogicalTypeToColumnType(LogicalType type) {

src/include/utils.hpp

@@ -5,5 +5,7 @@
 namespace duckdb {
 
 void warmup_gpu();
+void convertInt64ToInt128(uint8_t *input, uint8_t *output, size_t count);
+void convertInt32ToInt128(uint8_t *input, uint8_t *output, size_t count);
 
 } // namespace duckdb

src/operator/gpu_physical_grouped_aggregate.cpp

@@ -300,6 +300,10 @@ HandleGroupByAggregateCuDF(vector<shared_ptr<GPUColumn>> &group_by_keys, vector<
 			agg_mode[agg_idx] = AggregationType::SUM;
 		} else if (expr.function.name.compare("sum") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
 			agg_mode[agg_idx] = AggregationType::SUM;
+		} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data == nullptr && aggregate_keys[agg_idx]->column_length == 0) {
+			agg_mode[agg_idx] = AggregationType::SUM;
+		} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
+			agg_mode[agg_idx] = AggregationType::SUM;
 		} else if (expr.function.name.compare("avg") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
 			agg_mode[agg_idx] = AggregationType::AVERAGE;
 		} else if (expr.function.name.compare("max") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {

src/operator/gpu_physical_result_collector.cpp

@@ -9,6 +9,7 @@
 #include "gpu_buffer_manager.hpp"
 #include "gpu_materialize.hpp"
 #include "log/logging.hpp"
+#include "utils.hpp"
 
 namespace duckdb {
 
@@ -178,6 +179,8 @@ LogicalType ColumnTypeToLogicalType(ColumnType type) {
 			return LogicalType::DATE;
 		case ColumnType::VARCHAR:
 			return LogicalType::VARCHAR;
+		case ColumnType::INT128:
+			return LogicalType::HUGEINT;
 		default:
 			throw NotImplementedException("Unsupported sirius column type in `ColumnTypeToLogicalType`: %d",
 																		static_cast<int>(type));
@@ -198,6 +201,8 @@ Vector rawDataToVector(uint8_t* host_data, size_t vector_offset, ColumnType type
 			sizeof_type = sizeof(double); break;
 		case ColumnType::BOOLEAN:
 			sizeof_type = sizeof(uint8_t); break;
+		case ColumnType::INT128:
+			sizeof_type = 2 * sizeof(uint64_t); break;
 		default:
 			throw NotImplementedException("Unsupported sirius column type in `rawDataToVector`: %d",
 																		static_cast<int>(type));
@@ -238,8 +243,8 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
 	char* all_columns_chars = reinterpret_cast<char*>(combined_buffer + all_columns_strings_buffer_size);
 
 	size_t size_bytes = 0;
-	Allocator& allocator = Allocator::DefaultAllocator();
 	uint8_t** host_data = gpuBufferManager->customCudaHostAlloc<uint8_t*>(input_relation.columns.size());
+
 	GPUIntermediateRelation materialized_relation(input_relation.columns.size());
 	string_t** duckdb_strings = gpuBufferManager->customCudaHostAlloc<string_t*>(input_relation.columns.size());
 	string_t* curr_column_string_buffer = all_columns_string;
@@ -256,9 +261,26 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
 		ColumnType col_type = input_relation.columns[col]->data_wrapper.type;
 		bool is_string = false;
 		if(col_type != ColumnType::VARCHAR) {
-			// host_data[col] = allocator.AllocateData(size_bytes);
-			host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes);
-			callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], materialized_relation.columns[col]->data_wrapper.data, size_bytes, 0);
+			if (types[col].InternalType() == PhysicalType::INT128) {
+				if (materialized_relation.columns[col]->data_wrapper.type == ColumnType::INT64) {
+					SIRIUS_LOG_DEBUG("Converting INT64 to INT128 for column {}", col);
+					uint8_t* temp_int128 = gpuBufferManager->customCudaMalloc<uint8_t>(size_bytes * 2, 0, 0);
+					convertInt64ToInt128(materialized_relation.columns[col]->data_wrapper.data, temp_int128, materialized_relation.columns[col]->column_length);
+					host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes * 2);
+					callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], temp_int128, size_bytes * 2, 0);
+				} else if (materialized_relation.columns[col]->data_wrapper.type == ColumnType::INT32) {
+					SIRIUS_LOG_DEBUG("Converting INT32 to INT128 for column {}", col);
+					uint8_t* temp_int128 = gpuBufferManager->customCudaMalloc<uint8_t>(size_bytes * 4, 0, 0);
+					convertInt32ToInt128(materialized_relation.columns[col]->data_wrapper.data, temp_int128, materialized_relation.columns[col]->column_length);
+					host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes * 4);
+					callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], temp_int128, size_bytes * 4, 0);
+				} else {
+					throw NotImplementedException("Unsupported column type for INT128 conversion");
+				}
+			} else {
+				host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes);
+				callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], materialized_relation.columns[col]->data_wrapper.data, size_bytes, 0);
+			}
 		} else {
 			// Use the helper method to materialize the string on the GPU
 			shared_ptr<GPUColumn> str_column = materialized_relation.columns[col];
@@ -291,8 +313,13 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
 		chunk.InitializeEmpty(types);
 		for (int col = 0; col < materialized_relation.columns.size(); col++) {
 			if(materialized_relation.columns[col]->data_wrapper.type != ColumnType::VARCHAR) {
-				Vector vector = rawDataToVector(host_data[col], vec, materialized_relation.columns[col]->data_wrapper.type);
-				chunk.data[col].Reference(vector);
+				if (types[col].InternalType() == PhysicalType::INT128) {
+					Vector vector = rawDataToVector(host_data[col], vec, ColumnType::INT128);
+					chunk.data[col].Reference(vector);
+				} else {
+					Vector vector = rawDataToVector(host_data[col], vec, materialized_relation.columns[col]->data_wrapper.type);
+					chunk.data[col].Reference(vector);
+				}
 			} else {
 				// Add the strings to the vector
 				Vector str_vector(LogicalType::VARCHAR, reinterpret_cast<data_ptr_t>(duckdb_strings[col] + read_index));

src/operator/gpu_physical_ungrouped_aggregate.cpp

@@ -115,6 +115,10 @@ HandleAggregateExpressionCuDF(vector<shared_ptr<GPUColumn>> &aggregate_keys, GPU
 			agg_mode[agg_idx] = AggregationType::SUM;
 		} else if (expr.function.name.compare("sum") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
 			agg_mode[agg_idx] = AggregationType::SUM;
+		} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data == nullptr && aggregate_keys[agg_idx]->column_length == 0) {
+			agg_mode[agg_idx] = AggregationType::SUM;
+		} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
+			agg_mode[agg_idx] = AggregationType::SUM;
 		} else if (expr.function.name.compare("avg") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
 			agg_mode[agg_idx] = AggregationType::AVERAGE;
 		} else if (expr.function.name.compare("max") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {