pierrejeambrun commented on code in PR #73441:
URL: https://github.com/apache/airflow/pull/73441#discussion_r4142781517
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst:
##########
@@ -336,6 +336,29 @@ Pass ``{ prefixGroupId: false }`` to keep the ids declared
in a group as written
does in Python; they then have to be unique across the Dag. A group id is made
of letters, digits, dashes and
underscores, and is at most 200 characters.
+Serialization
+~~~~~~~~~~~~~
+
+A native Dag serializes into the same Dag JSON a Python Dag produces, so the
scheduler reads it
+without knowing which language declared it.
+
+``schedule`` accepts what maps to a stock timetable: unset, ``@once``,
``@continuous``, or a cron
+expression. A cron preset such as ``@daily`` is recorded as the expression it
stands for. Anything
+else names a Python object a TypeScript bundle cannot point at, and is
rejected.
+
+Every task of a native Dag runs on the Node coordinator, so it needs the queue
the deployment routes
+there. Set it once on the Dag and each task inherits it:
+
+.. code-block:: typescript
+
+ const dag = new Dag("ts_etl", { schedule: "@daily", queue: "typescript" });
Review Comment:
Can we make typescript 'queue' injected by the Dag constructor if the
`queue` option isn't supplied ?
This way someone having a specific typescript queue / setup can specify it
with:
```
const dag = new Dag("ts_etl", { schedule: "@daily", queue: "typescript"
});
```
But most users will use standard 'typescript' queue, and we can still do
```
const dag = new Dag("ts_etl", { schedule: "@daily"});
```
And Dag constructor will inject the missing `queue: "typescript"`.
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -0,0 +1,729 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Turns a Dag declared in TypeScript into Airflow's DagSerialization v3 JSON —
+// what the Dag processor stores and the scheduler reads. The format is
+// Airflow-internal rather than an SDK schema, so it is reimplemented per
+// language against `airflow-core/src/airflow/serialization/schema.json`; see
+// `airflow-core/adr/lang-sdk/0004-dag-parsing.md` for the field table this
+// follows, and the Java SDK's `Serde.kt` for the same job in another language.
+//
+// Byte-parity with Python's serializer is not the goal — Python omits fields
+// against a `client_defaults` table this SDK does not receive. What has to
hold
+// is that `DagSerialization.from_dict` rebuilds the same Dag, which
+// scripts/ci/lang_sdk_serialization/compare.py checks against Python's own
serialization.
+//
+// This module only produces the payload. Answering a Dag-parsing request with
+// it is the bundle's job, once the coordinator has a parse request to answer.
+
+import { relative as relativePath } from "node:path";
+
+import {
+ DAG_SCHEMA_FIELDS,
+ TASK_SCHEMA_FIELDS,
+ type SchemaField,
+} from "../generated/dag-schema-fields.js";
+import type { JsonValue } from "../sdk/client-types.js";
+import {
+ getDagOrderEdges,
+ getDagTaskGroups,
+ getDagTaskInputs,
+ getDagTaskRecords,
+ isTaskRef,
+ type Dag,
+ type RecordedInputs,
+ type TaskGroupRecord,
+} from "../sdk/dag.js";
+
+/** A serialized Dag: JSON, by the time it reaches the supervisor as msgpack.
*/
+type SerializedValue = JsonValue;
+
+/** Airflow's type/var encoding, as `BaseSerialization.serialize()` emits it.
*/
+interface TypeEncoded {
+ readonly __type: string;
+ readonly __var: SerializedValue;
+}
+
+/**
+ * Identity every TypeScript task carries, in place of the Python operator
class
+ * a Python Dag would name.
+ *
+ * Fixed rather than derived: nothing on the Airflow side imports
`_task_module`
+ * — `SerializedBaseOperator.populate_operator` only compares the pair as
+ * strings when matching plugin extra links — so the pair is free to name the
+ * coordinator that actually runs the task, which makes every TypeScript task
+ * greppable in the UI and the metadata DB.
+ */
+const TASK_TYPE = "TypeScriptOperator";
+const TASK_MODULE = "airflow.sdk.coordinators.node";
+
+/**
+ * Marks the tasks this SDK serialized, as the Java SDK marks its own.
+ *
+ * Nothing in airflow-core reads it today; the `operator` schema definition
+ * allows additional properties, so it rides along as a marker for tooling that
+ * wants to tell language-native tasks apart without parsing `_task_module`.
+ */
+const TASK_LANGUAGE = "typescript";
+
+// Python resolves these from [core]/[scheduler] config when the Dag leaves
them
+// unset, and its serializer always writes the resolved value — there is no
+// schema default to omit against. A bundle cannot read airflow.cfg, so the
+// stock defaults stand in.
+const DAG_CONFIG_FALLBACKS: Readonly<Record<string, SerializedValue>> = {
+ max_active_tasks: 16, // [core] max_active_tasks_per_dag
+ max_active_runs: 16, // [core] max_active_runs_per_dag
+ max_consecutive_failed_dag_runs: 0,
+ catchup: false, // [scheduler] catchup_by_default
+ disable_bundle_versioning: false,
+};
+
Review Comment:
Can't we instead leave this unset, and let the scheduler
`SerializedDAG(dag_id=...)` run the attrs factory which read the config?
because this will create a gap. (`max_active_tasks_per_dag` isn't honored by
TS native dags) otherwise.
--
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]
