Skip to content

Commit f100550

Browse files
bwyogatamabwyogatama
committed
Bug fixes for #14 
1 parent 54a4d58 commit f100550

6 files changed

Lines changed: 97 additions & 7 deletions

File tree

‎src/cuda/utils.cu‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,63 @@ __device__ __forceinline__ uint64_t convert_endianess(uint64_t value) {
2626
return (bottom_swapped << 32) | top_swapped;
2727
}
2828

29+
__global__ void convert_int64_to_int128(uint8_t *input, uint8_t *output, size_t count) {
30+
size_t idx = blockIdx.x * blockDim.x + threadIdx.x;
31+
if (idx < count) {
32+
// Store the converted value as 16 bytes in output
33+
uint64_t* input_ptr = reinterpret_cast<uint64_t*>(input + idx * 8);
34+
uint64_t* output_ptr = reinterpret_cast<uint64_t*>(output + idx * 16);
35+
// for (int i = 0; i < 8; ++i) {
36+
// output[idx * 16 + i] = input[idx * 8 + i];
37+
// output[idx * 16 + i + 8] = 0;
38+
output_ptr[0] = input_ptr[0];
39+
output_ptr[1] = 0; // Set the upper 64 bits to zero
40+
// }
41+
}
42+
}
43+
44+
__global__ void convert_int32_to_int128(uint8_t *input, uint8_t *output, size_t count) {
45+
size_t idx = blockIdx.x * blockDim.x + threadIdx.x;
46+
if (idx < count) {
47+
// Store the converted value as 16 bytes in output
48+
int32_t* input_ptr = reinterpret_cast<int32_t*>(input + idx * 4);
49+
int32_t* output_ptr = reinterpret_cast<int32_t*>(output + idx * 16);
50+
// for (int i = 0; i < 8; ++i) {
51+
output_ptr[0] = input_ptr[0];
52+
output_ptr[1] = 0; // Set the upper 64 bits to zero
53+
output_ptr[2] = 0; // Set the upper 64 bits to zero
54+
output_ptr[3] = 0; // Set the upper 64 bits to zero
55+
// }
56+
}
57+
}
58+
2959
void warmup_gpu() {
3060
// Perform the warmup
3161
cudaFree(0);
3262
warmup_kernel<<<WARMUP_BLOCKS, WARMUP_THREADS_PER_BLOCK>>>();
3363
cudaDeviceSynchronize();
3464
}
3565

66+
void convertInt64ToInt128(uint8_t *input, uint8_t *output, size_t count) {
67+
if (count == 0) return;
68+
69+
// Launch the kernel to convert the data
70+
size_t threads_per_block = 256;
71+
size_t blocks = (count + threads_per_block - 1) / threads_per_block;
72+
73+
convert_int64_to_int128<<<blocks, threads_per_block>>>(input, output, count);
74+
cudaDeviceSynchronize();
75+
}
76+
77+
void convertInt32ToInt128(uint8_t *input, uint8_t *output, size_t count) {
78+
if (count == 0) return;
79+
80+
// Launch the kernel to convert the data
81+
size_t threads_per_block = 256;
82+
size_t blocks = (count + threads_per_block - 1) / threads_per_block;
83+
84+
convert_int32_to_int128<<<blocks, threads_per_block>>>(input, output, count);
85+
cudaDeviceSynchronize();
86+
}
87+
3688
} // namespace duckdb

‎src/include/gpu_columns.hpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,8 @@ enum class ColumnType {
2121
FLOAT32,
2222
FLOAT64,
2323
BOOLEAN,
24-
VARCHAR
24+
VARCHAR,
25+
INT128
2526
};
2627

