code: def _sample_fragment(
      to_batches_kwargs,
      columns,
      schema,
      file_fragment: _SerializedFragment,
  ) -> float:
      # Sample the first rows batch from file fragment `serialized_fragment`.
      # Return the encoding ratio calculated from the sampled rows.
      fragment = _deserialize_fragments_with_retry([file_fragment])[0]
  
      # Only sample the first row group.
      fragment = fragment.subset(row_group_ids=[0])
      batch_size = max(
          min(
              fragment.metadata.num_rows,
              PARQUET_ENCODING_RATIO_ESTIMATE_NUM_ROWS,
          ),
          1,
      )
      # Use the batch_size calculated above, and ignore the one specified by user if set.
      # This is to avoid sampling too few or too many rows.
      to_batches_kwargs.pop("batch_size", None)
      batches = fragment.to_batches(
          columns=columns,
          schema=schema,
          batch_size=batch_size,
          **to_batches_kwargs,
      )
      # Use first batch in-memory size as ratio estimation.
      try:
          batch = next(batches)
      except StopIteration:
          ratio = PARQUET_ENCODING_RATIO_ESTIMATE_LOWER_BOUND
      else:
          if batch.num_rows > 0:
              in_memory_size = batch.nbytes / batch.num_rows
              metadata = fragment.metadata
              total_size = 0
              for idx in range(metadata.num_row_groups):
                  total_size += metadata.row_group(idx).total_byte_size
              file_size = total_size / metadata.num_rows
              ratio = in_memory_size / file_size
          else:
              ratio = PARQUET_ENCODING_RATIO_ESTIMATE_LOWER_BOUND
      logger.debug(
          f"Estimated Parquet encoding ratio is {ratio} for fragment {fragment} "
          f"with batch size {batch_size}."
      )
      return ratio
  
imports: import logging
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  import ray.cloudpickle
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource
  from ray.data.datasource._default_metadata_providers import get_generic_metadata_provider
  from ray.data.datasource.datasource import ReadTask
  from ray.data.datasource.file_meta_provider import DefaultParquetMetadataProvider, ParquetMetadataProvider, _handle_read_os_error
  from ray.data.datasource.partitioning import PathPartitionFilter
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 6 for calling a global function with example usage: _sample_fragment(to_batches_kwargs, columns, schema, file_fragment) and returns: ratio