| 600 | } |
| 601 | |
| 602 | fn fold_flat_map_constant( |
| 603 | func: &TableFunc, |
| 604 | exprs: &[MirScalarExpr], |
| 605 | rows: &[(Row, Diff)], |
| 606 | limit: Option<usize>, |
| 607 | ) -> Result<Option<Vec<(Row, Diff)>>, EvalError> { |
| 608 | // We cannot exceed `usize::MAX` in any array, so this is a fine upper bound. |
| 609 | let limit = limit.unwrap_or(usize::MAX); |
| 610 | let mut new_rows = Vec::new(); |
| 611 | let mut row_buf = Row::default(); |
| 612 | let mut datum_vec = mz_repr::DatumVec::new(); |
| 613 | for (input_row, diff) in rows { |
| 614 | let datums = datum_vec.borrow_with(input_row); |
| 615 | let temp_storage = RowArena::new(); |
| 616 | let datums = exprs |
| 617 | .iter() |
| 618 | .map(|expr| expr.eval(&datums, &temp_storage)) |
| 619 | .collect::<Result<Vec<_>, _>>()?; |
| 620 | let mut output_rows = func.eval(&datums, &temp_storage)?.fuse(); |
| 621 | for (output_row, diff2) in (&mut output_rows).take(limit - new_rows.len()) { |
| 622 | let mut packer = row_buf.packer(); |
| 623 | packer.extend_by_row(input_row); |
| 624 | packer.extend_by_row(&output_row); |
| 625 | new_rows.push((row_buf.clone(), diff2 * *diff)) |
| 626 | } |
| 627 | // If we still have records to enumerate, but dropped out of the iteration, |
| 628 | // it means we have exhausted `limit` and should stop. |
| 629 | if output_rows.next() != None { |
| 630 | return Ok(None); |
| 631 | } |
| 632 | } |
| 633 | Ok(Some(new_rows)) |
| 634 | } |
| 635 | |
| 636 | fn fold_filter_constant( |
| 637 | predicates: &[MirScalarExpr], |