Repository: modin-project/modin (python)

Issue:
HDK: Use MODIN_CPUS instead of os.cpu_count() for the fragment size calculation
When the HDK engine is initialized, it sets the environment variable OMP_NUM_THREADS = CpuCount.get(). By default this value is equal to the number of CPUs, but it could be changed by the user.

Pull request description:
PERF-#6614: HDK: Use MODIN_CPUS instead of os.cpu_count() for the fragment size calculation
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #6614 <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [ ] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Patch:
diff --git a/docs/flow/modin/core/execution/dispatching.rst b/docs/flow/modin/core/execution/dispatching.rst
index cf158c88..507bb21d 100644
--- a/docs/flow/modin/core/execution/dispatching.rst
+++ b/docs/flow/modin/core/execution/dispatching.rst
@@ -31,7 +31,7 @@ extends ``PandasDataframe``.
 In the scope of this module, each execution is represented with a factory class located in
 ``modin/core/execution/dispatching/factories/factories.py``. Each factory contains a field that identifies the IO module of the corresponding execution. This IO module is
 responsible for dispatching calls of IO functions to their actual implementations in the
-underlying IO module. For more information about IO module visit :doc:`related doc </flow/modin/core/io/index>`.
+underlying IO module. For more information about IO module visit :doc:`IO </flow/modin/core/io/index>` page.
 
 Factory Dispatcher
 ''''''''''''''''''
diff --git a/docs/flow/modin/core/storage_formats/base/query_compiler.rst b/docs/flow/modin/core/storage_formats/base/query_compiler.rst
index bfdb6a2b..999019cc 100644
--- a/docs/flow/modin/core/storage_formats/base/query_compiler.rst
+++ b/docs/flow/modin/core/storage_formats/base/query_compiler.rst
@@ -88,7 +88,7 @@ and already can be used in Modin DataFrame:
 To be able to select this query compiler as default via ``modin.config`` you also need
 to define the combination of your query compiler and pandas engine as an execution
 by adding the corresponding factory. To find more information about factories,
-visit :doc:`corresponding section </flow/modin/core/execution/dispatching>` of the flow documentation.
+visit :doc:`dispatching </flow/modin/core/execution/dispatching>` page.
 
 Query Compiler API
 ''''''''''''''''''
diff --git a/docs/flow/modin/pandas/dataframe.rst b/docs/flow/modin/pandas/dataframe.rst
index dd5c3eb9..3b74b54c 100644
--- a/docs/flow/modin/pandas/dataframe.rst
+++ b/docs/flow/modin/pandas/dataframe.rst
@@ -26,7 +26,7 @@ Usage Guide
 
 The most efficient way to create Modin ``DataFrame`` is to import data from external
 storage using the highly efficient Modin IO methods (for example using ``pd.read_csv``,
-see details for Modin IO methods in the :doc:`separate section </flow/modin/core/io/index>`),
+see details for Modin IO methods in the :doc:`IO </flow/modin/core/io/index>` page),
 but even if the data does not originate from a file, any pandas supported data type or
 ``pandas.DataFrame`` can be used. Internally, the ``DataFrame`` data is divided into
 partitions, which number along an axis usually corresponds to the number of the user's hardware CPUs. If needed,
diff --git a/docs/flow/modin/pandas/series.rst b/docs/flow/modin/pandas/series.rst
index 6d3010c7..7a0bd9a6 100644
--- a/docs/flow/modin/pandas/series.rst
+++ b/docs/flow/modin/pandas/series.rst
@@ -26,7 +26,7 @@ Usage Guide
 
 The most efficient way to create Modin ``Series`` is to import data from external
 storage using the highly efficient Modin IO methods (for example using ``pd.read_csv``,
-see details for Modin IO methods in the :doc:`separate section </flow/modin/core/io/index>`),
+see details for Modin IO methods in the :doc:`IO </flow/modin/core/io/index>` page),
 but even if the data does not originate from a file, any pandas supported data type or
 ``pandas.Series`` can be used. Internally, the ``Series`` data is divided into
 partitions, which number along an axis usually corresponds to the number of the user's hardware CPUs. If needed,
