zozo123 commented on code in PR #74330:
URL: https://github.com/apache/airflow/pull/74330#discussion_r4432225153
##########
providers/databricks/tests/system/databricks/example_databricks.py:
##########
@@ -157,6 +170,7 @@
databricks_conn_id="databricks_default",
statement="select * from default.my_airflow_table",
warehouse_id=WAREHOUSE_ID,
+ outlets=[UNITY_TABLE.to_asset()],
Review Comment:
This statement is a `SELECT`, so declaring the table as an outlet says this
task writes it. Since this snippet is rendered in the SQL operator docs, please
drop the outlet here (or make it `inlets=`).
##########
providers/databricks/docs/operators/sql_statements.rst:
##########
@@ -109,3 +109,47 @@ An example usage of the ``DatabricksSQLStatementsSensor``
is as follows:
:language: python
:start-after: [START howto_sensor_databricks_sql_statement]
:end-before: [END howto_sensor_databricks_sql_statement]
+
+Explicit assets
+---------------
+
+Use ``outlets`` to declare the Unity Catalog tables that the SQL statement
writes to,
+using
:class:`~airflow.providers.databricks.assets.databricks.UnityTableIdentity`::
+
+ from airflow.providers.databricks.assets.databricks import
UnityTableIdentity
+
+ table = UnityTableIdentity(
+ host="my-workspace.cloud.databricks.com",
+ catalog="main",
+ schema="default",
+ table="my_airflow_table",
+ )
+
+ DatabricksSQLStatementsOperator(
+ task_id="sql_statement",
+ statement="insert into default.my_airflow_table values (3, 'test 3')",
+ warehouse_id=WAREHOUSE_ID,
+ outlets=[
+ table.to_asset()
+ ],
+ )
+
+An outlet is a declared output asset. It does not by itself verify that the
+declared table was changed.
+
+When ``wait_for_termination=False``, successful task completion means that the
+SQL statement was submitted successfully. The resulting asset event does not
+confirm that the table was refreshed. A warning is emitted when outlets are
+configured with ``wait_for_termination=False``.
+
+For deferrable execution, the asset event is emitted only after the remote
+statement completes successfully. Remote failures and cancellations do not
+produce a successful outlet event.
+
+Mapped tasks and templated names
+--------------------------------
+
+For mapped tasks, declare the outlet explicitly in the mapped operator
Review Comment:
Suggest: "`outlets` is not a templated field, so the asset is fixed when the
Dag is parsed. Each successful mapped task instance emits an event for that
same asset. The operator does not read table names from rendered SQL."
##########
providers/databricks/docs/operators/sql_statements.rst:
##########
@@ -109,3 +109,47 @@ An example usage of the ``DatabricksSQLStatementsSensor``
is as follows:
:language: python
:start-after: [START howto_sensor_databricks_sql_statement]
:end-before: [END howto_sensor_databricks_sql_statement]
+
+Explicit assets
+---------------
+
+Use ``outlets`` to declare the Unity Catalog tables that the SQL statement
writes to,
+using
:class:`~airflow.providers.databricks.assets.databricks.UnityTableIdentity`::
+
+ from airflow.providers.databricks.assets.databricks import
UnityTableIdentity
+
+ table = UnityTableIdentity(
+ host="my-workspace.cloud.databricks.com",
+ catalog="main",
+ schema="default",
+ table="my_airflow_table",
+ )
+
+ DatabricksSQLStatementsOperator(
+ task_id="sql_statement",
+ statement="insert into default.my_airflow_table values (3, 'test 3')",
+ warehouse_id=WAREHOUSE_ID,
+ outlets=[
+ table.to_asset()
+ ],
+ )
+
+An outlet is a declared output asset. It does not by itself verify that the
+declared table was changed.
+
+When ``wait_for_termination=False``, successful task completion means that the
+SQL statement was submitted successfully. The resulting asset event does not
+confirm that the table was refreshed. A warning is emitted when outlets are
+configured with ``wait_for_termination=False``.
+
+For deferrable execution, the asset event is emitted only after the remote
Review Comment:
This only covers deferrable. Can you use the same paragraph as run_now.rst /
submit_run.rst ("With synchronous or deferrable execution, task success follows
successful remote completion...") so the three pages match?
##########
providers/databricks/docs/operators/run_now.rst:
##########
@@ -172,6 +172,44 @@ Durable execution requires Airflow 3.3 or newer, since it
relies on the task sta
operator always triggers a fresh run on retry, exactly as before this feature
existed.
+
+Explicit assets
+^^^^^^^^^^^^^^^
+
+Use ``outlets`` to declare the Unity Catalog tables that the job writes to,
+using
:class:`~airflow.providers.databricks.assets.databricks.UnityTableIdentity`::
+
+ from airflow.providers.databricks.assets.databricks import
UnityTableIdentity
+
+ table = UnityTableIdentity(
+ host="my-workspace.cloud.databricks.com",
+ catalog="main",
+ schema="default",
+ table="my_airflow_table",
+ )
+
+ DatabricksRunNowOperator(
+ task_id="run_now",
+ job_id=jobs_create_named.output,
+ outlets=[
+ table.to_asset()
+ ],
+ )
+
+An outlet is a declared output asset and does not automatically verify that the
+job produced a new table commit.
+
+When ``wait_for_termination=False``, a successful task means that the
Databricks
+run was submitted successfully, not that the declared table was refreshed.
+The operator emits a warning when outlets are configured in this mode.
+
+With synchronous or deferrable execution, task success follows successful
remote
+completion. A declared outlet does not prove that a new Delta commit was
written.
+If downstream tasks need the table to be ready, use
``wait_for_termination=True``
+or a Delta table sensor. Remote failure or cancellation does not produce a
Review Comment:
There's no Delta table sensor in the provider yet (#74225 / #74550 are
open). Point to `DatabricksSqlSensor` for now, or drop it until one lands. The
same goes for submit_run.rst.
##########
providers/databricks/docs/operators/sql_statements.rst:
##########
@@ -109,3 +109,47 @@ An example usage of the ``DatabricksSQLStatementsSensor``
is as follows:
:language: python
:start-after: [START howto_sensor_databricks_sql_statement]
:end-before: [END howto_sensor_databricks_sql_statement]
+
+Explicit assets
+---------------
+
+Use ``outlets`` to declare the Unity Catalog tables that the SQL statement
writes to,
+using
:class:`~airflow.providers.databricks.assets.databricks.UnityTableIdentity`::
+
+ from airflow.providers.databricks.assets.databricks import
UnityTableIdentity
+
+ table = UnityTableIdentity(
+ host="my-workspace.cloud.databricks.com",
+ catalog="main",
+ schema="default",
+ table="my_airflow_table",
+ )
+
+ DatabricksSQLStatementsOperator(
+ task_id="sql_statement",
+ statement="insert into default.my_airflow_table values (3, 'test 3')",
Review Comment:
Please make this `main.default.my_airflow_table` so the statement and the
declared `UnityTableIdentity` name the same table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
