def group_by(
        self,
        *,
        partition_by: Optional[Union[str, Func, Sequence[Union[str, Func]]]] = None,
        **kwargs: Func,
    ) -> "Self":
        """Group rows by specified set of signals and return new signals
        with aggregated values.

        The supported functions:
           count(), sum(), avg(), min(), max(), any_value(), collect(), concat()

        Example:
            ```py
            chain = chain.group_by(
                cnt=func.count(),
                partition_by=("file_source", "file_ext"),
            )
            ```
        """
        if partition_by is None:
            partition_by = []
        elif isinstance(partition_by, (str, Func)):
            partition_by = [partition_by]

        partition_by_columns: list[Column] = []
        signal_columns: list[Column] = []
        schema_fields: dict[str, DataType] = {}
        keep_columns: list[str] = []

        # validate partition_by columns and add them to the schema
        for col in partition_by:
            if isinstance(col, str):
                col_db_name = ColumnMeta.to_db_name(col)
                col_type = self.signals_schema.get_column_type(col_db_name)
                column = Column(col_db_name, python_to_sql(col_type))
                if col not in keep_columns:
                    keep_columns.append(col)
            elif isinstance(col, Function):
                column = col.get_column(self.signals_schema)
                col_db_name = column.name
                col_type = column.type.python_type
                schema_fields[col_db_name] = col_type
            else:
                raise DataChainColumnError(
                    col,
                    (
                        f"partition_by column {col} has type {type(col)}"
                        " but expected str or Function"
                    ),
                )
            partition_by_columns.append(column)

        # validate signal columns and add them to the schema
        if not kwargs:
            raise ValueError("At least one column should be provided for group_by")
        for col_name, func in kwargs.items():
            if not isinstance(func, Func):
                raise DataChainColumnError(
                    col_name,
                    f"Column {col_name} has type {type(func)} but expected Func object",
                )
            column = func.get_column(self.signals_schema, label=col_name)
            signal_columns.append(column)
            schema_fields[col_name] = func.get_result_type(self.signals_schema)

        signal_schema = SignalSchema(schema_fields)
        if keep_columns:
            signal_schema |= self.signals_schema.to_partial(*keep_columns)

        return self._evolve(
            query=self._query.group_by(signal_columns, partition_by_columns),
            signal_schema=signal_schema,
        )
Group rows by specified set of signals and return new signals
        with aggregated values.

        The supported functions:
           count(), sum(), avg(), min(), max(), any_value(), collect(), concat()

        Example:
            ```py
            chain = chain.group_by(
                cnt=func.count(),
                partition_by=("file_source", "file_ext"),
            )
            ```
        
src/datachain/lib/dc/datachain.py