class_docstr: Consider FunctionExpression: ObjDetector -> (labels, boxes)
  
  `output`: If the user wants only subset of outputs. Eg,
  ObjDetector.labels the parser with set output to 'labels'
  
  `output_objs`: It is populated by the binder. In case the
  output is None, the binder sets output_objs to list of all
  output columns of the FunctionExpression. Eg, ['labels',
  'boxes']. Otherwise, only the output columns.
  
  FunctionExpression also needs to prepend its alias to all the
  projected columns. This is important as other parts of the query
  might be assessing the results using alias. Eg,
  
  `Select Detector.labels
   FROM Video JOIN LATERAL ObjDetector AS Detector;`
class_docstr_tok: ["Consider","FunctionExpression",":","ObjDetector","-",">","(","labels",",","boxes",")","`","output","`",":","If","the","user","wants","only","subset","of","outputs",".","Eg",",","ObjDetector.labels","the","parser","with","set","output","to","'labels'","`","output_objs","`",":","It","is","populated","by","the","binder",".","In","case","the","output","is","None",",","the","binder","sets","output_objs","to","list","of","all","output","columns","of","the","FunctionExpression",".","Eg",",","[","'labels","'",",","'boxes","'","]",".","Otherwise",",","only","the","output","columns",".","FunctionExpression","also","needs","to","prepend","its","alias","to","all","the","projected","columns",".","This","is","important","as","other","parts","of","the","query","might","be","assessing","the","results","using","alias",".","Eg",",","`","Select","Detector.labels","FROM","Video","JOIN","LATERAL","ObjDetector","AS","Detector",";","`"]
docstr: If cache is not enabled, call the func on the batch and return.
  If cache is enabled:
  (1) iterate over the input batch rows and check if we have the value in the
  cache;
  (2) for all cache miss rows, call the func;
  (3) iterate over each cache miss row and store the results in the cache;
  (4) stitch back the partial cache results with the new func calls.
docstr_tok: ["If","cache","is","not","enabled",",","call","the","func","on","the","batch","and","return",".","If","cache","is","enabled",":","(","1",")","iterate","over","the","input","batch","rows","and","check","if","we","have","the","value","in","the","cache",";","(","2",")","for","all","cache","miss","rows",",","call","the","func",";","(","3",")","iterate","over","each","cache","miss","row","and","store","the","results","in","the","cache",";","(","4",")","stitch","back","the","partial","cache","results","with","the","new","func","calls","."]
code: def _apply_function_expression(
      self, func: Callable, batch: Batch, **kwargs
  ):
      """
      If cache is not enabled, call the func on the batch and return.
      If cache is enabled:
      (1) iterate over the input batch rows and check if we have the value in the
      cache;
      (2) for all cache miss rows, call the func;
      (3) iterate over each cache miss row and store the results in the cache;
      (4) stitch back the partial cache results with the new func calls.
      """
      func_args = Batch.merge_column_wise(
          [child.evaluate(batch, **kwargs) for child in self.children]
      )
  
      if not self._cache:
          return func_args.apply_function_expression(func)
  
      output_cols = [obj.name for obj in self.function_obj.outputs]
  
      # 1. check cache
      # We are required to iterate over the batch row by row and check the cache.
      # This can hurt performance, as we have to stitch together columns to generate
      # row tuples. Is there an alternative approach we can take?
  
      results = np.full([len(batch), len(output_cols)], None)
      cache_keys = func_args
      # cache keys can be different from func_args
      # see optimize_cache_key
      if self._cache.key:
          cache_keys = Batch.merge_column_wise(
              [
                  child.evaluate(batch, **kwargs)
                  for child in self._cache.key
              ]
          )
          assert len(cache_keys) == len(
              batch
          ), "Not all rows have the cache key"
  
      cache_miss = np.full(len(batch), True)
      for idx, (_, key) in enumerate(cache_keys.iterrows()):
          val = self._cache.store.get(key.to_numpy())
          results[idx] = val
          cache_miss[idx] = val is None
  
      # log the cache misses
      self._stats.cache_misses += sum(cache_miss)
  
      # 2. call func for cache miss rows
      if cache_miss.any():
          func_args = func_args[list(cache_miss)]
          cache_miss_results = func_args.apply_function_expression(func)
  
          # 3. set the cache results
          missing_keys = cache_keys[list(cache_miss)]
          for key, value in zip(
              missing_keys.iterrows(), cache_miss_results.iterrows()
          ):
              self._cache.store.set(
                  key[1].to_numpy(), value[1].to_numpy()
              )
  
          # 4. merge the cache results
          results[cache_miss] = cache_miss_results.to_numpy()
  
      # 5. return the correct batch
      return Batch(pd.DataFrame(results, columns=output_cols))
  
imports: from dataclasses import dataclass
  from typing import Callable, List, Tuple
  import numpy
  import pandas
  from evadb.catalog.models.function_catalog import FunctionCatalogEntry
  from evadb.catalog.models.function_io_catalog import FunctionIOCatalogEntry
  from evadb.constants import NO_GPU
  from evadb.executor.execution_context import Context
  from evadb.expression.abstract_expression import AbstractExpression, ExpressionType
  from evadb.functions.gpu_compatible import GPUCompatible
  from evadb.models.storage.batch import Batch
  from evadb.parser.alias import Alias
  from evadb.utils.kv_cache import DiskKVCache
  from evadb.utils.logging_manager import logger
  from evadb.utils.stats import FunctionStats
label_desc: Use image node_id 13 for calling the FunctionExpression obj's underlying member method code with example usage: obj._apply_function_expression(func, batch) and returns: Batch, func_args