DataFrame#

def __contains__(self, col_name: str) -> bool:
    """Returns whether the column exists in the dataframe.

    Args:
        col_name (str): column name

    Returns:
        bool: whether the column exists in the dataframe.

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"x": [1, 2, 3], "y": [4, 5, 6], "z": [7, 8, 9]})
        >>> "x" in df
        True

    """
    return col_name in self.column_names

def __getitem__(self, item: Union[slice, int, str, Iterable[Union[str, int]]]) -> Union[Expression, "DataFrame"]:
    """Gets a column from the DataFrame as an Expression (``df["mycol"]``)."""
    result: Optional[Expression]

    if isinstance(item, int):
        schema = self._builder.schema()
        if item < -len(schema) or item >= len(schema):
            raise ValueError(f"{item} out of bounds for {schema}")
        result = ExpressionsProjection.from_schema(schema)[item]
        assert result is not None
        return result
    elif isinstance(item, str):
        schema = self._builder.schema()
        if item not in schema.column_names() and item != "*":
            raise ValueError(f"{item} does not exist in schema {schema}")

        return col(item)
    elif isinstance(item, Iterable):
        schema = self._builder.schema()

        columns = []
        for it in item:
            if isinstance(it, str):
                result = col(schema[it].name)
                columns.append(result)
            elif isinstance(it, int):
                if it < -len(schema) or it >= len(schema):
                    raise ValueError(f"{it} out of bounds for {schema}")
                field = list(self._builder.schema())[it]
                columns.append(col(field.name))
            else:
                raise ValueError(f"unknown indexing type: {type(it)}")
        return self.select(*columns)
    elif isinstance(item, slice):
        schema = self._builder.schema()
        columns_exprs: ExpressionsProjection = ExpressionsProjection.from_schema(schema)
        selected_columns = columns_exprs[item]
        return self.select(*selected_columns)
    else:
        raise ValueError(f"unknown indexing type: {type(item)}")

@DataframePublicAPI
def __iter__(self) -> Iterator[Dict[str, Any]]:
    """Alias of `self.iter_rows()` with default arguments for convenient access of data."""
    return self.iter_rows(results_buffer_size=None)

def __len__(self):
    """Returns the count of rows when dataframe is materialized.

    If dataframe is not materialized yet, raises a runtime error.

    Returns:
        int: count of rows.

    """
    if self._result is not None:
        return len(self._result)

    message = (
        "Cannot call len() on an unmaterialized dataframe:"
        " either materialize your dataframe with df.collect() first before calling len(),"
        " or use `df.count_rows()` instead which will calculate the total number of rows."
    )
    raise RuntimeError(message)

@DataframePublicAPI
def agg(self, *to_agg: Union[Expression, Iterable[Expression]]) -> "DataFrame":
    """Perform aggregations on this DataFrame.

    Allows for mixed aggregations for multiple columns and will return a single row that aggregated the entire DataFrame.

    Args:
        *to_agg (Expression): aggregation expressions

    Returns:
        DataFrame: DataFrame with aggregated results

    Examples:
        >>> import daft
        >>> from daft import col
        >>> df = daft.from_pydict(
        ...     {"student_id": [1, 2, 3, 4], "test1": [0.5, 0.4, 0.6, 0.7], "test2": [0.9, 0.8, 0.7, 1.0]}
        ... )
        >>> agg_df = df.agg(
        ...     col("test1").mean(),
        ...     col("test2").mean(),
        ...     ((col("test1") + col("test2")) / 2).min().alias("total_min"),
        ...     ((col("test1") + col("test2")) / 2).max().alias("total_max"),
        ... )
        >>> agg_df.show()
        ╭─────────┬────────────────────┬────────────────────┬───────────╮
        │ test1   ┆ test2              ┆ total_min          ┆ total_max │
        │ ---     ┆ ---                ┆ ---                ┆ ---       │
        │ Float64 ┆ Float64            ┆ Float64            ┆ Float64   │
        ╞═════════╪════════════════════╪════════════════════╪═══════════╡
        │ 0.55    ┆ 0.8500000000000001 ┆ 0.6000000000000001 ┆ 0.85      │
        ╰─────────┴────────────────────┴────────────────────┴───────────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

    """
    to_agg_list = (
        list(to_agg[0])
        if (len(to_agg) == 1 and not isinstance(to_agg[0], Expression))
        else list(typing.cast("Tuple[Expression]", to_agg))
    )

    for expr in to_agg_list:
        if not isinstance(expr, Expression):
            raise ValueError(f"DataFrame.agg() only accepts expression type, received: {type(expr)}")

    return self._agg(to_agg_list, group_by=None)

