code: def test_lightgbm_dataset_partition(ray_cluster_2cpu):
      # Create a LightGBM model with a Ray backend
      config = {
          INPUT_FEATURES: [{"name": "in_column", "type": "binary"}],
          OUTPUT_FEATURES: [{"name": "out_column", "type": "binary"}],
          "model_type": "gbm",
          # Disable feature filtering to avoid having no features due to small test dataset,
          # see https://stackoverflow.com/a/66405983/5222402
          TRAINER: {"feature_pre_filter": False},
      }
      backend_config = {**RAY_BACKEND_CONFIG}
      backend_config["preprocessor_kwargs"] = {"num_cpu": 1}
      model = LudwigModel(config, backend=backend_config)
      lgbm_model = GBM(ModelConfig.from_dict(config))
      trainer = LightGBMRayTrainer(GBMTrainerConfig(), lgbm_model)
  
      def create_dataset(model: LudwigModel, size: int) -> RayDataset:
          df = pd.DataFrame(
              {
                  "in_column_lm_J5T": np.random.randint(
                      0, 1, size=(size,), dtype=np.uint8
                  ),
                  "out_column_2Xl8CP": np.random.randint(
                      0, 1, size=(size,), dtype=np.uint8
                  ),
              }
          )
          df = dask.dataframe.from_pandas(df, npartitions=1)
          return model.backend.dataset_manager.create(
              df, config=model.config, training_set_metadata={}
          )
  
      # Create synthetic train, val, and test datasets with one block
      train_ds = create_dataset(model, int(1e4))
      val_ds = create_dataset(model, int(1e4))
      test_ds = create_dataset(model, int(1e4))
  
      # Test with no repartition. This occurs when the number of dataset blocks
      # is equal to the number of ray actors.
      trainer.ray_params.num_actors = 1
      trainer._construct_lgb_datasets(
          train_ds, validation_set=val_ds, test_set=test_ds
      )
      assert train_ds.ds.num_blocks() == 1
      assert val_ds.ds.num_blocks() == 1
      assert test_ds.ds.num_blocks() == 1
  
      # Test with repartition. This occurs when the number of dataset blocks
      # is less than the number of ray actors.
      trainer.ray_params.num_actors = 2
      trainer._construct_lgb_datasets(
          train_ds, validation_set=val_ds, test_set=test_ds
      )
      assert train_ds.ds.num_blocks() == 2
      assert val_ds.ds.num_blocks() == 2
      assert test_ds.ds.num_blocks() == 2
  
      # Test again with no repartition. This also occurs when the number of dataset blocks
      # is greater than the number of ray actors.
      trainer.ray_params.num_actors = 1
      trainer._construct_lgb_datasets(
          train_ds, validation_set=val_ds, test_set=test_ds
      )
      assert train_ds.ds.num_blocks() == 2
      assert val_ds.ds.num_blocks() == 2
      assert test_ds.ds.num_blocks() == 2
  
imports: import logging
  import os
  import shutil
  from unittest import mock
  import numpy
  import pandas
  import pytest
  import torch
  from packaging.version import parse
  from ludwig.api import LudwigModel
  from ludwig.callbacks import Callback
  from ludwig.constants import BATCH_SIZE, EFFECTIVE_BATCH_SIZE, EPOCHS, EVAL_BATCH_SIZE, INPUT_FEATURES, MAX_BATCH_SIZE_DATASET_FRACTION, OUTPUT_FEATURES, TRAINER
  from ludwig.distributed import init_dist_strategy
  from tests.integration_tests.utils import binary_feature, category_feature, generate_data, LocalTestBackend, number_feature, RAY_BACKEND_CONFIG, sequence_feature, text_feature, vector_feature
label_desc: Use image node_id 5 for calling a global function with example usage: test_lightgbm_dataset_partition(ray_cluster_2cpu) and returns: model