code: def test_hash_split(df_engine, ray_cluster_2cpu):
      nrows = 100
      npartitions = 10
  
      df = pd.DataFrame(
          np.random.randint(0, 100, size=(nrows, 3)),
          columns=["A", "B", "C"],
      )
      df["id"] = np.arange(0, 100)
  
      if isinstance(df_engine, DaskEngine):
          df = df_engine.df_lib.from_pandas(df, npartitions=npartitions)
  
      probabilities = [0.8, 0.1, 0.1]
      split_params = {
          "type": "hash",
          "column": "id",
          "probabilities": probabilities,
      }
      splitter = get_splitter(**split_params)
  
      backend = Mock()
      backend.df_engine = df_engine
      splits = splitter.split(df, backend)
      assert len(splits) == 3
      if isinstance(df_engine, DaskEngine):
          splits = [split.compute() for split in splits]
  
      # IDs should not overlap between splits
      assert all(
          [
              set(split1["id"]).isdisjoint(set(split2["id"]))
              for split1, split2 in combinations(splits, 2)
          ]
      )
  
      for split, p in zip(splits, probabilities):
          # Should be approximately the same size as the desired proportion
          assert nrows * p - 5 <= len(split["id"]) <= nrows * p + 5
  
      # Need to ensure deterministic splitting even as we append data
      df2 = pd.DataFrame(
          np.random.randint(0, 100, size=(nrows, 3)),
          columns=["A", "B", "C"],
      )
      df2["id"] = np.arange(100, 200)
  
      nrows *= 2
  
      df = df_engine.df_lib.concat([df, df2])
  
      splits2 = splitter.split(df, backend)
      assert len(splits2) == 3
      if isinstance(df_engine, DaskEngine):
          splits2 = [split.compute() for split in splits2]
  
      # IDs should not overlap between splits
      assert all(
          [
              set(split1["id"]).isdisjoint(set(split2["id"]))
              for split1, split2 in combinations(splits2, 2)
          ]
      )
  
      for split1, split2, p in zip(splits, splits2, probabilities):
          ids1 = set(split1["id"].values.tolist())
          ids2 = set(split2["id"].values.tolist())
  
          assert nrows * p - 10 <= len(ids2) <= nrows * p + 10
  
          # All elements from the first round of splitting are in the same split, even after appending
          # more rows
          assert ids1.issubset(ids2)
  
file_path: ludwig/tests/ludwig/data/test_split.py
imports: from datetime import datetime, timedelta
  from itertools import combinations
  from random import randrange
  from unittest.mock import Mock
  import numpy
  import pandas
  import pytest
  from ludwig.data.dataframe.pandas import PandasEngine
  from ludwig.data.split import get_splitter
label_desc: Use image node_id 8 for calling a global function with example usage: test_hash_split(df_engine, ray_cluster_2cpu) without return types