@DataframePublicAPI
def agg_concat(self, *cols: ColumnInputType) -> "DataFrame":
    """Performs a global list concatenation agg on the DataFrame.

    Args:
        *cols (Union[str, Expression]): columns that are lists to concatenate
    Returns:
        DataFrame: Globally aggregated list. Should be a single row.
    """
    return self._apply_agg_fn(Expression.agg_concat, cols)

@DataframePublicAPI
def agg_list(self, *cols: ColumnInputType) -> "DataFrame":
    """Performs a global list agg on the DataFrame.

    Args:
        *cols (Union[str, Expression]): columns to form into a list
    Returns:
        DataFrame: Globally aggregated list. Should be a single row.
    """
    return self._apply_agg_fn(Expression.agg_list, cols)

@DataframePublicAPI
def agg_set(self, *cols: ColumnInputType) -> "DataFrame":
    """Performs a global set agg on the DataFrame (ignoring nulls).

    Args:
        *cols (Union[str, Expression]): columns to form into a set

    Returns:
        DataFrame: Globally aggregated set. Should be a single row.
    """
    return self._apply_agg_fn(Expression.agg_set, cols)

@DataframePublicAPI
def any_value(self, *cols: ColumnInputType) -> "DataFrame":
    """Returns an arbitrary value on this DataFrame.

    Values for each column are not guaranteed to be from the same row.

    Args:
        *cols (Union[str, Expression]): columns to get an arbitrary value from
    Returns:
        DataFrame: DataFrame with any values.
    """
    return self._apply_agg_fn(Expression.any_value, cols)

@DataframePublicAPI
def collect(self, num_preview_rows: Optional[int] = 8) -> "DataFrame":
    """Executes the entire DataFrame and materializes the results.

    Args:
        num_preview_rows: Number of rows to preview. Defaults to 8.

    Returns:
        DataFrame: DataFrame with materialized results.

    Note:
        This call is **blocking** and will execute the DataFrame when called
    """
    plan_time_start = _utc_now()
    self._materialize_results()
    plan_time_end = _utc_now()
    self._broadcast_query_plan(plan_time_start, plan_time_end)
    assert self._result is not None
    dataframe_len = len(self._result)
    if num_preview_rows is not None:
        self._num_preview_rows = num_preview_rows
    else:
        self._num_preview_rows = dataframe_len
    return self

@DataframePublicAPI
def concat(self, other: "DataFrame") -> "DataFrame":
    """Concatenates two DataFrames together in a "vertical" concatenation.

    The resulting DataFrame has number of rows equal to the sum of the number of rows of the input DataFrames.

    Args:
        other (DataFrame): other DataFrame to concatenate

    Returns:
        DataFrame: DataFrame with rows from `self` on top and rows from `other` at the bottom.

    Note:
        DataFrames being concatenated **must have exactly the same schema**. You may wish to use the
        [df.select()][daft.DataFrame.select] and [expr.cast()][daft.expressions.Expression.cast] methods
        to ensure schema compatibility before concatenation.
    """
    if self.schema() != other.schema():
        raise ValueError(
            f"DataFrames must have exactly the same schema for concatenation!\nExpected:\n{self.schema()}\n\nReceived:\n{other.schema()}"
        )
    builder = self._builder.concat(other._builder)
    return DataFrame(builder)