2728
inline ColumnType convertLogicalTypeToColumnType(LogicalType type) {

‎src/include/utils.hpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,5 +5,7 @@
55
namespace duckdb {
66

77
void warmup_gpu();
8+
void convertInt64ToInt128(uint8_t *input, uint8_t *output, size_t count);
9+
void convertInt32ToInt128(uint8_t *input, uint8_t *output, size_t count);
810

911
} // namespace duckdb

‎src/operator/gpu_physical_grouped_aggregate.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -297,6 +297,10 @@ HandleGroupByAggregateCuDF(vector<shared_ptr<GPUColumn>> &group_by_keys, vector<
297297
agg_mode[agg_idx] = AggregationType::SUM;
298298
} else if (expr.function.name.compare("sum") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
299299
agg_mode[agg_idx] = AggregationType::SUM;
300+
} 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) {
301+
agg_mode[agg_idx] = AggregationType::SUM;
302+
} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
303+
agg_mode[agg_idx] = AggregationType::SUM;
300304
} else if (expr.function.name.compare("avg") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
301305
agg_mode[agg_idx] = AggregationType::AVERAGE;
302306
} else if (expr.function.name.compare("max") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {

‎src/operator/gpu_physical_result_collector.cpp‎

Lines changed: 33 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
#include "gpu_buffer_manager.hpp"
1010
#include "gpu_materialize.hpp"
1111
#include "log/logging.hpp"
12+
#include "utils.hpp"
1213

1314
namespace duckdb {
1415

@@ -174,6 +175,8 @@ LogicalType ColumnTypeToLogicalType(ColumnType type) {
174175
return LogicalType::BOOLEAN;
175176
case ColumnType::VARCHAR:
176177
return LogicalType::VARCHAR;
178+
case ColumnType::INT128:
179+
return LogicalType::HUGEINT;
177180
default:
178181
throw NotImplementedException("Unsupported column type");
179182
}
@@ -192,6 +195,8 @@ Vector rawDataToVector(uint8_t* host_data, size_t vector_offset, ColumnType type
192195
sizeof_type = sizeof(double); break;
193196
case ColumnType::BOOLEAN:
194197
sizeof_type = sizeof(uint8_t); break;
198+
case ColumnType::INT128:
199+
sizeof_type = 2 * sizeof(uint64_t); break;
195200
default:
196201
throw NotImplementedException("Unsupported column type");
197202
}
@@ -231,8 +236,8 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
231236
char* all_columns_chars = reinterpret_cast<char*>(combined_buffer + all_columns_strings_buffer_size);
232237

233238
size_t size_bytes = 0;
234-
Allocator& allocator = Allocator::DefaultAllocator();
235239
uint8_t** host_data = gpuBufferManager->customCudaHostAlloc<uint8_t*>(input_relation.columns.size());
240+
236241
GPUIntermediateRelation materialized_relation(input_relation.columns.size());
237242
string_t** duckdb_strings = gpuBufferManager->customCudaHostAlloc<string_t*>(input_relation.columns.size());
238243
string_t* curr_column_string_buffer = all_columns_string;
@@ -249,9 +254,26 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
249254
ColumnType col_type = input_relation.columns[col]->data_wrapper.type;
250255
bool is_string = false;
251256
if(col_type != ColumnType::VARCHAR) {
252-
// host_data[col] = allocator.AllocateData(size_bytes);
253-
host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes);
254-
callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], materialized_relation.columns[col]->data_wrapper.data, size_bytes, 0);
257+
if (types[col].InternalType() == PhysicalType::INT128) {
258+
if (materialized_relation.columns[col]->data_wrapper.type == ColumnType::INT64) {
259+
SIRIUS_LOG_DEBUG("Converting INT64 to INT128 for column {}", col);
260+
uint8_t* temp_int128 = gpuBufferManager->customCudaMalloc<uint8_t>(size_bytes * 2, 0, 0);
261+
convertInt64ToInt128(materialized_relation.columns[col]->data_wrapper.data, temp_int128, materialized_relation.columns[col]->column_length);
262+
host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes * 2);
263+
callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], temp_int128, size_bytes * 2, 0);
264+
} else if (materialized_relation.columns[col]->data_wrapper.type == ColumnType::INT32) {
265+
SIRIUS_LOG_DEBUG("Converting INT32 to INT128 for column {}", col);
266+
uint8_t* temp_int128 = gpuBufferManager->customCudaMalloc<uint8_t>(size_bytes * 4, 0, 0);
267+
convertInt32ToInt128(materialized_relation.columns[col]->data_wrapper.data, temp_int128, materialized_relation.columns[col]->column_length);
268+
host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes * 4);
269+
callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], temp_int128, size_bytes * 4, 0);
270+
} else {
271+
throw NotImplementedException("Unsupported column type for INT128 conversion");
272+
}
273+
} else {
274+
host_data[col] = gpuBufferManager->customCudaHostAlloc<uint8_t>(size_bytes);
275+
callCudaMemcpyDeviceToHost<uint8_t>(host_data[col], materialized_relation.columns[col]->data_wrapper.data, size_bytes, 0);
276+
}
255277
} else {
256278
// Use the helper method to materialize the string on the GPU
257279
shared_ptr<GPUColumn> str_column = materialized_relation.columns[col];
@@ -284,8 +306,13 @@ SinkResultType GPUPhysicalMaterializedCollector::Sink(GPUIntermediateRelation &i
284306
chunk.InitializeEmpty(types);
285307
for (int col = 0; col < materialized_relation.columns.size(); col++) {
286308
if(materialized_relation.columns[col]->data_wrapper.type != ColumnType::VARCHAR) {
287-
Vector vector = rawDataToVector(host_data[col], vec, materialized_relation.columns[col]->data_wrapper.type);
288-
chunk.data[col].Reference(vector);
309+
if (types[col].InternalType() == PhysicalType::INT128) {
310+
Vector vector = rawDataToVector(host_data[col], vec, ColumnType::INT128);
311+
chunk.data[col].Reference(vector);
312+
} else {
313+
Vector vector = rawDataToVector(host_data[col], vec, materialized_relation.columns[col]->data_wrapper.type);
314+
chunk.data[col].Reference(vector);
315+
}
289316
} else {
290317
// Add the strings to the vector
291318
Vector str_vector(LogicalType::VARCHAR, reinterpret_cast<data_ptr_t>(duckdb_strings[col] + read_index));

‎src/operator/gpu_physical_ungrouped_aggregate.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,10 @@ HandleAggregateExpressionCuDF(vector<shared_ptr<GPUColumn>> &aggregate_keys, GPU
114114
agg_mode[agg_idx] = AggregationType::SUM;
115115
} else if (expr.function.name.compare("sum") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
116116
agg_mode[agg_idx] = AggregationType::SUM;
117+
} 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) {
118+
agg_mode[agg_idx] = AggregationType::SUM;
119+
} else if (expr.function.name.compare("sum_no_overflow") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
120+
agg_mode[agg_idx] = AggregationType::SUM;
117121
} else if (expr.function.name.compare("avg") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {
118122
agg_mode[agg_idx] = AggregationType::AVERAGE;
119123
} else if (expr.function.name.compare("max") == 0 && aggregate_keys[agg_idx]->data_wrapper.data != nullptr) {

0 commit comments

Comments
 (0)