code: def call(self, step):
      # set row_id
      data = self.steps_data[step.dataframe.step_num]
  
      params = step.params or {}
  
      for table in data.get_tables()[
          :1
      ]:  # add  __mindsdb_row_id only for first table
          row_id_col = Column(
              name="__mindsdb_row_id",
              database=table["database"],
              table_name=table["table_name"],
              table_alias=table["table_alias"],
          )
  
          row_id = self.context.get("row_id")
          values = list(range(row_id, row_id + data.length()))
          data.add_column(row_id_col, values)
          self.context["row_id"] += data.length()
  
      project_name = step.namespace
      predictor_name = step.predictor.parts[0]
  
      where_data = data.get_records()
  
      # add constants from where
      row_dict = {}
      if step.row_dict is not None:
          for k, v in step.row_dict.items():
              if isinstance(v, Result):
                  prev_result = self.steps_data[v.step_num]
                  # TODO we await only one value: model.param = (subselect)
                  v = prev_result.get_records_raw()[0][0]
              row_dict[k] = v
  
          for record in where_data:
              record.update(row_dict)
  
      predictor_metadata = {}
      for pm in self.context["predictor_metadata"]:
          if (
              pm["name"] == predictor_name
              and pm["integration_name"].lower() == project_name
          ):
              predictor_metadata = pm
              break
      is_timeseries = predictor_metadata["timeseries"]
      _mdb_forecast_offset = None
      if is_timeseries:
          if "> LATEST" in self.context["query_str"]:
              # stream mode -- if > LATEST, forecast starts on inferred next timestamp
              _mdb_forecast_offset = 1
          elif "= LATEST" in self.context["query_str"]:
              # override: when = LATEST, forecast starts on last provided timestamp instead of inferred next time
              _mdb_forecast_offset = 0
          else:
              # normal mode -- emit a forecast ($HORIZON data points on each) for each provided timestamp
              params["force_ts_infer"] = True
              _mdb_forecast_offset = None
          for row in where_data:
              if "__mdb_forecast_offset" not in row:
                  row["__mdb_forecast_offset"] = _mdb_forecast_offset
  
      # for row in where_data:
      #     for key in row:
      #         if isinstance(row[key], datetime.date):
      #             row[key] = str(row[key])
  
      table_name = get_preditor_alias(step, self.context["database"])
      result = ResultSet()
      result.is_prediction = True
  
      project_datanode = self.session.datahub.get(project_name)
      if len(where_data) == 0:
          cols = project_datanode.get_table_columns(predictor_name) + [
              "__mindsdb_row_id"
          ]
          for col in cols:
              result.add_column(
                  Column(
                      name=col,
                      database=table_name[0],
                      table_name=table_name[1],
                      table_alias=table_name[2],
                  )
              )
      else:
          predictor_id = predictor_metadata["id"]
          key = f"{predictor_name}_{predictor_id}_{json_checksum(where_data)}"
  
          if self.session.predictor_cache is False:
              data = None
          else:
              predictor_cache = get_cache("predict")
              data = predictor_cache.get(key)
  
          if data is None:
              version = None
              if (
                  len(step.predictor.parts) > 1
                  and step.predictor.parts[-1].isdigit()
              ):
                  version = int(step.predictor.parts[-1])
              predictions = project_datanode.predict(
                  model_name=predictor_name,
                  data=where_data,
                  version=version,
                  params=params,
              )
              data = predictions.to_dict(orient="records")
              columns_dtypes = dict(predictions.dtypes)
  
              if (
                  data is not None
                  and isinstance(data, list)
                  and self.session.predictor_cache is not False
              ):
                  predictor_cache.set(key, data)
          else:
              columns_dtypes = {}
          if len(data) > 0:
              cols = list(data[0].keys())
              for col in cols:
                  result.add_column(
                      Column(
                          name=col,
                          table_name=table_name[1],
                          table_alias=table_name[2],
                          database=table_name[0],
                          type=columns_dtypes.get(col),
                      )
                  )
          # apply filter
          if is_timeseries:
              data = self.apply_ts_filter(
                  data, where_data, step, predictor_metadata
              )
          result.add_records(data)
  
      return result
  
file_path: mindsdb/mindsdb/api/executor/sql_query/steps/apply_predictor_step.py
imports: import datetime
  import re
  import dateinfer
  import pandas
  from mindsdb_sql.parser.ast import BinaryOperation, Identifier, Constant, Latest, BetweenOperation, Parameter
  from mindsdb_sql.planner.step_result import Result
  from mindsdb_sql.planner.steps import ApplyTimeseriesPredictorStep, ApplyPredictorRowStep, ApplyPredictorStep
  from mindsdb.api.executor.sql_query.result_set import ResultSet, Column
  from mindsdb.utilities.cache import get_cache, json_checksum
  from .base import BaseStepCall
label_desc: Use image node_id 1 for calling the ApplyPredictorStepCall obj's underlying member method code with example usage: obj.call(step) and returns: result