code: def test_materialized_preprocessing(ray_start_4_cpus):
      # TODO(ekl) we should test all these configs with splitting enabled, but this
      # requires implementing deterministic streaming split.
      ds = ray.data.range(100, parallelism=100).randomize_block_order()
      ds = ds.materialize()
      test = TestRandom(
          2,
          False,
          datasets={"train": ds},
          dataset_config=DataConfig(datasets_to_split=[]),
      )
      test.fit()
  
      ds = ray.data.range(100, parallelism=100).random_shuffle()
      ds = ds.materialize()
      test = TestRandom(
          2,
          False,
          datasets={"train": ds},
          dataset_config=DataConfig(datasets_to_split=[]),
      )
      test.fit()
  
      ds = ray.data.range(100, parallelism=100).map(
          lambda x: {"id": x["id"] * random.random()}
      )
      ds = ds.materialize()
      test = TestRandom(
          2,
          False,
          datasets={"train": ds},
          dataset_config=DataConfig(datasets_to_split=[]),
      )
      test.fit()
  
imports: from typing import Optional
  import random
  import pytest
  from unittest.mock import MagicMock
  import ray
  from ray import train
  from ray.train import DataConfig, ScalingConfig
  from ray.data import DataIterator
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.data._internal.execution.interfaces.execution_options import ExecutionOptions
  from ray.tests.conftest import *
label_desc: Use image node_id 9 for calling a global function with example usage: test_materialized_preprocessing(ray_start_4_cpus) without return types