viirya commented on code in PR #27278:
URL: https://github.com/apache/datafusion/pull/27278#discussion_r4530641636
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1771,33 +2008,215 @@ impl MaterializingSortMergeJoinStream {
Ok(())
}
- /// Materializes right-side columns across all matched chunks.
+ /// Freezes the matched pairs of an Inner join with a filter. The filter
runs
+ /// on just the columns it reads, for the pairs of the chunks that
+ /// [`Self::pair_filtered_run`] did not filter already, and only the pairs
that
+ /// pass have their output columns materialized. A selective filter over a
+ /// many-to-many join rejects most pairs, so materializing every column of
+ /// every pair before filtering would spend most of the freeze on rows that
+ /// are then dropped. When no chunk was filtered already and earlier
freezes
+ /// kept enough of their pairs, [`Self::filter_first_pays`] says the
opposite,
+ /// and the batch is materialized whole and then filtered.
+ fn freeze_inner_filtered(
+ &mut self,
+ matched_chunks: &[MatchedChunk],
+ total_matched_rows: usize,
+ ) -> Result<()> {
+ let Some(filter) = &self.filter else {
+ return internal_err!("early filtering requires a join filter");
+ };
+ debug_assert_eq!(
+ self.join_type,
+ JoinType::Inner,
+ "only an Inner join filters early, so the streamed side is the
left one"
+ );
+
+ // Evaluate the filter over the pairs of the chunks not filtered yet.
+ let unfiltered: Vec<&MatchedChunk> =
+ matched_chunks.iter().filter(|c| !c.filtered).collect();
+ if unfiltered.len() == matched_chunks.len() &&
!self.filter_first_pays(filter) {
+ return self.freeze_inner_filtered_whole(matched_chunks,
total_matched_rows);
+ }
+ let passed = if unfiltered.is_empty() {
+ None
+ } else {
+ let num_rows = unfiltered.iter().map(|c| c.len()).sum();
+ let streamed_indices =
+ concat_indices(unfiltered.iter().map(|c|
&c.streamed_indices))?;
+ let mut buffered_arrays = self
+ .materialize_right_columns(
+ &unfiltered,
+ num_rows,
+ &self.buffered_filter_columns,
+ )?
+ .into_iter();
+ let columns = filter
+ .column_indices()
+ .iter()
+ .map(|column_index| match column_index.side {
+ JoinSide::Left => take_column(
+ self.streamed_batch.batch.column(column_index.index),
+ &streamed_indices,
+ ),
+ JoinSide::Right => buffered_arrays.next().ok_or_else(|| {
+ internal_datafusion_err!("missing buffered filter
column")
+ }),
+ JoinSide::None => {
+ internal_err!("Unexpected JoinSide::None in filter")
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ self.freeze_filter_evaluated += num_rows;
+ Some(evaluate_join_filter(
+ filter,
+ filter.schema(),
+ columns,
+ num_rows,
+ )?)
+ };
+
+ // Keep the pairs that passed, in order.
+ let mut kept: Vec<Cow<MatchedChunk>> =
Vec::with_capacity(matched_chunks.len());
+ let mut kept_rows = 0;
+ let mut offset = 0;
+ for chunk in matched_chunks {
+ let len = chunk.len();
+ match (&passed, chunk.filtered) {
+ (Some(passed), false) => {
+ let mask = passed.slice(offset, len);
+ offset += len;
+ let count = mask.count_set_bits();
+ if count == len {
+ kept.push(Cow::Borrowed(chunk));
+ } else if count > 0 {
+ kept.push(Cow::Owned(chunk.filter(&mask)?));
+ }
+ self.freeze_filter_passed += count;
+ kept_rows += count;
+ }
+ _ => {
+ kept.push(Cow::Borrowed(chunk));
+ kept_rows += len;
+ }
+ }
+ }
+ if kept_rows == 0 {
+ return Ok(());
+ }
+
+ let streamed_indices = concat_indices(kept.iter().map(|c|
&c.streamed_indices))?;
+ let streamed_columns =
+ materialize_left_columns(&self.streamed_batch.batch,
&streamed_indices)?;
+ let buffered_columns =
+ self.materialize_right_columns(&kept, kept_rows,
&self.buffered_columns)?;
+ let output_batch = RecordBatch::try_new(
+ Arc::clone(&self.schema),
+ [streamed_columns, buffered_columns].concat(),
+ )?;
+ self.joined_record_batches
+ .push_batch_without_metadata(output_batch);
+ Ok(())
+ }
+
+ /// Freezes the matched pairs of an Inner join with a filter by
materializing
+ /// every column of every pair, evaluating the filter on the columns it
reads,
+ /// and filtering the batch. None of the pairs was filtered already.
+ fn freeze_inner_filtered_whole(
+ &mut self,
+ matched_chunks: &[MatchedChunk],
+ total_matched_rows: usize,
+ ) -> Result<()> {
+ let streamed_indices =
+ concat_indices(matched_chunks.iter().map(|c|
&c.streamed_indices))?;
+ let streamed_columns =
+ materialize_left_columns(&self.streamed_batch.batch,
&streamed_indices)?;
+ let buffered_columns = self.materialize_right_columns(
+ matched_chunks,
+ total_matched_rows,
+ &self.buffered_columns,
+ )?;
+ let Some(filter) = &self.filter else {
+ return internal_err!("early filtering requires a join filter");
+ };
+ let filter_columns =
+ get_filter_columns(Some(filter), &streamed_columns,
&buffered_columns)?;
+ let passed = evaluate_join_filter(
+ filter,
+ filter.schema(),
+ filter_columns,
+ total_matched_rows,
+ )?;
+ let kept_rows = passed.count_set_bits();
+ self.freeze_filter_evaluated += total_matched_rows;
+ self.freeze_filter_passed += kept_rows;
+ if kept_rows == 0 {
+ return Ok(());
+ }
+
+ let output_batch = RecordBatch::try_new(
+ Arc::clone(&self.schema),
+ [streamed_columns, buffered_columns].concat(),
+ )?;
+ let output_batch = if kept_rows == total_matched_rows {
+ output_batch
+ } else {
+ filter_record_batch(&output_batch, &BooleanArray::new(passed,
None))?
+ };
+ self.joined_record_batches
+ .push_batch_without_metadata(output_batch);
+ Ok(())
+ }
+
+ /// Whether filtering the pairs of a freeze before materializing them is
cheaper
+ /// than materializing every column of every pair and filtering the batch.
+ ///
+ /// Filtering first gathers the filter's `f` columns for every pair,
filters the
+ /// pairs' two index arrays, and gathers all `k` output columns for the
share
+ /// `p` of pairs that pass. The other way gathers the `k` columns for every
+ /// pair and copies the share `p` of them to filter the batch. Counting a
+ /// gathered value as 1 and a copied one as 1/2, filtering first is cheaper
+ /// while `f + 1 + k·p < k + k·p/2`, that is while `p < 2·(k - f - 1)/k`.
`p` is
+ /// the share that passed in the earlier freezes, or 0 before the first.
+ fn filter_first_pays(&self, filter: &JoinFilter) -> bool {
+ let k = self.schema.fields().len() as f64;
+ let f = filter.column_indices().len() as f64;
+ let p = if self.freeze_filter_evaluated == 0 {
+ 0.0
+ } else {
+ self.freeze_filter_passed as f64 / self.freeze_filter_evaluated as
f64
+ };
+ p < 2.0 * (k - f - 1.0) / k
Review Comment:
+1 to the contiguous check. Two more things the model leaves out, in case it
stays: it counts columns rather than bytes, so a string payload weighs the same
as an `Int64`, and `p` is a lifetime average that pairing-time evaluations
never feed. Neither needs solving here, but the doc comment should say so.
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1771,33 +2008,215 @@ impl MaterializingSortMergeJoinStream {
Ok(())
}
- /// Materializes right-side columns across all matched chunks.
+ /// Freezes the matched pairs of an Inner join with a filter. The filter
runs
+ /// on just the columns it reads, for the pairs of the chunks that
+ /// [`Self::pair_filtered_run`] did not filter already, and only the pairs
that
+ /// pass have their output columns materialized. A selective filter over a
+ /// many-to-many join rejects most pairs, so materializing every column of
+ /// every pair before filtering would spend most of the freeze on rows that
+ /// are then dropped. When no chunk was filtered already and earlier
freezes
+ /// kept enough of their pairs, [`Self::filter_first_pays`] says the
opposite,
+ /// and the batch is materialized whole and then filtered.
+ fn freeze_inner_filtered(
+ &mut self,
+ matched_chunks: &[MatchedChunk],
+ total_matched_rows: usize,
+ ) -> Result<()> {
+ let Some(filter) = &self.filter else {
+ return internal_err!("early filtering requires a join filter");
+ };
+ debug_assert_eq!(
+ self.join_type,
+ JoinType::Inner,
+ "only an Inner join filters early, so the streamed side is the
left one"
+ );
+
+ // Evaluate the filter over the pairs of the chunks not filtered yet.
+ let unfiltered: Vec<&MatchedChunk> =
+ matched_chunks.iter().filter(|c| !c.filtered).collect();
+ if unfiltered.len() == matched_chunks.len() &&
!self.filter_first_pays(filter) {
+ return self.freeze_inner_filtered_whole(matched_chunks,
total_matched_rows);
+ }
+ let passed = if unfiltered.is_empty() {
+ None
+ } else {
+ let num_rows = unfiltered.iter().map(|c| c.len()).sum();
+ let streamed_indices =
+ concat_indices(unfiltered.iter().map(|c|
&c.streamed_indices))?;
+ let mut buffered_arrays = self
+ .materialize_right_columns(
+ &unfiltered,
+ num_rows,
+ &self.buffered_filter_columns,
+ )?
+ .into_iter();
+ let columns = filter
+ .column_indices()
+ .iter()
+ .map(|column_index| match column_index.side {
+ JoinSide::Left => take_column(
+ self.streamed_batch.batch.column(column_index.index),
+ &streamed_indices,
+ ),
+ JoinSide::Right => buffered_arrays.next().ok_or_else(|| {
+ internal_datafusion_err!("missing buffered filter
column")
+ }),
+ JoinSide::None => {
+ internal_err!("Unexpected JoinSide::None in filter")
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ self.freeze_filter_evaluated += num_rows;
+ Some(evaluate_join_filter(
+ filter,
+ filter.schema(),
+ columns,
+ num_rows,
+ )?)
+ };
+
+ // Keep the pairs that passed, in order.
+ let mut kept: Vec<Cow<MatchedChunk>> =
Vec::with_capacity(matched_chunks.len());
Review Comment:
`MatchedChunk::clone` only bumps the refcounts of its two index buffers, so
`Vec<MatchedChunk>` with a plain clone would do here. That drops `Cow`, and
`materialize_right_columns` can go back to taking `&[MatchedChunk]` instead of
being generic over `Borrow`. With @andygrove's diff above,
`MatchedChunk::filter` goes away as well.
##########
datafusion/physical-plan/src/joins/sort_merge_join/tests.rs:
##########
@@ -8395,3 +8399,322 @@ async fn
left_join_with_partial_filter_preserves_streamed_order() -> Result<()>
);
Ok(())
}
+
+// ============================================================================
+// Early filtering of Inner joins
+// ============================================================================
+
+/// Rows for the early-filtering tests, sorted by the key column `names[0]`:
key
+/// `k` of `0..keys` has `rows_per_key(k)` rows. `names[1]` is the row number
and
+/// `names[2]` a pseudo-random value in `0..1000` that is NULL in every 37th
row.
+/// The rows are split into batches of `rows_per_batch`.
+fn early_filtering_batches(
+ names: [&str; 3],
+ keys: i32,
+ rows_per_key: impl Fn(i32) -> usize,
+ seed: u64,
+ rows_per_batch: usize,
+) -> (Vec<RecordBatch>, SchemaRef) {
+ let mut key_values = vec![];
+ let mut row_numbers = vec![];
+ let mut values = vec![];
+ let mut state = seed;
+ for key in 0..keys {
+ for _ in 0..rows_per_key(key) {
+ state = state
+ .wrapping_mul(6364136223846793005)
+ .wrapping_add(1442695040888963407);
+ let row = key_values.len() as i32;
+ key_values.push(key);
+ row_numbers.push(row);
+ values.push((row % 37 != 0).then_some(((state >> 33) % 1000) as
i32));
+ }
+ }
+ let schema = Arc::new(Schema::new(vec![
+ Field::new(names[0], DataType::Int32, false),
+ Field::new(names[1], DataType::Int32, false),
+ Field::new(names[2], DataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(key_values)),
+ Arc::new(Int32Array::from(row_numbers)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap();
+ let batches = (0..batch.num_rows())
+ .step_by(rows_per_batch)
+ .map(|offset| batch.slice(offset, rows_per_batch.min(batch.num_rows()
- offset)))
+ .collect();
+ (batches, schema)
+}
+
+/// Inputs for the early-filtering tests. Every third key has a group of 600
+/// buffered rows, long enough to be filtered as it is paired, and the 250-row
+/// buffered batches split each such group. The other keys have groups of one
or
+/// two buffered rows, whose pairs are filtered at freeze. Keys repeat up to
four
+/// times on the streamed side.
+fn early_filtering_inputs()
+-> ((Vec<RecordBatch>, SchemaRef), (Vec<RecordBatch>, SchemaRef)) {
+ let left =
+ early_filtering_batches(["a1", "b1", "c1"], 40, |k| 1 + (k % 4) as
usize, 1, 7);
+ let right = early_filtering_batches(
+ ["a2", "b2", "c2"],
+ 40,
+ |k| if k % 3 == 0 { 600 } else { (k % 3) as usize },
+ 2,
+ 250,
+ );
+ (left, right)
+}
+
+/// The filter `c2 > c1 AND b1 % 2 = b2 % 2`, whose intermediate columns are in
+/// the order `c2, c1, b1, b2`, so that they alternate between the join sides.
+fn interleaved_early_filter(left_schema: &Schema, right_schema: &Schema) ->
JoinFilter {
+ let parity = |name: &str, index: usize| -> PhysicalExprRef {
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new(name, index)),
+ Operator::Modulo,
+ Arc::new(Literal::new(ScalarValue::Int32(Some(2)))),
+ ))
+ };
+ let expression = Arc::new(BinaryExpr::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c2", 0)),
+ Operator::Gt,
+ Arc::new(Column::new("c1", 1)),
+ )),
+ Operator::And,
+ Arc::new(BinaryExpr::new(
+ parity("b1", 2),
+ Operator::Eq,
+ parity("b2", 3),
+ )),
+ ));
+ let column_indices = vec![
+ ColumnIndex {
+ index: 2,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 2,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ];
+ let schema = Schema::new(vec![
+ right_schema.field_with_name("c2").unwrap().clone(),
+ left_schema.field_with_name("c1").unwrap().clone(),
+ left_schema.field_with_name("b1").unwrap().clone(),
+ right_schema.field_with_name("b2").unwrap().clone(),
+ ]);
+ JoinFilter::new(expression, column_indices, Arc::new(schema))
+}
+
+/// The filter `b2 <> b1 OR c2 < c1`, which nearly every pair passes, with its
+/// intermediate columns in the order `b2, b1, c2, c1`. It reads four of the
six
+/// output columns, so after the first freeze the join materializes whole
batches
+/// and filters them (see `filter_first_pays`).
+fn high_pass_early_filter(left_schema: &Schema, right_schema: &Schema) ->
JoinFilter {
+ let expression = Arc::new(BinaryExpr::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b2", 0)),
+ Operator::NotEq,
+ Arc::new(Column::new("b1", 1)),
+ )),
+ Operator::Or,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c2", 2)),
+ Operator::Lt,
+ Arc::new(Column::new("c1", 3)),
+ )),
+ ));
+ let column_indices = vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 2,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 2,
+ side: JoinSide::Left,
+ },
+ ];
+ let schema = Schema::new(vec![
+ right_schema.field_with_name("b2").unwrap().clone(),
+ left_schema.field_with_name("b1").unwrap().clone(),
+ right_schema.field_with_name("c2").unwrap().clone(),
+ left_schema.field_with_name("c1").unwrap().clone(),
+ ]);
+ JoinFilter::new(expression, column_indices, Arc::new(schema))
+}
+
+/// A filter that reads no columns and always evaluates to `value`.
+fn column_free_early_filter(value: bool) -> JoinFilter {
+ JoinFilter::new(
+ Arc::new(Literal::new(ScalarValue::Boolean(Some(value)))),
+ vec![],
+ Arc::new(Schema::empty()),
+ )
+}
+
+fn early_filtering_on(left_schema: &Schema, right_schema: &Schema) ->
Result<JoinOn> {
+ Ok(vec![(
+ Arc::new(Column::new_with_schema("a1", left_schema)?) as _,
+ Arc::new(Column::new_with_schema("a2", right_schema)?) as _,
+ )])
+}
+
+/// Asserts that the key column of `batches` is ascending, the order of the
+/// streamed side that the join maintains.
+fn assert_streamed_order(batches: &[RecordBatch]) {
+ let keys: Vec<i32> = batches
+ .iter()
+ .flat_map(|b| {
+ b.column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ assert!(
+ keys.is_sorted(),
Review Comment:
This checks only that keys ascend, so a reorder within a key passes. `b1`
and `b2` are row numbers on each side, so an Inner join's output should ascend
on `(b1, b2)`. Asserting that covers the order within a key too.
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1771,33 +2008,215 @@ impl MaterializingSortMergeJoinStream {
Ok(())
}
- /// Materializes right-side columns across all matched chunks.
+ /// Freezes the matched pairs of an Inner join with a filter. The filter
runs
+ /// on just the columns it reads, for the pairs of the chunks that
+ /// [`Self::pair_filtered_run`] did not filter already, and only the pairs
that
+ /// pass have their output columns materialized. A selective filter over a
+ /// many-to-many join rejects most pairs, so materializing every column of
+ /// every pair before filtering would spend most of the freeze on rows that
+ /// are then dropped. When no chunk was filtered already and earlier
freezes
+ /// kept enough of their pairs, [`Self::filter_first_pays`] says the
opposite,
+ /// and the batch is materialized whole and then filtered.
+ fn freeze_inner_filtered(
+ &mut self,
+ matched_chunks: &[MatchedChunk],
+ total_matched_rows: usize,
+ ) -> Result<()> {
+ let Some(filter) = &self.filter else {
+ return internal_err!("early filtering requires a join filter");
+ };
+ debug_assert_eq!(
+ self.join_type,
+ JoinType::Inner,
+ "only an Inner join filters early, so the streamed side is the
left one"
+ );
+
+ // Evaluate the filter over the pairs of the chunks not filtered yet.
+ let unfiltered: Vec<&MatchedChunk> =
+ matched_chunks.iter().filter(|c| !c.filtered).collect();
+ if unfiltered.len() == matched_chunks.len() &&
!self.filter_first_pays(filter) {
+ return self.freeze_inner_filtered_whole(matched_chunks,
total_matched_rows);
+ }
+ let passed = if unfiltered.is_empty() {
+ None
+ } else {
+ let num_rows = unfiltered.iter().map(|c| c.len()).sum();
+ let streamed_indices =
+ concat_indices(unfiltered.iter().map(|c|
&c.streamed_indices))?;
+ let mut buffered_arrays = self
+ .materialize_right_columns(
+ &unfiltered,
+ num_rows,
+ &self.buffered_filter_columns,
+ )?
+ .into_iter();
+ let columns = filter
+ .column_indices()
+ .iter()
+ .map(|column_index| match column_index.side {
+ JoinSide::Left => take_column(
+ self.streamed_batch.batch.column(column_index.index),
+ &streamed_indices,
+ ),
+ JoinSide::Right => buffered_arrays.next().ok_or_else(|| {
+ internal_datafusion_err!("missing buffered filter
column")
+ }),
+ JoinSide::None => {
+ internal_err!("Unexpected JoinSide::None in filter")
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ self.freeze_filter_evaluated += num_rows;
+ Some(evaluate_join_filter(
+ filter,
+ filter.schema(),
+ columns,
+ num_rows,
+ )?)
+ };
+
+ // Keep the pairs that passed, in order.
+ let mut kept: Vec<Cow<MatchedChunk>> =
Vec::with_capacity(matched_chunks.len());
+ let mut kept_rows = 0;
+ let mut offset = 0;
+ for chunk in matched_chunks {
+ let len = chunk.len();
+ match (&passed, chunk.filtered) {
+ (Some(passed), false) => {
+ let mask = passed.slice(offset, len);
+ offset += len;
+ let count = mask.count_set_bits();
+ if count == len {
+ kept.push(Cow::Borrowed(chunk));
+ } else if count > 0 {
+ kept.push(Cow::Owned(chunk.filter(&mask)?));
+ }
+ self.freeze_filter_passed += count;
+ kept_rows += count;
+ }
+ _ => {
+ kept.push(Cow::Borrowed(chunk));
+ kept_rows += len;
+ }
+ }
+ }
+ if kept_rows == 0 {
+ return Ok(());
+ }
+
+ let streamed_indices = concat_indices(kept.iter().map(|c|
&c.streamed_indices))?;
+ let streamed_columns =
+ materialize_left_columns(&self.streamed_batch.batch,
&streamed_indices)?;
+ let buffered_columns =
+ self.materialize_right_columns(&kept, kept_rows,
&self.buffered_columns)?;
Review Comment:
I don't think this is specific to this PR. A 256-row batch with 200 distinct
values cannot be encoded with `Int8` keys by any path. The base passes with
these inputs because each freeze leaves 150 rows, which is over half of
`batch_size`, so the coalescer passes them through without concatenating. On
`main`, would 128-row buffered batches (two chunks per freeze, through
`interleave`) or a filter keeping 100 rows per batch (through the coalescer's
`concat`) fail the same way?
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -805,6 +967,91 @@ impl MaterializingSortMergeJoinStream {
true
}
+ /// [`Self::pair_streamed_row_with_group`] for an Inner join with a filter,
+ /// which keeps only the pairs that pass it. For a run of at least
+ /// [`FILTERED_RUN_MIN_ROWS`] buffered rows, the filter runs as the run is
+ /// paired, and the pairs it rejects are never recorded.
+ fn pair_streamed_row_with_group_filtered(&mut self) -> Result<bool> {
+ let mut room =
self.batch_size.saturating_sub(self.num_unfrozen_pairs());
+ while !self.buffered_data.scanning_finished() && room > 0 {
+ let rows = self.buffered_data.scanning_remaining();
+ let len = rows.len().min(room);
+ let filtered = if len >= FILTERED_RUN_MIN_ROWS {
+ self.pair_filtered_run(rows.start, len)?
+ } else {
+ None
+ };
+ match filtered {
+ Some(appended) => room -= appended,
Review Comment:
`main` already polls the streamed child while holding fewer than
`batch_size` unfrozen pairs: an unfiltered join with 200 and then 360 pairs
from two one-row streamed batches would hold 560 rows the same way. The bound
is still about 2 × `batch_size`. Freezing before the poll when the coalescer
plus unfrozen pairs reach `batch_size` would help both cases, so it may fit
better as its own change.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
