Skip to content

Commit 4f41d62

Browse files
poyrazKpoyrazK
authored
Merge pull request #163 from poyrazK/column-projection-pushdown-v2
Column projection pushdown for GROUP BY
2 parents e46a1f6 + 2453262 commit 4f41d62

4 files changed

Lines changed: 235 additions & 5 deletions

File tree

‎include/executor/vectorized_operator.hpp‎

Lines changed: 29 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ class VectorizedSeqScanOperator : public VectorizedOperator {
7373
size_t num_threads_ = 1;
7474
std::vector<std::unique_ptr<VectorBatch>> parallel_results_;
7575
size_t parallel_idx_ = 0;
76+
std::vector<size_t> required_col_indices_;
77+
executor::Schema reduced_schema_;
7678

7779
public:
7880
VectorizedSeqScanOperator(std::string table_name, std::shared_ptr<storage::ColumnarTable> table,
@@ -94,12 +96,26 @@ class VectorizedSeqScanOperator : public VectorizedOperator {
9496
return next_batch_parallel(out_batch);
9597
}
9698

99+
void set_required_columns(std::vector<size_t> col_indices, executor::Schema reduced_schema) {
100+
required_col_indices_ = std::move(col_indices);
101+
reduced_schema_ = std::move(reduced_schema);
102+
}
103+
97104
private:
98105
bool next_batch_sequential(VectorBatch& out_batch) {
99106
if (current_row_ >= table_->row_count()) {
100107
return false;
101108
}
102109

110+
if (!required_col_indices_.empty()) {
111+
out_batch.init_from_schema(reduced_schema_);
112+
if (table_->read_batch(current_row_, batch_size_, out_batch, required_col_indices_)) {
113+
current_row_ += out_batch.row_count();
114+
return true;
115+
}
116+
return false;
117+
}
118+
103119
if (table_->read_batch(current_row_, batch_size_, out_batch)) {
104120
current_row_ += out_batch.row_count();
105121
return true;
@@ -128,7 +144,8 @@ class VectorizedSeqScanOperator : public VectorizedOperator {
128144
size_t end = std::min(start + range_size, total_rows);
129145
current_row_ = end;
130146

131-
auto batch = VectorBatch::create(output_schema_);
147+
auto batch = VectorBatch::create(required_col_indices_.empty() ? output_schema_
148+
: reduced_schema_);
132149
parallel_results_.push_back(std::move(batch));
133150
}
134151

@@ -139,10 +156,17 @@ class VectorizedSeqScanOperator : public VectorizedOperator {
139156
parallel_results_[t]->set_row_count(0);
140157
continue;
141158
}
142-
thread_pool_->submit([this, t, start, rows_to_read]() {
143-
table_->read_batch(start, static_cast<uint32_t>(rows_to_read),
144-
*parallel_results_[t]);
145-
});
159+
if (!required_col_indices_.empty()) {
160+
thread_pool_->submit([this, t, start, rows_to_read]() {
161+
table_->read_batch(start, static_cast<uint32_t>(rows_to_read),
162+
*parallel_results_[t], required_col_indices_);
163+
});
164+
} else {
165+
thread_pool_->submit([this, t, start, rows_to_read]() {
166+
table_->read_batch(start, static_cast<uint32_t>(rows_to_read),
167+
*parallel_results_[t]);
168+
});
169+
}
146170
}
147171

148172
thread_pool_->wait();

‎include/storage/columnar_table.hpp‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,16 @@ class ColumnarTable {
3737
*/
3838
bool read_batch(uint64_t start_row, uint32_t batch_size, executor::VectorBatch& out_batch);
3939

40+
/**
41+
* @brief Load a batch of data for only the specified columns
42+
* @param start_row Starting row index
43+
* @param batch_size Maximum rows to read
44+
* @param out_batch Output batch (pre-initialized with required schema)
45+
* @param col_indices Columns to read (indices into table's schema)
46+
*/
47+
bool read_batch(uint64_t start_row, uint32_t batch_size, executor::VectorBatch& out_batch,
48+
const std::vector<size_t>& col_indices);
49+
4050
/**
4151
* @brief Append a batch of data to the table
4252
*/

‎src/executor/query_executor.cpp‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1512,6 +1512,8 @@ std::unique_ptr<VectorizedOperator> QueryExecutor::build_vectorized_plan(
15121512
auto thread_pool = std::make_shared<executor::ThreadPool>(std::thread::hardware_concurrency());
15131513
std::unique_ptr<VectorizedOperator> current_root =
15141514
std::make_unique<VectorizedSeqScanOperator>(base_table_name, col_table, thread_pool);
1515+
VectorizedSeqScanOperator* base_scan =
1516+
static_cast<VectorizedSeqScanOperator*>(current_root.get());
15151517

15161518
// Track estimated output rows for join reordering decisions
15171519
uint64_t current_est_rows = optimizer::RowEstimator::estimate_scan_rows(*base_table_meta);
@@ -1720,20 +1722,41 @@ std::unique_ptr<VectorizedOperator> QueryExecutor::build_vectorized_plan(
17201722
}
17211723

17221724
executor::Schema output_schema;
1725+
std::vector<size_t> required_col_indices;
17231726
for (const auto& gb : stmt.group_by()) {
17241727
const auto& gb_name = gb->to_string();
17251728
size_t idx = current_root->output_schema().find_column(gb_name);
17261729
if (idx != static_cast<size_t>(-1)) {
17271730
output_schema.add_column(current_root->output_schema().get_column(idx).name(),
17281731
current_root->output_schema().get_column(idx).type(),
17291732
current_root->output_schema().get_column(idx).nullable());
1733+
required_col_indices.push_back(idx);
17301734
}
17311735
}
17321736
for (size_t i = 0; i < agg_infos.size(); ++i) {
17331737
output_schema.add_column("agg_" + std::to_string(i), common::ValueType::TYPE_FLOAT64,
17341738
false);
1739+
if (agg_infos[i].input_col_idx >= 0) {
1740+
required_col_indices.push_back(static_cast<size_t>(agg_infos[i].input_col_idx));
1741+
}
17351742
}
17361743

1744+
// Deduplicate required_col_indices (same column may appear in GROUP BY and aggregate)
1745+
sort(required_col_indices.begin(), required_col_indices.end());
1746+
required_col_indices.erase(unique(required_col_indices.begin(), required_col_indices.end()),
1747+
required_col_indices.end());
1748+
1749+
// Build scan's reduced schema (table columns only, not aggregate output columns)
1750+
executor::Schema scan_reduced_schema;
1751+
for (size_t idx : required_col_indices) {
1752+
scan_reduced_schema.add_column(
1753+
current_root->output_schema().get_column(idx).name(),
1754+
current_root->output_schema().get_column(idx).type(),
1755+
current_root->output_schema().get_column(idx).nullable());
1756+
}
1757+
1758+
base_scan->set_required_columns(required_col_indices, scan_reduced_schema);
1759+
17371760
auto agg_op = std::make_unique<VectorizedGroupByOperator>(
17381761
std::move(current_root), std::move(group_by), std::move(agg_infos), output_schema,
17391762
thread_pool);

‎src/storage/columnar_table.cpp‎

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -295,4 +295,177 @@ bool ColumnarTable::read_batch(uint64_t start_row, uint32_t batch_size,
295295
return true;
296296
}
297297

298+
bool ColumnarTable::read_batch(uint64_t start_row, uint32_t batch_size,
299+
executor::VectorBatch& out_batch,
300+
const std::vector<size_t>& col_indices) {
301+
if (start_row >= row_count_) return false;
302+
if (col_indices.empty()) return false;
303+
304+
uint32_t actual_rows =
305+
static_cast<uint32_t>(std::min(static_cast<uint64_t>(batch_size), row_count_ - start_row));
306+
307+
// out_batch is pre-initialized with the reduced schema by the caller
308+
// (VectorizedSeqScanOperator via set_required_columns)
309+
310+
for (size_t idx = 0; idx < col_indices.size(); ++idx) {
311+
size_t col_idx = col_indices[idx];
312+
const std::string base = name_ + ".col" + std::to_string(col_idx);
313+
std::ifstream n_in(storage_manager_.get_full_path(base + ".nulls.bin"), std::ios::binary);
314+
std::ifstream d_in(storage_manager_.get_full_path(base + ".data.bin"), std::ios::binary);
315+
if (!n_in.is_open() || !d_in.is_open()) return false;
316+
317+
auto& target_col = out_batch.get_column(idx);
318+
const auto type = schema_.get_column(col_idx).type();
319+
320+
if (type == common::ValueType::TYPE_INT64) {
321+
auto& num_vec = dynamic_cast<executor::NumericVector<int64_t>&>(target_col);
322+
323+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
324+
std::vector<uint8_t> nulls(actual_rows);
325+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
326+
327+
d_in.seekg(static_cast<std::streamoff>(start_row * 8), std::ios::beg);
328+
std::vector<int64_t> data(actual_rows);
329+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows * 8);
330+
331+
for (uint32_t r = 0; r < actual_rows; ++r) {
332+
if (nulls[r] != 0U) {
333+
num_vec.append(common::Value::make_null());
334+
} else {
335+
num_vec.append(common::Value::make_int64(data[r]));
336+
}
337+
}
338+
} else if (type == common::ValueType::TYPE_INT32 || type == common::ValueType::TYPE_INT16 ||
339+
type == common::ValueType::TYPE_INT8) {
340+
auto& num_vec = dynamic_cast<executor::NumericVector<int64_t>&>(target_col);
341+
342+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
343+
std::vector<uint8_t> nulls(actual_rows);
344+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
345+
346+
d_in.seekg(static_cast<std::streamoff>(start_row * 8), std::ios::beg);
347+
std::vector<int64_t> data(actual_rows);
348+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows * 8);
349+
350+
for (uint32_t r = 0; r < actual_rows; ++r) {
351+
if (nulls[r] != 0U) {
352+
num_vec.append(common::Value::make_null());
353+
} else if (type == common::ValueType::TYPE_INT32) {
354+
num_vec.append(common::Value(static_cast<int32_t>(data[r])));
355+
} else if (type == common::ValueType::TYPE_INT16) {
356+
num_vec.append(common::Value(static_cast<int16_t>(data[r])));
357+
} else {
358+
num_vec.append(common::Value(static_cast<int8_t>(data[r])));
359+
}
360+
}
361+
} else if (type == common::ValueType::TYPE_FLOAT64) {
362+
auto& num_vec = dynamic_cast<executor::NumericVector<double>&>(target_col);
363+
364+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
365+
std::vector<uint8_t> nulls(actual_rows);
366+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
367+
368+
d_in.seekg(static_cast<std::streamoff>(start_row * 8), std::ios::beg);
369+
std::vector<double> data(actual_rows);
370+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows * 8);
371+
372+
for (uint32_t r = 0; r < actual_rows; ++r) {
373+
if (nulls[r] != 0U) {
374+
num_vec.append(common::Value::make_null());
375+
} else {
376+
num_vec.append(common::Value::make_float64(data[r]));
377+
}
378+
}
379+
} else if (type == common::ValueType::TYPE_FLOAT32) {
380+
auto& num_vec = dynamic_cast<executor::NumericVector<float>&>(target_col);
381+
382+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
383+
std::vector<uint8_t> nulls(actual_rows);
384+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
385+
386+
d_in.seekg(static_cast<std::streamoff>(start_row * 8), std::ios::beg);
387+
std::vector<double> data(actual_rows);
388+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows * 8);
389+
390+
for (uint32_t r = 0; r < actual_rows; ++r) {
391+
if (nulls[r] != 0U) {
392+
num_vec.append(common::Value::make_null());
393+
} else {
394+
num_vec.append(common::Value(static_cast<float>(data[r])));
395+
}
396+
}
397+
} else if (type == common::ValueType::TYPE_DECIMAL) {
398+
auto& num_vec = dynamic_cast<executor::NumericVector<double>&>(target_col);
399+
400+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
401+
std::vector<uint8_t> nulls(actual_rows);
402+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
403+
404+
d_in.seekg(static_cast<std::streamoff>(start_row * 8), std::ios::beg);
405+
std::vector<double> data(actual_rows);
406+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows * 8);
407+
408+
for (uint32_t r = 0; r < actual_rows; ++r) {
409+
if (nulls[r] != 0U) {
410+
num_vec.append(common::Value::make_null());
411+
} else {
412+
num_vec.append(common::Value::make_float64(data[r]));
413+
}
414+
}
415+
} else if (type == common::ValueType::TYPE_BOOL) {
416+
auto& num_vec = dynamic_cast<executor::NumericVector<bool>&>(target_col);
417+
418+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
419+
std::vector<uint8_t> nulls(actual_rows);
420+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
421+
422+
d_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
423+
std::vector<uint8_t> data(actual_rows);
424+
d_in.read(reinterpret_cast<char*>(data.data()), actual_rows);
425+
426+
for (uint32_t r = 0; r < actual_rows; ++r) {
427+
if (nulls[r] != 0U) {
428+
num_vec.append(common::Value::make_null());
429+
} else {
430+
num_vec.append(common::Value(data[r] != 0));
431+
}
432+
}
433+
} else if (type == common::ValueType::TYPE_TEXT ||
434+
type == common::ValueType::TYPE_VARCHAR ||
435+
type == common::ValueType::TYPE_CHAR) {
436+
auto& str_vec = dynamic_cast<executor::StringVector&>(target_col);
437+
438+
n_in.seekg(static_cast<std::streamoff>(start_row), std::ios::beg);
439+
std::vector<uint8_t> nulls(actual_rows);
440+
n_in.read(reinterpret_cast<char*>(nulls.data()), actual_rows);
441+
442+
if (start_row > 0) {
443+
for (uint32_t r = 0; r < start_row; ++r) {
444+
uint32_t len = 0;
445+
if (!d_in.read(reinterpret_cast<char*>(&len), 4)) break;
446+
if (len > 0) {
447+
d_in.seekg(static_cast<std::streamoff>(len), std::ios::cur);
448+
}
449+
}
450+
}
451+
452+
for (uint32_t r = 0; r < actual_rows; ++r) {
453+
uint32_t len = 0;
454+
d_in.read(reinterpret_cast<char*>(&len), 4);
455+
std::string s(len, '\0');
456+
d_in.read(s.data(), len);
457+
if (nulls[r] != 0U) {
458+
str_vec.append(common::Value::make_null());
459+
} else {
460+
str_vec.append(common::Value::make_text(s));
461+
}
462+
}
463+
} else {
464+
throw std::runtime_error("ColumnarTable::read_batch(col_indices): Unsupported type " +
465+
std::to_string(static_cast<int>(type)));
466+
}
467+
}
468+
out_batch.set_row_count(actual_rows);
469+
return true;
470+
}
298471
} // namespace cloudsql::storage

0 commit comments

Comments
 (0)