diff --git a/docs/getting_started/examples.rst b/docs/getting_started/examples.rst
index 3c361495..cf436cdd 100644
--- a/docs/getting_started/examples.rst
+++ b/docs/getting_started/examples.rst
@@ -1,8 +1,8 @@
 Examples and Resources
 ======================
 
-Here you can find additional resources to learn about Modin. To learn more about
-advanced usage for Modin, please refer to :doc:`this section </usage_guide/index>`.
+Here you can find additional resources to learn about Modin. To learn more about 
+advanced usage for Modin, please refer to :doc:`Usage Guide </usage_guide/index>` section..
 
 Usage Examples
 ''''''''''''''
diff --git a/docs/getting_started/faq.rst b/docs/getting_started/faq.rst
index 724b5711..bca4e36f 100644
--- a/docs/getting_started/faq.rst
+++ b/docs/getting_started/faq.rst
@@ -21,8 +21,8 @@ The :py:class:`~modin.pandas.dataframe.DataFrame` is a highly
 scalable, parallel DataFrame. Modin transparently distributes the data and computation so
 that you can continue using the same pandas API while being able to work with more data faster.
 Modin lets you use all the CPU cores on your machine, and because it is lightweight, it
-often has less memory overhead than pandas. See this :doc:`page </getting_started/why_modin/pandas>` to
-learn more about how Modin is different from pandas.
+often has less memory overhead than pandas. See :doc:` Why Modin? </getting_started/why_modin/pandas>`
+page to learn more about how Modin is different from pandas.
 
 Why not just improve pandas?
 """"""""""""""""""""""""""""
@@ -61,7 +61,8 @@ How does Modin compare to Dask DataFrame and Koalas?
 
 TLDR: Modin has better coverage of the pandas API, has a flexible backend, better ordering semantics,
 and supports both row and column-parallel operations.
-Check out this :doc:`page </getting_started/why_modin/modin_vs_dask_vs_koalas>` detailing the differences!
+Check out :doc:`Modin vs Dask vs Koalas </getting_started/why_modin/modin_vs_dask_vs_koalas>` page detailing
+the differences!
 
 How does Modin work under the hood?
 """""""""""""""""""""""""""""""""""
@@ -135,8 +136,8 @@ This can also be done with:
     modin_cfg.Engine.put('unidist') # Modin will use Unidist
     unidist_cfg.Backend.put('mpi') # Unidist will use MPI backend
 
-We also have an experimental HDK-based engine of Modin you can read about :doc:`here </development/using_hdk>`.
-We plan to support more execution engines in future. If you have a specific request,
+We also have an experimental HDK-based engine of Modin, which you can read about on :doc:`Using HDK </development/using_hdk>`
+page. We plan to support more execution engines in future. If you have a specific request,
 please post on the #feature-requests channel on our Slack_ community.
 
 How do I connect Modin to a database via `read_sql`?
diff --git a/docs/getting_started/using_modin/using_modin_locally.rst b/docs/getting_started/using_modin/using_modin_locally.rst
index 23d48693..4ea09518 100644
--- a/docs/getting_started/using_modin/using_modin_locally.rst
+++ b/docs/getting_started/using_modin/using_modin_locally.rst
@@ -67,7 +67,7 @@ cluster for you:
 Finally, if you already have an Ray or Dask engine initialized, Modin will
 automatically attach to whichever engine is available. If you are interested in using
 Modin with HDK engine, please refer to :doc:`these instructions </development/using_hdk>`. For additional information on other settings you can configure, see
-:doc:`this page </flow/modin/config>` for more details.
+:doc:`Modin's config </flow/modin/config>` page for more details.
 
 Advanced: Configuring the resources Modin uses
 ----------------------------------------------
@@ -81,7 +81,8 @@ the following code:
    import modin
    print(modin.config.NPartitions.get()) #prints 16 on a laptop with 16 physical cores
 
-Modin fully utilizes the resources on your machine. To read more about how this works, see :doc:`this page</getting_started/why_modin/pandas/>` for more details.
+Modin fully utilizes the resources on your machine. To read more about how this works, see :doc:`Why Modin? </getting_started/why_modin/pandas/>`
+page for more details.
 
 Since Modin will use all of the resources available on your machine by default, at
 times, it is possible that you may like to limit the amount of resources Modin uses to
