def start(self, input_df: Optional[pd.DataFrame] = None):
        """
        Start the run by invoking the main method of the user's app with the input data

        Args:
            input_df (Optional[pd.DataFrame], optional): user provided input dataframe.
        """
        current_status = self.get_status()
        logger.info(f"Current run status: {current_status}")
        if not self._can_start_new_invocation(current_status):
            return f"Cannot start a new invocation when in run status: {current_status}. Valid statuses are: {RunStatus.CREATED}, {RunStatus.INVOCATION_PARTIALLY_COMPLETED}, or {RunStatus.FAILED}."

        if input_df is None:
            logger.info(
                "No input dataframe provided. Fetching input data from source."
            )
            rows = self.run_dao.session.sql(
                f"SELECT * FROM {self.source_info.name}"
            ).collect()
            input_df = pd.DataFrame([row.as_dict() for row in rows])

        dataset_spec = self.source_info.column_spec

        # Preprocess the dataset_spec to create mappings for input columns
        # and map the inputs for reserved fields only once, before the iteration over rows.

        reserved_field_column_mapping = {}

        # Process dataset column spec to handle subscripting logic for input columns
        for reserved_field, user_column in dataset_spec.items():
            reserved_field_column_mapping[reserved_field] = user_column

        input_records_count = len(input_df)

        invocation_metadata_id = self.run_dao._compute_invocation_metadata_id(
            dataset_name=self.source_info.name,
            input_records_count=input_records_count,
        )
        start_time_ms = self._get_current_time_in_ms()

        logger.info(
            f"Creating or updating invocation metadata with {input_records_count} records from input."
        )

        self.run_dao.upsert_run_metadata_fields(
            entry_type=SupportedEntryType.INVOCATIONS.value,
            entry_id=invocation_metadata_id,
            start_time_ms=start_time_ms,
            end_time_ms=0,  # required field
            run_name=self.run_name,
            input_records_count=input_records_count,
            object_name=self.object_name,
            object_type=self.object_type,
            object_version=self.object_version,
        )

        # user app invocation - will block until the app completes
        try:
            for i, row in input_df.iterrows():
                main_method_args = []

                # Call the instrumented main method with the arguments
                # TODO (dhuang) better way to check span attributes, also is this all we need to support?
                input_id = (
                    row[dataset_spec["input_id"]]
                    if "input_id" in dataset_spec
                    else None
                )
                input_col = None
                if input_id is None:
                    if "input" in dataset_spec:
                        input_col = dataset_spec["input"]
                    elif "record_root.input" in dataset_spec:
                        input_col = dataset_spec["record_root.input"]
                    if input_col:
                        input_id = obj_id_of_obj(row[input_col])
                        main_method_args.append(row[input_col])

                ground_truth_output = row.get(
                    dataset_spec.get("ground_truth_output")
                    or dataset_spec.get("record_root.ground_truth_output")
                )

                self.app.instrumented_invoke_main_method(
                    run_name=self.run_name,
                    input_id=input_id,
                    ground_truth_output=ground_truth_output,
                    main_method_args=tuple(
                        main_method_args
                    ),  # Ensure correct order
                    main_method_kwargs=None,  # don't take any kwargs for now so we don't break TruChain / TruLlama where input argument name cannot be defined by users.
                )
        except Exception as e:
            logger.exception(
                f"Error encountered during invoking app main method: {e}."
            )

            self.run_dao.upsert_run_metadata_fields(
                entry_type=SupportedEntryType.INVOCATIONS.value,
                entry_id=invocation_metadata_id,
                start_time_ms=start_time_ms,
                input_records_count=input_records_count,
                end_time_ms=self._get_current_time_in_ms(),
                completion_status=Run.CompletionStatus(
                    status=Run.CompletionStatusStatus.FAILED,
                ).model_dump(),
                run_name=self.run_name,
                object_name=self.object_name,
                object_type=self.object_type,
                object_version=self.object_version,
            )

            raise

        self.tru_session.force_flush()
        logger.info("Run started, invocation done and ingestion in process.")

        Start the run by invoking the main method of the user's app with the input data

        Args:
            input_df (Optional[pd.DataFrame], optional): user provided input dataframe.
        
src/core/trulens/core/run.py