Check initialization of serializables on CI (#3007)

Author: wjsi

.github/workflows/benchmark-ci.yml

@@ -48,6 +48,7 @@ jobs:
       - name: Run ASV benchmarks
         run: |
           source ./ci/reload-env.sh
+          unset CI
           cd benchmarks/asv_bench
           asv check -E existing
           git remote add upstream https://github.com/mars-project/mars.git

azure-pipelines.yml

@@ -87,6 +87,7 @@ jobs:
     - bash: |
         set -e
         source ci/reload-env.sh
+        export CI=true
         mkdir -p build
         pytest $PYTEST_CONFIG mars/$(mars.test.module)
         mv .coverage build/.coverage.main.file

mars/dataframe/base/value_counts.py

@@ -30,94 +30,37 @@
 class DataFrameValueCounts(DataFrameOperand, DataFrameOperandMixin):
     _op_type_ = opcodes.VALUE_COUNTS
 
-    _input = KeyField("input")
-    _normalize = BoolField("normalize")
-    _sort = BoolField("sort")
-    _ascending = BoolField("ascending")
-    _bins = Int64Field("bins")
-    _dropna = BoolField("dropna")
-    _method = StringField("method")
-    _convert_index_to_interval = BoolField("convert_index_to_interval")
-    _nrows = Int64Field("nrows")
-
-    def __init__(
-        self,
-        normalize=None,
-        sort=None,
-        ascending=None,
-        bins=None,
-        dropna=None,
-        method=None,
-        convert_index_to_interval=None,
-        nrows=None,
-        **kw
-    ):
-        super().__init__(
-            _normalize=normalize,
-            _sort=sort,
-            _ascending=ascending,
-            _bins=bins,
-            _dropna=dropna,
-            _method=method,
-            _convert_index_to_interval=convert_index_to_interval,
-            _nrows=nrows,
-            **kw
-        )
+    input = KeyField("input")
+    normalize = BoolField("normalize")
+    sort = BoolField("sort")
+    ascending = BoolField("ascending")
+    bins = Int64Field("bins")
+    dropna = BoolField("dropna")
+    method = StringField("method")
+    convert_index_to_interval = BoolField("convert_index_to_interval", default=None)
+    nrows = Int64Field("nrows", default=None)
+
+    def __init__(self, **kw):
+        super().__init__(**kw)
         self.output_types = [OutputType.series]
 
-    @property
-    def input(self):
-        return self._input
-
-    @property
-    def normalize(self):
-        return self._normalize
-
-    @property
-    def sort(self):
-        return self._sort
-
-    @property
-    def ascending(self):
-        return self._ascending
-
-    @property
-    def bins(self):
-        return self._bins
-
-    @property
-    def dropna(self):
-        return self._dropna
-
-    @property
-    def method(self):
-        return self._method
-
-    @property
-    def convert_index_to_interval(self):
-        return self._convert_index_to_interval
-
-    @property
-    def nrows(self):
-        return self._nrows
-
     def _set_inputs(self, inputs):
         super()._set_inputs(inputs)
-        self._input = self._inputs[0]
+        self.input = self._inputs[0]
 
     def __call__(self, inp):
         test_series = build_series(inp).value_counts(normalize=self.normalize)
-        if self._bins is not None:
+        if self.bins is not None:
             from .cut import cut
 
             # cut
             try:
-                inp = cut(inp, self._bins, include_lowest=True)
+                inp = cut(inp, self.bins, include_lowest=True)
             except TypeError:  # pragma: no cover
                 raise TypeError("bins argument only works with numeric data.")
 
-            self._bins = None
-            self._convert_index_to_interval = True
+            self.bins = None
+            self.convert_index_to_interval = True
             return self.new_series(
                 [inp],
                 shape=(np.nan,),
@@ -174,7 +117,7 @@ def tile(cls, op: "DataFrameValueCounts"):
 
             if op.nrows:
                 # set to sort_values
-                inp.op._nrows = op.nrows
+                inp.op.nrows = op.nrows
         elif op.nrows:
             inp = inp.iloc[: op.nrows]
 

mars/dataframe/datasource/core.py

@@ -36,11 +36,7 @@ class HeadOptimizedDataSource(DataFrameOperand, DataFrameOperandMixin):
     # First, it will try to trigger first_chunk.head() and raise TilesError,
     # When iterative tiling is triggered,
     # check if the first_chunk.head() meets requirements.
-    _nrows = Int64Field("nrows", default=None)
-
-    @property
-    def nrows(self):
-        return self._nrows
+    nrows = Int64Field("nrows", default=None)
 
     @property
     def first_chunk(self):
@@ -67,7 +63,7 @@ def _tile_head(cls, op: "HeadOptimizedDataSource"):
             tileds[0]._shape = chunk_shape
         else:
             for chunk in tileds[0].chunks:
-                chunk.op._nrows = None
+                chunk.op.nrows = None
             # otherwise
             tiled = yield from recursive_tile(tileds[0].iloc[: op.nrows])
             tileds = [tiled]

mars/dataframe/datasource/from_index.py

@@ -14,7 +14,7 @@
 
 from ... import opcodes
 from ...core import recursive_tile
-from ...serialization.serializables import KeyField, StringField
+from ...serialization.serializables import AnyField, KeyField
 from ..initializer import Index
 from ..operands import DataFrameOperand, DataFrameOperandMixin
 
@@ -24,7 +24,7 @@ class SeriesFromIndex(DataFrameOperand, DataFrameOperandMixin):
 
     input_ = KeyField("input_")
     index = KeyField("index")
-    name = StringField("name", default=None)
+    name = AnyField("name", default=None)
 
     def _set_inputs(self, inputs):
         super()._set_inputs(inputs)

mars/dataframe/datasource/from_records.py

@@ -28,46 +28,15 @@
 class DataFrameFromRecords(DataFrameOperand, DataFrameOperandMixin):
     _op_type_ = OperandDef.DATAFRAME_FROM_RECORDS
 
-    _columns = ListField("columns")
-    _exclude = ListField("exclude")
-    _coerce_float = BoolField("coerce_float")
-    _nrows = Int32Field("nrows")
-
-    def __init__(
-        self,
-        index=None,
-        columns=None,
-        exclude=None,
-        coerce_float=False,
-        nrows=None,
-        **kw
-    ):
+    columns = ListField("columns", default=None)
+    exclude = ListField("exclude", default=None)
+    coerce_float = BoolField("coerce_float", default=False)
+    nrows = Int32Field("nrows", default=None)
+
+    def __init__(self, index=None, columns=None, **kw):
         if index is not None or columns is not None:
             raise NotImplementedError("Specifying index value is not supported for now")
-        super().__init__(
-            _exclude=exclude,
-            _columns=columns,
-            _coerce_float=coerce_float,
-            _nrows=nrows,
-            _output_types=[OutputType.dataframe],
-            **kw
-        )
-
-    @property
-    def columns(self):
-        return self._columns
-
-    @property
-    def exclude(self):
-        return self._exclude
-
-    @property
-    def coerce_float(self):
-        return self._coerce_float
-
-    @property
-    def nrows(self):
-        return self._nrows
+        super().__init__(columns=columns, _output_types=[OutputType.dataframe], **kw)
 
     def __call__(self, data):
         if self.nrows is None:

mars/dataframe/datasource/read_raydataset.py

@@ -46,33 +46,10 @@ class DataFrameReadRayDataset(
 ):
     _op_type_ = OperandDef.READ_RAYDATASET
 
-    _refs = AnyField("refs")
-    _columns = ListField("columns")
-    _incremental_index = BoolField("incremental_index")
-    _nrows = Int64Field("nrows")
-
-    def __init__(
-        self, refs=None, columns=None, incremental_index=None, nrows=None, **kw
-    ):
-        super().__init__(
-            _refs=refs,
-            _columns=columns,
-            _incremental_index=incremental_index,
-            _nrows=nrows,
-            **kw,
-        )
-
-    @property
-    def refs(self):
-        return self._refs
-
-    @property
-    def columns(self):
-        return self._columns
-
-    @property
-    def incremental_index(self):
-        return self._incremental_index
+    refs = AnyField("refs", default=None)
+    columns = ListField("columns", default=None)
+    incremental_index = BoolField("incremental_index", default=None)
+    nrows = Int64Field("nrows", default=None)
 
     @classmethod
     def _tile_partitioned(cls, op: "DataFrameReadRayDataset"):
@@ -157,20 +134,12 @@ def read_raydataset(*args, **kwargs):
 
 class DataFrameReadMLDataset(HeadOptimizedDataSource):
     _op_type_ = OperandDef.READ_MLDATASET
-    _mldataset = ReferenceField("mldataset", "ray.util.data.MLDataset")
-    _columns = ListField("columns")
 
-    def __init__(self, mldataset=None, columns=None, **kw):
-        super().__init__(
-            _mldataset=mldataset,
-            _columns=columns,
-            _output_types=[OutputType.dataframe],
-            **kw,
-        )
+    mldataset = ReferenceField("mldataset", "ray.util.data.MLDataset", default=None)
+    columns = ListField("columns", default=None)
 
-    @property
-    def mldataset(self):
-        return self._mldataset
+    def __init__(self, **kw):
+        super().__init__(_output_types=[OutputType.dataframe], **kw)
 
     def _update_key(self):
         """We can't direct generate token for mldataset when we use

mars/dataframe/datasource/read_sql.py

@@ -51,15 +51,15 @@ def _get_logic_key_token_values(self):
         fields_to_tokenize = [
             getattr(self, k, None)
             for k in [
-                "_table_or_sql",
-                "_schema",
-                "_coerce_float",
-                "_parse_dates",
-                "_columns",
-                "_method",
-                "_incremental_index",
-                "_use_arrow_dtype",
-                "_partition_col",
+                "table_or_sql",
+                "schema",
+                "coerce_float",
+                "parse_dates",
+                "columns",
+                "method",
+                "incremental_index",
+                "use_arrow_dtype",
+                "partition_col",
             ]
         ]
         return super()._get_logic_key_token_values() + fields_to_tokenize
@@ -104,7 +104,7 @@ class DataFrameReadSQL(
     nrows = Int64Field("nrows", default=None)
 
     def get_columns(self):
-        return self._columns
+        return self.columns
 
     def set_pruned_columns(self, columns, *, keep_order=None):
         self.columns = columns