Repository: modin-project/modin

Issue:
TEST: Rewrite --extra-test-parameters parsing so we can collect pytests
If I go to the modin root and run `pytest --collect-only`, I get an error because `config.option` has no attribute `extra_test_parameters`: https://github.com/modin-project/modin/blob/3ee4fa02ee69272aebf4fea005fe77268b166fbc/modin/conftest.py#L315

We should rewrite the code so that we don't assume we can always get `config.option.extra_test_parameters`. I need to collect pytests without running them so I can debug them in vscode.

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/modin-project/modin/pull/5075?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project) Report
> Merging [#5075](https://codecov.io/gh/modin-project/modin/pull/5075?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project) (b7bff34) into [master](https://codecov.io/gh/modin-project/modin/commit/3ee4fa02ee69272aebf4fea005fe77268b166fbc?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project) (3ee4fa0) will **decrease** coverage by `57.39%`.
> The diff coverage is `100.00%`.

> :exclamation: Current head b7bff34 differs from pull request most recent head 1231002. Consider uploading reports for the commit 1231002 to get more accurate results

```diff
@@             Coverage Diff             @@
##           master    #5075       +/-   ##
===========================================
- Coverage   85.00%   27.60%   -57.40%     
===========================================
  Files         253      253               
  Lines       19130    19129        -1     
===========================================
- Hits        16261     5281    -10980     
- Misses       2869    13848    +10979     
```


| [Impacted Files](https://codecov.io/gh/modin-project/modin/pull/5075?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project) | Coverage Δ | |
|---|---|---|
| [modin/conftest.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vY29uZnRlc3QucHk=) | `59.44% <100.00%> (-22.91%)` | :arrow_down: |
| [modin/pandas/window.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vcGFuZGFzL3dpbmRvdy5weQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/distributed/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZGlzdHJpYnV0ZWQvX19pbml0X18ucHk=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/experimental/core/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZXhwZXJpbWVudGFsL2NvcmUvX19pbml0X18ucHk=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/core/execution/dask/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vY29yZS9leGVjdXRpb24vZGFzay9fX2luaXRfXy5weQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/experimental/pandas/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZXhwZXJpbWVudGFsL3BhbmRhcy9fX2luaXRfXy5weQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/experimental/sklearn/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZXhwZXJpbWVudGFsL3NrbGVhcm4vX19pbml0X18ucHk=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/experimental/xgboost/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZXhwZXJpbWVudGFsL3hnYm9vc3QvX19pbml0X18ucHk=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/distributed/dataframe/\_\_init\_\_.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vZGlzdHJpYnV0ZWQvZGF0YWZyYW1lL19faW5pdF9fLnB5) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [modin/core/execution/ray/generic/io/io.py](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project#diff-bW9kaW4vY29yZS9leGVjdXRpb24vcmF5L2dlbmVyaWMvaW8vaW8ucHk=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [162 more](https://codecov.io/gh/modin-project/modin/pull/5075/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project) | |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=modin-project)


Patch:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 978b3650..91159468 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -55,8 +55,10 @@ Key Features and Updates
   * FIX-#4996: Evaluate BenchmarkMode at each function call (#4997)
   * FIX-#4022: Fixed empty data frame with index (#4910)
   * FIX-#4090: Fixed check if the index is trivial (#4936)
+  * FIX-#4966: Fix `to_timedelta` to return Series instead of TimedeltaIndex (#5028)
   * FIX-#5042: Fix series __getitem__ with invalid strings (#5048)
   * FIX-#4691: Fix binary operations between virtual partitions (#5049)  
+  * FIX-#5045: Fix ray virtual_partition.wait with duplicate object refs (#5058)
 * Performance enhancements
   * PERF-#4182: Add cell-wise execution for binary ops, fix bin ops for empty dataframes (#4391)
   * PERF-#4288: Improve perf of `groupby.mean` for narrow data (#4591)
diff --git a/modin/core/execution/ray/common/utils.py b/modin/core/execution/ray/common/utils.py
index acba8870..9b51e916 100644
--- a/modin/core/execution/ray/common/utils.py
+++ b/modin/core/execution/ray/common/utils.py
@@ -246,3 +246,26 @@ def deserialize(obj):
         return dict(zip(obj.keys(), RayWrapper.materialize(list(obj.values()))))
     else:
         return obj
+
+
+def wait(obj_ids):
+    """
+    Wrap ``ray.wait`` to handle duplicate object references.
+
+    ``ray.wait`` assumes a list of unique object references: see
+    https://github.com/modin-project/modin/issues/5045
+
+    Parameters
+    ----------
+    obj_ids : List[ObjectIDType]
+        The object IDs to wait on.
+
+    Returns
+    -------
+    Tuple[List[ObjectIDType], List[ObjectIDType]]
+        A list of object IDs that are ready, and a list of object IDs remaining (this
+        is the same as for ``ray.wait``). Unlike ``ray.wait``, the order of these IDs is not
+        guaranteed.
+    """
+    unique_ids = list(set(obj_ids))
+    return ray.wait(unique_ids, num_returns=len(unique_ids))
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
index 876a1059..6561fcda 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
@@ -16,7 +16,7 @@
 import ray
 from ray.util import get_node_ip_address
 import uuid
-from modin.core.execution.ray.common.utils import deserialize, ObjectIDType
+from modin.core.execution.ray.common.utils import deserialize, ObjectIDType, wait
 from modin.core.execution.ray.common import RayWrapper
 
 from modin.core.dataframe.pandas.partitioning.partition import PandasDataframePartition
@@ -197,7 +197,7 @@ class PandasOnRayDataframePartition(PandasDataframePartition):
     def wait(self):
         """Wait completing computations on the object wrapped by the partition."""
         self.drain_call_queue()
-        ray.wait([self._data])
+        wait([self._data])
 
     def __copy__(self):
         """
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
index f74b15f3..91a720bc 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
@@ -16,13 +16,12 @@
 import inspect
 import threading
 
-import ray
-
 from modin.config import ProgressBar
 from modin.core.execution.ray.generic.partitioning import (
     GenericRayDataframePartitionManager,
 )
 from modin.core.execution.ray.common import RayWrapper
+from modin.core.execution.ray.common.utils import wait
 from .virtual_partition import (
     PandasOnRayDataframeColumnPartition,
     PandasOnRayDataframeRowPartition,
@@ -131,7 +130,7 @@ class PandasOnRayDataframePartitionManager(GenericRayDataframePartitionManager):
         blocks = [
             block for partition in partitions for block in partition.list_of_blocks
         ]
-        ray.wait(blocks, num_returns=len(blocks))
+        wait(blocks)
 
 
 def _make_wrapped_method(name: str):
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/virtual_partition.py b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/virtual_partition.py
index bc359c06..fc7aea7a 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/virtual_partition.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/virtual_partition.py
@@ -20,7 +20,7 @@ from ray.util import get_node_ip_address
 from modin.core.dataframe.pandas.partitioning.axis_partition import (
     PandasDataframeAxisPartition,
 )
-from modin.core.execution.ray.common.utils import deserialize
+from modin.core.execution.ray.common.utils import deserialize, wait
 from .partition import PandasOnRayDataframePartition
 from modin.utils import _inherit_docstrings
 
@@ -469,7 +469,7 @@ class PandasOnRayDataframeVirtualPartition(PandasDataframeAxisPartition):
         """Wait completing computations on the object wrapped by the partition."""
         self.drain_call_queue()
         futures = self.list_of_blocks
-        ray.wait(futures, num_returns=len(futures))
+        wait(futures)
 
     def add_to_apply_calls(self, func, *args, length=None, width=None, **kwargs):
         """
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 9ae4fedd..daf6f745 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -1320,6 +1320,27 @@ class BaseQueryCompiler(ClassLogger, abc.ABC):
         """
         return SeriesDefault.register(pandas.to_numeric)(self, *args, **kwargs)
 
+    @doc_utils.add_one_column_warning
+    @doc_utils.add_refer_to("to_timedelta")
+    def to_timedelta(self, unit="ns", errors="raise"):  # noqa: PR02
+        """
+        Convert argument to timedelta.
+
+        Parameters
+        ----------
+        unit : str, default: "ns"
+            Denotes the unit of the arg for numeric arg. Defaults to "ns".
+        errors : {"ignore", "raise", "coerce"}, default: "raise"
+
+        Returns
+        -------
+        BaseQueryCompiler
+            New QueryCompiler with converted to timedelta values.
+        """
+        return SeriesDefault.register(pandas.to_timedelta)(
+            self, unit=unit, errors=errors
+        )
+
     # FIXME: get rid of `**kwargs` parameter (Modin issue #3108).
     @doc_utils.add_one_column_warning
     @doc_utils.add_refer_to("Series.unique")
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 5c2e21b0..de442670 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -1360,6 +1360,12 @@ class PandasQueryCompiler(BaseQueryCompiler):
             pandas.to_numeric(df.squeeze(axis=1), *args, **kwargs)
         )
     )
+    to_timedelta = Map.register(
+        lambda s, *args, **kwargs: pandas.to_timedelta(
+            s.squeeze(axis=1), *args, **kwargs
+        ).to_frame(),
+        dtypes="timedelta64[ns]",
+    )
 
     # END Map partitions operations
 
diff --git a/modin/pandas/__init__.py b/modin/pandas/__init__.py
index 3965e34a..26c7d314 100644
--- a/modin/pandas/__init__.py
+++ b/modin/pandas/__init__.py
@@ -55,7 +55,6 @@ with warnings.catch_warnings():
         DatetimeIndex,
         Timedelta,
         Timestamp,
-        to_timedelta,
         set_eng_float_format,
         options,
         set_option,
@@ -251,6 +250,7 @@ from .general import (
     crosstab,
     lreshape,
     wide_to_long,
+    to_timedelta,
 )
 
 from modin._compat.pandas_api.namespace import pivot_table
diff --git a/modin/pandas/general.py b/modin/pandas/general.py
index 8cbc0bb5..a34dac3a 100644
--- a/modin/pandas/general.py
+++ b/modin/pandas/general.py
@@ -759,3 +759,18 @@ def _determine_name(objs: Iterable[BaseQueryCompiler], axis: Union[int, str]):
         return list(names[0]) if is_list_like(names[0]) else [names[0]]
     else:
         return None
+
+
+@_inherit_docstrings(pandas.to_datetime, apilink="pandas.to_timedelta")
+@enable_logging
+def to_timedelta(arg, unit=None, errors="raise"):  # noqa: PR01, RT01, D200
+    """
+    Convert argument to timedelta.
+
+    Accepts str, timedelta, list-like or Series for arg parameter.
+    Returns a Series if and only if arg is provided as a Series.
+    """
+    if isinstance(arg, Series):
+        query_compiler = arg._query_compiler.to_timedelta(unit=unit, errors=errors)
+        return Series(query_compiler=query_compiler)
+    return pandas.to_timedelta(arg, unit=unit, errors=errors)


Test patch:
diff --git a/modin/conftest.py b/modin/conftest.py
index d875eb95..4a9687a2 100644
--- a/modin/conftest.py
+++ b/modin/conftest.py
@@ -83,6 +83,7 @@ def pytest_addoption(parser):
         "--extra-test-parameters",
         action="store_true",
         help="activate extra test parameter combinations",
+        default=False,
     )
 
 
@@ -309,10 +310,9 @@ def get_unique_base_execution():
 
 
 def pytest_configure(config):
-    if config.option.extra_test_parameters is not None:
-        import modin.pandas.test.utils as utils
+    import modin.pandas.test.utils as utils
 
-        utils.extra_test_parameters = config.option.extra_test_parameters
+    utils.extra_test_parameters = config.getoption("--extra-test-parameters")
 
     execution = config.option.execution
 
diff --git a/modin/pandas/test/test_general.py b/modin/pandas/test/test_general.py
index 89d63c5b..6d075076 100644
--- a/modin/pandas/test/test_general.py
+++ b/modin/pandas/test/test_general.py
@@ -26,6 +26,7 @@ from .utils import (
     test_data_keys,
     df_equals,
     sort_index_for_equal_values,
+    eval_general,
 )
 
 
@@ -801,3 +802,25 @@ def test_empty_dataframe():
 def test_empty_series():
     s = pd.Series([])
     pd.to_numeric(s)
+
+
+@pytest.mark.parametrize(
+    "arg",
+    [[1, 2], ["a"], 1, "a"],
+    ids=["list_of_ints", "list_of_invalid_strings", "scalar", "invalid_scalar"],
+)
+def test_to_timedelta(arg):
+    # This test case comes from
+    # https://github.com/modin-project/modin/issues/4966
+    eval_general(pd, pandas, lambda lib: lib.to_timedelta(arg))
+
+
+@pytest.mark.parametrize("data", test_data_values, ids=test_data_keys)
+def test_series_to_timedelta(data):
+    def make_frame(lib):
+        series = lib.Series(
+            next(iter(data.values())) if isinstance(data, dict) else data
+        )
+        return lib.to_timedelta(series).to_frame(name="timedelta")
+
+    eval_general(pd, pandas, make_frame)
diff --git a/modin/test/storage_formats/pandas/test_internals.py b/modin/test/storage_formats/pandas/test_internals.py
index d553660b..92c91027 100644
--- a/modin/test/storage_formats/pandas/test_internals.py
+++ b/modin/test/storage_formats/pandas/test_internals.py
@@ -19,6 +19,7 @@ from modin.pandas.test.utils import (
 )
 from modin.config import NPartitions, Engine
 
+import numpy as np
 import pandas
 import pytest
 
@@ -414,3 +415,21 @@ def test_virtual_partition_apply_not_returning_pandas_dataframe(
 
     apply_result = partition.apply(lambda df: 1).get()
     assert apply_result == 1
+
+
+@pytest.mark.skipif(
+    Engine.get() != "Ray",
+    reason="Only ray.wait() does not take duplicate object refs.",
+)
+def test_virtual_partition_dup_object_ref():
+    # See https://github.com/modin-project/modin/issues/5045
+    frame_c = pd.DataFrame(np.zeros((100, 20), dtype=np.float32, order="C"))
+    frame_c = [frame_c] * 20
+    df = pd.concat(frame_c)
+    partition = df._query_compiler._modin_frame._partitions.flatten()[0]
+    obj_refs = partition.list_of_blocks
+    assert len(obj_refs) != len(
+        set(obj_refs)
+    ), "Test setup did not contain duplicate objects"
+    # The below call to wait() should not crash
+    partition.wait()