+------------------------------------------------------------------+
|  Pipeline DSL 定位                                                 |
|                                                                   |
|  低代码 UI ←──────── Pipeline DSL ──────────→ 原生 API             |
|  (易用/受限)         (平衡点)                  (强大/复杂)          |
|                                                                   |
|  特点：                                                            |
|  - Python 链式语法，代码即文档                                      |
|  - Schema 类型安全                                                 |
|  - 多引擎编译（Flink / Spark / DuckDB）                            |
|  - 自动 DAG 可视化                                                 |
|  - Git 版本控制友好                                                |
+------------------------------------------------------------------+

from onto_pipeline import Pipeline, Source, Transform, Sink
from onto_pipeline.transforms import filter, map, join, aggregate, window
from onto_pipeline.sources import mysql_cdc, kafka, csv_file
from onto_pipeline.sinks import iceberg, doris, ontology

# 定义一条完整的 Pipeline
pipeline = (
    Pipeline.create("equipment-sync")
    .description("设备数据实时同步管道")
    .owner("data-team")

    # 数据源
    .source(
        mysql_cdc("factory_db.equipment")
        .host("mysql.internal")
        .port(3306)
        .startup_mode("initial")
    )

    # 转换链
    .transform(
        filter(lambda row: row["status"] != "decommissioned")
    )
    .transform(
        map(lambda row: {
            **row,
            "display_name": f"{row['type']}-{row['serial_number']}",
            "is_active": row["status"] in ("running", "idle"),
        })
    )
    .transform(
        join(
            right=Source.lookup("dim_department"),
            on="department_id",
            fields=["department_name", "department_manager"]
        )
    )

    # 输出
    .sink(
        ontology("Equipment")
        .primary_key("equipment_id")
        .property_mapping({
            "id": "equipment_id",
            "name": "display_name",
            "status": "status",
            "department_name": "department",
        })
    )

    # 构建
    .build()
)

# 执行
pipeline.run()

from __future__ import annotations
from typing import Optional, List, Dict, Callable, Any, Union
from pydantic import BaseModel
from abc import ABC, abstractmethod


class Pipeline:
    """Pipeline 构建器"""

    def __init__(self, name: str):
        self._name = name
        self._description: Optional[str] = None
        self._owner: Optional[str] = None
        self._source: Optional[SourceNode] = None
        self._transforms: List[TransformNode] = []
        self._sinks: List[SinkNode] = []
        self._config: PipelineConfig = PipelineConfig()

    @staticmethod
    def create(name: str) -> Pipeline:
        return Pipeline(name)

    def description(self, desc: str) -> Pipeline:
        self._description = desc
        return self

    def owner(self, owner: str) -> Pipeline:
        self._owner = owner
        return self

    def source(self, source: SourceNode) -> Pipeline:
        self._source = source
        return self

    def transform(self, transform: TransformNode) -> Pipeline:
        self._transforms.append(transform)
        return self

    def sink(self, sink: SinkNode) -> Pipeline:
        self._sinks.append(sink)
        return self

    def config(self, **kwargs) -> Pipeline:
        self._config = PipelineConfig(**kwargs)
        return self

    def build(self) -> CompiledPipeline:
        """编译 Pipeline"""
        # 验证
        self._validate()
        # 构建 DAG
        dag = self._build_dag()
        # Schema 推导
        self._infer_schemas(dag)
        # 优化
        optimized_dag = self._optimize(dag)
        # 返回编译后的 Pipeline
        return CompiledPipeline(
            name=self._name,
            dag=optimized_dag,
            config=self._config
        )

    def _validate(self):
        if self._source is None:
            raise PipelineBuildError("Pipeline must have a source")
        if not self._sinks:
            raise PipelineBuildError("Pipeline must have at least one sink")

    def _build_dag(self) -> PipelineDAG:
        """构建有向无环图"""
        dag = PipelineDAG()
        prev_node = dag.add_node(self._source)
        for transform in self._transforms:
            current = dag.add_node(transform)
            dag.add_edge(prev_node, current)
            prev_node = current
        for sink in self._sinks:
            sink_node = dag.add_node(sink)
            dag.add_edge(prev_node, sink_node)
        return dag

    def _infer_schemas(self, dag: PipelineDAG):
        """Schema 推导：从 Source 开始逐步推导每个节点的输出 Schema"""
        for node in dag.topological_sort():
            input_schemas = [dag.get_output_schema(pred) for pred in dag.predecessors(node)]
            node.infer_output_schema(input_schemas)

    def _optimize(self, dag: PipelineDAG) -> PipelineDAG:
        """Pipeline 优化"""
        optimizer = PipelineOptimizer()
        dag = optimizer.fuse_filters(dag)        # 合并连续的 filter
        dag = optimizer.push_down_predicates(dag) # 谓词下推
        dag = optimizer.eliminate_projections(dag) # 投影消除
        return dag