@DataframePublicAPI
def count(self, *cols: ColumnInputType) -> "DataFrame":
    """Performs a global count on the DataFrame.

    Args:
        *cols (Union[str, Expression]): columns to count
    Returns:
        DataFrame: Globally aggregated count. Should be a single row.

    Examples:
        If no columns are specified (i.e. in the case you call `df.count()`), or only the literal string "*",
        this functions very similarly to a COUNT(*) operation in SQL and will return a new dataframe with a
        single column with the name "count".

        >>> import daft
        >>> from daft import col
        >>> df = daft.from_pydict({"foo": [1, None, None], "bar": [None, 2, 2], "baz": [3, 4, 5]})
        >>> df.count().show()  # equivalent to df.count("*").show()
        ╭────────╮
        │ count  │
        │ ---    │
        │ UInt64 │
        ╞════════╡
        │ 3      │
        ╰────────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

        However, specifying some column names would instead change the behavior to count all non-null values,
        similar to a SQL command for `SELECT COUNT(foo), COUNT(bar) FROM df`. Also, using `df.count(col("*"))`
        will expand out into count() for each column.

        >>> df.count("foo", "bar").show()
        ╭────────┬────────╮
        │ foo    ┆ bar    │
        │ ---    ┆ ---    │
        │ UInt64 ┆ UInt64 │
        ╞════════╪════════╡
        │ 1      ┆ 2      │
        ╰────────┴────────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

        >>> df.count(col("*")).show()
        ╭────────┬────────┬────────╮
        │ foo    ┆ bar    ┆ baz    │
        │ ---    ┆ ---    ┆ ---    │
        │ UInt64 ┆ UInt64 ┆ UInt64 │
        ╞════════╪════════╪════════╡
        │ 1      ┆ 2      ┆ 3      │
        ╰────────┴────────┴────────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

    """
    # Special case: treat this as a COUNT(*) operation which is likely what most people would expect
    # If user passes in "*", also do this behavior (by default it would count each column individually)
    if len(cols) == 0 or (len(cols) == 1 and isinstance(cols[0], str) and cols[0] == "*"):
        builder = self._builder.count()
        return DataFrame(builder)

    if any(isinstance(c, str) and c == "*" for c in cols):
        # we do not support hybrid count-all and count-nonnull
        raise ValueError("Cannot call count() with both * and column names")

    # Otherwise, perform a column-wise count on the specified columns
    return self._apply_agg_fn(Expression.count, cols)

@DataframePublicAPI
def count_rows(self) -> int:
    """Executes the Dataframe to count the number of rows.

    Returns:
        int: count of the number of rows in this DataFrame.
    """
    builder = self._builder.count()
    count_df = DataFrame(builder)
    # Expects builder to produce a single-partition, single-row DataFrame containing
    # a "count" column, where the lone value represents the row count for the DataFrame.
    return count_df.to_pydict()["count"][0]

@DataframePublicAPI
def describe(self) -> "DataFrame":
    """Returns the Schema of the DataFrame, which provides information about each column, as a new DataFrame.

    Returns:
        DataFrame: A dataframe where each row is a column name and its corresponding type.

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
        >>> df.describe().show()
        ╭─────────────┬───────╮
        │ column_name ┆ type  │
        │ ---         ┆ ---   │
        │ Utf8        ┆ Utf8  │
        ╞═════════════╪═══════╡
        │ a           ┆ Int64 │
        ├╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ b           ┆ Utf8  │
        ╰─────────────┴───────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)
    """
    builder = self.__builder.describe()
    return DataFrame(builder)

@DataframePublicAPI
def distinct(self) -> "DataFrame":
    """Computes distinct rows, dropping duplicates.

    Returns:
        DataFrame: DataFrame that has only distinct rows.

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"x": [1, 2, 2], "y": [4, 5, 5], "z": [7, 8, 8]})
        >>> distinct_df = df.distinct()
        >>> distinct_df = distinct_df.sort("x")
        >>> distinct_df.show()
        ╭───────┬───────┬───────╮
        │ x     ┆ y     ┆ z     │
        │ ---   ┆ ---   ┆ ---   │
        │ Int64 ┆ Int64 ┆ Int64 │
        ╞═══════╪═══════╪═══════╡
        │ 1     ┆ 4     ┆ 7     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 2     ┆ 5     ┆ 8     │
        ╰───────┴───────┴───────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)
    """
    ExpressionsProjection.from_schema(self._builder.schema())
    builder = self._builder.distinct()
    return DataFrame(builder)

