Method deduplicate

datafusion/physical-plan/src/recursive_query.rs:481–491

Remove duplicated rows from the given batch, keeping a state between batches. We use a hash table to allocate new group ids for the new rows. [`GroupValues`] allocate increasing group ids. Hence, if groups (i.e., rows) are new, then they have ids >= length before interning, we keep them. We also detect duplicates by enforcing that group ids are increasing.

(&mut self, batch: &RecordBatch)

    /// Hence, if groups (i.e., rows) are new, then they have ids >= length before interning, we keep them.
    /// We also detect duplicates by enforcing that group ids are increasing.
    fn deduplicate(&mut self, batch: &RecordBatch) -> Result<RecordBatch> {
        let size_before = self.group_values.len();
        self.intern_output_buffer.reserve(batch.num_rows());
        self.group_values
            .intern(batch.columns(), &mut self.intern_output_buffer)?;
        let mask = new_groups_mask(&self.intern_output_buffer, size_before);
        self.intern_output_buffer.clear();
        // We update the reservation to reflect the new size of the hash table.
        self.reservation.try_resize(self.group_values.size())?;
        Ok(filter_record_batch(batch, &mask)?)
    }
}

/// Return a mask, each element being true if, and only if, the element is greater than all previous elements and greater or equal than the provided max_already_seen_group_id