class SourceNode(ABC):
    """数据源节点基类"""

    @abstractmethod
    def output_schema(self) -> Schema:
        """输出 Schema"""
        ...

    @abstractmethod
    def to_execution_plan(self, engine: str) -> ExecutionPlan:
        """编译为执行计划"""
        ...


class MySQLCDCSource(SourceNode):
    """MySQL CDC 数据源"""

    def __init__(self, table: str):
        self._table = table
        self._host: Optional[str] = None
        self._port: int = 3306
        self._startup_mode: str = "initial"
        self._schema: Optional[Schema] = None

    def host(self, h: str) -> MySQLCDCSource:
        self._host = h
        return self

    def port(self, p: int) -> MySQLCDCSource:
        self._port = p
        return self

    def startup_mode(self, mode: str) -> MySQLCDCSource:
        self._startup_mode = mode
        return self

    def output_schema(self) -> Schema:
        if self._schema is None:
            # 从数据库元数据推导 Schema
            self._schema = self._introspect_schema()
        return self._schema

    def _introspect_schema(self) -> Schema:
        """连接数据库获取表结构"""
        connector = MySQLConnector(self._host, self._port)
        columns = connector.describe_table(self._table)
        fields = []
        for col in columns:
            fields.append(SchemaField(
                name=col.name,
                data_type=self._map_mysql_type(col.type),
                nullable=col.nullable,
                comment=col.comment
            ))
        return Schema(fields=fields)


class KafkaSource(SourceNode):
    """Kafka 数据源"""

    def __init__(self, topic: str):
        self._topic = topic
        self._brokers: List[str] = []
        self._format: str = "json"
        self._schema: Optional[Schema] = None

    def brokers(self, brokers: List[str]) -> KafkaSource:
        self._brokers = brokers
        return self

    def format(self, fmt: str) -> KafkaSource:
        self._format = fmt
        return self

    def schema(self, schema: Schema) -> KafkaSource:
        self._schema = schema
        return self

    def output_schema(self) -> Schema:
        if self._schema is None:
            raise PipelineBuildError("Kafka source requires explicit schema")
        return self._schema


class CSVFileSource(SourceNode):
    """CSV 文件数据源"""

    def __init__(self, path: str):
        self._path = path
        self._delimiter: str = ","
        self._header: bool = True
        self._schema: Optional[Schema] = None

    def delimiter(self, d: str) -> CSVFileSource:
        self._delimiter = d
        return self

    def header(self, h: bool) -> CSVFileSource:
        self._header = h
        return self

    def output_schema(self) -> Schema:
        if self._schema is None:
            self._schema = self._infer_from_file()
        return self._schema

def mysql_cdc(table: str) -> MySQLCDCSource:
    """创建 MySQL CDC 数据源"""
    return MySQLCDCSource(table)

def kafka(topic: str) -> KafkaSource:
    """创建 Kafka 数据源"""
    return KafkaSource(topic)

def csv_file(path: str) -> CSVFileSource:
    """创建 CSV 文件数据源"""
    return CSVFileSource(path)

def ontology_source(object_type: str) -> OntologySource:
    """从 Ontology 读取数据"""
    return OntologySource(object_type)

