code: def _tile_reservoirs(cls, op: "DataFrameSample", in_df, weights):
      out_data = op.outputs[0]
      input_dfs = [in_df]
      size = op.size
  
      weight_chunks = itertools.repeat(None)
      if isinstance(weights, ENTITY_TYPE):
          input_dfs.append(weights)
          weight_chunks = weights.chunks
  
      if any(cs < size for cs in in_df.nsplits[op.axis]):
          # make sure all chunk > m
          n_records = in_df.shape[op.axis]
          n_chunk = min(
              max(ceildiv(n_records, size), 1), in_df.chunk_shape[0]
          )
          chunk_size = ceildiv(in_df.shape[op.axis], n_chunk)
          chunk_sizes = list(
              normalize_chunk_sizes(n_records, chunk_size)[0]
          )
          if chunk_sizes[-1] < size and len(chunk_sizes) > 1:
              # the last chunk may still less than m
              # merge it into previous one
              chunk_sizes[-2] += chunk_sizes[-1]
              chunk_sizes = chunk_sizes[:-1]
          in_df = yield from recursive_tile(
              in_df.rechunk({0: tuple(chunk_sizes)})
          )
          if isinstance(weights, ENTITY_TYPE):
              weights = yield from recursive_tile(
                  weights.rechunk({0: tuple(chunk_sizes)})
              )
          if len(chunk_sizes) == 1:
              return cls._tile_one_chunk(op, in_df, weights)
  
      # for each chunk in a, do regular sampling
      sampled_chunks = []
      seeds = gen_random_seeds(len(in_df.chunks), op.random_state)
      for data_chunk, weights_chunk, seed in zip(
          in_df.chunks, weight_chunks, seeds
      ):
          input_chunks = [data_chunk]
  
          chunk_op = op.copy().reset_key()
          chunk_op._random_state = None
          chunk_op._seed = seed
          if isinstance(op.weights, ENTITY_TYPE):
              input_chunks.append(weights_chunk)
              chunk_op._weights = weights_chunk
  
          params = data_chunk.params
          new_shape = list(data_chunk.shape)
          new_shape[op.axis] = size
          params["shape"] = tuple(new_shape)
          sampled_chunks.append(
              chunk_op.new_chunk(input_chunks, **params)
          )
  
      # generate a random variable for samples in every chunk
      state = TensorRandomState.from_numpy(op.random_state)
      indices = state.rand(size)
  
      if weights is None:
          # weights not specified, use nsplits to calculate cumulative probability
          # to distribute samples in each chunk
          cum_offsets = np.cumsum(in_df.nsplits[op.axis])
          cum_offsets = cum_offsets * 1.0 / cum_offsets[-1]
      else:
          # weights specified, use weights to calculate cumulative probability
          # to distribute samples in each chunk
          chunk_weights = yield from recursive_tile(
              weights.to_tensor().map_chunk(
                  lambda x: x.sum(keepdims=True)
              )
          )
          chunk_weights_chunk = TensorConcatenate(
              dtype=chunk_weights.dtype
          ).new_chunk(
              chunk_weights.chunks,
              shape=(len(chunk_weights.chunks),),
              index=(0,),
          )
  
          cum_chunk = TensorMapChunk(
              func=lambda x: (x / x.sum()).cumsum()
          ).new_chunk(
              [chunk_weights_chunk],
              shape=(len(chunk_weights.chunks),),
              index=(0,),
          )
          cum_offsets = TensorMapChunk(
              func=cum_chunk.op.func
          ).new_tensor(
              [weights],
              chunks=[cum_chunk],
              nsplits=((s,) for s in cum_chunk.shape),
              **cum_chunk.params,
          )
  
      index_chunks = []
      # seek which chunk the final sample will select
      chunk_sel = yield from recursive_tile(
          searchsorted(cum_offsets, indices, side="right")
      )
      # for every chunk, select samples with bool indexing
      for idx, sampled_chunk in enumerate(sampled_chunks):
          chunk_index = chunk_sel.map_chunk(
              func=lambda x, i: x == i,
              args=(idx,),
              elementwise=True,
              dtype=bool,
          )
          sampled_df_op = sampled_chunk.op.copy().reset_key()
          sampled_chunk._index = (0,) * sampled_chunk.ndim
          sampled_df = sampled_df_op.new_tileable(
              input_dfs,
              chunks=[sampled_chunk],
              nsplits=((s,) for s in sampled_chunk.shape),
              **sampled_chunk.params,
          )
          index_chunk = (
              yield from recursive_tile(sampled_df.iloc[chunk_index])
          ).chunks[0]
  
          chunk_idx = [0] * sampled_chunk.ndim
          chunk_idx[op.axis] = idx
          index_chunk._index = tuple(chunk_idx)
          index_chunks.append(index_chunk)
  
      params = out_data.params
      new_shape = list(in_df.shape)
      new_shape[op.axis] = size
      params["shape"] = tuple(new_shape)
  
      new_nsplits = list(in_df.nsplits)
      new_nsplits[op.axis] = (np.nan,) * len(index_chunks)
  
      df_op = op.copy().reset_key()
      return df_op.new_tileables(
          input_dfs,
          chunks=index_chunks,
          nsplits=tuple(new_nsplits),
          **params,
      )
  
imports: import copy
  import itertools
  import numpy
  from ...None import opcodes
  from ...core import ENTITY_TYPE, get_output_types, recursive_tile
  from ...serialization.serializables import BoolField, AnyField, Int8Field, Int64Field, Float64Field, KeyField
  from ...tensor import searchsorted
  from ...tensor.base import TensorMapChunk
  from ...tensor.merge import TensorConcatenate
  from ...tensor.random import RandomState, RandomStateField
  from ...tensor.utils import normalize_chunk_sizes, gen_random_seeds
  from ...utils import has_unknown_shape, ceildiv
  from ..operands import DataFrameOperandMixin, DataFrameOperand
  from ..utils import validate_axis, parse_index
label_desc: Use image node_id 15 for calling the DataFrameSample obj's underlying member method code with example usage: obj._tile_reservoirs(cls, op, in_df, weights) and returns: df_op, cls