diff --git a/docs/getting_started/why_modin/modin_vs_dask_vs_koalas.rst b/docs/getting_started/why_modin/modin_vs_dask_vs_koalas.rst
index a8482ce3..5fb95a09 100644
--- a/docs/getting_started/why_modin/modin_vs_dask_vs_koalas.rst
+++ b/docs/getting_started/why_modin/modin_vs_dask_vs_koalas.rst
@@ -46,7 +46,7 @@ Execution Semantics
 
 **DaskDF and Koalas make use of lazy evaluation, which means that the computation is delayed until users explicitly evaluate the results.** This mode of evaluation places a lot of optimization responsibility on the user, forcing them to think about when it would be useful to inspect the intermediate results or delay doing so. Specifically, DaskDF's API differs from pandas in that it requires users to explicitly call ``.compute()`` to materialize the result of the computation. Often if that computation corresponds to a long chain of operators, this call can take a very long time to execute. Overall, the need to explicitly trigger computation makes the API less convenient to work with, but gives DaskDF and Koalas the opportunity to perform holistic optimizations over the entire dataflow graph. However, to the best of our knowledge, neither DaskDF nor Koalas actually leverage holistic optimizations.
 
-**Modin employs eager evaluation, like pandas.** Eager evaluation is the default mode of operation for data scientists when working with pandas in an interactive environment, such as Jupyter Notebooks. Modin reproduces this familiar behavior by performing all computations eagerly as soon as it is issued, so that users can inspect intermediate results and quickly see the results of their computations without having to wait or explicitly trigger computation. This is especially useful during interactive data analysis, where users often iterate on their dataframe workflows or build up their dataframe queries in an incremental fashion. Modin also supports lazy evaluation via the HDK engine, you can learn more about it :doc:`here </flow/modin/experimental/core/execution/native/implementations/hdk_on_native/index>`. We also have developed techniques for `opportunistic evaluation <https://arxiv.org/pdf/2103.02145.pdf>`_ that bridges the gap between lazy and eager evaluation that will be incorporated in Modin in the future.
+**Modin employs eager evaluation, like pandas.** Eager evaluation is the default mode of operation for data scientists when working with pandas in an interactive environment, such as Jupyter Notebooks. Modin reproduces this familiar behavior by performing all computations eagerly as soon as it is issued, so that users can inspect intermediate results and quickly see the results of their computations without having to wait or explicitly trigger computation. This is especially useful during interactive data analysis, where users often iterate on their dataframe workflows or build up their dataframe queries in an incremental fashion. Modin also supports lazy evaluation via the HDK engine, you can learn more about it on :doc:`HDK </flow/modin/experimental/core/execution/native/implementations/hdk_on_native/index>` page. We also have developed techniques for `opportunistic evaluation <https://arxiv.org/pdf/2103.02145.pdf>`_ that bridges the gap between lazy and eager evaluation that will be incorporated in Modin in the future.
 
 Ordering Semantics
 ------------------
diff --git a/docs/usage_guide/optimization_notes/index.rst b/docs/usage_guide/optimization_notes/index.rst
index 8eed7f77..52fd9b19 100644
--- a/docs/usage_guide/optimization_notes/index.rst
+++ b/docs/usage_guide/optimization_notes/index.rst
@@ -196,7 +196,7 @@ Use Modin's Dataframe Algebra API to implement custom parallel functions
 Modin provides a set of low-level parallel-implemented operators which can be used to build most of the
 aggregation functions. These operators are present in the :doc:`algebra module </flow/modin/core/dataframe/algebra>`.
 Modin DataFrame allows users to use their own aggregations built with this module. Visit the