class FilterTransform(TransformNode):
    """过滤变换"""

    def __init__(self, predicate: Callable[[Dict], bool]):
        self._predicate = predicate

    def infer_output_schema(self, input_schemas: List[Schema]) -> Schema:
        # Filter 不改变 Schema
        return input_schemas[0]

    def to_execution_plan(self, engine: str) -> ExecutionPlan:
        if engine == "flink":
            return FlinkFilterPlan(self._predicate)
        elif engine == "spark":
            return SparkFilterPlan(self._predicate)
        elif engine == "duckdb":
            return DuckDBFilterPlan(self._predicate)


class MapTransform(TransformNode):
    """映射变换"""

    def __init__(self, mapper: Callable[[Dict], Dict]):
        self._mapper = mapper
        self._output_schema: Optional[Schema] = None

    def output_schema(self, schema: Schema) -> MapTransform:
        """手动指定输出 Schema"""
        self._output_schema = schema
        return self

    def infer_output_schema(self, input_schemas: List[Schema]) -> Schema:
        if self._output_schema:
            return self._output_schema
        # 尝试从 mapper 函数推导 Schema
        sample_input = input_schemas[0].generate_sample()
        sample_output = self._mapper(sample_input)
        return Schema.from_dict(sample_output)


class JoinTransform(TransformNode):
    """关联变换"""

    def __init__(self, right: SourceNode, on: str, fields: List[str]):
        self._right = right
        self._on = on
        self._fields = fields
        self._join_type: str = "left"

    def join_type(self, jt: str) -> JoinTransform:
        self._join_type = jt
        return self

    def infer_output_schema(self, input_schemas: List[Schema]) -> Schema:
        left_schema = input_schemas[0]
        right_schema = self._right.output_schema()
        # 合并 Schema：左表所有字段 + 右表指定字段
        merged_fields = list(left_schema.fields)
        for field in right_schema.fields:
            if field.name in self._fields:
                merged_fields.append(field)
        return Schema(fields=merged_fields)


class AggregateTransform(TransformNode):
    """聚合变换"""

    def __init__(self):
        self._group_by: List[str] = []
        self._aggregations: List[Aggregation] = []

    def group_by(self, *fields: str) -> AggregateTransform:
        self._group_by = list(fields)
        return self

    def agg(self, field: str, func: str, alias: Optional[str] = None) -> AggregateTransform:
        self._aggregations.append(Aggregation(
            field=field, function=func, alias=alias or f"{func}_{field}"
        ))
        return self

    def infer_output_schema(self, input_schemas: List[Schema]) -> Schema:
        fields = []
        # Group By 字段保留
        for gb in self._group_by:
            field = input_schemas[0].get_field(gb)
            fields.append(field)
        # 聚合字段
        for agg in self._aggregations:
            fields.append(SchemaField(
                name=agg.alias,
                data_type=self._infer_agg_type(agg.function, input_schemas[0].get_field(agg.field)),
                nullable=False
            ))
        return Schema(fields=fields)


class WindowTransform(TransformNode):
    """窗口变换"""

    def __init__(self, window_type: str, size: str):
        self._window_type = window_type  # tumbling | sliding | session
        self._size = size                 # "5m" | "1h" | "1d"
        self._slide: Optional[str] = None
        self._gap: Optional[str] = None
        self._time_field: str = "event_time"

    def slide(self, s: str) -> WindowTransform:
        self._slide = s
        return self

    def gap(self, g: str) -> WindowTransform:
        self._gap = g
        return self

    def time_field(self, f: str) -> WindowTransform:
        self._time_field = f
        return self


class UnionTransform(TransformNode):
    """合并变换"""

    def __init__(self, *sources: SourceNode):
        self._sources = sources

    def infer_output_schema(self, input_schemas: List[Schema]) -> Schema:
        # 验证所有输入 Schema 兼容
        base = input_schemas[0]
        for schema in input_schemas[1:]:
            if not base.is_compatible(schema):
                raise SchemaError(f"Union schemas are incompatible: {base} vs {schema}")
        return base