@DataframePublicAPI
def drop_nan(self, *cols: ColumnInputType):
    """Drops rows that contains NaNs. If cols is None it will drop rows with any NaN value.

    If column names are supplied, it will drop only those rows that contains NaNs in one of these columns.

    Args:
        *cols (str): column names by which rows containing nans/NULLs should be filtered

    Returns:
        DataFrame: DataFrame without NaNs in specified/all columns

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"a": [1.0, 2.2, 3.5, float("nan")]})
        >>> df.drop_nan().collect()  # drops rows where any column contains NaN values
        ╭─────────╮
        │ a       │
        │ ---     │
        │ Float64 │
        ╞═════════╡
        │ 1       │
        ├╌╌╌╌╌╌╌╌╌┤
        │ 2.2     │
        ├╌╌╌╌╌╌╌╌╌┤
        │ 3.5     │
        ╰─────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)

        >>> import daft
        >>> df = daft.from_pydict({"a": [1.6, 2.5, 3.3, float("nan")]})
        >>> df.drop_nan("a").collect()  # drops rows where column `a` contains NaN values
        ╭─────────╮
        │ a       │
        │ ---     │
        │ Float64 │
        ╞═════════╡
        │ 1.6     │
        ├╌╌╌╌╌╌╌╌╌┤
        │ 2.5     │
        ├╌╌╌╌╌╌╌╌╌┤
        │ 3.3     │
        ╰─────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)

    """
    if len(cols) == 0:
        columns = self.__column_input_to_expression(self.column_names)
    else:
        columns = self.__column_input_to_expression(cols)
    float_columns = [
        column
        for column in columns
        if (
            column._to_field(self.schema()).dtype == DataType.float32()
            or column._to_field(self.schema()).dtype == DataType.float64()
        )
    ]

    # avoid superfluous .where with empty iterable when nothing to filter.
    if not float_columns:
        return self

    return self.where(
        ~reduce(
            lambda x, y: x.is_null().if_else(lit(False), x) | y.is_null().if_else(lit(False), y),
            (x.float.is_nan() for x in float_columns),
        )
    )

@DataframePublicAPI
def drop_null(self, *cols: ColumnInputType):
    """Drops rows that contains NaNs or NULLs. If cols is None it will drop rows with any NULL value.

    If column names are supplied, it will drop only those rows that contains NULLs in one of these columns.

    Args:
        *cols (str): column names by which rows containing nans should be filtered

    Returns:
        DataFrame: DataFrame without missing values in specified/all columns

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"a": [1.6, 2.5, None, float("NaN")]})
        >>> df.drop_null("a").collect()
        ╭─────────╮
        │ a       │
        │ ---     │
        │ Float64 │
        ╞═════════╡
        │ 1.6     │
        ├╌╌╌╌╌╌╌╌╌┤
        │ 2.5     │
        ├╌╌╌╌╌╌╌╌╌┤
        │ NaN     │
        ╰─────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)


    """
    if len(cols) == 0:
        columns = self.__column_input_to_expression(self.column_names)
    else:
        columns = self.__column_input_to_expression(cols)
    return self.where(~reduce(lambda x, y: x | y, (x.is_null() for x in columns)))

@DataframePublicAPI
def except_all(self, other: "DataFrame") -> "DataFrame":
    """Returns the set difference of two DataFrames, considering duplicates.

    Args:
        other (DataFrame): DataFrame to except with

    Returns:
        DataFrame: DataFrame with the set difference of the two DataFrames, considering duplicates

    Examples:
        >>> import daft
        >>> df1 = daft.from_pydict({"a": [1, 1, 2, 2], "b": [4, 4, 6, 6]})
        >>> df2 = daft.from_pydict({"a": [1, 2, 2], "b": [4, 6, 6]})
        >>> df1.except_all(df2).collect()
        ╭───────┬───────╮
        │ a     ┆ b     │
        │ ---   ┆ ---   │
        │ Int64 ┆ Int64 │
        ╞═══════╪═══════╡
        │ 1     ┆ 4     │
        ╰───────┴───────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

    """
    builder = self._builder.except_all(other._builder)
    return DataFrame(builder)

@DataframePublicAPI
def except_distinct(self, other: "DataFrame") -> "DataFrame":
    """Returns the set difference of two DataFrames.

    Args:
        other (DataFrame): DataFrame to except with

    Returns:
        DataFrame: DataFrame with the set difference of the two DataFrames

    Examples:
        >>> import daft
        >>> df1 = daft.from_pydict({"a": [1, 2, 3], "b": [4, 5, 6]})
        >>> df2 = daft.from_pydict({"a": [1, 2, 3], "b": [4, 8, 6]})
        >>> df1.except_distinct(df2).collect()
        ╭───────┬───────╮
        │ a     ┆ b     │
        │ ---   ┆ ---   │
        │ Int64 ┆ Int64 │
        ╞═══════╪═══════╡
        │ 2     ┆ 5     │
        ╰───────┴───────╯
        <BLANKLINE>
        (Showing first 1 of 1 rows)

    """
    builder = self._builder.except_distinct(other._builder)
    return DataFrame(builder)

