Class FromArray

dask/dataframe/dask_expr/io/io.py:633–719

class FromArray(PartitionsFiltered, BlockwiseIO):
    _parameters = [
        "frame",
        "chunksize",
        "original_columns",
        "meta",
        "columns",
        "_partitions",
    ]
    _defaults = {
        "chunksize": 50_000,
        "original_columns": None,
        "meta": None,
        "columns": None,
        "_partitions": None,
    }
    _absorb_projections = True

    @functools.cached_property
    def _meta(self):
        meta = _meta_from_array(
            self.frame, self.operand("original_columns"), self.operand("meta")
        )
        if self.operand("columns") is not None:
            return meta[self.operand("columns")]
        return meta

    @functools.cached_property
    def original_columns(self):
        if self.operand("original_columns") is None:
            if is_series_like(self._meta):
                return [0]
            return list(range(len(self._meta.columns)))
        return self.operand("original_columns")

    @functools.cached_property
    def _column_indices(self):
        if self.operand("columns") is None:
            return slice(0, len(self.original_columns))
        return [
            i
            for i, col in enumerate(self.original_columns)
            if col in self.operand("columns")
        ]

    def _divisions(self):
        divisions = tuple(range(0, len(self.frame), self.chunksize))
        divisions = divisions + (len(self.frame) - 1,)
        return divisions

    @functools.cached_property
    def unfiltered_divisions(self):
        return self._divisions()

    def _filtered_task(self, name: Key, index: int) -> Task:
        data = self.frame[slice(index * self.chunksize, (index + 1) * self.chunksize)]
        if index == len(self.unfiltered_divisions) - 2:
            idx = range(