def filter(predicate: Callable[[Dict], bool]) -> FilterTransform:
    return FilterTransform(predicate)

def map(mapper: Callable[[Dict], Dict]) -> MapTransform:
    return MapTransform(mapper)

def join(right: SourceNode, on: str, fields: List[str]) -> JoinTransform:
    return JoinTransform(right, on, fields)

def aggregate() -> AggregateTransform:
    return AggregateTransform()

def window(window_type: str, size: str) -> WindowTransform:
    return WindowTransform(window_type, size)

def union(*sources: SourceNode) -> UnionTransform:
    return UnionTransform(*sources)

class OntologySink(SinkNode):
    """Ontology Sink：写入 Ontology 实例"""

    def __init__(self, object_type: str):
        self._object_type = object_type
        self._primary_key: Optional[str] = None
        self._property_mapping: Dict[str, str] = {}
        self._write_mode: str = "upsert"  # upsert | append | overwrite

    def primary_key(self, pk: str) -> OntologySink:
        self._primary_key = pk
        return self

    def property_mapping(self, mapping: Dict[str, str]) -> OntologySink:
        self._property_mapping = mapping
        return self

    def write_mode(self, mode: str) -> OntologySink:
        self._write_mode = mode
        return self

    def validate_schema(self, input_schema: Schema):
        """验证输入 Schema 与目标 ObjectType 的兼容性"""
        target_schema = self._get_object_type_schema()
        for source_field, target_property in self._property_mapping.items():
            # 验证源字段存在
            source = input_schema.get_field(source_field)
            if source is None:
                raise SchemaError(f"Source field '{source_field}' not found in input schema")
            # 验证目标属性存在
            target = target_schema.get_property(target_property)
            if target is None:
                raise SchemaError(f"Target property '{target_property}' not found in ObjectType '{self._object_type}'")
            # 验证类型兼容
            if not source.data_type.is_assignable_to(target.property_type):
                raise SchemaError(
                    f"Type mismatch: {source_field}({source.data_type}) -> {target_property}({target.property_type})"
                )


class IcebergSink(SinkNode):
    """Iceberg Sink：写入 Iceberg 表"""

    def __init__(self, table: str):
        self._table = table
        self._catalog: str = "nessie"
        self._branch: str = "main"
        self._partition_spec: Optional[str] = None

    def catalog(self, c: str) -> IcebergSink:
        self._catalog = c
        return self

    def branch(self, b: str) -> IcebergSink:
        self._branch = b
        return self

    def partition_by(self, spec: str) -> IcebergSink:
        self._partition_spec = spec
        return self


class DorisSink(SinkNode):
    """Doris Sink：写入 Doris 表"""

    def __init__(self, table: str):
        self._table = table
        self._fe_endpoints: List[str] = []
        self._batch_size: int = 1000
        self._flush_interval_ms: int = 5000

    def endpoints(self, eps: List[str]) -> DorisSink:
        self._fe_endpoints = eps
        return self

    def batch_size(self, size: int) -> DorisSink:
        self._batch_size = size
        return self


def ontology(object_type: str) -> OntologySink:
    return OntologySink(object_type)

def iceberg(table: str) -> IcebergSink:
    return IcebergSink(table)

def doris(table: str) -> DorisSink:
    return DorisSink(table)

class DataType(str, Enum):
    STRING = "string"
    INT32 = "int32"
    INT64 = "int64"
    FLOAT32 = "float32"
    FLOAT64 = "float64"
    BOOLEAN = "boolean"
    TIMESTAMP = "timestamp"
    DATE = "date"
    DECIMAL = "decimal"
    BINARY = "binary"
    ARRAY = "array"
    MAP = "map"
    STRUCT = "struct"

    def is_assignable_to(self, target: DataType) -> bool:
        """检查类型是否可以安全赋值"""
        promotions = {
            DataType.INT32: {DataType.INT64, DataType.FLOAT64, DataType.DECIMAL, DataType.STRING},
            DataType.INT64: {DataType.FLOAT64, DataType.DECIMAL, DataType.STRING},
            DataType.FLOAT32: {DataType.FLOAT64, DataType.STRING},
            DataType.FLOAT64: {DataType.STRING},
            DataType.BOOLEAN: {DataType.STRING},
            DataType.DATE: {DataType.TIMESTAMP, DataType.STRING},
        }
        if self == target:
            return True
        return target in promotions.get(self, set())