@DataframePublicAPI
def exclude(self, *names: str) -> "DataFrame":
    """Drops columns from the current DataFrame by name.

    This is equivalent of performing a select with all the columns but the ones excluded.

    Args:
        *names (str): names to exclude

    Returns:
        DataFrame: DataFrame with some columns excluded.

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"x": [1, 2, 3], "y": [4, 5, 6], "z": [7, 8, 9]})
        >>> df_without_x = df.exclude("x")
        >>> df_without_x.show()
        ╭───────┬───────╮
        │ y     ┆ z     │
        │ ---   ┆ ---   │
        │ Int64 ┆ Int64 │
        ╞═══════╪═══════╡
        │ 4     ┆ 7     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 5     ┆ 8     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 6     ┆ 9     │
        ╰───────┴───────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)
    """
    builder = self._builder.exclude(list(names))
    return DataFrame(builder)

@DataframePublicAPI
def explain(
    self, show_all: bool = False, format: str = "ascii", simple: bool = False, file: Optional[io.IOBase] = None
) -> Any:
    """Prints the (logical and physical) plans that will be executed to produce this DataFrame.

    Defaults to showing the unoptimized logical plan. Use `show_all=True` to show the unoptimized logical plan,
    the optimized logical plan, and the physical plan.

    Args:
        show_all (bool): Whether to show the optimized logical plan and the physical plan in addition to the
            unoptimized logical plan.
        format (str): The format to print the plan in. one of 'ascii' or 'mermaid'
        simple (bool): Whether to only show the type of op for each node in the plan, rather than showing details
            of how each op is configured.

        file (Optional[io.IOBase]): Location to print the output to, or defaults to None which defaults to the default location for
            print (in Python, that should be sys.stdout)
    """
    is_cached = self._result_cache is not None
    if format == "mermaid":
        from daft.dataframe.display import MermaidFormatter
        from daft.utils import in_notebook

        instance = MermaidFormatter(self.__builder, show_all, simple, is_cached)
        if file is not None:
            # if we are printing to a file, we print the markdown representation of the plan
            text = instance._repr_markdown_()
            print(text, file=file)
        if in_notebook():
            # if in a notebook, we return the class instance and let jupyter display it
            return instance
        else:
            # if we are not in a notebook, we return the raw markdown instead of the class instance
            return repr(instance)

    print_to_file = partial(print, file=file)

    if self._result_cache is not None:
        print_to_file("Result is cached and will skip computation\n")
        print_to_file(self._builder.pretty_print(simple, format=format))

        print_to_file("However here is the logical plan used to produce this result:\n", file=file)

    builder = self.__builder
    print_to_file("== Unoptimized Logical Plan ==\n")
    print_to_file(builder.pretty_print(simple, format=format))
    if show_all:
        print_to_file("\n== Optimized Logical Plan ==\n")
        builder = builder.optimize()
        print_to_file(builder.pretty_print(simple))
        print_to_file("\n== Physical Plan ==\n")
        if get_context().get_or_create_runner().name != "native":
            physical_plan_scheduler = builder.to_physical_plan_scheduler(get_context().daft_execution_config)
            print_to_file(physical_plan_scheduler.pretty_print(simple, format=format))
        else:
            native_executor = NativeExecutor()
            print_to_file(
                native_executor.pretty_print(builder, get_context().daft_execution_config, simple, format=format)
            )
    else:
        print_to_file(
            "\n \nSet `show_all=True` to also see the Optimized and Physical plans. This will run the query optimizer.",
        )
    return None

@DataframePublicAPI
def explode(self, *columns: ColumnInputType) -> "DataFrame":
    """Explodes a List column, where every element in each row's List becomes its own row, and all other columns in the DataFrame are duplicated across rows.

    If multiple columns are specified, each row must contain the same number of items in each specified column.

    Exploding Null values or empty lists will create a single Null entry (see example below).

    Args:
        *columns (ColumnInputType): columns to explode

    Returns:
        DataFrame: DataFrame with exploded column

    Examples:
        >>> import daft
        >>> df = daft.from_pydict(
        ...     {
        ...         "x": [[1], [2, 3]],
        ...         "y": [["a"], ["b", "c"]],
        ...         "z": [
        ...             [1.0],
        ...             [2.0, 2.0],
        ...         ],
        ...     }
        ... )
        >>> df.explode(col("x"), col("y")).collect()
        ╭───────┬──────┬───────────────╮
        │ x     ┆ y    ┆ z             │
        │ ---   ┆ ---  ┆ ---           │
        │ Int64 ┆ Utf8 ┆ List[Float64] │
        ╞═══════╪══════╪═══════════════╡
        │ 1     ┆ a    ┆ [1]           │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
        │ 2     ┆ b    ┆ [2, 2]        │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
        │ 3     ┆ c    ┆ [2, 2]        │
        ╰───────┴──────┴───────────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)

    """
    parsed_exprs = self.__column_input_to_expression(columns)
    builder = self._builder.explode(parsed_exprs)
    return DataFrame(builder)

