jason810496 commented on code in PR #65956: URL: https://github.com/apache/airflow/pull/65956#discussion_r3158812381
########## java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt: ########## @@ -0,0 +1,294 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk.execution + +import org.apache.airflow.sdk.Dag +import org.apache.airflow.sdk.Task +import java.nio.file.Path +import java.time.Duration +import java.time.Instant + +/** + * Serialization logic decoupled from user-facing SDK classes. + * + * Produces output compatible with Python Airflow's DagSerialization format (version 3). + */ +typealias Serialized = Map<String, Any?> + +private object SerdeScope + +private val logger = Logger(SerdeScope::class) + +// --------------------------------------------------------------------------- +// Value encoding — matches Python's BaseSerialization.serialize +// --------------------------------------------------------------------------- + +/** + * Recursively serialize a value with Airflow's type/var encoding. + * + * Primitives pass through; complex types are wrapped in {"__type": ..., "__var": ...}. + * This matches the Python BaseSerialization.serialize() output exactly: + * - dict -> {"__type": "dict", "__var": {k: serialize(v), ...}} + * - set -> {"__type": "set", "__var": [sorted items]} + * - datetime -> {"__type": "datetime", "__var": epoch_seconds} + * - timedelta -> {"__type": "timedelta", "__var": total_seconds} + */ Review Comment: IMHO, having multi-lang serialization like gRPC wouldn't help here due to Airflow internal SDK serialization. The goal of SDK serialization for each language SDK here is to translate the user-facing API (e.g. the `@task` for Python, `func transform(ctx context.Context, client sdk.VariableClient, log *slog.Logger) error` for Go, `public static class Transform implements Task {` for Java) into some kind of format for representing "what is the Dag and Task structure". There're couple of reason that lead to having SDK-compatible serialization is the only way to achieve it. 1. The IPC format of `DagFileParseRequest` for dag-processor 2. If we use ProtoBuf as IPC format, this means we're designing another intermediate format for representing the final SDK serialized DAG JSON format. 3. Each language SDK still require to implement the serialization between user-facing API and the ProtoBuf. The https://github.com/apache/airflow/pull/65956#discussion_r3158477228 comment also point out the current restriction (we're using custom serialization for Python SDK already) and should answer partial of the question. -- 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]