class SchemaField(BaseModel):
    name: str
    data_type: DataType
    nullable: bool = True
    comment: Optional[str] = None
    metadata: Dict[str, Any] = {}


class Schema(BaseModel):
    fields: List[SchemaField]

    def get_field(self, name: str) -> Optional[SchemaField]:
        return next((f for f in self.fields if f.name == name), None)

    def is_compatible(self, other: Schema) -> bool:
        """检查两个 Schema 是否兼容（用于 Union）"""
        if len(self.fields) != len(other.fields):
            return False
        for a, b in zip(self.fields, other.fields):
            if a.name != b.name:
                return False
            if not a.data_type.is_assignable_to(b.data_type):
                return False
        return True

    def generate_sample(self) -> Dict[str, Any]:
        """生成 Schema 的样例数据（用于 Map 变换的 Schema 推导）"""
        sample = {}
        defaults = {
            DataType.STRING: "sample",
            DataType.INT32: 0,
            DataType.INT64: 0,
            DataType.FLOAT32: 0.0,
            DataType.FLOAT64: 0.0,
            DataType.BOOLEAN: False,
            DataType.TIMESTAMP: datetime.utcnow(),
            DataType.DATE: date.today(),
        }
        for field in self.fields:
            sample[field.name] = defaults.get(field.data_type)
        return sample

    @staticmethod
    def from_dict(d: Dict[str, Any]) -> Schema:
        """从字典推导 Schema"""
        type_map = {
            str: DataType.STRING,
            int: DataType.INT64,
            float: DataType.FLOAT64,
            bool: DataType.BOOLEAN,
            datetime: DataType.TIMESTAMP,
            date: DataType.DATE,
        }
        fields = []
        for key, value in d.items():
            dt = type_map.get(type(value), DataType.STRING)
            fields.append(SchemaField(name=key, data_type=dt))
        return Schema(fields=fields)

class PipelineOptimizer:
    """Pipeline DAG 优化器"""

    def fuse_filters(self, dag: PipelineDAG) -> PipelineDAG:
        """合并连续的 Filter 节点"""
        nodes = dag.topological_sort()
        i = 0
        while i < len(nodes) - 1:
            if isinstance(nodes[i], FilterTransform) and isinstance(nodes[i + 1], FilterTransform):
                # 合并两个 Filter 为一个
                fused = FusedFilterTransform(nodes[i]._predicate, nodes[i + 1]._predicate)
                dag.replace_nodes([nodes[i], nodes[i + 1]], fused)
                nodes = dag.topological_sort()  # 重新排序
            else:
                i += 1
        return dag

    def push_down_predicates(self, dag: PipelineDAG) -> PipelineDAG:
        """谓词下推：将 Filter 尽可能推向数据源"""
        for node in dag.topological_sort():
            if isinstance(node, FilterTransform):
                # 检查是否可以下推到 Source
                pred = dag.predecessors(node)
                if pred and isinstance(pred[0], SourceNode):
                    if pred[0].supports_pushdown():
                        pred[0].add_pushdown_filter(node._predicate)
                        dag.remove_node(node)
        return dag

    def eliminate_projections(self, dag: PipelineDAG) -> PipelineDAG:
        """投影消除：只查询下游需要的字段"""
        # 从 Sink 反向推导需要的字段
        for node in reversed(dag.topological_sort()):
            if isinstance(node, SinkNode):
                required_fields = node.required_fields()
            elif isinstance(node, TransformNode):
                required_fields = node.required_input_fields(required_fields)
            elif isinstance(node, SourceNode):
                node.set_projection(required_fields)
        return dag


