| 552 | } |
| 553 | |
| 554 | inline void table_data::create_streamer(int32_t idx, int32_t worker_id) |
| 555 | { |
| 556 | const auto col_count = num_columns(); |
| 557 | if (streamers_.streamers.empty()) { |
| 558 | streamers_.streamers.resize(col_count); |
| 559 | streamers_.column_to_batches.resize(col_count); |
| 560 | } |
| 561 | ASSERT(idx >= 0 && idx < col_count); |
| 562 | if (streamers_.streamers[idx]) { |
| 563 | return; // Already created |
| 564 | } |
| 565 | if (pg::memory_tracker::has_memory_limit()) { |
| 566 | const auto column_size = pg::utils::get_column_width(get_base_atttypid(idx), get_atttypmod(idx)) * num_total_rows(); |
| 567 | pg::memory_tracker::ensure_memory_available(column_size); |
| 568 | } |
| 569 | heimdall::column_view_ptr cv = get_column_view(idx); |
| 570 | if (worker_id != -1) { |
| 571 | auto [start_row, end_row] = get_row_range(worker_id); |
| 572 | cv = heimdall_common::create_filtered_column(*(cv), |
| 573 | icm::index_mapping_t<int64_t>::slice({start_row, end_row, 1})); |
| 574 | } |
| 575 | streamers_.streamers[idx] = std::make_unique<bifrost::column_streamer>(cv, batch_size_); |
| 576 | const int64_t row_count = num_total_rows(); |
| 577 | const int64_t batch_count = (row_count + batch_size_ - 1) / batch_size_; |
| 578 | streamers_.column_to_batches[idx].batches.resize(batch_count); |
| 579 | } |
| 580 | |
| 581 | inline nd::array table_data::streamer_info::get_sample(int32_t column_number, int64_t row_number) |
| 582 | { |
no test coverage detected