dijiekstra opened a new issue, #18627:
URL: https://github.com/apache/dolphinscheduler/issues/18627
# DolphinScheduler Lakehouse Asset Snapshot Event-Driven Scheduling Design
Document
> Status: Draft / RFC
> Author: Dolphin Agent
> Related modules: dolphinscheduler-dao, dolphinscheduler-service,
dolphinscheduler-master, dolphinscheduler-api, dolphinscheduler-task-plugin
## 1. Background and Goals
### 1.1 Background
Modern CDC Lakehouse architectures (based on table formats such as Paimon /
Iceberg / Hudi) are driving the evolution of data warehouse modeling and data
platform scheduling paradigms:
- Data warehouse modeling is evolving from a single “table” into a
three-layer semantic model of “entity state table (latest-state) + event fact
table (event fact) + change history table (changelog),” relying on capabilities
provided by table formats such as Snapshot, Manifest, Primary Key Table,
Changelog, Time Travel/Tag, and Schema Evolution.
- The scheduling paradigm is evolving from a “time-based DAG” to “data asset
state scheduling”: in other words, scheduling is no longer triggered only by
cron time points, but by whether a data asset has advanced to a usable version.
Typical scenarios include:
- Trigger downstream processing after a snapshot is generated
- Trigger windowed computation output after a watermark arrives
- Allow metric publication only after a quality check passes
- Trigger financial statement posting after a tag is solidified
- Trigger downstream model upgrades after schema change approval passes
- Trigger recomputation and reconciliation after backfill completes
### 1.2 Goals
Without breaking DolphinScheduler’s existing scheduling semantics (cron
scheduling, manual runs, backfill, failure recovery, DEPENDENT tasks, and
normal DAG execution), add “data asset version advancement event-driven
scheduling” capabilities to achieve:
```text
湖仓表 snapshot/instant 推进
-> 事件被感知并持久化为 AssetEvent
-> 更新资产的当前状态 AssetState（latest snapshot / watermark / quality / schema）
-> 按工作流/任务声明的资产依赖条件进行匹配评估
-> 依赖满足后，通过 Command 机制触发/释放 DolphinScheduler 工作流或任务实例
-> 执行、观测、失败重试与补偿
```
This design follows the following non-negotiable principles:
1. It must not be “trigger when the event arrives”; it must be “trigger when
the asset state reaches the condition.”
2. The same version / same dependency combination can only trigger once
(idempotency).
3. Events may be duplicated, out of order, delayed, or lost. Asset state is
the basis for scheduling; events are only input evidence.
4. Existing infrastructure such as Command / CommandType / DEPENDENT tasks /
task instance state machines should be reused as much as possible. Do not build
an independent scheduling kernel.
5. In phase one, only implement a small, verifiable closed loop (Paimon
snapshot -> AssetEvent -> AssetState -> dependency matching -> Command
trigger), then extend to Iceberg/Hudi and multi-asset AND/OR combinations later.
### 1.3 Applicability Boundaries and Problem Definition
This design targets **next-generation data warehouse architectures** (based
on Lakehouse table formats such as Paimon/Iceberg/Hudi), whose core
characteristics are:
- **The entire DAG chain is driven by snapshot versions**: from ODS (data
ingestion into the Lakehouse) to DWD (dimensional processing) to DWS
(subject-area aggregation) to ADS (serving/output), each layer is the result of
data asset version advancement in the previous layer, eliminating the mismatch
between time and data availability.
- **No reliance on cron or time expressions**: the trigger for any task in
the DAG is based on its dependent assets (snapshot/watermark/quality/tag), not
on “a fixed point in time.”
- **A downstream task is triggered only when multiple upstream assets are
ready together**: for example, a task in DWS may depend on snapshots of
multiple DWD tables having advanced, and these dependencies are expressed
through snapshot version combinations rather than time windows.
- **Keep orchestration capabilities such as cross-flow dependencies**: if a
task in the DAG declares cross-flow dependencies (depending on tasks in other
workflows/projects), those dependencies should also be upgraded to a
snapshot-based model to preserve the data-driven nature of overall scheduling.
**Applicable scenarios**: any end-to-end data processing workflow in which
ODS/DWD/DWS/ADS is carried by Lakehouse tables.
**Not applicable**: traditional data warehouses still using “batch
processing at fixed times” (even if already migrated to the cloud).
## 2. Current-State Analysis: Existing Event/Dependency Trigger
Infrastructure in DolphinScheduler
Based on an actual inspection of the codebase (version: current master
branch), the following reusable capabilities already exist:
### 2.1 CommandType / Command (`dolphinscheduler-dao`,
`dolphinscheduler-common`)
- `org.apache.dolphinscheduler.common.enums.CommandType` (file:
`dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java`)
already defines multiple trigger source types such as `START_PROCESS`,
`SCHEDULER`, `COMPLEMENT_DATA`, and `RECOVER_TOLERANCE_FAULT_PROCESS` (the
comment already contains `// todo: rename to WorkflowTriggerType`, indicating
that Command is already treated internally as an abstraction of a “trigger
type,” not just a “start action”).
- `org.apache.dolphinscheduler.dao.entity.Command`
(`dolphinscheduler-dao/.../entity/Command.java`) is the persisted entity mapped
to table `t_ds_command`, with key fields including `commandType`,
`workflowDefinitionCode`, `workflowDefinitionVersion`, `commandParam`
(JSON-formatted startup parameters), and `workflowInstancePriority`.
- `org.apache.dolphinscheduler.service.command.CommandService`
(`dolphinscheduler-service/.../command/CommandService.java`) exposes `int
createCommand(Command command)`, which is currently the unified entry point for
“initiating one workflow instance run.” On the Master side, the
`CommandService`/scanner consumes the `t_ds_command` table to create
WorkflowInstance records.
**Conclusion**: we do not need to build a parallel execution path for
“event-triggered workflows.” Add a new `CommandType.ASSET_EVENT_TRIGGER` (or
reuse `START_PROCESS` while marking the trigger source in `commandParam`), and
write into `t_ds_command` through the existing `CommandService#createCommand`.
This fully reuses the existing Master-side Command consumption,
WorkflowInstance creation, and DAG execution chain. This is the lowest-risk
integration point.
### 2.2 DEPENDENT Task (task-level dependency waiting)
- `DependentType`, `DependentRelation`
(`dolphinscheduler-task-plugin/dolphinscheduler-task-api/.../enums/DependentType.java`,
`DependentRelation.java`): the current DEPENDENT task only supports
dependencies on “whether another workflow/task succeeded within a certain
cycle” (`DependentItem` contains `projectCode` / `definitionCode` /
`depTaskCode` / `cycle` / `dateValue` / `dependResult`), and it is **oriented
toward task instance execution results, not external data asset state**.
- `DependentParameters` (same directory,
`parameters/DependentParameters.java`) supports `DependentRelation` (AND/OR)
for combining multiple `DependentTaskModel` entries. This structure for
“evaluating multiple dependency items with AND/OR combinations” can be directly
borrowed for expressing combinations of asset dependencies.
- `DependentLogicTask` / `DependentTaskTracker`
(`dolphinscheduler-master/.../executor/plugin/dependent/`): after a task
instance enters the execution state, `DependentTaskTracker` polls for
dependency completion, and `getDependentTaskStatus()` returns
`TaskExecutionStatus`. In essence, this means “a task instance occupies an
execution slot and internally performs polling waits,” rather than using event
callback-driven execution.
**Conclusion**: the “multiple dependency AND/OR combination expression”
design from DEPENDENT tasks is worth reusing (specifically
`DependentRelation`), but its “dependency object” is task execution result
rather than external data asset version. Therefore, its `DependentItem` /
`DependentTaskTracker` implementation cannot be reused directly. A new
asset-oriented dependency model and evaluator must be added. Its execution form
can, however, be used as a reference for a new task type `ASSET_SENSOR`
(Strategy A, described later), reusing its “occupy task instance slot + poll
for condition + continue downstream after success” running mode and state
machine (`onTaskRunning` / `onTaskPaused` / `onTaskKilled`, etc., whose
lifecycle hooks already exist in `AbstractLogicTask`).
### 2.3 Missing Capabilities (need to be added)
- There is no first-class “Asset” model in the style of Airflow Datasets;
there is no table such as `t_ds_asset`.
- There is no unified ingestion layer for external events (messages /
callbacks / polling); the current `dolphinscheduler-extract` module mainly
contains internal RPC contracts between Master/Worker/API (`extract-master`,
`extract-worker`, `extract-alert`, etc.), not event ingestion interfaces for
external Lakehouse systems.
- There is no “event deduplication + idempotent trigger history” table. The
current `t_ds_command` table has no unique constraint preventing duplicate
Command insertion for the same (workflow, trigger condition) pair. Existing
business-level controls are implemented through scheduler/manual trigger points
and do not cover “externally event-driven” scenarios.
- There are no fields or state machines representing Lakehouse semantics
such as watermark / snapshot / quality / schema.
In summary, the **overall strategy** is to add three logical
modules—`event-source`, `asset-state`, and `dependency-resolver`—plus the
corresponding DAO tables. On the trigger side, integrate with the existing
`CommandService.createCommand` (Strategy C, directly creating workflow
instances) and add a new `ASSET_SENSOR` task type (Strategy A, task-level
waiting that reuses embedded DAG waiting semantics). These two strategies
should be implemented in phases, without invasive modifications to the Master
scheduling kernel or task state machine.
## 3. Overall Architecture Design
### 3.1 End-to-End Flow
```text
┌───────────────┐ ┌────────────────┐ ┌───────────────┐
┌─────────────────────┐ ┌───────────────┐
│ Lakehouse │ │ Event Ingestion│ │ Asset State │ │ Dependency
Resolver │ │ Scheduler │
│ (Paimon/Iceberg│──▶│ (Poll Scanner /│──▶│ Store │──▶│ (Match
AssetDependency│──▶│ Adapter │
│ /Hudi commit) │ │ Report API) │ │ (t_ds_asset_ │ │ ->
READY/BLOCKED) │ │ (CommandService│
└───────────────┘ └────────────────┘ │ state) │
└─────────────────────┘ │ / ASSET_SENSOR│
└───────────────┘
│ Task) │
└───────┬────────┘
▼
Master 正常 Command/DAG 执行链路
```
### 3.2 Domain Model of Asset and Asset Event
```java
// 资产唯一标识：catalog.database.table[/partition]
public class AssetIdentifier {
private String assetKey; // 规范化后的唯一键，例如
paimon://catalog/db/table/dt=2024-05-20
private String storageFormat; // PAIMON / ICEBERG / HUDI
private String catalogName;
private String databaseName;
private String tableName;
private String partitionExpr; // 可为空，支持分区级资产
}
// 原始事件，来源可为轮询扫描或外部主动上报，仅作为“输入证据”，不可直接触发调度
public class AssetEvent {
private String eventId; // 幂等键之一，来源系统内可重复计算得到（见 4.2）
private String assetKey;
private String eventType; // SNAPSHOT_COMMITTED /
WATERMARK_ADVANCED / TAG_CREATED /
// QUALITY_CHECK_PASSED /
QUALITY_CHECK_FAILED /
// SCHEMA_CHANGE_APPROVED /
BACKFILL_COMPLETED / CDC_LAG_RECOVERED
private Long snapshotId; // Paimon snapshot-id / Iceberg
snapshot-id / Hudi instant time 映射值
private Long schemaId;
private Long watermark; // epoch millis
private String commitKind; // APPEND / COMPACT / OVERWRITE / ...
private Long commitTime;
private Long deltaRecordCount;
private Boolean schemaChanged;
private String payload; // 原始 JSON，便于排障，不参与判定逻辑
private Long receiveTime;
}
// 资产当前状态：调度判定的唯一事实来源（Source of Truth）
public class AssetState {
private String assetKey;
private Long latestSnapshotId;
private Long latestSchemaId;
private Long latestWatermark;
private String latestTag;
private String qualityStatus; // UNKNOWN / PASSED / FAILED
private String schemaStatus; // UNKNOWN / COMPATIBLE /
PENDING_APPROVAL / REJECTED
private String backfillStatus; // NONE / RUNNING / COMPLETED
private Long updateTime;
private Long version; // 乐观锁，防止并发覆盖（见 4.4）
}
```
### 3.3 Event Ingestion Design
The two methods should coexist in parallel and compensate for each other. In
the MVP phase, prioritize method one:
**Method 1: Incremental Polling Scanner (Polling Scanner, priority
implementation)**
- Independently deploy a lightweight `AssetEventScanner` (either as a
background task inside DolphinScheduler or as an independent process /
scheduled API task) to periodically execute the following against registered
Paimon tables:
```sql
SELECT snapshot_id, schema_id, commit_user, commit_identifier,
commit_kind, commit_time,
watermark, total_record_count, delta_record_count,
changelog_record_count
FROM catalog_name.database_name.`table_name$snapshots`
WHERE snapshot_id > ?
ORDER BY snapshot_id;
```
Here `?` comes from `last_scanned_snapshot_id` recorded in
`t_ds_asset_event_consumer_offset`.
- Advantages: no need to modify the writers; existing Paimon/Iceberg tables
can be connected directly; it naturally provides “compensation” because it is
inherently a pull model that can replay the full history.
- Disadvantages: there is delay introduced by the scan interval (typically
on the order of 10 seconds to 1 minute).
**Method 2: Active Reporting by Writers (Push, for low-latency scenarios,
introduced in a later phase)**
- After a successful commit, Flink/Spark writing jobs call a new REST
endpoint `POST /dolphinscheduler/asset-events` to actively report events (see
Section 4.6 API design).
- Advantages: low latency (second-level).
- Disadvantages: requires changes to application code on the writer side;
and it must coexist with polling scanner as a fallback, otherwise missed
reports cannot be detected.
**Conclusion (trade-off)**: in the MVP phase, implement only Method 1
(Polling Scanner), because it is zero-intrusion to existing Flink/Paimon
production pipelines, independently verifiable, and naturally compensating.
Method 2 is a phase-two enhancement. Both methods share the same processing
chain—“event deduplication -> state update -> dependency evaluation”—with only
the source type (`sourceType=POLL` / `sourceType=PUSH`) differing.
Deduplication is handled naturally by the unique constraint on
`t_ds_asset_event`, so the two paths will not cause duplicate triggers.
### 3.4 Database Table Design (DDL draft)
Add 5 new tables, all using the `t_ds_` prefix to align with existing naming
conventions, and place them under new/upgrade SQL directories in
`dolphinscheduler-dao` (following the organization of existing upgrade scripts
such as
`dolphinscheduler-dao/src/main/resources/sql/upgrade/<version>_schema/{mysql,postgresql}/dolphinscheduler_ddl.sql`).
```sql
-- 资产注册表：描述一个可被依赖的湖仓数据资产
CREATE TABLE t_ds_asset (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
asset_key VARCHAR(512) NOT NULL, --
catalog.db.table[/partition_expr] 规范化后的唯一键
storage_format VARCHAR(32) NOT NULL, -- PAIMON / ICEBERG / HUDI
catalog_name VARCHAR(128) NOT NULL,
database_name VARCHAR(128) NOT NULL,
table_name VARCHAR(128) NOT NULL,
partition_expr VARCHAR(256),
description VARCHAR(512),
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
UNIQUE KEY uk_asset_key (asset_key)
);
-- 原始事件表：仅作为证据留存与排障，不作为调度判定依据
CREATE TABLE t_ds_asset_event (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
event_id VARCHAR(128) NOT NULL,
source_type VARCHAR(16) NOT NULL, -- POLL / PUSH
asset_key VARCHAR(512) NOT NULL,
event_type VARCHAR(32) NOT NULL, -- SNAPSHOT_COMMITTED /
WATERMARK_ADVANCED / ...
snapshot_id BIGINT,
schema_id BIGINT,
watermark BIGINT,
commit_kind VARCHAR(32),
commit_time BIGINT,
delta_record_count BIGINT,
schema_changed TINYINT,
payload TEXT,
receive_time BIGINT NOT NULL,
create_time DATETIME NOT NULL,
UNIQUE KEY uk_dedup (source_type, asset_key, event_type, snapshot_id),
KEY idx_asset_snapshot (asset_key, snapshot_id),
KEY idx_event_type_time (event_type, receive_time)
);
-- 资产状态表：调度判定的唯一事实来源
CREATE TABLE t_ds_asset_state (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
asset_key VARCHAR(512) NOT NULL,
latest_snapshot_id BIGINT,
latest_schema_id BIGINT,
latest_watermark BIGINT,
latest_tag VARCHAR(128),
quality_status VARCHAR(16) DEFAULT 'UNKNOWN',
schema_status VARCHAR(32) DEFAULT 'UNKNOWN',
backfill_status VARCHAR(16) DEFAULT 'NONE',
version BIGINT NOT NULL DEFAULT 0, -- 乐观锁
update_time DATETIME NOT NULL,
UNIQUE KEY uk_asset_key (asset_key)
);
-- 工作流/任务对资产的依赖声明
CREATE TABLE t_ds_asset_dependency (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
workflow_definition_code BIGINT NOT NULL,
task_definition_code BIGINT, -- 为空表示对整个工作流生效（Strategy C）
asset_key VARCHAR(512) NOT NULL,
dependency_group VARCHAR(64) NOT NULL DEFAULT 'default', --
支持多资产 AND 分组
condition_json TEXT NOT NULL, -- 见 3.6 依赖表达式
enabled TINYINT NOT NULL DEFAULT 1,
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
KEY idx_asset_key (asset_key),
KEY idx_workflow (workflow_definition_code)
);
-- 幂等触发历史 + 审计
CREATE TABLE t_ds_asset_trigger_history (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
trigger_key VARCHAR(256) NOT NULL, -- 幂等键，见 4.1
event_id VARCHAR(128),
asset_key VARCHAR(512) NOT NULL,
workflow_definition_code BIGINT NOT NULL,
workflow_instance_id BIGINT,
task_definition_code BIGINT,
task_instance_id BIGINT,
trigger_status VARCHAR(16) NOT NULL, -- TRIGGERING / SUCCESS
/ FAILED / SKIPPED
reason VARCHAR(512),
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
UNIQUE KEY uk_trigger_key (trigger_key),
KEY idx_status_time (trigger_status, create_time)
);
-- 事件消费位点（用于 Polling Scanner 的断点续扫）
CREATE TABLE t_ds_asset_event_consumer_offset (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
source_type VARCHAR(16) NOT NULL,
asset_key VARCHAR(512) NOT NULL,
consumer_group VARCHAR(64) NOT NULL DEFAULT 'default',
last_snapshot_id BIGINT,
last_scan_time DATETIME,
UNIQUE KEY uk_offset (source_type, asset_key, consumer_group)
);
```
### 3.5 Scheduling Trigger Flow (integrated with the existing Command
mechanism)
```text
1. AssetEventScanner 定时扫描 Paimon $snapshots -> 产出候选 AssetEvent
2. 写入 t_ds_asset_event（利用唯一约束 uk_dedup 做插入去重，冲突则说明该事件已处理过，直接跳过）
3. 事件写入成功后，进入 AssetState 更新：
- 读取当前 t_ds_asset_state（带 version）
- 若 event.snapshotId <= state.latestSnapshotId，丢弃（乱序保护，见 4.3）
- 否则以乐观锁 UPDATE ... WHERE asset_key=? AND version=? 更新状态，version+1
4. 状态更新成功后，触发 DependencyResolver：
- 查询该 asset_key 关联的 t_ds_asset_dependency
- 对每条依赖，按 condition_json
求值（快照存在/watermark>=X/quality=PASSED/schema=COMPATIBLE）
- 对同一 dependency_group 下的多条依赖做 AND 聚合（同一工作流等待多个上游资产都就绪）
5. 若判定为 READY：
a. 计算 trigger_key（见 4.1），向 t_ds_asset_trigger_history 做唯一插入
- 插入成功 => 本次获得触发权，继续步骤 b
- 插入失败（唯一冲突）=> 说明已被其他并发的 Master/Scanner 实例触发过，直接返回，不重复触发
b. 根据依赖类型选择触发方式：
- Strategy C（工作流级）：调用现有 CommandService#createCommand，
使用新增 CommandType（如 ASSET_EVENT_TRIGGER，或复用 START_PROCESS 并在
commandParam 中标记
triggerSource=ASSET_EVENT、assetKey、snapshotId 等审计信息），
由 Master 侧既有 Command 消费链路完成 WorkflowInstance 创建与 DAG 执行，不改动 Master
内核。
- Strategy A（任务级）：若工作流已在运行且包含 ASSET_SENSOR 任务，
则更新对应任务实例的依赖满足标记，由 ASSET_SENSOR 任务（运行时轮询/订阅 AssetState 变化）
自行判定成功，走入现有任务状态机（复用 DependentLogicTask 同款生命周期钩子）。
c. 在真正触发前执行"原生依赖门禁校验"：
- 若用户配置了必须先满足的原生依赖（上游任务成功、DEPENDENT 条件、运行窗口限制），则先校验通过再触发；
- 仅在显式配置为 snapshot 直驱模式时，可由资产版本推进直接作为主触发条件。
d. 触发结果（成功/失败）回写 t_ds_asset_trigger_history.trigger_status，
失败的记录由 Compensation Scanner 定期扫描重试。
6. Master 按照现有正常链路执行 WorkflowInstance / TaskInstance，无需感知事件来源。
```
### 3.6 Dependency Expression (structured configuration, not a general DSL)
```json
{
"assetKey": "paimon://catalog/db/ods_order",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
}
```
`dependency_group` is used to express multi-asset AND semantics: all
dependencies within the same group must evaluate to READY before that group is
considered ready; a workflow can configure multiple groups, and groups are
OR-related (any one group being ready is sufficient to trigger). The semantics
remain aligned with `DependentParameters.Dependence.relation` (AND/OR), making
it easier for users to understand and migrate. In the initial phase, support
only the two-level structure of AND-within-group / OR-across-group, without
supporting general nested Boolean expressions.
### 3.7 Idempotency Design
`trigger_key` composition (see 4.1) together with the unique constraint on
`t_ds_asset_trigger_history` is the core guarantee for idempotency:
- The same combination of workflowDefinitionCode + dependency_group +
snapshot set for all related assets can trigger successfully only once.
- When multiple Masters/Scanners concurrently scan the same batch of events,
the database unique constraint naturally ensures that “only one winner” exists,
with no need for a distributed lock.
- If `createCommand` fails (network/DB jitter), `trigger_status` remains
`TRIGGERING`. The Compensation Scanner identifies records stuck without a
terminal state and safely retries them (before retrying, it must confirm
whether the corresponding Command/WorkflowInstance has already been created, to
avoid duplicate creation—for example by writing back `trigger_key` into
`commandParam` and checking for an associated Command/WorkflowInstance before
retrying).
### 3.8 Multi-Asset Alignment (AND/OR)
- Dependencies are stored in `t_ds_asset_dependency`, grouped by
`dependency_group`. Each evaluation-trigger attempt must read the entire group,
evaluate `condition_json` item by item, and enter the trigger flow only if all
are satisfied.
- To avoid the issue of “asset A becomes ready -> trigger evaluation runs ->
asset B is not ready -> give up” and then never re-triggers after B becomes
ready, **the arrival of any asset event re-evaluates all dependency groups
associated with that asset** (not only the asset itself). In other words,
DependencyResolver uses the reverse index `asset_key -> dependency_group` to
drive re-evaluation.
### 3.9 Out-of-Order Event Handling
- At the event layer: the unique constraint on `t_ds_asset_event` is defined
by `(source_type, asset_key, event_type, snapshot_id)`, so duplicate reports
are rejected directly by the database and are considered successfully
deduplicated.
- At the state layer: when updating `AssetState`, version numbers
(`snapshotId` / `watermark`) are compared strictly, and the strategy is
“advance only, never roll back”:
```sql
UPDATE t_ds_asset_state
SET latest_snapshot_id = ?, latest_watermark = GREATEST(latest_watermark,
?),
version = version + 1, update_time = ?
WHERE asset_key = ? AND version = ? AND ? > latest_snapshot_id;
```
If the affected row count of `UPDATE` is 0, it indicates either a
concurrent conflict or an expired event (an old snapshot arrived late). The
event is marked `IGNORED` and the reason is recorded, without affecting
scheduling decisions.
### 3.10 Failure Retry and Compensation
- **Compensation Scanner** (reusing the Polling Scanner infrastructure from
3.3, with an independent schedule such as once every 5 to 15 minutes) is
responsible for:
1. Comparing the actual latest Paimon snapshot with
`t_ds_asset_state.latest_snapshot_id`. If it is behind, a collection gap exists
(Push failed or the Scanner was temporarily down), so the missing snapshots are
re-fetched and backfilled into `t_ds_asset_event`.
2. Scanning `t_ds_asset_trigger_history` for records that remain in
`TRIGGERING` / `FAILED` state for a long time, and re-executing the trigger
(first checking whether the corresponding Command/WorkflowInstance already
exists to avoid duplicate triggering).
3. For dependencies in `BLOCKED_BY_*` status, providing a query interface
explaining “why it has not triggered yet” (see observability).
- This compensation path shares exactly the same deduplication and state
update path as Push events, ensuring that the two paths do not cause duplicate
triggers.
## 4. Integration Points with Existing Modules
### 4.1 How workflow/task definitions declare asset dependencies
Two implementation forms are recommended, and should be delivered
incrementally by phase:
**(1) Strategy A: add a new task type `ASSET_SENSOR` (recommended MVP first
choice, lowest risk)**
- Following the plugin mechanism of the existing `DEPENDENT` task
(`DependentLogicTaskChannel` / `DependentLogicTaskChannelFactory` /
`DependentLogicTask`), add a new module `dolphinscheduler-task-asset-sensor`
under `dolphinscheduler-task-plugin`:
- `AssetSensorParameters extends AbstractParameters`: the field structure
follows the dependency expression JSON in 3.6, plus `checkIntervalSeconds` and
`timeoutMinutes`.
- `AssetSensorLogicTask extends AbstractLogicTask<AssetSensorParameters>`:
reuse the lifecycle of `AbstractLogicTask`
(`onTaskRunning/onTaskPaused/onTaskKilled`), and internally poll
`AssetStateDao` to determine whether the condition is satisfied (or register a
callback and let DependencyResolver actively push READY status to reduce
polling latency).
- Users add an `ASSET_SENSOR` node in the workflow DAG as an “entry
sentinel task,” and downstream tasks are linked using normal DAG dependencies.
After the task succeeds, the DAG proceeds normally.
- By default, it is recommended to place `ASSET_SENSOR` before existing
business tasks as a data-readiness gate, rather than replacing existing DAG
dependencies. This allows both “data version readiness” and “user-configured
native dependencies” to be satisfied together.
- Advantages: fully reuse the existing WorkflowInstance/TaskInstance state
machines, retries, timeouts, alerts, and UI display (through the task type
registration mechanism); minimal change scope; effectively equivalent to
“adding a new task plugin to the DAG.”
**(2) Strategy C: workflow-level events directly create workflow instances
(for pure event-driven scenarios, phase-two implementation)**
- Add `assetTriggerEnabled` and the associated `dependency_group` to the
extended properties of workflow definitions (records with empty
`task_definition_code` in `t_ds_asset_dependency` can represent workflow-level
dependencies).
- After DependencyResolver determines READY, call
`CommandService#createCommand`, passing audit fields related to triggering such
as `assetKey` and `snapshotId` in `commandParam`.
- Risk note: the mapping between workflow instances and snapshot versions
must be explicitly defined (see Section 11 risks).
### 4.2 New API endpoints on the API layer
Add a new Controller in `dolphinscheduler-api` (following the layering style
of the existing `ExecutorController`: Controller -> Service -> DAO):
```text
POST /projects/{projectCode}/assets 注册/更新资产
GET /projects/{projectCode}/assets 查询资产列表
GET /projects/{projectCode}/assets/{assetKey}/state 查询资产当前状态
POST /projects/{projectCode}/asset-events 外部系统主动上报事件（Push
方式）
GET /projects/{projectCode}/asset-events 查询事件历史（排障）
POST /projects/{projectCode}/asset-dependencies 为工作流/任务声明资产依赖
GET /projects/{projectCode}/asset-dependencies/{workflowDefinitionCode}
查询依赖及其当前判定状态（为什么没触发）
GET /projects/{projectCode}/asset-trigger-history 查询触发历史（幂等审计）
```
### 4.3 UI layer suggestions (not the focus of this round, brief only)
- In the workflow definition canvas, an `ASSET_SENSOR` task node should
display the currently waited asset(s), condition expression(s), and current
AssetState snapshot comparison (how many snapshots behind / how far watermark
lags).
- Add an “asset dashboard” page: asset list, recent events, current state,
associated downstream workflows, and trigger history timeline, similar to
Airflow’s Dataset view.
## 4.4 Asset dependency patterns within the DAG chain
In next-generation data warehouse DAGs, the trigger for every task should
follow a **pure snapshot model**, without time expressions or native dependency
gates:
| DAG Layer | Dependency Source | Example | Description |
|---|---|---|---|
| **ODS (data ingestion into the Lakehouse, first layer)** | Snapshot
advancement from external data sources | Flink CDC -> Paimon ods_order_latest
snapshot | The initial source of data asset versions, typically with snapshot
events produced by external data engines (Flink/Spark) |
| **DWD (dimensional processing, middle layer)** | Depends on snapshots from
the previous ODS layer | Wait until snapshots of ods_order_latest and
ods_payment_latest have both advanced | Multiple upstream assets are combined
using AND conditions |
| **DWS (subject-area aggregation, middle layer)** | Depends on multiple DWD
snapshots | Wait until snapshots of dwd_trade_event and dwd_order_dim advance |
Also a multi-asset AND dependency, aligned with DAG predecessor/successor
dependencies |
| **ADS (serving/output, terminal layer)** | Depends on DWS snapshots | Wait
until snapshots of dws_gmv_daily and dws_order_stat advance | The final data
product layer |
**Key principles**:
- The trigger for any task in the DAG is **entirely driven by the snapshot
versions of its dependent assets**, without time expressions or cron.
- “Predecessor/successor dependencies” in the DAG are automatically
transformed into “snapshot dependencies”: the later task waits for the snapshot
produced by the earlier task.
- Cross-flow dependencies (depending on tasks in other DAGs) are also
upgraded to a snapshot model by referencing the asset snapshots produced by
external DAGs.
- **There is no need to mix native dependencies and asset dependencies**: in
next-generation data warehouse architectures, asset snapshots are the sole
trigger criterion.
## 4.5 Impact on and Adaptations to the Existing Master Scheduling
Architecture
### 4.5.1 Overview of the current Master scheduling flow
The core scheduling flow of DolphinScheduler Master is as follows:
```text
1. CommandService 定时扫描 t_ds_command 表（按 id 递增）
↓
2. 对每条 Command 记录：
a) 检查 commandType（START_PROCESS / SCHEDULER / COMPLEMENT_DATA / ...）
b) 读取关联的 workflow_definition（工作流定义）
c) 创建 WorkflowInstance（工作流实例，初始状态=SUBMITTED）
d) 生成该工作流的 DAG 实例（TaskInstance + TaskDependency）
↓
3. MasterScheduler 定时扫描 WorkflowInstance / TaskInstance（按状态分组）
↓
4. 工作流实例状态转移：
SUBMITTED -> RUNNING（任何非跳过的任务已提交）
-> SUCCESS（所有任务成功）
-> FAILURE（任何关键路径任务失败）
↓
5. 任务实例状态转移（核心）：
SUBMITTED -> READY/WAITING_DEPENDENCY（依赖不满足）
-> RUNNING（所有依赖满足，发送给 Worker）
-> SUCCESS/FAILED/...
↓
6. 依赖判定逻辑（当前）：
- DEPENDENT 任务：轮询查询上游任务实例状态（是否 SUCCESS）
- DAG 前后置依赖：检查前置任务是否 SUCCESS
- 其他：判定条件立即满足，直接 READY
↓
7. Worker 执行 TaskInstance，汇报结果
↓
8. 重复步骤 3-7，直到工作流完成
```
### 4.5.2 Integration points for asset-event scheduling
After asset event-driven scheduling is introduced, Master needs to handle
**two trigger sources** and **three layers of dependency evaluation**:
**Trigger sources (Command creation)**:
```text
Original: START_PROCESS / SCHEDULER / COMPLEMENT_DATA / ...
↓
Enhanced: 新增 ASSET_EVENT_TRIGGER / 或复用 START_PROCESS + 特殊 commandParam
↓
由 DependencyResolver 模块判定资产就绪后调用 CommandService#createCommand
写入 t_ds_command（与现有流程完全相同）
```
**Three layers of dependency evaluation (from outer to inner)**:
```text
第一层：工作流级依赖（Strategy C，二期实现）
├─ 资产依赖判定：DependencyResolver 评估
t_ds_asset_dependency（task_definition_code=null 的记录）
└─ 不满足 => 不创建 Command / 工作流实例
第二层：任务级依赖（Strategy A，MVP）
├─ 原有 DAG 前后置依赖：Master 现有逻辑，检查前置任务是否 SUCCESS
├─ ASSET_SENSOR 任务依赖：轮询 t_ds_asset_state，判定条件（snapshot / watermark /
quality）
└─ DEPENDENT 任务依赖：轮询上游任务实例状态
第三层：工作流内部 DAG 依赖
└─ 任务间的串联 / 并联，正常的 DAG 拓扑执行
```
### 4.5.3 Specific Master-side adaptations
#### Adaptation 1: integrate DependencyResolver into Master
**Conceptual model**:
```java
public interface DependencyResolver {
/**
* 评估一个 dependency_group 是否 READY
* @param workflowCode 工作流定义编码（用于工作流级依赖评估）
* @param taskCode 任务定义编码（如果为 null，表示工作流级依赖）
* @param dependencyGroup 依赖分组名
* @return 依赖是否 READY（true 可以触发 / 释放，false 继续等待）
*/
DependencyStatus resolveDependency(Long workflowCode, Long taskCode,
String dependencyGroup);
/**
* 注册资产状态变化的监听器
* DependencyResolver 内部可基于事件回调而非轮询，降低延迟
*/
void registerAssetStateChangeListener(AssetStateChangeListener listener);
}
```
**Timing for workflow-level triggering (Strategy C)**:
```text
在 DependencyResolver 判定工作流级依赖 READY 后，立即调用：
commandService.createCommand(
Command.builder()
.commandType(CommandType.ASSET_EVENT_TRIGGER)
.workflowDefinitionCode(workflowCode)
.commandParam(JSON 序列化的 {assetKey, snapshotId, triggerKey, ...})
.build()
);
然后在 t_ds_asset_trigger_history 中 INSERT 占位记录（trigger_key 唯一约束）。
Master 的现有 CommandService scanner 会照常消费这条 Command，创建 WorkflowInstance，执行 DAG。
无需在 Master 侧做特殊识别——完全复用现有流程。
```
#### Adaptation 2: execution lifecycle of the `ASSET_SENSOR` task type
**Behavior during the Master execution phase**:
```text
1. TaskInstance 状态 = SUBMITTED，判定是否可进入 READY
↓
2. MasterScheduler 调用 task plugin 的 tracker（AssetSensorTracker extends
TaskTracker）
↓
3. AssetSensorTracker.checkTaskStatus() 周期性调用
DependencyResolver#resolveDependency()
返回值：
- DependencyStatus.READY => 任务转为 READY，发送给 Worker 执行
- DependencyStatus.BLOCKED => 保持 WAITING_DEPENDENCY，继续轮询
- DependencyStatus.TIMEOUT => 任务转为 ERROR（超时失败）
↓
4. Worker 侧照常执行该任务（可能是 Shell / SQL，或空操作代表"依赖已满足"）
↓
5. 下游任务通过 DAG 前后置依赖关系正常推进
```
**Code integration points** (`dolphinscheduler-master` module):
```text
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/
├── engine/
│ └── DAGExecutionEngine.java (现有，DAG 状态机)
│ ├── 改造点1：在 commitTask / dispatchTask 前检查 ASSET_SENSOR 任务
│ │ if (isAssetSensorTask) {
│ │ // 调用 DependencyResolver 检查依赖
│ │ if (!dependencyResolver.resolveDependency(...).isReady()) {
│ │ return; // 不提交给 Worker，保持 WAITING_DEPENDENCY
│ │ }
│ │ }
│ └── 改造点2：在工作流 SUBMITTED -> RUNNING 转移前
│ if (isAssetEventTrigger && hasWorkflowLevelAssetDependency) {
│ // 可选：在 Master 侧提前检查，尽早发现不满足的依赖
│ // （否则直接创建实例，在工作流执行时逐个任务检查）
│ }
├── processor/
│ ├── WorkflowProcessor.java (工作流实例处理)
│ │ └── 改造点3：处理 CommandType.ASSET_EVENT_TRIGGER
│ │ 解析 commandParam 提取 assetKey / snapshotId 等信息，
│ │ 存入 WorkflowInstance 的 commandParam 或扩展字段
│ │
│ └── TaskProcessor.java (任务实例处理)
│ └── 改造点4：任务状态转移时，对 ASSET_SENSOR 任务特殊处理
│ 调用 AssetSensorTracker.checkTaskStatus()
│
└── service/
├── DependencyResolverService.java (新增)
│ ├── 注入 AssetStateService (DAO 层)
│ ├── 注入 AssetDependencyService (DAO 层)
│ └── 实现 resolveDependency 逻辑
│ a) 查询 t_ds_asset_dependency(task_definition_code=?,
dependency_group=?)
│ b) 逐一查询 t_ds_asset_state 对应的 asset_key
│ c) 对每个资产条件求值（snapshot exists, watermark gte, quality status, ...）
│ d) 组合结果（同 group 所有条件 AND，不同 group 间 OR）
│
└── TrackerManager.java (现有，但需扩展)
└── 改造点5：注册 ASSET_SENSOR 对应的 tracker
trackerFactory.get(TaskType.ASSET_SENSOR)
=> new AssetSensorTracker(dependencyResolverService)
```
#### Adaptation 3: workflow instance startup and handling of SUBMITTED status
**Current logic** (simplified):
```java
// MasterScheduler#submitWorkflow
Command command = commandService.findOne(commandId);
WorkflowInstance instance = new WorkflowInstance();
instance.setWorkflowDefinitionCode(command.getWorkflowDefinitionCode());
instance.setState(WorkflowExecutionStatus.SUBMITTED);
// ... 生成 TaskInstance，立即进入 DAG 执行
```
**New logic** (if workflow-level asset dependency checks are required):
```java
// 在 WorkflowInstance 创建后、生成 DAG 任务前
if (command.getCommandType() == CommandType.ASSET_EVENT_TRIGGER) {
// 可选：提前验证工作流级资产依赖是否已满足
// （这是一个检查点，如果仍未满足则可能是 Compensation 重试时的竞态）
boolean workflowLevelAssetReady = dependencyResolverService
.resolveDependency(workflowCode, null, "default")
.isReady();
if (!workflowLevelAssetReady) {
// 记录告警，可选地延迟创建 TaskInstance
log.warn("Workflow {} triggered by asset event but asset dependency
no longer ready. " +
"This may indicate compensation retry or state backtrack.",
workflowCode);
}
}
// 正常流程继续，生成 DAG 任务
for (TaskDefinition taskDef : dag.getTasks()) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setTaskDefinitionCode(taskDef.getCode());
taskInstance.setState(TaskExecutionStatus.SUBMITTED);
if (isAssetSensorTask(taskDef)) {
// ASSET_SENSOR 任务状态初始化为 WAITING_DEPENDENCY
taskInstance.setState(TaskExecutionStatus.WAITING_DEPENDENCY);
}
// ... 保存 taskInstance
}
```
#### Adaptation 4: Master’s loop scanning logic
**Existing scan cycle**:
```text
CommandService.scanner() // 扫描 t_ds_command，周期=配置（默认 5s）
↓
WorkflowProcessor.processWorkflow() // 处理工作流状态转移
↓
TaskProcessor.processTask() // 处理任务状态转移，检查依赖
↓
AbstractTaskTracker.checkTaskStatus() // 调用具体 task plugin 的 tracker（如
DependentTaskTracker）
```
**New scan logic** (does not conflict with the existing one):
```text
新增 AssetEventCompensationScanner（可选，作为 Master 内部的后台线程）
周期：5~15 分钟（独立于 CommandService 扫描）
↓
扫描 t_ds_asset_event 和 t_ds_asset_state，检查：
1. 是否有漏采集的事件（e.g. Paimon 实际最新 snapshot_id >
t_ds_asset_state.latest_snapshot_id）
2. 是否有卡顿的 trigger_history（TRIGGERING 或 FAILED 状态超过 N 分钟）
↓
对于漏采集：补写 t_ds_asset_event（去重自动处理）-> 触发 DependencyResolver 重新评估
对于卡顿：可选地重试 createCommand（检查唯一约束冲突，确保不重复）
↓
该扫描是"兜底补偿"，不影响主调度流程
```
### 4.5.4 Minimal impact on the existing architecture
**No changes required**:
- ✅ The core state machines of WorkflowInstance and TaskInstance (SUBMITTED
/ RUNNING / SUCCESS / FAILED / ...)
- ✅ The expression and execution of DAG predecessor/successor dependencies
(kept as-is)
- ✅ Existing logic for DEPENDENT tasks (fully independent, no interference)
- ✅ Worker task execution lifecycle (transparent / unaffected)
- ✅ Peripheral capabilities such as Alert / Monitor / Backfill (compatible)
**Changes required** (but all are plugin-based and non-invasive):
- Add the new `ASSET_SENSOR` task type (as a task plugin similar to
`DependentLogicTask`)
- Add `AssetSensorTracker` as the tracker for that task (implementing the
TaskTracker interface)
- Add `DependencyResolverService` to handle dependency evaluation logic
(Service layer, without changing the Master kernel)
- Add `CommandType.ASSET_EVENT_TRIGGER` (optional, or reuse START_PROCESS)
- Add a single check before `TaskProcessor#submitTask()`:
```java
if (task is ASSET_SENSOR) {
if (!dependencyResolver.resolveDependency(...).isReady()) {
return; // 不提交
}
}
```
**Estimated code changes**:
Estimated **< 500 lines** (including the new `DependencyResolverService`,
`AssetSensorTracker`, and the required integration checks), which is a very
small change relative to DolphinScheduler’s overall codebase size.
### 4.5.5 Concurrency and consistency guarantees
**Multi-Master concurrency scenario**:
```text
场景：5 个 Master 实例同时运行，都在扫描 t_ds_command 和执行工作流
冲突点1：CommandService.createCommand() 写入同一条 Command
=> 依赖 t_ds_command 的 PK 或 sequence，一般由应用侧写入，不会重复
冲突点2：多个 Master 同时评估 ASSET_SENSOR 任务的依赖，决定是否提交给 Worker
=> 每个 Master 各自维护 TaskTracker，评估结果本地缓存
=> 当依赖首次从 NOT_READY 转为 READY 时，最快的 Master 抢到 Worker 资源
=> TaskProcessor#submitTask() 会检查 TaskInstance 是否已有 Worker 分配，重复提交会被幂等处理
=> 无需分布式锁，因为 TaskInstance 表会做状态检查
冲突点3：多个 Master 同时插入 t_ds_asset_trigger_history
=> 唯一约束 uk_trigger_key 保证只有一个成功
=> 失败的 Master 捕获唯一冲突异常，记录日志后继续（不影响调度）
```
**Idempotency guarantee**:
```text
trigger_key 形如：
workflow_12345 + dependency_group_default + snapshot_1001 + snapshot_1002
+ ...
= 完整标识"该工作流依赖这一组快照组合"的唯一键
同一 trigger_key 最多被成功创建一次 t_ds_asset_trigger_history 记录。
对应的 Command 也最多被创建一次（或可在 commandParam 中埋入 trigger_key 做检查）。
因此，无论有多少个 Master 并发评估，最终只会创建一个 WorkflowInstance。
```
### 4.5.6 Key configuration items
New Master-side configuration is required (to be written into Master’s
`application.yaml`):
```yaml
# 资产事件驱动调度开关
dolphinscheduler:
asset-event-scheduling:
enabled: true # 全局开关
# DependencyResolver 配置
dependency-resolver:
cache-expire-seconds: 60 # 资产状态本地缓存过期时间
resolve-timeout-seconds: 5 # 单次依赖评估的超时时间
# AssetEventCompensationScanner 配置
compensation-scanner:
enabled: true
interval-seconds: 300 # 补偿扫描周期（5分钟）
batch-size: 100 # 每次扫描的批量大小
# 事件源配置
event-source:
paimon-polling:
enabled: true
interval-seconds: 30 # Paimon $snapshots 轮询周期
batch-size: 500
push-receiver:
enabled: true # 是否启用 HTTP/Webhook 接收事件
```
All of these configurations are **optional and incremental**. In the MVP
phase, it is enough to enable only `enabled: true` and use the default values.
---
## 5. End-to-End Example: GMV Scenario Based on a Paimon Order Table
### 5.1 Business process and data assets
```text
MySQL 订单库（order / payment / refund 表）
-> Flink CDC 持续消费 binlog，生产 Paimon snapshot
-> Paimon 湖仓表：
ods_order_latest（主键表，最新订单状态）→ snapshot推进
ods_payment_latest（主键表，最新支付状态）→ snapshot推进
-> DWD 层任务：dwd_trade_event_fact（识别支付/退款事件）
依赖 ods_order_latest + ods_payment_latest 快照都推进
-> DWS 层任务：dws_gmv_daily（每日GMV聚合）
依赖 dwd_trade_event_fact 快照推进
-> ADS 层任务：ads_gmv_dashboard（对外报表/API）
依赖 dws_gmv_daily 快照推进
```
### 5.2 Workflow definition (DAG structure) — pure snapshot model
**Next-generation data warehouse DAG structure** (no cron, no time
expressions):
```text
Flink CDC 持续运行，生产 ods_order_latest、ods_payment_latest snapshot
↓
DAG工作流启动（由外部系统触发，例如Airflow或DolphinScheduler的API）
↓
Task 1: DWD_trade_event_fact
依赖：ods_order_latest snapshot_id >= N1 AND ods_payment_latest snapshot_id
>= N2
执行：读两张表的最新快照，识别交易事件
产出：dwd_trade_event_fact snapshot
↓
Task 2: DWS_gmv_daily
依赖：dwd_trade_event_fact snapshot_id >= M1
执行：聚合每日GMV
产出：dws_gmv_daily snapshot
↓
Task 3: ADS_gmv_dashboard
依赖：dws_gmv_daily snapshot_id >= P1
执行：同步到报表库
产出：ads_gmv_dashboard 就绪
```
**Core improvements**:
- Completely cron-free, with no fixed execution time. Once the DAG starts,
execution of each task depends on advancement of its dependent snapshots.
- If Flink CDC is delayed and ODS snapshots advance slowly, DWD waits; once
the snapshots arrive, DWD executes immediately.
- Snapshots from multiple upstream assets (`ods_order_latest +
ods_payment_latest`) are combined using AND conditions.
### 5.3 Example asset dependency declarations
**Dependencies of the DWD layer on two ODS tables**:
```json
[
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 100, // DWD_trade_event_fact 任务
"assetKey": "paimon://ods_db/ods_order_latest",
"dependencyGroup": "trade_event_group",
"conditionJson": {
"assetKey": "paimon://ods_db/ods_order_latest",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
},
"enabledFlag": 1
},
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 100, // 同一任务的第二个依赖
"assetKey": "paimon://ods_db/ods_payment_latest",
"dependencyGroup": "trade_event_group",
"conditionJson": {
"assetKey": "paimon://ods_db/ods_payment_latest",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
},
"enabledFlag": 1
}
]
```
**Dependency of the DWS layer on the DWD layer**:
```json
[
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 200, // DWS_gmv_daily 任务
"assetKey": "paimon://dw_db/dwd_trade_event_fact",
"dependencyGroup": "gmv_group",
"conditionJson": {
"assetKey": "paimon://dw_db/dwd_trade_event_fact",
"snapshotRequired": true,
"qualityStatus": "PASSED"
},
"enabledFlag": 1
}
]
```
**Trigger behavior**:
1. Flink CDC continuously consumes MySQL binlog and produces an event
whenever a new snapshot is generated (for example, `ods_order_latest
snapshot_id=1001`).
2. AssetEventScanner or the Push endpoint captures the event and writes it
into `t_ds_asset_event`.
3. AssetStateService uses optimistic locking to update `t_ds_asset_state`:
`asset_key=paimon://ods_db/ods_order_latest, latest_snapshot_id=1001`.
4. DependencyResolver scans via reverse index: which dependency groups are
associated with this `asset_key`? (for example, `trade_event_group`).
5. Evaluate all dependencies in `trade_event_group`:
- Does `ods_order_latest` snapshot exist? ✓
- Does `ods_payment_latest` snapshot exist? ✓ (assuming it has also
advanced)
- Are both `qualityStatus` values equal to `PASSED`? ✓
→ Dependency group READY
6. Compute
`trigger_key=workflow_12345_taskdef_100_group_trade_event_group_snapshot_1001_1002`
(including all key snapshot IDs), and uniquely insert it into
`t_ds_asset_trigger_history`.
7. Insert succeeds → the `DWD_trade_event_fact` task instance is notified
that the dependency is satisfied and executes immediately.
8. DWD produces a new snapshot (for example, `dwd_trade_event_fact
snapshot_id=2001`), and steps 2-7 repeat, triggering DWS.
9. ADS similarly waits for DWS snapshot advancement.
## 6. Observability Design
### 6.1 Key metrics (Metrics)
The system should expose the following Prometheus metrics (Micrometer) for
convenient Grafana dashboard integration:
```text
# 事件接入层
lakehouse_asset_event_received_total{source_type="paimon", asset_key="..."}
- 意义：从指定资产收到的事件总数
lakehouse_asset_event_deduplicated_total{source_type="paimon",
asset_key="..."}
- 意义：被去重丢弃的重复事件数
lakehouse_asset_event_invalid_total{source_type="paimon", asset_key="...",
reason="xxx"}
- 意义：被拒绝的无效事件（格式错误、资产不存在等），按原因分类
# 资产状态层
lakehouse_asset_state_updated_total{asset_key="..."}
- 意义：资产状态被推进的次数
lakehouse_asset_state_lag_seconds{asset_key="..."}
- 意义：资产最新快照的年龄（当前时间 - snapshot commit_time）
lakehouse_asset_watermark_lag_seconds{asset_key="..."}
- 意义：资产 watermark 相对于业务日期的延迟（例如 watermark=10-10 23:59:59，当前日期 10-11，则
lag=-10秒表示超前）
# 依赖与触发层
lakehouse_asset_dependency_ready_total{asset_key="...", workflow_code="...",
dependency_group="..."}
- 意义：依赖就绪的次数
lakehouse_asset_dependency_blocked_by{asset_key="...", workflow_code="...",
blocked_by="snapshot|watermark|quality|schema"}
- 意义：依赖被卡住的原因分布
lakehouse_asset_trigger_success_total{workflow_code="...",
trigger_source="asset_event"}
- 意义：由资产事件成功触发的工作流实例数
lakehouse_asset_trigger_failed_total{workflow_code="...", reason="xxx"}
- 意义：触发失败的工作流数，按原因分类（command_create_failed、duplicate_trigger 等）
# 补偿层
lakehouse_asset_compensation_scan_total{asset_key="..."}
- 意义：补偿扫描器运行次数
lakehouse_asset_compensation_missed_events{asset_key="..."}
- 意义：补偿发现的漏采集事件数
lakehouse_asset_trigger_history_retry_total{trigger_status="..."}
- 意义：触发历史被重试的次数
```
### 6.2 Key log fields
All log output should include the following contextual fields for end-to-end
tracing and troubleshooting:
```json
{
"timestamp": "2024-09-10T10:15:30.123Z",
"level": "INFO",
"logger": "org.apache.dolphinscheduler.service.asset.AssetEventScanner",
"traceId": "abc123xyz",
"assetKey": "paimon://ods_db/ods_order_latest",
"partitionKey": "dt=20240910",
"snapshotId": 1001,
"schemaId": 5,
"watermark": "2024-09-10T23:59:59Z",
"eventType": "SNAPSHOT_COMMITTED",
"eventId": "paimon_evt_20240910_1001_xxx",
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 54321,
"workflowInstanceId": 67890,
"taskInstanceId": 11111,
"triggerKey": "workflow_12345_group_gmv_default_snapshot_1001",
"status": "READY | BLOCKED_BY_QUALITY | TRIGGERED | FAILED",
"reason": "Asset quality check passed, dependency READY",
"message": "Asset snapshot processed, dependency READY, triggering
workflow instance..."
}
```
### 6.3 Alerting rules
| Alert Condition | Threshold | Priority | Recommended Action |
|---|---|---|---|
| `lakehouse_asset_state_lag_seconds > 3600` | 1 hour | P2 | Check whether
the Paimon table is producing new snapshots; check scanner health |
| `lakehouse_asset_watermark_lag_seconds > 600` | 10 minutes | P2 | Check
whether the CDC/Flink source is delayed; check the `_watermark` field in the
Paimon table |
| `lakehouse_asset_trigger_failed_total increase > 3/5min` | 3 failures
within 5 minutes | P1 | Check whether Command creation is failing; check Master
availability |
| `lakehouse_asset_dependency_blocked_by{blocked_by="quality"} > 0`
sustained for > 30min | 30 minutes | P2 | Quality check failure; intervention
is needed from data producers or the quality team |
| `lakehouse_asset_compensation_missed_events > 10/hour` | > 10 per hour |
P1 | Indicates a systemic issue in the Push or Poll path; compensation is
masking the issue but a root fix is required |
## 7. Asset Identification Specification
### 7.1 AssetKey naming convention
In next-generation data warehouses, `assetKey` identifies the globally
unique location of a Lakehouse table and does not need to include partition
variables:
```text
paimon://<catalog_name>/<database_name>/<table_name>
```
Examples:
- `paimon://ods_db/ods_order_latest` — ODS-layer order table
- `paimon://dw_db/dwd_trade_event_fact` — DWD-layer trade event table
- `paimon://dws_db/dws_gmv_daily` — DWS-layer daily GMV table
**Notes**:
- `assetKey` points to the table level and does not include
`partition_spec`, because in next-generation data warehouses all partitions of
a table reside in the same Paimon/Iceberg/Hudi primary-key table, and version
advancement is table-level.
- If the snapshot independence of different partitions needs to be tracked,
it should be recorded in the `partitionKey` field of `t_ds_asset_event` (for
example, `dt=20240910`).
### 7.2 AssetKey registration and discovery
- Assets must be registered in the `t_ds_asset` table (via API or batch
scripts), including metadata such as sourceType (paimon/iceberg/hudi), catalog,
database, and table.
- When defining workflow dependencies, users query and select existing
assets through the UI “asset dashboard” or API, similar to the Airflow Dataset
selector.
- If a dependent `assetKey` does not exist, DependencyResolver records a
warning and returns `BLOCKED_BY_UNKNOWN_ASSET`, preventing the workflow from
being stuck forever without explanation.
## 8. Phased Implementation Roadmap
### Phase 1 (MVP, 2–3 weeks): Paimon snapshot -> ASSET_SENSOR single-asset
trigger closed loop
- Add tables: `t_ds_asset`, `t_ds_asset_event`, `t_ds_asset_state`,
`t_ds_asset_dependency` (only the scenario where `task_definition_code` is
non-null), `t_ds_asset_trigger_history`, `t_ds_asset_event_consumer_offset`.
- Implement `AssetEventScanner` (polling Paimon `$snapshots`).
- Implement `AssetStateService` (optimistic-lock update and out-of-order
discard).
- Implement `DependencyResolver` (support only single-asset snapshot
existence checks).
- Implement the `ASSET_SENSOR` task plugin (Strategy A integration).
- Clarify the MVP problem boundary: prioritize solving the scenario where
“tasks should trigger on snapshot readiness rather than time,” without changing
the semantics of existing native DAG dependencies.
- Complete the end-to-end flow: manually advance a Paimon table snapshot ->
observe an `ASSET_SENSOR` task transition from waiting to success within tens
of seconds -> downstream task executes.
- Complete unit tests for idempotency / out-of-order handling / basic
observability (log fields; see 5.3).
Key interface sketches:
```java
public interface AssetEventSource {
List<AssetEvent> pollNewEvents(Asset asset, Long sinceSnapshotId);
}
public interface AssetStateService {
/** 乱序保护 + 乐观锁更新，返回 true 表示状态被推进 */
boolean applyEvent(AssetEvent event);
AssetState getState(String assetKey);
}
public interface AssetDependencyResolver {
/** 返回 READY / BLOCKED_BY_WATERMARK / BLOCKED_BY_QUALITY /
BLOCKED_BY_SCHEMA / WAITING */
DependencyEvalResult evaluate(AssetDependency dependency);
}
public interface AssetTriggerService {
/** 幂等触发，内部完成 trigger_key 唯一插入 + Command 创建或 ASSET_SENSOR 状态回写 */
TriggerResult triggerIfReady(AssetDependencyGroup group);
}
```
### Phase 2: multi-asset AND/OR, watermark/quality/schema conditions,
Compensation Scanner
- Extend `condition_json` to support evaluation of watermark / qualityStatus
/ schemaStatus.
- Support multi-asset AND aggregation within `dependency_group` and OR
across groups.
- Bring Compensation Scanner online to periodically compare actual Paimon
snapshots against AssetState, and retry failed triggers.
- Add observability metrics and Grafana panels.
### Phase 3: Push reporting API, workflow-level Strategy C, Iceberg/Hudi
event source integration
- Add `POST /asset-events` API so Flink/Spark jobs can actively report
events, coexisting with Polling and sharing deduplication.
- Support workflow-level direct creation of WorkflowInstance (Strategy C),
and add documentation for rules around bizDate derivation and backfill boundary
scenarios.
- Extend the `AssetEventSource` interface with Iceberg/Hudi implementations.
### Phase 4: governance and observability enhancements (not the core of this
round, reserved)
- Asset lineage, full asset lifecycle dashboards, cross-project asset
dependencies, etc.
## 9. Refactoring and Compatibility for Next-Generation Data Warehouse DAGs
### 9.1 Relationship with existing cron scheduling
Next-generation data warehouse DAGs are **entirely based on snapshot version
advancement**, not time triggers. This means:
- **No cron retained**: under the new architecture, a DAG no longer has the
concept of a “scheduled execution time.”
- **No time-based dependencies retained**: tasks in the DAG no longer depend
on the original DEPENDENT model (which targets historical instance results) or
time dimensions such as “business date.”
- **Completely driven by data asset versions**: after the DAG starts,
execution of each task depends 100% on the advancement of upstream asset
snapshots.
For existing users:
- To upgrade to the next-generation data warehouse architecture, the DAG
must be **redefined in full**, replacing every “cron + DEPENDENT” organization
with “snapshot + assetKey.”
- This is an architectural upgrade rather than an incremental retrofit, so
it requires unified planning and execution.
### 9.2 Registration of the `ASSET_SENSOR` task type
- `ASSET_SENSOR` is a newly added task type and must be registered,
packaged, and released within `dolphinscheduler-task-plugin`.
- Existing clusters automatically support it after upgrading to the new
version.
- If a workflow definition references `ASSET_SENSOR` but the cluster is not
upgraded/installed, Master records an error and skips the task.
### 9.3 Database upgrade path
- New tables such as `t_ds_asset` are automatically created via SQL
migration scripts in `dolphinscheduler-dao`.
- The upgrade does not require downtime; migration scripts are idempotent
and non-invasive to existing data.
- Compatible with common databases such as MySQL 5.7+ and PostgreSQL 10+.
### 9.4 Difference from the existing DependentTask
| Feature | DEPENDENT Task (old) | ASSET_SENSOR Task (new) |
|---|---|---|
| Dependency object | Upstream task/workflow instance execution result |
External data asset snapshot version |
| Applicability | Task orchestration within traditional DAGs | Data
version-driven scheduling in next-generation data warehouses |
| Trigger condition | Upstream task success/failure | Asset snapshot
advancement / quality pass |
## 10. Frequently Asked Questions (FAQ)
**Q1: How are dependencies on multiple upstream assets handled in a DAG (for
example, DWS depending on multiple DWD tables)?**
A1: Through `dependencyGroup` combined with AND conditions. All asset
dependencies within the same `dependencyGroup` must be READY for the group to
become READY. For example:
```json
[
{ "taskDefinitionCode": 200, "assetKey": "paimon://dw_db/dwd_trade_event",
"dependencyGroup": "dws_gmv" },
{ "taskDefinitionCode": 200, "assetKey": "paimon://dw_db/dwd_order_dim",
"dependencyGroup": "dws_gmv" }
]
```
Only when both DWD table snapshots advance will DWS be triggered.
---
**Q2: If an asset event is lost (for example, a Paimon snapshot is not
captured by the Scanner), will downstream be stuck forever?**
A2: No. There are two layers of protection:
1. Each task has `timeoutMinutes` configured (for example, 60 minutes).
After timeout, the task enters ERROR state and triggers an alert.
2. Compensation Scanner runs every 5 to 15 minutes, compares actual Paimon
snapshots with AssetState, and if it finds missed collections, it backfills the
event and retriggers evaluation.
---
**Q3: If multiple workflows depend on the same asset, will they be triggered
repeatedly?**
A3: No. Each workflow has an independent `trigger_key` for that asset, for
example:
```text
workflow_A_group_default_snapshot_1001
workflow_B_group_default_snapshot_1001
```
Although both depend on the same snapshot, their trigger records are
separate and each triggers once.
---
**Q4: If a task retries multiple times due to network issues, will it
repeatedly trigger downstream?**
A4: No. Task instance retry is handled on the Worker side. As long as the
task eventually succeeds, downstream tasks are released once. Downstream tasks
are triggered through DAG dependencies and follow the same “task instance
uniqueness” principle, so they will not be triggered repeatedly.
---
**Q5: Does it support mixed dependencies across multiple data Lakehouse
table formats (Paimon, Iceberg, Hudi)?**
A5: MVP (Phase 1) supports only Paimon. Phase 2 extends unified processing
for watermark/quality/schema conditions. Phase 3 adds Iceberg/Hudi event
sources. In theory mixed dependencies are supported, but the quality and test
coverage of the EventSource implementation for each format must be ensured.
---
**Q6: How are cross-flow dependencies (task dependencies between DAGs)
supported in the new architecture?**
A6: Cross-flow dependencies are upgraded to “depending on the asset
snapshots produced by external DAGs.” For example, a task in DAG-B may depend
on a table snapshot produced by DAG-A:
```json
{
"assetKey": "paimon://dw_db/external_dag_a_output_table",
"dependencyGroup": "cross_flow",
"conditionJson": {
"snapshotRequired": true,
"qualityStatus": "PASSED"
}
}
```
This way, DAG-B waits for the external DAG-A table snapshot to advance,
rather than waiting for a task instance.
---
**Q7: If INSERT conflicts occur on database unique constraints, will
exception handling affect performance?**
A7: Database unique constraint conflicts are usually very fast (< 1ms) and
are part of the expected normal path. They are logged and execution continues,
with no obvious performance impact. However, if concurrency is very high (for
example, 10+ Master instances competing to trigger), it is recommended to
monitor the database connection pool and index performance.
## 11. Risks and Open Questions
1. **Semantic definition of backfill and event-driven behavior**: manual
backfill can reproduce historical snapshots, so it must be clarified whether
those historical snapshots should trigger downstream tasks. It is recommended
to define a `backfill_mode` marker telling the system whether historical
snapshots should trigger.
2. **Race window in multi-Master concurrent triggering**: although the
unique constraint guarantees that only one insert into `trigger_history`
ultimately succeeds, there is still a short inconsistency window between
`createCommand` and persistence of `trigger_history`. The ordering constraint
“insert a placeholder into `trigger_history` first, then call `createCommand`”
must be explicitly defined, and compensation must be added for `createCommand`
failures.
3. **Task polling frequency and database pressure**: if many task instances
poll the same batch of asset states simultaneously, batch-fetching/caching
strategies must be considered to avoid excessive query pressure on
`t_ds_asset_state`. It is recommended to introduce an active callback mechanism
in DependencyResolver to replace per-task polling.
4. **Compatibility of extending `CommandType` semantics**: adding
`ASSET_EVENT_TRIGGER` requires evaluating its impact on existing code paths
that exhaustively switch over `CommandType` (such as alerting, logging, and UI
display). All usages of `CommandType` across the repository should be searched
and adapted for compatibility.
5. **Permission model for cross-catalog / cross-project asset
dependencies**: assets may belong to different DolphinScheduler projects or
even different tenants. It is not yet designed whether dependency declaration
and triggering require cross-project authorization. This requires separate
follow-up discussion.
6. **Clock/order dependence in deduplicating Push and Poll paths**: if the
same snapshot event arrives through the two paths within an extremely short
interval, the unique constraint prevents duplicate persistence, but correctness
and performance of database unique-index conflict handling under high
concurrency still need to be validated.
7. **Scope of exactly-once claims**: this design can guarantee only that
“the trigger record for the same `trigger_key` is successfully created once,”
but cannot guarantee strict exactly-once semantics for downstream Command
consumption and WorkflowInstance creation under extreme failure scenarios. This
boundary must be explicitly documented.
8. **Mixed dependencies between partitioned and non-partitioned tables**:
when a task in the DAG simultaneously depends on snapshots from partitioned and
non-partitioned tables, how to express and evaluate them uniformly remains
open. Clear rules need to be defined in dependency expressions.
## 12. Summary
This design upgrades DolphinScheduler from a “time-based DAG scheduling
system” to a “data asset version-driven scheduling system.” The core principles
are:
- **The entire workflow (DAG) represents the full data processing chain from
ODS -> DWD -> DWS -> ADS**.
- **The trigger of every task layer in the DAG is entirely determined by the
advancement of snapshot versions of its dependent assets**, rather than time or
manual triggering.
- **By introducing the new `ASSET_SENSOR` task type**, the existing task
instance state machine and DAG execution mechanism are reused.
- **By introducing new asset tables and an event deduplication mechanism**,
idempotent triggering and multi-Master concurrency safety are guaranteed.
- **Without breaking the existing DolphinScheduler kernel**, integration is
achieved through plugin-based tasks and the existing Command entry point.
This is a native scheduling solution for next-generation Lakehouse
architectures (Paimon/Iceberg/Hudi), and it addresses the root problem of
mismatch between traditional time-based triggering and actual data availability.
--
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]