class FusedFilterTransform(FilterTransform):
    """合并后的 Filter"""

    def __init__(self, pred1: Callable, pred2: Callable):
        super().__init__(lambda row: pred1(row) and pred2(row))
        self._original_predicates = [pred1, pred2]

class PipelineCompiler(ABC):
    """Pipeline 编译器基类"""

    @abstractmethod
    def compile(self, dag: PipelineDAG) -> ExecutionPlan:
        ...


class FlinkCompiler(PipelineCompiler):
    """Flink 编译器"""

    def compile(self, dag: PipelineDAG) -> FlinkExecutionPlan:
        env = StreamExecutionEnvironment.get_execution_environment()
        env.set_parallelism(dag.config.parallelism)
        env.enable_checkpointing(dag.config.checkpoint_interval_ms)

        flink_nodes = {}
        for node in dag.topological_sort():
            if isinstance(node, MySQLCDCSource):
                flink_nodes[node.id] = self._compile_mysql_source(env, node)
            elif isinstance(node, FilterTransform):
                input_stream = flink_nodes[dag.predecessors(node)[0].id]
                flink_nodes[node.id] = input_stream.filter(
                    FlinkFilterFunction(node._predicate)
                )
            elif isinstance(node, MapTransform):
                input_stream = flink_nodes[dag.predecessors(node)[0].id]
                flink_nodes[node.id] = input_stream.map(
                    FlinkMapFunction(node._mapper)
                )
            elif isinstance(node, OntologySink):
                input_stream = flink_nodes[dag.predecessors(node)[0].id]
                input_stream.add_sink(FlinkOntologySink(node))

        return FlinkExecutionPlan(env)


class SparkCompiler(PipelineCompiler):
    """Spark 编译器"""

    def compile(self, dag: PipelineDAG) -> SparkExecutionPlan:
        spark = SparkSession.builder.appName(dag.name).getOrCreate()

        spark_dfs = {}
        for node in dag.topological_sort():
            if isinstance(node, SourceNode):
                spark_dfs[node.id] = self._compile_source(spark, node)
            elif isinstance(node, FilterTransform):
                input_df = spark_dfs[dag.predecessors(node)[0].id]
                spark_dfs[node.id] = input_df.filter(
                    self._to_spark_expression(node._predicate)
                )
            elif isinstance(node, MapTransform):
                input_df = spark_dfs[dag.predecessors(node)[0].id]
                spark_dfs[node.id] = self._apply_map(input_df, node._mapper)
            elif isinstance(node, SinkNode):
                input_df = spark_dfs[dag.predecessors(node)[0].id]
                self._compile_sink(input_df, node)

        return SparkExecutionPlan(spark)


class DuckDBCompiler(PipelineCompiler):
    """DuckDB 编译器（用于本地开发和测试）"""

    def compile(self, dag: PipelineDAG) -> DuckDBExecutionPlan:
        import duckdb
        conn = duckdb.connect()

        sql_parts = []
        for node in dag.topological_sort():
            if isinstance(node, SourceNode):
                sql_parts.append(self._source_to_sql(node))
            elif isinstance(node, FilterTransform):
                sql_parts.append(f"WHERE {self._predicate_to_sql(node._predicate)}")
            elif isinstance(node, AggregateTransform):
                sql_parts.append(self._aggregate_to_sql(node))

        full_sql = self._assemble_sql(sql_parts)
        return DuckDBExecutionPlan(conn, full_sql)

+------------------------------------------------------------------+
|  引擎选择策略                                                      |
|                                                                   |
|  场景                          推荐引擎    原因                    |
|  ─────────────────            ────────    ─────────────────       |
|  实时 CDC 同步                 Flink       流处理原生支持           |
|  批量 ETL（T+1）               Spark       批处理性能优越           |
|  本地开发/测试                 DuckDB      零依赖，毫秒级启动       |
|  小数据量临时分析              DuckDB      嵌入式，无需集群         |
|  大规模批处理 + 流处理混合      Flink       统一批流引擎             |
+------------------------------------------------------------------+

