jason810496 commented on code in PR #74674:
URL: https://github.com/apache/airflow/pull/74674#discussion_r4412668151
##########
go-sdk/sdk/context.go:
##########
@@ -85,13 +87,22 @@ type TaskInstance struct {
}
// DagRun identifies the Dag run the current task instance belongs to and
-// carries its scheduling timestamps. The *time.Time fields are nil when the
-// supervisor did not provide a value (for example, a manually triggered run
-// without a logical date).
+// carries its scheduling timestamps, how it was created, and its conf. The
+// *time.Time fields are nil when the supervisor did not provide a value (for
+// example, a manually triggered run without a logical date).
type DagRun struct {
DagID string
RunID string
LogicalDate *time.Time
DataIntervalStart *time.Time
DataIntervalEnd *time.Time
+ RunAfter time.Time
+ // RunType says how the run was created, such as
+ // [github.com/apache/airflow/go-sdk/airflow.DagRunTypeManual].
+ RunType genmodels.DagRunType
Review Comment:
We shouldn't expose`genmodels.DagRunType` to the users.
IIUC, we could declare `type DagRunType = genmodels.DagRunType` in `sdk` too
and type the field `DagRunType`. Godoc then reads sdk.DagRunType, and since
everything is one alias chain, `actx.DagRun().RunType ==
airflow.DagRunTypeManual` still compiles.
##########
go-sdk/pkg/execution/task_runner.go:
##########
@@ -125,6 +133,58 @@ func RunTask(
return executeTask(ctx, task, args, details.TIContext.ShouldRetry,
logger)
}
+func normalizeDagRunConf(conf map[string]any) map[string]any {
+ if conf == nil {
+ return nil
+ }
+ normalized := make(map[string]any, len(conf))
+ for key, value := range conf {
+ normalized[key] = normalizeDagRunConfValue(value)
+ }
+ return normalized
+}
+
+func normalizeDagRunConfValue(value any) any {
Review Comment:
Yes, it makes sense to have a follow-up to move it to decode layer, thanks.
##########
go-sdk/sdk/context.go:
##########
@@ -85,13 +87,22 @@ type TaskInstance struct {
}
// DagRun identifies the Dag run the current task instance belongs to and
-// carries its scheduling timestamps. The *time.Time fields are nil when the
-// supervisor did not provide a value (for example, a manually triggered run
-// without a logical date).
+// carries its scheduling timestamps, how it was created, and its conf. The
+// *time.Time fields are nil when the supervisor did not provide a value (for
+// example, a manually triggered run without a logical date).
type DagRun struct {
DagID string
RunID string
LogicalDate *time.Time
DataIntervalStart *time.Time
DataIntervalEnd *time.Time
+ RunAfter time.Time
+ // RunType says how the run was created, such as
+ // [github.com/apache/airflow/go-sdk/airflow.DagRunTypeManual].
+ RunType genmodels.DagRunType
+ // Conf is the configuration the run was triggered with. It may be nil
or
+ // empty when no configuration was supplied. The runtime copies its
nested
+ // maps and slices from the startup details. Integers are int64, except
values
+ // above math.MaxInt64, which are uint64; floating-point values are
float64.
+ Conf map[string]any
Review Comment:
We should rephrase the "may be nil or empty map" from the user doc and the
docstring here.
Since the `Conf` will never be nil from the server side IIUC.
https://github.com/apache/airflow/blob/8401a190fa8174e8089914408fe82797a202f71b/airflow-core/src/airflow/models/dagrun.py#L528
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst:
##########
@@ -467,14 +467,21 @@ Its first parameter is an ``airflow.Context``, which
gives it what Airflow provi
- ``DagID``, ``RunID``, ``TaskID`` (including any task group prefix),
``MapIndex`` (``nil`` for an unmapped
task) and ``TryNumber``.
* - ``actx.DagRun()``
- - ``DagID``, ``RunID``, and the ``*time.Time`` fields ``LogicalDate``,
``DataIntervalStart`` and
- ``DataIntervalEnd`` (``nil`` when the run has no such value, such as a
manual run).
+ - ``DagID``, ``RunID``, ``RunAfter``, ``RunType`` (an
``airflow.DagRunType`` such as
+ ``airflow.DagRunTypeManual``), ``Conf`` (may be ``nil`` or empty when
no configuration was supplied), and
+ the ``*time.Time`` fields ``LogicalDate``, ``DataIntervalStart`` and
``DataIntervalEnd`` (``nil`` when the
+ run has no such value, such as a manual run).
``airflow.Context`` is a ``context.Context``, so pass it to a client call or
to ``http.NewRequestWithContext``. Its
``Done()`` fires when Airflow stops the task, so long-running work can stop
cleanly. Cleanup that must outlive
that runs under ``context.WithoutCancel(actx)``. A helper that takes a plain
``context.Context`` gets the
``airflow.Context`` back with ``airflow.FromContext``.
+``Conf`` is a ``map[string]any``. Its integers are ``int64``, except positive
values above ``math.MaxInt64``, which
+are ``uint64``; floating-point values are ``float64``. This also applies to
values in nested maps and arrays. The
+runtime copies these containers from the startup details. Use a map lookup or
``len`` to check for configuration
+entries; a run without configuration may have a ``nil`` or empty map.
Review Comment:
We need to fix the "may have a ``nil`` or empty map." statement and perhaps
no need to mention that much detail for the value normalization here. Let's
keep it in the independent section (or consolidate into the XCom value
section).
--
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]
