code: def _handle_deferrable_databricks_operator_execution(
      operator, hook, log, context
  ) -> None:
      """
      Handle the Airflow + Databricks lifecycle logic for deferrable Databricks operators.
  
      :param operator: Databricks async operator being handled
      :param context: Airflow context
      """
      job_id = hook.get_job_id(operator.run_id)
      if operator.do_xcom_push and context is not None:
          context["ti"].xcom_push(key=XCOM_JOB_ID_KEY, value=job_id)
      if operator.do_xcom_push and context is not None:
          context["ti"].xcom_push(
              key=XCOM_RUN_ID_KEY, value=operator.run_id
          )
      log.info("Run submitted with run_id: %s", operator.run_id)
  
      run_page_url = hook.get_run_page_url(operator.run_id)
      if operator.do_xcom_push and context is not None:
          context["ti"].xcom_push(
              key=XCOM_RUN_PAGE_URL_KEY, value=run_page_url
          )
      log.info(
          "View run status, Spark UI, and logs at %s", run_page_url
      )
  
      if operator.wait_for_termination:
          operator.defer(
              trigger=DatabricksExecutionTrigger(
                  run_id=operator.run_id,
                  databricks_conn_id=operator.databricks_conn_id,
                  polling_period_seconds=operator.polling_period_seconds,
                  retry_limit=operator.databricks_retry_limit,
                  retry_delay=operator.databricks_retry_delay,
                  retry_args=operator.databricks_retry_args,
                  run_page_url=run_page_url,
              ),
              method_name=DEFER_METHOD_NAME,
          )
  
imports: from __future__ import annotations
  import time
  import warnings
  from functools import cached_property
  from logging import Logger
  from typing import TYPE_CHECKING, Any, Sequence
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.models import BaseOperator, BaseOperatorLink, XCom
  from airflow.providers.databricks.hooks.databricks import DatabricksHook, RunState
  from airflow.providers.databricks.triggers.databricks import DatabricksExecutionTrigger
  from airflow.providers.databricks.utils.databricks import normalise_json_content, validate_trigger_event
label_desc: Use image node_id 2 for calling a global function with example usage: _handle_deferrable_databricks_operator_execution(operator, hook, log, context) without return types