Fix reduction agg with numpy udf

python/xorbits/_mars/dataframe/reduction/max.py

@@ -27,7 +27,15 @@ def is_atomic(self):
         return True
 
 
-def max_series(df, axis=None, skipna=True, level=None, combine_size=None, method=None):
+def max_series(
+    df,
+    axis=None,
+    skipna=True,
+    level=None,
+    combine_size=None,
+    method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
+):
     op = DataFrameMax(
         axis=axis,
         skipna=skipna,
@@ -47,6 +55,7 @@ def max_dataframe(
     numeric_only=None,
     combine_size=None,
     method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
 ):
     op = DataFrameMax(
         axis=axis,

python/xorbits/_mars/dataframe/reduction/mean.py

@@ -32,7 +32,15 @@ def mean(x):
         return mean
 
 
-def mean_series(df, axis=None, skipna=True, level=None, combine_size=None, method=None):
+def mean_series(
+    df,
+    axis=None,
+    skipna=True,
+    level=None,
+    combine_size=None,
+    method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
+):
     op = DataFrameMean(
         axis=axis,
         skipna=skipna,
@@ -52,6 +60,7 @@ def mean_dataframe(
     numeric_only=None,
     combine_size=None,
     method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
 ):
     op = DataFrameMean(
         axis=axis,

python/xorbits/_mars/dataframe/reduction/min.py

@@ -27,7 +27,15 @@ def is_atomic(self):
         return True
 
 
-def min_series(df, axis=None, skipna=True, level=None, combine_size=None, method=None):
+def min_series(
+    df,
+    axis=None,
+    skipna=True,
+    level=None,
+    combine_size=None,
+    method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
+):
     op = DataFrameMin(
         axis=axis,
         skipna=skipna,
@@ -47,6 +55,7 @@ def min_dataframe(
     numeric_only=None,
     combine_size=None,
     method=None,
+    **kwargs,  # kwargs for compatible with numpy reduction
 ):
     op = DataFrameMin(
         axis=axis,

python/xorbits/_mars/dataframe/reduction/tests/test_reduction_execution.py

@@ -781,6 +781,27 @@ def realized_volatility(series):
         result.execute().fetch(), data.agg(realized_volatility)
     )
 
+    def trip_type(x):
+        return np.min(x)
+
+    df = md.DataFrame(data)
+    result = df.agg(trip_type)
+    pd.testing.assert_series_equal(result.execute().fetch(), data.agg(trip_type))
+
+    def trip_type_max(x):
+        return np.max(x)
+
+    df = md.DataFrame(data)
+    result = df.agg(trip_type_max)
+    pd.testing.assert_series_equal(result.execute().fetch(), data.agg(trip_type_max))
+
+    def trip_type_mean(x):
+        return np.mean(x)
+
+    df = md.DataFrame(data)
+    result = df.agg(trip_type_mean)
+    pd.testing.assert_series_equal(result.execute().fetch(), data.agg(trip_type_mean))
+
     df = md.DataFrame(data)
     result = df.agg(all_aggs)
     pd.testing.assert_frame_equal(result.execute().fetch(), data.agg(all_aggs))