MCPcopy Create free account
hub / github.com/dask/dask / _lower

Method _lower

dask/dataframe/dask_expr/_expr.py:3543–3582  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

3541 return [op for op in dfs if not is_broadcastable(dfs, op)]
3542
3543 def _lower(self):
3544 # This can be expensive when something that has expensive division
3545 # calculation is in the Expression
3546 dfs = self.args
3547 if (
3548 len(dfs) == 1
3549 or all(
3550 dfs[0].divisions == df.divisions and df.known_divisions for df in dfs
3551 )
3552 or len(self.divisions) == 2
3553 and max(map(lambda x: len(x.divisions), dfs)) == 2
3554 ):
3555 return self._expr_cls(*self.operands)
3556 elif self.divisions[0] is None:
3557 # We have to shuffle
3558 npartitions = max(df.npartitions for df in dfs)
3559 dtypes = {df._meta.index.dtype for df in dfs}
3560 if not _are_dtypes_shuffle_compatible(dtypes):
3561 raise TypeError(
3562 "DataFrames are not aligned. We need to shuffle to align partitions "
3563 "with each other. This is not possible because the indexes of the "
3564 f"DataFrames have differing dtypes={dtypes}. Please ensure that "
3565 "all Indexes have the same dtype or align manually for this to "
3566 "work."
3567 )
3568
3569 from dask.dataframe.dask_expr._shuffle import RearrangeByColumn
3570
3571 args = [
3572 (
3573 RearrangeByColumn(df, None, npartitions, index_shuffle=True)
3574 if isinstance(df, Expr)
3575 else df
3576 )
3577 for df in self.operands
3578 ]
3579 return self._expr_cls(*args)
3580
3581 args = maybe_align_partitions(*self.operands, divisions=self.divisions)
3582 return self._expr_cls(*args)
3583
3584 @functools.cached_property
3585 def _meta(self):

Callers

nothing calls this directly

Calls 5

RearrangeByColumnClass · 0.90
allFunction · 0.85
maxFunction · 0.85
maybe_align_partitionsFunction · 0.85

Tested by

no test coverage detected