@@ -86,67 +86,80 @@ double CKMSQuantiles::allowableError(int rank) {
8686}
8787
8888bool CKMSQuantiles::insertBatch () {
89+ // If there is no data to insert return false
8990 if (buffer_count_ == 0 ) {
9091 return false ;
9192 }
9293
94+ // Sort the buffer upto buffer_count_ to prepare for inserting items
9395 std::sort (buffer_.begin (), buffer_.begin () + buffer_count_);
9496
9597 std::size_t start = 0 ;
98+
99+ sample_.reserve (buffer_count_);
100+ // If the sample set is empty, add the first item
96101 if (sample_.empty ()) {
97102 sample_.emplace_back (buffer_[0 ], 1 , 0 );
98- ++start;
103+ ++start; // Skip the first item since it's already added to the sample
99104 ++count_;
100105 }
101106
102- std::size_t idx = 0 ;
103- std::size_t item = idx++;
104-
107+ // Loop through the buffer and insert the items into the sample set
105108 for (std::size_t i = start; i < buffer_count_; ++i) {
106- double v = buffer_[i];
107- while (idx < sample_.size () && sample_[item].value < v) {
108- item = idx++;
109- }
110-
111- if (sample_[item].value > v) {
112- --idx;
109+ float value = buffer_[i];
110+
111+ auto iterator = std::lower_bound (
112+ sample_.begin (), sample_.end (), value,
113+ [](const Item& item, float val) { return item.value < val; });
114+ std::size_t idx = std::distance (sample_.begin (), iterator);
115+
116+ int delta = 0 ;
117+ if (idx > 0 && idx < sample_.size ()) {
118+ delta = static_cast <int >(
119+ std::floor (allowableError (static_cast <int >(idx) + 1 ))) +
120+ 1 ;
113121 }
114122
115- int delta;
116- if (idx - 1 == 0 || idx + 1 == sample_.size ()) {
117- delta = 0 ;
118- } else {
119- delta = static_cast <int >(std::floor (allowableError (idx + 1 ))) + 1 ;
120- }
121-
122- sample_.emplace (sample_.begin () + idx, v, 1 , delta);
123- count_++;
124- item = idx++;
123+ sample_.emplace (iterator, value, 1 , delta);
124+ ++count_;
125125 }
126126
127127 buffer_count_ = 0 ;
128128 return true ;
129129}
130130
131131void CKMSQuantiles::compress () {
132+ // If there are less than 2 items in the sample set, there's nothing to
133+ // compress
132134 if (sample_.size () < 2 ) {
133135 return ;
134136 }
135137
136- std::size_t idx = 0 ;
137- std::size_t prev;
138- std::size_t next = idx++;
138+ std::vector<Item> compressed_samples; // Vector to hold compressed samples
139+ compressed_samples.reserve (
140+ sample_.size ()); // Reserve space to avoid multiple allocations
141+
142+ // Start with the first sample
143+ compressed_samples.push_back (sample_[0 ]);
139144
140- while ( idx < sample_.size ()) {
141- prev = next ;
142- next = idx++ ;
145+ for (std:: size_t idx = 1 ; idx < sample_.size (); ++idx ) {
146+ const Item& current_sample = sample_[idx] ;
147+ Item& last_compressed_sample = compressed_samples. back () ;
143148
144- if (sample_[prev].g + sample_[next].g + sample_[next].delta <=
145- allowableError (idx - 1 )) {
146- sample_[next].g += sample_[prev].g ;
147- sample_.erase (sample_.begin () + prev);
149+ // Check if we can compress the current sample into the last compressed
150+ // sample
151+ if (last_compressed_sample.g + current_sample.g + current_sample.delta <=
152+ allowableError (static_cast <int >(compressed_samples.size ()) - 1 )) {
153+ // Merge current sample into last compressed sample
154+ last_compressed_sample.g += current_sample.g ; // Update weight
155+ } else {
156+ // If not compressible, add current sample to compressed samples
157+ compressed_samples.push_back (current_sample);
148158 }
149159 }
160+
161+ // Replace old samples with new compressed samples
162+ sample_ = std::move (compressed_samples);
150163}
151164
152165} // namespace detail
0 commit comments