Fix backward compatibility for pandas 1.0 (#2628)

Author: wjsi

azure-pipelines.yml

@@ -94,8 +94,18 @@ jobs:
     - bash: |
         set -e
         source ci/reload-env.sh
+        mkdir -p build
         pytest $PYTEST_CONFIG mars/$(mars.test.module)
-        coverage report
+        mv .coverage build/.coverage.main.file
+        
+        # do compatibility test for earliest supported pandas release
+        if [[ "$(mars.test.module)" == "dataframe" ]]; then
+          pip install pandas==1.0.5
+          pytest $PYTEST_CONFIG -m pd_compat mars/dataframe
+          mv .coverage build/.coverage.pd_compat.file
+        fi
+        
+        coverage combine build/ && coverage report
         coverage xml
       displayName: 'Run tests'
 

mars/conftest.py

@@ -18,6 +18,7 @@
 import pytest
 
 from mars.config import option_context
+from mars.core.mode import is_kernel_mode, is_build_mode
 from mars.lib.aio import stop_isolation
 from mars.oscar.backends.router import Router
 from mars.oscar.backends.ray.communication import RayServer
@@ -168,6 +169,7 @@ def _new_gpu_test_session(_stop_isolation):  # pragma: no cover
 def setup(_new_test_session):
     _new_test_session.as_default()
     yield _new_test_session
+    assert not (is_build_mode() or is_kernel_mode())
 
 
 @pytest.fixture

mars/core/graph/builder/chunk.py

@@ -259,8 +259,14 @@ def _if_add_node(self, node: EntityType, visited: Set):
         return node not in visited and node not in self._processed_chunks
 
     def _build(self) -> Iterable[Union[TileableGraph, ChunkGraph]]:
-        yield from self.tiler
+        tile_iterator = iter(self.tiler)
+        while True:
+            try:
+                with enter_mode(build=True, kernel=True):
+                    graph = next(tile_iterator)
+                yield graph
+            except StopIteration:
+                break
 
     def build(self) -> Generator[Union[TileableGraph, ChunkGraph], None, None]:
-        with enter_mode(build=True, kernel=True):
-            yield from self._build()
+        yield from self._build()

mars/core/graph/core.pyx

@@ -143,7 +143,7 @@ cdef class DirectedGraph:
         try:
             return list(self._successors[n])
         except KeyError:
-            return KeyError(f'Node {n} does not exist in the directed graph')
+            raise KeyError(f'Node {n} does not exist in the directed graph')
 
     def iter_predecessors(self, n):
         try:

mars/core/mode.py

@@ -18,7 +18,6 @@
 
 from ..config import options
 
-
 _internal_mode = threading.local()
 
 
@@ -69,18 +68,19 @@ def __exit__(self, *_):
             setattr(_internal_mode, mode_name, mode_name_to_old_value[mode_name])
 
     def __call__(self, func):
+        mode_name_to_value = self.mode_name_to_value.copy()
         if not inspect.iscoroutinefunction(func):
             # sync
             @functools.wraps(func)
             def _inner(*args, **kwargs):
-                with self:
+                with enter_mode(**mode_name_to_value):
                     return func(*args, **kwargs)
 
         else:
             # async
             @functools.wraps(func)
             async def _inner(*args, **kwargs):
-                with self:
+                with enter_mode(**mode_name_to_value):
                     return await func(*args, **kwargs)
 
         return _inner

mars/dataframe/arrays.py

@@ -54,10 +54,9 @@
 
 from ..config import options
 from ..core import is_kernel_mode
-from ..lib.version import parse as parse_version
-from ..utils import tokenize
+from ..utils import pd_release_version, tokenize
 
-_use_bool_any_all = parse_version(pd.__version__) >= parse_version("1.3.0")
+_use_bool_any_all = pd_release_version >= (1, 3, 0)
 
 
 class ArrowDtype(ExtensionDtype):

mars/dataframe/base/astype.py

@@ -18,14 +18,14 @@
 
 from ... import opcodes as OperandDef
 from ...core import recursive_tile
-from ...lib.version import parse as parse_version
 from ...serialization.serializables import AnyField, StringField, ListField
 from ...tensor.base import sort
+from ...utils import pd_release_version
 from ..core import DATAFRAME_TYPE, SERIES_TYPE
 from ..operands import DataFrameOperand, DataFrameOperandMixin
 from ..utils import build_empty_df, build_empty_series, parse_index
 
-_need_astype_contiguous = parse_version(pd.__version__) == parse_version("1.3.0")
+_need_astype_contiguous = pd_release_version == (1, 3, 0)
 
 
 class DataFrameAstype(DataFrameOperand, DataFrameOperandMixin):

mars/dataframe/base/duplicated.py

@@ -148,19 +148,20 @@ def _execute_tree_map(cls, ctx, op):
         duplicated = cls._duplicated(inp, op)
         if not duplicated.name:
             duplicated.name = "_duplicated_"
-        result.iloc[duplicated] = None
+        result.iloc[duplicated.values] = None
         result = xdf.concat([result, duplicated], axis=1)
         ctx[op.outputs[0].key] = result
 
     @classmethod
     def _execute_tree_combine(cls, ctx, op):
         inp = ctx[op.input.key]
         result = inp.copy()
-        duplicates = inp[~inp.iloc[:, -1]]
+        duplicated_filter = ~inp.iloc[:, -1]
+        duplicates = inp.loc[duplicated_filter]
         dup_on_duplicated = cls._duplicated(duplicates, op)
-        result.iloc[~inp.iloc[:, -1], -1] = dup_on_duplicated
+        result.iloc[duplicated_filter.values, -1] = dup_on_duplicated
         duplicated = result.iloc[:, -1]
-        result.iloc[duplicated, :-1] = None
+        result.iloc[duplicated.values, :-1] = None
         ctx[op.outputs[0].key] = result
 
     @classmethod

mars/dataframe/base/tests/test_base_execution.py

@@ -29,7 +29,7 @@
 from ....tensor import arange, tensor
 from ....tensor.random import rand
 from ....tests.core import require_cudf
-from ....utils import lazy_import
+from ....utils import lazy_import, pd_release_version
 from ... import eval as mars_eval, cut, qcut, get_dummies
 from ...datasource.dataframe import from_pandas as from_pandas_df
 from ...datasource.series import from_pandas as from_pandas_series
@@ -38,8 +38,12 @@
 from ..to_numeric import to_numeric
 from ..rebalance import DataFrameRebalance
 
+pytestmark = pytest.mark.pd_compat
+
 cudf = lazy_import("cudf", globals=globals())
 
+_explode_with_ignore_index = pd_release_version[:2] >= (1, 1)
+
 
 @require_cudf
 def test_to_gpu_execution(setup_gpu):
@@ -1968,7 +1972,12 @@ def test_stack_execution(setup):
             assert_method(result, expected)
 
 
-def test_explode_execution(setup):
+@pytest.mark.parametrize(
+    "ignore_index", [False, True] if _explode_with_ignore_index else [False]
+)
+def test_explode_execution(setup, ignore_index):
+    explode_kw = {"ignore_index": True} if ignore_index else {}
+
     raw = pd.DataFrame(
         {
             "a": np.random.rand(10),
@@ -1978,20 +1987,12 @@ def test_explode_execution(setup):
         }
     )
     df = from_pandas_df(raw, chunk_size=(4, 2))
-
-    for ignore_index in [False, True]:
-        r = df.explode("b", ignore_index=ignore_index)
-        pd.testing.assert_frame_equal(
-            r.execute().fetch(), raw.explode("b", ignore_index=ignore_index)
-        )
+    r = df.explode("b", ignore_index=ignore_index)
+    pd.testing.assert_frame_equal(r.execute().fetch(), raw.explode("b", **explode_kw))
 
     series = from_pandas_series(raw.b, chunk_size=4)
-
-    for ignore_index in [False, True]:
-        r = series.explode(ignore_index=ignore_index)
-        pd.testing.assert_series_equal(
-            r.execute().fetch(), raw.b.explode(ignore_index=ignore_index)
-        )
+    r = series.explode(ignore_index=ignore_index)
+    pd.testing.assert_series_equal(r.execute().fetch(), raw.b.explode(**explode_kw))
 
 
 def test_eval_query_execution(setup):

mars/dataframe/base/transform.py

@@ -19,9 +19,8 @@
 from ...config import options
 from ...core import OutputType, recursive_tile
 from ...core.custom_log import redirect_custom_log
-from ...lib.version import parse as parse_version
 from ...serialization.serializables import AnyField, BoolField, TupleField, DictField
-from ...utils import enter_current_session, quiet_stdio
+from ...utils import enter_current_session, quiet_stdio, pd_release_version
 from ..core import DATAFRAME_CHUNK_TYPE, DATAFRAME_TYPE
 from ..operands import DataFrameOperandMixin, DataFrameOperand
 from ..utils import (
@@ -33,6 +32,8 @@
     make_dtypes,
 )
 
+_with_convert_dtype = pd_release_version < (1, 2, 0)
+
 
 class TransformOperand(DataFrameOperand, DataFrameOperandMixin):
     _op_type_ = opcodes.TRANSFORM
@@ -236,7 +237,7 @@ def _infer_df_func_returns(self, df, dtypes):
                     if self.call_agg:
                         infer_df = test_df.agg(self._func, args=self.args, **self.kwds)
                     else:
-                        if parse_version(pd.__version__) >= parse_version("1.2.0"):
+                        if not _with_convert_dtype:
                             infer_df = test_df.transform(
                                 self._func, *self.args, **self.kwds
                             )