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

Method _lower

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

Source from the content-addressed store, hash-verified

3518 return [op for op in dfs if not is_broadcastable(dfs, op)]
3519
3520 def _lower(self):
3521 # This can be expensive when something that has expensive division
3522 # calculation is in the Expression
3523 dfs = self.args
3524 if (
3525 len(dfs) == 1
3526 or all(
3527 dfs[0].divisions == df.divisions and df.known_divisions for df in dfs
3528 )
3529 or len(self.divisions) == 2
3530 and max(map(lambda x: len(x.divisions), dfs)) == 2
3531 ):
3532 return self._expr_cls(*self.operands)
3533 elif self.divisions[0] is None:
3534 # We have to shuffle
3535 npartitions = max(df.npartitions for df in dfs)
3536 dtypes = {df._meta.index.dtype for df in dfs}
3537 if not _are_dtypes_shuffle_compatible(dtypes):
3538 raise TypeError(
3539 "DataFrames are not aligned. We need to shuffle to align partitions "
3540 "with each other. This is not possible because the indexes of the "
3541 f"DataFrames have differing dtypes={dtypes}. Please ensure that "
3542 "all Indexes have the same dtype or align manually for this to "
3543 "work."
3544 )
3545
3546 from dask.dataframe.dask_expr._shuffle import RearrangeByColumn
3547
3548 args = [
3549 (
3550 RearrangeByColumn(df, None, npartitions, index_shuffle=True)
3551 if isinstance(df, Expr)
3552 else df
3553 )
3554 for df in self.operands
3555 ]
3556 return self._expr_cls(*args)
3557
3558 args = maybe_align_partitions(*self.operands, divisions=self.divisions)
3559 return self._expr_cls(*args)
3560
3561 @functools.cached_property
3562 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