code: def test_compile_function():
      compiler = ReductionCompiler()
      ms = md.Series([1, 2, 3])
      # no Mars objects inside closures
      with pytest.raises(ValueError):
          compiler.add_function(
              functools.partial(lambda x: (x + ms).sum()), ndim=2
          )
      # function should return a Mars object
      with pytest.raises(ValueError):
          compiler.add_function(lambda x: x is not None, ndim=2)
      # function should perform some sort of reduction in dimensionality
      with pytest.raises(ValueError):
          compiler.add_function(lambda x: x, ndim=2)
      # function should only contain acceptable operands
      with pytest.raises(ValueError):
          compiler.add_function(lambda x: x.sort_values().max(), ndim=1)
      with pytest.raises(ValueError):
          compiler.add_function(lambda x: x.max().shift(1), ndim=2)
  
      # test agg for all data
      for ndim in [1, 2]:
          compiler = ReductionCompiler(store_source=True)
          compiler.add_function(
              lambda x: (x**2).count() + 1, ndim=ndim
          )
          result = compiler.compile()
          # check pre_funcs
          assert len(result.pre_funcs) == 1
          assert "pow" in result.pre_funcs[0].func.__source__
          # check agg_funcs
          assert len(result.agg_funcs) == 1
          assert result.agg_funcs[0].map_func_name == "count"
          assert result.agg_funcs[0].agg_func_name == "sum"
          # check post_funcs
          assert len(result.post_funcs) == 1
          assert result.post_funcs[0].func_name == "<lambda>"
          assert "add" in result.post_funcs[0].func.__source__
  
          compiler.add_function(
              lambda x: -x.prod() ** 2 + (1 + (x**2).count()),
              ndim=ndim,
          )
          result = compiler.compile()
          # check pre_funcs
          assert len(result.pre_funcs) == 2
          assert (
              "pow" in result.pre_funcs[0].func.__source__
              or "pow" in result.pre_funcs[1].func.__source__
          )
          assert (
              "pow" not in result.pre_funcs[0].func.__source__
              or "pow" not in result.pre_funcs[1].func.__source__
          )
          # check agg_funcs
          assert len(result.agg_funcs) == 2
          assert set(
              result.agg_funcs[i].map_func_name for i in range(2)
          ) == {
              "count",
              "prod",
          }
          assert set(
              result.agg_funcs[i].agg_func_name for i in range(2)
          ) == {
              "sum",
              "prod",
          }
          # check post_funcs
          assert len(result.post_funcs) == 2
          assert result.post_funcs[0].func_name == "<lambda_0>"
          assert "add" in result.post_funcs[0].func.__source__
          assert "add" in result.post_funcs[1].func.__source__
  
          compiler = ReductionCompiler(store_source=True)
          compiler.add_function(
              lambda x: where_function(x.all(), x.count(), 0), ndim=ndim
          )
          result = compiler.compile()
          # check pre_funcs
          assert len(result.pre_funcs) == 1
          assert (
              result.pre_funcs[0].input_key
              == result.pre_funcs[0].output_key
          )
          # check agg_funcs
          assert len(result.agg_funcs) == 2
          assert set(
              result.agg_funcs[i].map_func_name for i in range(2)
          ) == {
              "all",
              "count",
          }
          assert set(
              result.agg_funcs[i].agg_func_name for i in range(2)
          ) == {
              "sum",
              "all",
          }
          # check post_funcs
          assert len(result.post_funcs) == 1
          if ndim == 1:
              assert "np.where" in result.post_funcs[0].func.__source__
          else:
              assert (
                  "np.where" not in result.post_funcs[0].func.__source__
              )
              assert ".where" in result.post_funcs[0].func.__source__
  
          # check boolean expressions
          compiler = ReductionCompiler(store_source=True)
          compiler.add_function(lambda x: (x == "1").sum(), ndim=ndim)
          result = compiler.compile()
          # check pre_funcs
          assert len(result.pre_funcs) == 1
          assert "eq" in result.pre_funcs[0].func.__source__
          # check agg_funcs
          assert len(result.agg_funcs) == 1
          assert result.agg_funcs[0].map_func_name == "sum"
          assert result.agg_funcs[0].agg_func_name == "sum"
  
      # test agg for specific columns
      compiler = ReductionCompiler(store_source=True)
      compiler.add_function(
          lambda x: 1 + x.sum(), ndim=2, cols=["a", "b"]
      )
      compiler.add_function(
          lambda x: -1 + x.sum(), ndim=2, cols=["b", "c"]
      )
      result = compiler.compile()
      # check pre_funcs
      assert len(result.pre_funcs) == 1
      assert set(result.pre_funcs[0].columns) == set("abc")
      # check agg_funcs
      assert len(result.agg_funcs) == 1
      assert result.agg_funcs[0].map_func_name == "sum"
      assert result.agg_funcs[0].agg_func_name == "sum"
      # check post_funcs
      assert len(result.post_funcs) == 2
      assert set(
          "".join(sorted(result.post_funcs[i].columns))
          for i in range(2)
      ) == {
          "ab",
          "bc",
      }
  
      # test agg for multiple columns
      compiler = ReductionCompiler(store_source=True)
      compiler.add_function(lambda x: x.sum(), ndim=2, cols=["a"])
      compiler.add_function(lambda x: x.sum(), ndim=2, cols=["b"])
      compiler.add_function(lambda x: x.min(), ndim=2, cols=["c"])
      result = compiler.compile()
      # check pre_funcs
      assert len(result.pre_funcs) == 1
      assert set(result.pre_funcs[0].columns) == set("abc")
      # check agg_funcs
      assert len(result.agg_funcs) == 2
      assert result.agg_funcs[0].map_func_name == "sum"
      assert result.agg_funcs[0].agg_func_name == "sum"
      # check post_funcs
      assert len(result.post_funcs) == 2
      assert set(result.post_funcs[0].columns) == set("ab")
  
imports: import functools
  import operator
  from functools import reduce
  from typing import NamedTuple
  import numpy
  import pandas
  import pytest
  from ....None import dataframe
  from ....core import tile
  from ....core.operand import OperandStage
  from ....tensor import Tensor
  from ...core import DataFrame, IndexValue, Series, OutputType
  from ...datasource.series import from_pandas
  from ...datasource.dataframe import from_pandas
  from ...merge import DataFrameConcat
  from ..None import DataFrameSum, DataFrameProd, DataFrameMin, DataFrameMax, DataFrameCount, DataFrameMean, DataFrameVar, DataFrameAll, DataFrameAny, DataFrameSkew, DataFrameKurtosis, DataFrameSem, DataFrameAggregate, DataFrameCummin, DataFrameCummax, DataFrameCumprod, DataFrameCumsum, DataFrameNunique, CustomReduction
  from ..aggregation import where_function
  from ..core import ReductionCompiler
label_desc: Use image node_id 8 for calling a global function with example usage: test_compile_function() without return types