save-buffer commented on code in PR #13669:
URL: https://github.com/apache/arrow/pull/13669#discussion_r1068646222
##########
cpp/src/arrow/compute/exec/accumulation_queue.h:
##########
@@ -42,16 +45,88 @@ class AccumulationQueue {
void Concatenate(AccumulationQueue&& that);
void InsertBatch(ExecBatch batch);
- int64_t row_count() { return row_count_; }
- size_t batch_count() { return batches_.size(); }
+ void SetBatch(size_t idx, ExecBatch batch);
+ size_t batch_count() const { return batches_.size(); }
bool empty() const { return batches_.empty(); }
+ size_t CalculateRowCount() const;
+
+ // Resizes the accumulation queue to contain size batches. The
+ // new batches will be empty and have length 0, but they will be
+ // usable (useful for concurrent modification of the AccumulationQueue
+ // of separate elements).
+ void Resize(size_t size) { batches_.resize(size); }
void Clear();
- ExecBatch& operator[](size_t i);
+ ExecBatch& operator[](size_t i) { return batches_[i]; }
+ const ExecBatch& operator[](size_t i) const { return batches_[i]; }
private:
- int64_t row_count_;
std::vector<ExecBatch> batches_;
};
-} // namespace util
+class SpillingAccumulationQueue {
+ public:
+ // Number of partitions must be a power of two, since we assign partitions by
+ // looking at bottom few bits.
+ static constexpr int kLogNumPartitions = 6;
+ static constexpr int kNumPartitions = 1 << kLogNumPartitions;
+ Status Init(QueryContext* ctx);
+ // Assumes that the final column in batch contains 64-bit hashes of the
columns.
+ Status InsertBatch(size_t thread_index, ExecBatch batch);
+ Status GetPartition(size_t thread_index, size_t partition,
Review Comment:
Added a comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