class PipelineVisualizer:
    """Pipeline DAG 可视化"""

    def to_mermaid(self, dag: PipelineDAG) -> str:
        """生成 Mermaid 格式的 DAG 图"""
        lines = ["graph LR"]
        for node in dag.nodes():
            label = self._node_label(node)
            lines.append(f"    {node.id}[{label}]")
        for edge in dag.edges():
            lines.append(f"    {edge.source.id} --> {edge.target.id}")
        return "\n".join(lines)

    def to_ascii(self, dag: PipelineDAG) -> str:
        """生成 ASCII 格式的 DAG 图"""
        nodes = dag.topological_sort()
        lines = []
        for i, node in enumerate(nodes):
            prefix = "    " * node.depth
            connector = "|-- " if i > 0 else ""
            lines.append(f"{prefix}{connector}{self._node_label(node)}")
        return "\n".join(lines)

    def _node_label(self, node) -> str:
        if isinstance(node, SourceNode):
            return f"Source: {node.__class__.__name__}"
        elif isinstance(node, TransformNode):
            return f"Transform: {node.__class__.__name__}"
        elif isinstance(node, SinkNode):
            return f"Sink: {node.__class__.__name__}"
        return str(node)

class PipelineDryRunner:
    """Pipeline Dry Run：不实际执行，只验证和预览"""

    async def dry_run(self, pipeline: CompiledPipeline) -> DryRunReport:
        """执行 Dry Run"""
        report = DryRunReport(pipeline_name=pipeline.name)

        # 1. Schema 验证
        schema_errors = self._validate_schemas(pipeline.dag)
        report.schema_errors = schema_errors

        # 2. 数据源连通性检查
        connectivity = await self._check_connectivity(pipeline.dag)
        report.connectivity = connectivity

        # 3. 预估数据量
        estimated_rows = await self._estimate_rows(pipeline.dag)
        report.estimated_rows = estimated_rows

        # 4. 预估执行时间
        estimated_duration = self._estimate_duration(pipeline.dag, estimated_rows)
        report.estimated_duration = estimated_duration

        # 5. 生成执行计划预览
        report.execution_plan = pipeline.to_execution_plan_preview()

        return report


class DryRunReport(BaseModel):
    pipeline_name: str
    schema_errors: List[str] = []
    connectivity: Dict[str, bool] = {}
    estimated_rows: int = 0
    estimated_duration: str = ""
    execution_plan: str = ""
    is_valid: bool = True

    def __str__(self):
        lines = [
            f"Pipeline: {self.pipeline_name}",
            f"Valid: {self.is_valid}",
            f"Schema Errors: {len(self.schema_errors)}",
            f"Estimated Rows: {self.estimated_rows:,}",
            f"Estimated Duration: {self.estimated_duration}",
            "",
            "Connectivity:",
        ]
        for source, ok in self.connectivity.items():
            status = "OK" if ok else "FAIL"
            lines.append(f"  {source}: {status}")
        return "\n".join(lines)

# 一个 Source 分流到多个 Sink
pipeline = (
    Pipeline.create("multi-sink-pipeline")
    .source(mysql_cdc("factory_db.sensor_data"))
    .transform(filter(lambda r: r["value"] is not None))
    .branch(
        # 分支 1：写入 Iceberg（全量存档）
        lambda p: p.sink(iceberg("warehouse.sensor_archive")),
        # 分支 2：聚合后写入 Doris（实时仪表盘）
        lambda p: p
            .transform(
                window("tumbling", "5m")
            )
            .transform(
                aggregate()
                .group_by("sensor_id")
                .agg("value", "avg", "avg_value")
                .agg("value", "max", "max_value")
            )
            .sink(doris("realtime.sensor_dashboard")),
    )
    .build()
)

# 子 Pipeline 作为可复用的组件
cleansing_pipeline = (
    Pipeline.fragment("data-cleansing")
    .transform(filter(lambda r: r["id"] is not None))
    .transform(map(lambda r: {**r, "name": r["name"].strip().title()}))
    .transform(filter(lambda r: len(r["name"]) > 0))
)