@DataframePublicAPI
def filter(self, predicate: Union[Expression, str]) -> "DataFrame":
    """Filters rows via a predicate expression, similar to SQL ``WHERE``.

    Alias for [daft.DataFrame.where][daft.DataFrame.where].

    Args:
        predicate (Expression): expression that keeps row if evaluates to True.

    Returns:
        DataFrame: Filtered DataFrame.

    Tip:
        See also [.where(predicate)][daft.DataFrame.where]

    """
    return self.where(predicate)

@DataframePublicAPI
def groupby(self, *group_by: ManyColumnsInputType) -> "GroupedDataFrame":
    """Performs a GroupBy on the DataFrame for aggregation.

    Args:
        *group_by (Union[str, Expression]): columns to group by

    Returns:
        GroupedDataFrame: DataFrame to Aggregate

    Examples:
        >>> import daft
        >>> from daft import col
        >>> df = daft.from_pydict(
        ...     {
        ...         "pet": ["cat", "dog", "dog", "cat"],
        ...         "age": [1, 2, 3, 4],
        ...         "name": ["Alex", "Jordan", "Sam", "Riley"],
        ...     }
        ... )
        >>> grouped_df = df.groupby("pet").agg(
        ...     col("age").min().alias("min_age"),
        ...     col("age").max().alias("max_age"),
        ...     col("pet").count().alias("count"),
        ...     col("name").any_value(),
        ... )
        >>> grouped_df = grouped_df.sort("pet")
        >>> grouped_df.show()
        ╭──────┬─────────┬─────────┬────────┬────────╮
        │ pet  ┆ min_age ┆ max_age ┆ count  ┆ name   │
        │ ---  ┆ ---     ┆ ---     ┆ ---    ┆ ---    │
        │ Utf8 ┆ Int64   ┆ Int64   ┆ UInt64 ┆ Utf8   │
        ╞══════╪═════════╪═════════╪════════╪════════╡
        │ cat  ┆ 1       ┆ 4       ┆ 2      ┆ Alex   │
        ├╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ dog  ┆ 2       ┆ 3       ┆ 2      ┆ Jordan │
        ╰──────┴─────────┴─────────┴────────┴────────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)

    """
    return GroupedDataFrame(self, ExpressionsProjection(self._wildcard_inputs_to_expressions(group_by)))

@DataframePublicAPI
def intersect(self, other: "DataFrame") -> "DataFrame":
    """Returns the intersection of two DataFrames.

    Args:
        other (DataFrame): DataFrame to intersect with

    Returns:
        DataFrame: DataFrame with the intersection of the two DataFrames

    Examples:
        >>> import daft
        >>> df1 = daft.from_pydict({"a": [1, 2, 3], "b": [4, 5, 6]})
        >>> df2 = daft.from_pydict({"a": [1, 2, 3], "b": [4, 8, 6]})
        >>> df = df1.intersect(df2)
        >>> df = df.sort("a")
        >>> df.show()
        ╭───────┬───────╮
        │ a     ┆ b     │
        │ ---   ┆ ---   │
        │ Int64 ┆ Int64 │
        ╞═══════╪═══════╡
        │ 1     ┆ 4     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 3     ┆ 6     │
        ╰───────┴───────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)

    """
    builder = self._builder.intersect(other._builder)
    return DataFrame(builder)

@DataframePublicAPI
def intersect_all(self, other: "DataFrame") -> "DataFrame":
    """Returns the intersection of two DataFrames, including duplicates.

    Args:
        other (DataFrame): DataFrame to intersect with

    Returns:
        DataFrame: DataFrame with the intersection of the two DataFrames, including duplicates

    Examples:
        >>> import daft
        >>> df1 = daft.from_pydict({"a": [1, 2, 2], "b": [4, 6, 6]})
        >>> df2 = daft.from_pydict({"a": [1, 1, 2, 2], "b": [4, 4, 6, 6]})
        >>> df1.intersect_all(df2).sort("a").collect()
        ╭───────┬───────╮
        │ a     ┆ b     │
        │ ---   ┆ ---   │
        │ Int64 ┆ Int64 │
        ╞═══════╪═══════╡
        │ 1     ┆ 4     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 2     ┆ 6     │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
        │ 2     ┆ 6     │
        ╰───────┴───────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)

    """
    builder = self._builder.intersect_all(other._builder)
    return DataFrame(builder)

