code: def tile(cls, op: "ProximaSearcher"):
      tensor = op.tensor
      index = op.index
      topk = op.topk
      outs = op.outputs
      row_number = op.row_number
  
      ctx = get_context()
  
      # make sure all inputs have known chunk sizes
      if has_unknown_shape(*op.inputs):
          yield
  
      rechunk_size = dict()
      if tensor.chunk_shape[1] > 1:
          rechunk_size[1] = tensor.shape[1]
      if row_number is not None:
          rechunk_size[0] = tensor.shape[0] // row_number
      if len(rechunk_size) > 0:
          tensor = yield from recursive_tile(
              tensor.rechunk(rechunk_size)
          )
  
      logger.warning(f"query chunks count: {len(tensor.chunks)} ")
  
      if hasattr(index, "op"):
          built_indexes = [index.chunks] * len(tensor.chunks)
      else:
          # index path
          fs: FileSystem = get_fs(index, op.storage_options)
          index_paths = [
              f
              for f in fs.ls(index)
              if f.rsplit("/", 1)[-1].startswith("proxima_")
          ]
          download_chunks = cls._build_download_chunks(op, index_paths)
          iters = [
              iter(itertools.cycle(i)) for i in download_chunks.values()
          ]
          built_indexes = []
          for _ in range(len(tensor.chunks)):
              built_indexes.append([next(it) for it in iters])
  
      if hasattr(index, "op"):
          index_chunks_workers = [
              m["bands"][0][0]
              for m in ctx.get_chunks_meta(
                  [c.key for c in index.chunks], fields=["bands"]
              )
          ]
      else:
          index_chunks_workers = [None] * len(built_indexes[0])
  
      out_chunks = [], []
      for i, tensor_chunk in enumerate(tensor.chunks):
          pk_chunks, distance_chunks = [], []
          for j, chunk_index, worker in zip(
              itertools.count(), built_indexes[i], index_chunks_workers
          ):
              chunk_op = op.copy().reset_key()
              chunk_op.stage = OperandStage.map
              if hasattr(index, "op"):
                  chunk_op.expect_worker = worker
              else:
                  chunk_op.expect_worker = chunk_index.op.expect_worker
              chunk_op._index = chunk_index
              chunk_op._tensor = None
              chunk_kws = [
                  {
                      "index": (tensor_chunk.index[0], j),
                      "dtype": outs[0].dtype,
                      "shape": (tensor_chunk.shape[0], topk),
                      "order": TensorOrder.C_ORDER,
                  },
                  {
                      "index": (tensor_chunk.index[0], j),
                      "dtype": outs[1].dtype,
                      "shape": (tensor_chunk.shape[0], topk),
                      "order": TensorOrder.C_ORDER,
                  },
              ]
              chunk_inputs = [tensor_chunk, chunk_index]
              pk_chunk, distance_chunk = chunk_op.new_chunks(
                  chunk_inputs, kws=chunk_kws
              )
              pk_chunks.append(pk_chunk)
              distance_chunks.append(distance_chunk)
  
          if len(pk_chunks) == 1:
              out_chunks[0].append(pk_chunks[0])
              out_chunks[1].append(distance_chunks[0])
              continue
  
          # combine topk results
          combine_size = options.combine_size
  
          tensor_out_chunks = [pk_chunks, distance_chunks]
          while True:
              chunk_size = ceildiv(
                  len(tensor_out_chunks[0]), combine_size
              )
              cur_out_chunks = [[], []]
              for k in range(chunk_size):
                  to_combine_pks = tensor_out_chunks[0][
                      k * combine_size : (k + 1) * combine_size
                  ]
                  to_combine_distances = tensor_out_chunks[1][
                      k * combine_size : (k + 1) * combine_size
                  ]
  
                  chunk_op = op.copy().reset_key()
                  chunk_op.stage = OperandStage.agg
                  chunk_op._tensor = None
                  chunk_op._index = None
                  agg_chunk_kws = [
                      {
                          "index": (i, 0),
                          "dtype": outs[0].dtype,
                          "shape": (tensor_chunk.shape[0], topk),
                          "order": outs[0].order,
                      },
                      {
                          "index": (i, 0),
                          "dtype": outs[1].dtype,
                          "shape": (tensor_chunk.shape[0], topk),
                          "order": outs[1].order,
                      },
                  ]
                  (
                      pk_result_chunk,
                      distance_result_chunk,
                  ) = chunk_op.new_chunks(
                      to_combine_pks + to_combine_distances,
                      kws=agg_chunk_kws,
                  )
                  cur_out_chunks[0].append(pk_result_chunk)
                  cur_out_chunks[1].append(distance_result_chunk)
              tensor_out_chunks = cur_out_chunks
              if len(tensor_out_chunks[0]) == 1:
                  break
          out_chunks[0].append(tensor_out_chunks[0][0])
          out_chunks[1].append(tensor_out_chunks[1][0])
  
      kws = []
      pk_params = outs[0].params
      pk_params["chunks"] = out_chunks[0]
      pk_params["nsplits"] = (tensor.nsplits[0], (topk,))
      kws.append(pk_params)
      distance_params = outs[1].params
      distance_params["chunks"] = out_chunks[1]
      distance_params["nsplits"] = (tensor.nsplits[0], (topk,))
      kws.append(distance_params)
      new_op = op.copy()
      return new_op.new_tileables(op.inputs, kws=kws)
  
imports: import itertools
  import logging
  import os
  import pickle
  import random
  from hashlib import md5
  from collections import defaultdict
  import numpy
  from ....None import opcodes
  from ....None import tensor
  from ....config import options
  from ....core import ENTITY_TYPE, OutputType, recursive_tile
  from ....core.context import get_context
  from ....core.operand import OperandStage
  from ....lib.filesystem import get_fs, FileSystem
  from ....serialization.serializables import KeyField, StringField, Int32Field, Int64Field, DictField, AnyField, BytesField, BoolField
  from ....tensor.core import TensorOrder
  from ....utils import has_unknown_shape, Timer, ceildiv
  from ...operands import LearnOperand, LearnOperandMixin
  from ..core import proxima, validate_tensor, get_proxima_type
label_desc: Use image node_id 19 for calling the ProximaSearcher obj's underlying member method code with example usage: obj.tile(cls, op) and returns: new_op