-:doc:`appropriate section </flow/modin/core/dataframe/algebra>` of the documentation for the steps to do it.
+:doc:`DataFrame's algebra </flow/modin/core/dataframe/algebra>` page of the documentation for the steps to do it.
 
 Avoid mixing pandas and Modin DataFrames
 """"""""""""""""""""""""""""""""""""""""
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index a642f945..904a2809 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -257,8 +257,16 @@ class PandasDataframePartitionManager(ClassLogger, ABC):
             )
         else:
             mapped_partitions = cls.map_partitions(partitions, map_func)
+
+        # Assuming, that the output will not be larger than the input,
+        # keep the current number of partitions.
+        num_splits = min(len(partitions), NPartitions.get())
         return cls.map_axis_partitions(
-            axis, mapped_partitions, reduce_func, enumerate_partitions=True
+            axis,
+            mapped_partitions,
+            reduce_func,
+            enumerate_partitions=True,
+            num_splits=num_splits,
         )
 
     @classmethod
diff --git a/modin/experimental/core/execution/native/implementations/hdk_on_native/hdk_worker.py b/modin/experimental/core/execution/native/implementations/hdk_on_native/hdk_worker.py
index aad566ef..4db43ffa 100644
--- a/modin/experimental/core/execution/native/implementations/hdk_on_native/hdk_worker.py
+++ b/modin/experimental/core/execution/native/implementations/hdk_on_native/hdk_worker.py
@@ -12,7 +12,6 @@
 # governing permissions and limitations under the License.
 
 """Module provides ``HdkWorker`` class."""
-import os
 from typing import List, Optional, Tuple, Union
 
 import pyarrow as pa
@@ -20,7 +19,7 @@ import pyhdk
 from packaging import version
 from pyhdk.hdk import HDK, ExecutionResult, QueryNode, RelAlgExecutor
 
-from modin.config import HdkFragmentSize, HdkLaunchParameters
+from modin.config import CpuCount, HdkFragmentSize, HdkLaunchParameters
 from modin.utils import _inherit_docstrings
 
 from .base_worker import BaseDbWorker, DbTable
@@ -143,7 +142,7 @@ class HdkWorker(BaseDbWorker):  # noqa: PR01
         fragment_size = HdkFragmentSize.get()
         if fragment_size is None:
             if cls._preferred_device == "CPU":
-                cpu_count = os.cpu_count()
+                cpu_count = CpuCount.get()
                 if cpu_count is not None:
                     fragment_size = table.num_rows // cpu_count
                     fragment_size = min(fragment_size, 2**25)
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
index b9c4f210..ff708bce 100644
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -1824,14 +1824,13 @@ class SeriesGroupBy(DataFrameGroupBy):
         Returns
         -------
         str, list
-            If `fn` is a callable, return its name if it's a method of the groupby
-            object, otherwise return `fn` itself. If `fn` is a string, return it.
-            If `fn` is an Iterable, return a list of _try_get_str_func applied to
-            each element of `fn`.
+            If `fn` is a callable, return its name, otherwise return `fn` itself.
+            If `fn` is a string, return it. If `fn` is an Iterable, return a list
+            of _try_get_str_func applied to each element of `fn`.
         """
         if not isinstance(fn, str) and isinstance(fn, Iterable):
             return [self._try_get_str_func(f) for f in fn]
-        return fn.__name__ if callable(fn) and fn.__name__ in dir(self) else fn
+        return fn.__name__ if callable(fn) else fn
 
     def value_counts(
         self,


Test patch:
diff --git a/modin/pandas/test/test_groupby.py b/modin/pandas/test/test_groupby.py
index 2406cf20..52aeeae0 100644
--- a/modin/pandas/test/test_groupby.py
+++ b/modin/pandas/test/test_groupby.py
@@ -1127,6 +1127,24 @@ def test_series_groupby(by, as_index_series_or_dataframe):
         eval_groups(modin_groupby, pandas_groupby)
 
 
+def test_agg_udf_6600():
+    data = {
+        "name": ["Mariners", "Lakers"] * 50,
+        "league_abbreviation": ["MLB", "NBA"] * 50,
+    }
+    modin_teams, pandas_teams = create_test_dfs(data)
+
+    def my_first_item(s):
+        return s.iloc[0]
+
+    for agg in (my_first_item, [my_first_item], ["nunique", my_first_item]):
+        eval_general(
+            modin_teams,
+            pandas_teams,
+            operation=lambda df: df.groupby("league_abbreviation").name.agg(agg),
+        )
+
+
 def test_multi_column_groupby():
     pandas_df = pandas.DataFrame(
         {