@DataframePublicAPI
def into_partitions(self, num: int) -> "DataFrame":
    """Splits or coalesces DataFrame to ``num`` partitions. Order is preserved.

    This will naively greedily split partitions in a round-robin fashion to hit the targeted number of partitions.
    The number of rows/size in a given partition is not taken into account during the splitting.

    Args:
        num (int): number of target partitions.

    Returns:
        DataFrame: Dataframe with `num` partitions.

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"x": [1, 2, 3], "y": [4, 5, 6], "z": [7, 8, 9]})
        >>> df_with_5_partitions = df.into_partitions(5)
        >>> df_with_5_partitions.num_partitions()
        5
    """
    builder = self._builder.into_partitions(num)
    return DataFrame(builder)

@DataframePublicAPI
def iter_partitions(
    self, results_buffer_size: Union[Optional[int], Literal["num_cpus"]] = "num_cpus"
) -> Iterator[Union[MicroPartition, "ray.ObjectRef[MicroPartition]"]]:
    """Begin executing this dataframe and return an iterator over the partitions.

    Each partition will be returned as a daft.recordbatch object (if using Python runner backend)
    or a ray ObjectRef (if using Ray runner backend).

    Args:
        results_buffer_size: how many partitions to allow in the results buffer (defaults to the total number of CPUs
            available on the machine).

    Note: A quick note on configuring asynchronous/parallel execution using `results_buffer_size`.
        The `results_buffer_size` kwarg controls how many results Daft will allow to be in the buffer while iterating.
        Once this buffer is filled, Daft will not run any more work until some partition is consumed from the buffer.

        * Increasing this value means the iterator will consume more memory and CPU resources but have higher throughput
        * Decreasing this value means the iterator will consume lower memory and CPU resources, but have lower throughput
        * Setting this value to `None` means the iterator will consume as much resources as it deems appropriate per-iteration

        The default value is the total number of CPUs available on the current machine.

    Examples:
        >>> import daft
        >>>
        >>> daft.context.set_runner_ray()  # doctest: +SKIP
        >>>
        >>> df = daft.from_pydict({"foo": [1, 2, 3], "bar": ["a", "b", "c"]}).into_partitions(2)
        >>> for part in df.iter_partitions():
        ...     print(part)  # doctest: +SKIP
        MicroPartition with 2 rows:
        TableState: Loaded. 1 tables
        ╭───────┬──────╮
        │ foo   ┆ bar  │
        │ ---   ┆ ---  │
        │ Int64 ┆ Utf8 │
        ╞═══════╪══════╡
        │ 1     ┆ a    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌┤
        │ 2     ┆ b    │
        ╰───────┴──────╯
        <BLANKLINE>
        <BLANKLINE>
        Statistics: missing
        <BLANKLINE>
        MicroPartition with 1 rows:
        TableState: Loaded. 1 tables
        ╭───────┬──────╮
        │ foo   ┆ bar  │
        │ ---   ┆ ---  │
        │ Int64 ┆ Utf8 │
        ╞═══════╪══════╡
        │ 3     ┆ c    │
        ╰───────┴──────╯
        <BLANKLINE>
        <BLANKLINE>
        Statistics: missing
        <BLANKLINE>
    """
    if results_buffer_size == "num_cpus":
        results_buffer_size = multiprocessing.cpu_count()
    elif results_buffer_size is not None and not results_buffer_size > 0:
        raise ValueError(f"Provided `results_buffer_size` value must be > 0, received: {results_buffer_size}")

    if self._result is not None:
        # If the dataframe has already finished executing,
        # use the precomputed results.
        for mat_result in self._result.values():
            yield mat_result.partition()

    else:
        # Execute the dataframe in a streaming fashion.
        context = get_context()
        results_iter = context.get_or_create_runner().run_iter(
            self._builder, results_buffer_size=results_buffer_size
        )
        for result in results_iter:
            yield result.partition()