# 在主 Pipeline 中使用
main_pipeline = (
    Pipeline.create("main-pipeline")
    .source(mysql_cdc("crm.customers"))
    .include(cleansing_pipeline)  # 嵌入子 Pipeline
    .sink(ontology("Customer"))
    .build()
)

pipeline = (
    Pipeline.create("daily-sync")
    .source(csv_file("s3://data-lake/daily-export/*.csv"))
    .transform(map(lambda r: {**r, "import_date": date.today().isoformat()}))
    .sink(ontology("DailyReport"))
    .schedule(cron="0 2 * * *")  # 每天凌晨 2 点
    .build()
)

+------------------------------------------------------------------+
|  Pipeline DSL 对比                                                |
|                                                                   |
|  能力                  Palantir Transforms    智策 Pipeline DSL    |
|  ──────────────       ──────────────────    ─────────────────     |
|  Python API            ✅                    ✅                   |
|  链式调用              ❌ (装饰器模式)        ✅                   |
|  类型安全              部分                   ✅ (Schema 推导)     |
|  多引擎支持            ❌ (Spark only)        ✅ (Flink/Spark/DuckDB)|
|  Schema 推导           ❌                    ✅                   |
|  DAG 可视化            ✅                    ✅                   |
|  版本控制              ✅ (Code Repo)         ✅ (Git)             |
|  Dry Run              ✅                    ✅                   |
|  分支 Pipeline         ✅                    ✅                   |
|  增量 / CDC            ✅                    ✅                   |
|  开源                  ❌                    ✅                   |
+------------------------------------------------------------------+

class TestPipelineDSL:

    def test_basic_pipeline_build(self):
        """验证基本 Pipeline 构建"""
        pipeline = (
            Pipeline.create("test")
            .source(csv_file("/data/test.csv"))
            .transform(filter(lambda r: r["age"] > 18))
            .sink(ontology("Person"))
            .build()
        )
        assert pipeline.name == "test"
        assert len(pipeline.dag.nodes()) == 3

    def test_schema_inference(self):
        """验证 Schema 推导"""
        source = csv_file("/data/test.csv").schema(Schema(fields=[
            SchemaField(name="name", data_type=DataType.STRING),
            SchemaField(name="age", data_type=DataType.INT32),
        ]))
        pipeline = (
            Pipeline.create("test")
            .source(source)
            .transform(map(lambda r: {**r, "is_adult": r["age"] >= 18}))
            .sink(ontology("Person"))
            .build()
        )
        output_schema = pipeline.dag.get_node_by_index(1).output_schema()
        assert output_schema.get_field("is_adult") is not None

    def test_schema_type_mismatch(self):
        """验证类型不匹配时报错"""
        with pytest.raises(SchemaError):
            Pipeline.create("test") \
                .source(csv_file("/data/test.csv").schema(Schema(fields=[
                    SchemaField(name="name", data_type=DataType.STRING),
                ]))) \
                .sink(ontology("Person").property_mapping({"name": "age"})) \
                .build()

    def test_multi_engine_compilation(self):
        """验证多引擎编译"""
        pipeline = (
            Pipeline.create("test")
            .source(csv_file("/data/test.csv"))
            .transform(filter(lambda r: r["status"] == "active"))
            .sink(doris("test.output"))
            .build()
        )
        # 可以编译为不同引擎
        flink_plan = FlinkCompiler().compile(pipeline.dag)
        assert flink_plan is not None

        duckdb_plan = DuckDBCompiler().compile(pipeline.dag)
        assert duckdb_plan is not None

    def test_filter_fusion_optimization(self):
        """验证 Filter 合并优化"""
        pipeline = (
            Pipeline.create("test")
            .source(csv_file("/data/test.csv"))
            .transform(filter(lambda r: r["age"] > 18))
            .transform(filter(lambda r: r["status"] == "active"))
            .sink(doris("test.output"))
            .build()
        )
        # 两个 Filter 应被合并为一个
        filter_nodes = [n for n in pipeline.dag.nodes() if isinstance(n, FilterTransform)]
        assert len(filter_nodes) == 1