Skip to content

Commit 7fb8b98

Browse files
bwyogatamabwyogatama
committed
Use Sirius materialize kernel instead of thrust in expression executor. Fix issue #51 
1 parent eb1cd81 commit 7fb8b98

17 files changed

Lines changed: 64 additions & 129 deletions

‎src/cuda/operator/materialize.cu‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ template
151151
__global__ void materialize_expression_with_null<uint8_t, BLOCK_THREADS, ITEMS_PER_THREAD>(const uint8_t *a, uint8_t* result, uint32_t* mask, uint32_t* out_mask, uint64_t *row_ids, uint64_t N);
152152

153153
template <typename T>
154-
void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_len) {
154+
void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_len) {
155155
CHECK_ERROR();
156156
if (result_len == 0) {
157157
SIRIUS_LOG_DEBUG("result_len is 0");
@@ -170,7 +170,7 @@ void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_
170170
gpuBufferManager->customCudaFree(reinterpret_cast<uint8_t*>(a), 0);
171171
}
172172

173-
void materializeString(uint8_t* data, uint64_t* offset, uint8_t* &result, uint64_t* &result_offset, uint64_t* row_ids, uint64_t* &result_bytes, uint64_t result_len, uint64_t input_size, uint64_t input_bytes) {
173+
void materializeString(uint8_t* data, uint64_t* offset, uint8_t* &result, uint64_t* &result_offset, uint64_t* row_ids, uint64_t* &result_bytes, uint64_t result_len) {
174174
CHECK_ERROR();
175175
if (result_len == 0) {
176176
SIRIUS_LOG_DEBUG("result_len is 0");
@@ -269,18 +269,18 @@ void materializeStringColumnToDuckdbFormat(shared_ptr<GPUColumn> column, char* c
269269
}
270270

271271
template
272-
void materializeExpression<int>(int *a, int*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
272+
void materializeExpression<int>(int *a, int*& result, uint64_t *row_ids, uint64_t result_len);
273273
template
274-
void materializeExpression<uint64_t>(uint64_t *a, uint64_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
274+
void materializeExpression<uint64_t>(uint64_t *a, uint64_t*& result, uint64_t *row_ids, uint64_t result_len);
275275
template
276-
void materializeExpression<float>(float *a, float*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
276+
void materializeExpression<float>(float *a, float*& result, uint64_t *row_ids, uint64_t result_len);
277277
template
278-
void materializeExpression<double>(double *a, double*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
278+
void materializeExpression<double>(double *a, double*& result, uint64_t *row_ids, uint64_t result_len);
279279
template
280-
void materializeExpression<uint8_t>(uint8_t *a, uint8_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
280+
void materializeExpression<uint8_t>(uint8_t *a, uint8_t*& result, uint64_t *row_ids, uint64_t result_len);
281281
template
282-
void materializeExpression<int64_t>(int64_t *a, int64_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
282+
void materializeExpression<int64_t>(int64_t *a, int64_t*& result, uint64_t *row_ids, uint64_t result_len);
283283
template
284-
void materializeExpression<__int128_t>(__int128_t *a, __int128_t*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_size);
284+
void materializeExpression<__int128_t>(__int128_t *a, __int128_t*& result, uint64_t *row_ids, uint64_t result_len);
285285

286286
} // namespace duckdb

‎src/cuda/operator/unused/optimized_grouped_aggregate.cu‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -529,7 +529,7 @@ void optimizedGroupedStringAggregate(uint8_t **keys, uint8_t **aggregate_keys, u
529529
// Materialize the string column
530530
uint8_t* result; uint64_t* result_offset; uint64_t* new_num_bytes;
531531
// void materializeString(uint8_t* data, uint64_t* offset, uint8_t* &result, uint64_t* &result_offset, uint64_t* row_ids, uint64_t* &result_bytes, uint64_t result_len, uint64_t input_size, uint64_t input_bytes)
532-
materializeString(group_key_chars, group_key_offsets, result, result_offset, d_original_row_ids, new_num_bytes, num_groups, N, num_bytes[i]);
532+
materializeString(group_key_chars, group_key_offsets, result, result_offset, d_original_row_ids, new_num_bytes, num_groups);
533533

534534
// Write back the result
535535
keys[i] = result;

‎src/cuda/operator/unused/string_order_by.cu‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ void orderByString(uint8_t** col_keys, uint64_t** col_offsets, int* sort_orders,
162162

163163
// Materialize the column in the new order
164164
uint8_t* sorted_chars; uint64_t* sorted_offsets; uint64_t* new_num_bytes;
165-
materializeString(unsorted_col_chars, unsorted_col_offsets, sorted_chars, sorted_offsets, d_row_ids, new_num_bytes, num_rows, num_rows, col_num_bytes[i]);
165+
materializeString(unsorted_col_chars, unsorted_col_offsets, sorted_chars, sorted_offsets, d_row_ids, new_num_bytes, num_rows);
166166

167167
// Write back the result
168168
col_keys[i] = sorted_chars;

‎src/expression_executor/specializations/gpu_execute_operator.cpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -73,21 +73,21 @@ struct ExecuteStringIn
7373

7474
// We need to convert to cudf/arrow format...
7575
std::vector<char> chars;
76-
std::vector<cudf::size_type> offsets;
77-
cudf::size_type offset = 0;
76+
std::vector<int64_t> offsets;
77+
int64_t offset = 0;
7878
for (idx_t child = 1; child < expr.children.size(); ++child)
7979
{
8080
const auto& child_expression = expr.children[child]->Cast<BoundConstantExpression>();
8181
const auto& child_string = child_expression.value.GetValue<std::string>();
8282
chars.insert(chars.end(), child_string.begin(), child_string.end());
8383
offsets.push_back(offset);
84-
offset += static_cast<cudf::size_type>(child_string.size());
84+
offset += static_cast<int64_t>(child_string.size());
8585
}
8686
offsets.push_back(offset);
8787

8888
// Allocate buffers and copy to device
8989
rmm::device_uvector<char> chars_buffer(offset, stream, mr);
90-
rmm::device_uvector<cudf::size_type> offsets_buffer(num_offsets,
90+
rmm::device_uvector<int64_t> offsets_buffer(num_offsets,
9191
stream,
9292
mr);
9393
CUDF_CUDA_TRY(cudaMemcpyAsync(chars_buffer.data(),
@@ -97,12 +97,12 @@ struct ExecuteStringIn
9797
stream));
9898
CUDF_CUDA_TRY(cudaMemcpyAsync(offsets_buffer.data(),
9999
offsets.data(),
100-
offsets.size() * sizeof(cudf::size_type),
100+
offsets.size() * sizeof(int64_t),
101101
cudaMemcpyHostToDevice,
102102
stream));
103103

104104
// Make CuDF things
105-
auto offsets_col = std::make_unique<cudf::column>(cudf::data_type(cudf::type_id::INT32),
105+
auto offsets_col = std::make_unique<cudf::column>(cudf::data_type(cudf::type_id::INT64),
106106
num_offsets,
107107
std::move(offsets_buffer).release(),
108108
rmm::device_buffer{},

‎src/expression_executor/specializations/gpu_execute_reference.cpp‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
#include "duckdb/planner/expression/bound_reference_expression.hpp"
22
#include "expression_executor/gpu_dispatcher.hpp"
33
#include "expression_executor/gpu_expression_executor.hpp"
4+
#include "operator/gpu_materialize.hpp"
45

56
namespace duckdb
67
{
@@ -26,7 +27,13 @@ std::unique_ptr<cudf::column> GpuExpressionExecutor::Execute(const BoundReferenc
2627
throw NotImplementedException(
2728
"Execute[Reference]: row_id_counts larger than int32_t are not supported in libcudf");
2829
}
29-
return GpuDispatcher::DispatchMaterialize(input_column.get(), resource_ref, execution_stream);
30+
// return GpuDispatcher::DispatchMaterialize(input_column.get(), resource_ref, execution_stream);
31+
auto gpuBufferManager = &GPUBufferManager::GetInstance();
32+
auto output_column = HandleMaterializeExpression(input_column, gpuBufferManager);
33+
auto output_column_view = output_column->convertToCudfColumn();
34+
return std::make_unique<cudf::column>(output_column_view,
35+
execution_stream,
36+
resource_ref);
3037
}
3138
if (input_column->column_length > INT32_MAX)
3239
{

‎src/include/gpu_buffer_manager.hpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@ template <typename T> T* callCudaHostAlloc(size_t size, bool return_dev_ptr);
1515
template <typename T> void callCudaFree(T* ptr, int gpu);
1616
template <typename T> void callCudaMemcpyHostToDevice(T* dest, T* src, size_t size, int gpu);
1717
template <typename T> void callCudaMemcpyDeviceToHost(T* dest, T* src, size_t size, int gpu);
18-
template <typename T> void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_len, uint64_t input_len);
19-
void materializeString(uint8_t* data, uint64_t* offset, uint8_t* &result, uint64_t* &result_offset, uint64_t* row_ids, uint64_t* &result_bytes, uint64_t result_len, uint64_t input_size, uint64_t input_bytes);
18+
template <typename T> void materializeExpression(T *a, T*& result, uint64_t *row_ids, uint64_t result_len);
19+
void materializeString(uint8_t* data, uint64_t* offset, uint8_t* &result, uint64_t* &result_offset, uint64_t* row_ids, uint64_t* &result_bytes, uint64_t result_len);
2020
template <typename T> void printGPUColumn(T* a, size_t N, int gpu);
2121
void cudaMemmove(uint8_t* destination, uint8_t* source, size_t num);
2222
uint8_t* allocatePinnedCPUMemory(size_t size);

‎src/include/operator/gpu_materialize.hpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,10 @@ void materializeStringColumnToDuckdbFormat(shared_ptr<GPUColumn> column, char* c
1010

1111
template <typename T>
1212
shared_ptr<GPUColumn>
13-
ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExpression& bound_ref, GPUBufferManager* gpuBufferManager);
13+
ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, GPUBufferManager* gpuBufferManager);
1414

1515
shared_ptr<GPUColumn>
16-
HandleMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExpression& bound_ref, GPUBufferManager* gpuBufferManager);
16+
HandleMaterializeExpression(shared_ptr<GPUColumn> column, GPUBufferManager* gpuBufferManager);
1717

1818
void
1919
HandleMaterializeRowIDs(GPUIntermediateRelation& input_relation, GPUIntermediateRelation& output_relation, uint64_t count, uint64_t* row_ids, GPUBufferManager* gpuBufferManager, bool maintain_unique);

‎src/operator/gpu_materialize.cpp‎

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

66
template <typename T>
77
shared_ptr<GPUColumn>
8-
ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExpression& bound_ref, GPUBufferManager* gpuBufferManager) {
8+
ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, GPUBufferManager* gpuBufferManager) {
99
size_t size;
1010
T* a;
1111
if (column->data_wrapper.data == nullptr || column->column_length == 0) {
@@ -14,7 +14,7 @@ ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExp
1414
if (column->row_ids != nullptr) {
1515
T* temp = reinterpret_cast<T*> (column->data_wrapper.data);
1616
uint64_t* row_ids_input = reinterpret_cast<uint64_t*> (column->row_ids);
17-
materializeExpression<T>(temp, a, row_ids_input, column->row_id_count, column->column_length);
17+
materializeExpression<T>(temp, a, row_ids_input, column->row_id_count);
1818
size = column->row_id_count;
1919
} else {
2020
a = reinterpret_cast<T*> (column->data_wrapper.data);
@@ -26,7 +26,7 @@ ResolveTypeMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExp
2626
}
2727

2828
shared_ptr<GPUColumn>
29-
ResolveTypeMaterializeString(shared_ptr<GPUColumn> column, BoundReferenceExpression& bound_ref, GPUBufferManager* gpuBufferManager) {
29+
ResolveTypeMaterializeString(shared_ptr<GPUColumn> column, GPUBufferManager* gpuBufferManager) {
3030
size_t size;
3131
uint8_t* a;
3232
uint64_t* result_offset;
@@ -40,7 +40,7 @@ ResolveTypeMaterializeString(shared_ptr<GPUColumn> column, BoundReferenceExpress
4040
uint64_t* offset = column->data_wrapper.offset;
4141
uint64_t* row_ids = column->row_ids;
4242
size = column->row_id_count;
43-
materializeString(data, offset, a, result_offset, row_ids, new_num_bytes, size, column->column_length, column->data_wrapper.num_bytes);
43+
materializeString(data, offset, a, result_offset, row_ids, new_num_bytes, size);
4444
} else {
4545
a = column->data_wrapper.data;
4646
result_offset = column->data_wrapper.offset;
@@ -54,29 +54,29 @@ ResolveTypeMaterializeString(shared_ptr<GPUColumn> column, BoundReferenceExpress
5454
}
5555

5656
shared_ptr<GPUColumn>
57-
HandleMaterializeExpression(shared_ptr<GPUColumn> column, BoundReferenceExpression& bound_ref, GPUBufferManager* gpuBufferManager) {
57+
HandleMaterializeExpression(shared_ptr<GPUColumn> column, GPUBufferManager* gpuBufferManager) {
5858
switch(column->data_wrapper.type.id()) {
5959
case GPUColumnTypeId::INT32:
6060
case GPUColumnTypeId::DATE:
61-
return ResolveTypeMaterializeExpression<int>(column, bound_ref, gpuBufferManager);
61+
return ResolveTypeMaterializeExpression<int>(column, gpuBufferManager);
6262
case GPUColumnTypeId::INT64:
63-
return ResolveTypeMaterializeExpression<uint64_t>(column, bound_ref, gpuBufferManager);
63+
return ResolveTypeMaterializeExpression<uint64_t>(column, gpuBufferManager);
6464
case GPUColumnTypeId::FLOAT32:
65-
return ResolveTypeMaterializeExpression<float>(column, bound_ref, gpuBufferManager);
65+
return ResolveTypeMaterializeExpression<float>(column, gpuBufferManager);
6666
case GPUColumnTypeId::FLOAT64:
67-
return ResolveTypeMaterializeExpression<double>(column, bound_ref, gpuBufferManager);
67+
return ResolveTypeMaterializeExpression<double>(column, gpuBufferManager);
6868
case GPUColumnTypeId::BOOLEAN:
69-
return ResolveTypeMaterializeExpression<uint8_t>(column, bound_ref, gpuBufferManager);
69+
return ResolveTypeMaterializeExpression<uint8_t>(column, gpuBufferManager);
7070
case GPUColumnTypeId::VARCHAR:
71-
return ResolveTypeMaterializeString(column, bound_ref, gpuBufferManager);
71+
return ResolveTypeMaterializeString(column, gpuBufferManager);
7272
case GPUColumnTypeId::DECIMAL: {
7373
switch (column->data_wrapper.getColumnTypeSize()) {
7474
case sizeof(int32_t):
75-
return ResolveTypeMaterializeExpression<int32_t>(column, bound_ref, gpuBufferManager);
75+
return ResolveTypeMaterializeExpression<int32_t>(column, gpuBufferManager);
7676
case sizeof(int64_t):
77-
return ResolveTypeMaterializeExpression<int64_t>(column, bound_ref, gpuBufferManager);
77+
return ResolveTypeMaterializeExpression<int64_t>(column, gpuBufferManager);
7878
case sizeof(__int128_t):
79-
return ResolveTypeMaterializeExpression<__int128_t>(column, bound_ref, gpuBufferManager);
79+
return ResolveTypeMaterializeExpression<__int128_t>(column, gpuBufferManager);
8080
default:
8181
throw NotImplementedException("Unsupported sirius DECIMAL column type size in `HandleMaterializeExpression`: %zu",
8282
column->data_wrapper.getColumnTypeSize());
@@ -123,7 +123,7 @@ HandleMaterializeRowIDs(GPUIntermediateRelation& input_relation, GPUIntermediate
123123
} else {
124124
uint64_t* temp_prev_row_ids = reinterpret_cast<uint64_t*> (input_relation.columns[i]->row_ids);
125125
uint64_t* temp_new_row_ids;
126-
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count, input_relation.columns[i]->column_length);
126+
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count);
127127
output_relation.columns[i]->row_ids = temp_new_row_ids;
128128
new_row_ids.push_back(temp_new_row_ids);
129129
prev_row_ids.push_back(temp_prev_row_ids);
@@ -170,7 +170,7 @@ HandleMaterializeRowIDsRHS(GPUIntermediateRelation& hash_table_result, GPUInterm
170170
} else {
171171
uint64_t* temp_prev_row_ids = reinterpret_cast<uint64_t*> (hash_table_result.columns[rhs_col]->row_ids);
172172
uint64_t* temp_new_row_ids;
173-
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count, hash_table_result.columns[rhs_col]->column_length);
173+
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count);
174174
output_relation.columns[offset + i]->row_ids = temp_new_row_ids;
175175
new_row_ids.push_back(temp_new_row_ids);
176176
prev_row_ids.push_back(temp_prev_row_ids);
@@ -218,7 +218,7 @@ HandleMaterializeRowIDsLHS(GPUIntermediateRelation& input_relation, GPUIntermedi
218218
} else {
219219
uint64_t* temp_prev_row_ids = reinterpret_cast<uint64_t*> (input_relation.columns[lhs_col]->row_ids);
220220
uint64_t* temp_new_row_ids;
221-
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count, input_relation.columns[lhs_col]->column_length);
221+
materializeExpression<uint64_t>(temp_prev_row_ids, temp_new_row_ids, row_ids, count);
222222
output_relation.columns[i]->row_ids = temp_new_row_ids;
223223
new_row_ids.push_back(temp_new_row_ids);
224224
prev_row_ids.push_back(temp_prev_row_ids);

‎src/operator/gpu_physical_grouped_aggregate.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -627,7 +627,7 @@ GPUPhysicalGroupedAggregate::Sink(GPUIntermediateRelation& input_relation) const
627627
D_ASSERT(group->type == ExpressionType::BOUND_REF);
628628
auto &bound_ref_expr = group->Cast<BoundReferenceExpression>();
629629
SIRIUS_LOG_DEBUG("Passing input column index {} to group by column index {}", bound_ref_expr.index, idx);
630-
group_by_column[idx] = HandleMaterializeExpression(input_relation.columns[bound_ref_expr.index], bound_ref_expr, gpuBufferManager);
630+
group_by_column[idx] = HandleMaterializeExpression(input_relation.columns[bound_ref_expr.index], gpuBufferManager);
631631
idx++;
632632
}
633633
}
@@ -641,7 +641,7 @@ GPUPhysicalGroupedAggregate::Sink(GPUIntermediateRelation& input_relation) const
641641
D_ASSERT(child_expr->type == ExpressionType::BOUND_REF);
642642
auto &bound_ref_expr = child_expr->Cast<BoundReferenceExpression>();
643643
SIRIUS_LOG_DEBUG("Passing input column index {} to aggregate column index {}", bound_ref_expr.index, aggr_idx);
644-
aggregate_column[aggr_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref_expr.index], bound_ref_expr, gpuBufferManager);
644+
aggregate_column[aggr_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref_expr.index], gpuBufferManager);
645645
}
646646
aggr_idx++;
647647
}
@@ -790,7 +790,7 @@ GPUPhysicalGroupedAggregate::SinkDistinctGrouping(GPUIntermediateRelation& input
790790
for (idx_t group_idx = 0; group_idx < grouped_aggregate_data.groups.size(); group_idx++) {
791791
auto &group = grouped_aggregate_data.groups[group_idx];
792792
auto &bound_ref = group->Cast<BoundReferenceExpression>();
793-
group_by_column[group_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref.index], bound_ref, gpuBufferManager);
793+
group_by_column[group_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref.index], gpuBufferManager);
794794
}
795795

796796
int aggr_idx = 0;
@@ -829,7 +829,7 @@ GPUPhysicalGroupedAggregate::SinkDistinctGrouping(GPUIntermediateRelation& input
829829
SIRIUS_LOG_DEBUG("Reading aggregation column from index {} and passing it to index {} in distinct aggregatec column", bound_ref.index, aggr_idx);
830830
// input_relation.checkLateMaterialization(bound_ref.index);
831831
// group_by_result->columns[grouped_aggregate_data.groups.size() + idx] = input_relation.columns[bound_ref.index];
832-
distinct_aggregate_columns[aggr_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref.index], bound_ref, gpuBufferManager);
832+
distinct_aggregate_columns[aggr_idx] = HandleMaterializeExpression(input_relation.columns[bound_ref.index], gpuBufferManager);
833833
}
834834
aggr_idx++;
835835
}

0 commit comments

Comments
 (0)