From 98575b4967c49374ee5adb088f3983d1fb71a358 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:33:06 +0200 Subject: [PATCH 1/5] fix(local): land the last build that produced a table, not the raw input _determine_source_table took the output table of the contract's last build. When that build produced none in the local planner (a dbt model, which it does not run), the fallback picked the first table it had loaded, an input, and the expose landed it raw: columns the expose does not declare, PII among them, included. It now takes the output of the last build that has one. --- fluid_build/providers/local/planner.py | 11 ++-- .../test_local_planner_source_table.py | 59 +++++++++++++++++++ 2 files changed, 65 insertions(+), 5 deletions(-) create mode 100644 tests/providers/test_local_planner_source_table.py diff --git a/fluid_build/providers/local/planner.py b/fluid_build/providers/local/planner.py index 34eb7b88..5a94de3b 100644 --- a/fluid_build/providers/local/planner.py +++ b/fluid_build/providers/local/planner.py @@ -511,7 +511,7 @@ def _determine_source_table( Priority: 1. expose.source_table (explicit) - 2. Last build's output table + 2. The output table of the last build that produced one 3. Table matching expose ID 4. None (will fail at apply time) """ @@ -520,10 +520,11 @@ def _determine_source_table( if source: return source - # Use last build's output - if builds: - last_build = builds[-1] - output_table = last_build.get("output_table") or f"result_{last_build.get('id', '0')}" + # The last build that produced a table: one this planner skipped (a Flink + # job, a dbt model) produced none, and taking its absence for "no build + # output" landed the first loaded input, raw, in the expose. + for build in reversed(builds): + output_table = build.get("output_table") or f"result_{build.get('id', '0')}" if output_table in loaded_tables: return output_table diff --git a/tests/providers/test_local_planner_source_table.py b/tests/providers/test_local_planner_source_table.py new file mode 100644 index 00000000..c9e90830 --- /dev/null +++ b/tests/providers/test_local_planner_source_table.py @@ -0,0 +1,59 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed 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. + +"""The local planner lands the last build that produced a table. + +It took the output of the contract's last build. A last build the local +planner does not run (a dbt model, for one) produced no table, so it fell back +to the first input it had loaded and landed that, raw, in the expose: columns +the expose does not declare, PII among them, included. +""" + +from __future__ import annotations + +import logging + +from fluid_build.providers.local.planner import plan_actions + + +def test_a_trailing_build_without_a_table_does_not_land_the_raw_input() -> None: + contract = { + "fluidVersion": "0.7.5", + "id": "demo.customers", + "builds": [ + { + "id": "clean", + "engine": "sql", + "pattern": "embedded-logic", + "properties": { + "sql": "SELECT id FROM customers", + "parameters": {"inputs": [{"name": "customers", "path": "data/customers.csv"}]}, + }, + }, + {"id": "docs", "engine": "dbt", "pattern": "hybrid-reference"}, + ], + "exposes": [ + { + "exposeId": "customers_clean", + "binding": { + "platform": "local", + "format": "csv", + "location": {"path": "out/clean.csv"}, + }, + } + ], + } + actions = plan_actions(contract, logger=logging.getLogger("t")) + (materialize,) = [a for a in actions if a.get("op") == "materialize"] + assert materialize["payload"]["source_table"] == "result_clean" From 8389cec3a7a7da27ac9df75c43c5dadbce6825e9 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:45:34 +0200 Subject: [PATCH 2/5] feat(schema): engine: flink in the 0.7.6 preview, run outside forge-cli A build can now declare Apache Flink as its engine. forge-cli validates and plans it but ships no runner: the job runs on the Flink Kubernetes Operator, AWS Managed Service for Apache Flink, or whatever platform runs it. 0.7.5 is unchanged. Deployment settings go under properties.parameters, because embeddedLogicPattern is closed; the schema description says so and names the operator's job spec keys. Before, the only option was engine: custom or engine: sql with properties.language: flink_sql, and either took the inline-SQL path: is_embedded_sql_build never looked at the language, so `fluid apply --mode amend-and-build` ran the Flink SQL on DuckDB when no platform was declared, refused kubernetes as an unknown embedded-SQL platform, and the local, AWS, GCP and Snowflake planners turned it into DuckDB, Athena, BigQuery or Snowflake statements. util/contract.is_externally_run_build is the one test: engine flink, or properties.language flink_sql, the way a 0.7.5 contract declares the same job. Every consumer that ran or planned a build's SQL checks it: the build runner, the local planner and provider, and the AWS, GCP and Snowflake planners (Snowflake's replace mode still ensures a Flink output table, since no CTAS recreates it). The runner reports such a build on its own summary line, not as a skip, so a contract whose only builds run elsewhere is not "every build was skipped"; the run report records it as skipped with reason runs_outside_forge. The local planner lands nothing for an expose such a build writes, instead of a loaded input in its place; when builds[].outputs does not say whether a Flink build or a local one writes an expose, it lands neither and warns rather than guess. Scheduler synthesis and the Airflow fluid-apply DAGs give it no task, and a contract with no other build and no hand-declared orchestration tasks has nothing to schedule (`fluid generate schedule` exits 0; an unknown --scheduler is still an error); `fluid generate transformation` says there is no code to generate for it. Other properties.language values keep the behaviour they had. --- fluid_build/build_runners/base.py | 42 +- fluid_build/cli/_apply_cc_report.py | 5 + fluid_build/cli/generate_schedule.py | 21 + .../cli/generate_speed_transformation.py | 26 + fluid_build/providers/aws/plan/planner.py | 6 +- fluid_build/providers/gcp/plan/planner.py | 4 +- fluid_build/providers/local/local.py | 7 +- fluid_build/providers/local/planner.py | 47 ++ .../providers/snowflake/plan/planner.py | 10 +- fluid_build/schedulers/airflow/fluid_apply.py | 5 + fluid_build/schedulers/synthesis.py | 10 +- fluid_build/schemas/fluid-schema-0.7.6.json | 5 +- fluid_build/util/contract.py | 31 ++ tests/test_flink_engine.py | 521 ++++++++++++++++++ 14 files changed, 727 insertions(+), 13 deletions(-) create mode 100644 tests/test_flink_engine.py diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py index 4ebe432a..c0d3dd04 100644 --- a/fluid_build/build_runners/base.py +++ b/fluid_build/build_runners/base.py @@ -38,6 +38,7 @@ from fluid_build._console import error as console_error from fluid_build._contract_loader import CLIError, load_contract_with_overlay from fluid_build.util.binding_paths import ENV_PLACEHOLDER_RE +from fluid_build.util.contract import describe_externally_run_build, is_externally_run_build LOG = logging.getLogger("fluid.build_runners") @@ -87,6 +88,24 @@ def is_dbt_build(build: Dict[str, Any]) -> bool: return engine == "dbt" or engine.startswith("dbt-") +def _report_externally_run_build(build: Dict[str, Any], build_id: str, report: Any) -> None: + """Say that a build runs on its own platform; it is not run here. + + A Flink job runs on the Flink Kubernetes Operator or AWS Managed Service + for Apache Flink. forge-cli has no runner for it, and its SQL is not + DuckDB's or a warehouse's. The summary counts it on its own line, apart + from skips; the run report, whose statuses are succeeded, failed and + skipped, records it as skipped with ``reason: runs_outside_forge``. + """ + platform = embedded_sql_platform(build) or "its own platform" + cprint( + f"\n⏭️ Build '{build_id}' - {describe_externally_run_build(build)} runs on " + f"{platform}, outside forge-cli; there is nothing for forge-cli to run." + ) + if report is not None: + report.record_build(build_id=build_id, status="skipped", reason="runs_outside_forge") + + def is_embedded_sql_build(build: Dict[str, Any]) -> bool: """Return True for builds with inline SQL (``engine: sql`` or ``pattern: embedded-logic`` + inline ``properties.sql``). @@ -95,14 +114,19 @@ def is_embedded_sql_build(build: Dict[str, Any]) -> bool: by the runtime platform the build declares — the local provider's DuckDB engine for ``platform: local`` (or an unset platform), the warehouse itself for a declared warehouse (see - :data:`_EMBEDDED_SQL_EXECUTORS`). They do NOT require an external + :func:`_execute_embedded_sql_build`). They do NOT require an external Python script, so the python-runner's ``resolve_script_path`` lookup is the wrong dispatch — it always returns ``None`` and emits the confusing "Script not found: ingest.py" warning. Bug A4-A: ``--mode amend-and-build`` delegated to ``run_builds_from_args`` which fell through to the Python-runner branch for embedded-SQL builds. + + A Flink build (``engine: flink``, or ``properties.language: flink_sql``) + is not one: its SQL is Flink's dialect, and the job runs outside forge-cli. """ + if is_externally_run_build(build): + return False engine = (build.get("engine") or "").strip().lower() pattern = (build.get("pattern") or "").strip().lower() if engine == "sql": @@ -1171,11 +1195,19 @@ def run_builds_from_args( total_executed = 0 total_failed = 0 total_skipped = 0 + # Builds whose engine runs on another platform: not run here, and not a + # skip either, which means a missing artifact (see the guard below). + total_external = 0 for build in builds: build_id = build.get("id", "unknown") runs_before = _run_ids(contract_path.parent, product_id, build_id) if report else None + if is_externally_run_build(build): + _report_externally_run_build(build, build_id, report) + total_external += 1 + continue + if is_acquisition_build(build): sample_rows = getattr(args, "sample_rows", None) # A plan's contract had its ``{{ env.* }}`` templates resolved @@ -1297,6 +1329,8 @@ def run_builds_from_args( success(f"Executed: {total_executed}") console_error(f"Failed: {total_failed}") cprint(f"⏭️ Skipped: {total_skipped}") + if total_external: + cprint(f"🛰️ Run outside forge-cli: {total_external}") cprint(f"{'=' * 80}\n") if total_failed: @@ -1319,12 +1353,14 @@ def run_builds_from_args( if report is not None: report.build_failed("builds_all_skipped") console_error( - f"Every build was skipped ({total_skipped}/{len(builds)}) — nothing " + f"Every build was skipped ({total_skipped}/{len(builds) - total_external}) — nothing " "was transformed and no rows were produced. Fix the missing build " "artifact(s) reported above, or pass --allow-skipped-builds if the " "skip is expected." ) - LOG.error("build.all_skipped skipped=%d total=%d", total_skipped, len(builds)) + LOG.error( + "build.all_skipped skipped=%d total=%d", total_skipped, len(builds) - total_external + ) return 1 return 0 diff --git a/fluid_build/cli/_apply_cc_report.py b/fluid_build/cli/_apply_cc_report.py index 399207a1..2b872086 100644 --- a/fluid_build/cli/_apply_cc_report.py +++ b/fluid_build/cli/_apply_cc_report.py @@ -190,14 +190,19 @@ def record_build( build_id: str, status: str, run: Optional[Mapping[str, Any]] = None, + reason: Optional[str] = None, ) -> None: """One build of a build-augmented apply: ``succeeded``, ``failed`` or ``skipped``. ``run`` is the run record the build wrote, if it wrote one; only its id, the table it loaded, where it landed and the rows are kept. + ``reason`` says why a build was not run, such as ``runs_outside_forge`` + for a job that runs on its own platform. """ try: entry: Dict[str, Any] = {"build_id": str(build_id), "status": str(status)} + if reason: + entry["reason"] = str(reason) entry.update(_build_facts(run)) self.result.setdefault("builds", []).append(entry) if status == "failed": diff --git a/fluid_build/cli/generate_schedule.py b/fluid_build/cli/generate_schedule.py index 2a4dd126..8a72412f 100644 --- a/fluid_build/cli/generate_schedule.py +++ b/fluid_build/cli/generate_schedule.py @@ -50,6 +50,14 @@ SUBCOMMAND = "schedule" +def _only_externally_run_builds(contract: Dict[str, Any]) -> bool: + """True when the contract has builds and every one runs outside forge-cli.""" + from fluid_build.util.contract import is_externally_run_build + + builds = [b for b in contract.get("builds") or [] if isinstance(b, dict)] + return bool(builds) and all(is_externally_run_build(b) for b in builds) + + def register_subcommand(subparsers: argparse._SubParsersAction): """Register the schedule subcommand under ``fluid generate``.""" p = subparsers.add_parser( @@ -157,6 +165,19 @@ def run(args: Any, logger: logging.Logger) -> int: # the operator names, so defaulting WHO renders it commits nothing. scheduler_name = fluid_apply.DEFAULT_ENGINE + # A contract whose every build runs on its own platform (a Flink job) + # has nothing to schedule, which is not an error: unless it hand-declares + # orchestration tasks, or names a scheduler that does not exist. + if _only_externally_run_builds(contract) and not fluid_apply.has_explicit_tasks(contract): + from fluid_build.schedulers import has_scheduler + + if not scheduler_name or has_scheduler(scheduler_name): + cprint( + "Every build runs outside forge-cli (a Flink job, for example), so " + "there is nothing to schedule. No files written." + ) + return 0 + if not scheduler_name: cprint( "No scheduler engine specified.\n" diff --git a/fluid_build/cli/generate_speed_transformation.py b/fluid_build/cli/generate_speed_transformation.py index 18847324..13dbc6d5 100644 --- a/fluid_build/cli/generate_speed_transformation.py +++ b/fluid_build/cli/generate_speed_transformation.py @@ -221,6 +221,9 @@ def run(args: Any, logger: logging.Logger) -> int: if getattr(args, "all_builds", False) else [(build_index, builds[build_index])] ) + selected_builds = _drop_externally_run_builds(selected_builds) + if not selected_builds: + return 0 if getattr(args, "all_builds", False): results = _generate_all_builds( args, @@ -521,6 +524,29 @@ def _topological_build_waves( return waves +def _drop_externally_run_builds( + selected: List[Tuple[int, Dict[str, Any]]], +) -> List[Tuple[int, Dict[str, Any]]]: + """Leave out builds whose job runs outside forge-cli, saying so for each. + + A Flink job's SQL is inline and runs on Flink: there is no transformation + code for forge-cli to generate. + """ + from fluid_build.util.contract import describe_externally_run_build, is_externally_run_build + + kept = [] + for index, build in selected: + if is_externally_run_build(build): + cprint( + f"Build '{build.get('id', index)}' runs outside forge-cli " + f"({describe_externally_run_build(build)}); there is no transformation " + "code to generate for it." + ) + else: + kept.append((index, build)) + return kept + + def _generate_single_build( args: Any, logger: logging.Logger, diff --git a/fluid_build/providers/aws/plan/planner.py b/fluid_build/providers/aws/plan/planner.py index b82a200c..9f78013d 100644 --- a/fluid_build/providers/aws/plan/planner.py +++ b/fluid_build/providers/aws/plan/planner.py @@ -27,6 +27,7 @@ from typing import Any, Dict, List, Optional from fluid_build.providers._planner_base import BasePlanner +from fluid_build.util.contract import is_externally_run_build from ..util import warehouse as _warehouse from ..util.logging import format_event @@ -560,11 +561,12 @@ def _plan_build_transformations( ) elif pattern == "embedded-logic": - # Inline SQL execution via Athena + # Inline SQL execution via Athena. Flink SQL is not Athena SQL: that + # job runs outside forge-cli. props = build_config.get("properties", {}) sql = props.get("sql") - if sql: + if sql and not is_externally_run_build(build_config): actions.append( { "op": "athena.execute_query", diff --git a/fluid_build/providers/gcp/plan/planner.py b/fluid_build/providers/gcp/plan/planner.py index 64f70c50..a4d4ce62 100644 --- a/fluid_build/providers/gcp/plan/planner.py +++ b/fluid_build/providers/gcp/plan/planner.py @@ -27,6 +27,7 @@ from fluid_build.providers._planner_base import BasePlanner from fluid_build.providers._sql_safety import validate_ident +from fluid_build.util.contract import is_externally_run_build from ..util.logging import format_event from ..util.names import normalize_bucket_name, normalize_dataset_name @@ -429,7 +430,8 @@ def _plan_build_transformations( # the table's rows (destructive). Multi-output builds emit one action # per output. for build_idx, build_entry in enumerate(contract.get("builds", []) or []): - if not isinstance(build_entry, Mapping): + # Flink SQL is not BigQuery SQL: that job runs outside forge-cli. + if not isinstance(build_entry, Mapping) or is_externally_run_build(build_entry): continue props = build_entry.get("properties", {}) if not isinstance(props, Mapping): diff --git a/fluid_build/providers/local/local.py b/fluid_build/providers/local/local.py index 1dd81ea8..52e1bee5 100644 --- a/fluid_build/providers/local/local.py +++ b/fluid_build/providers/local/local.py @@ -620,10 +620,15 @@ def _derive_actions_from_contract(self, contract: Dict[str, Any]) -> List[Dict[s - Write to exposes[0] path if provided, else default under runtime/out/. """ # Import contract utilities for version-agnostic access - from fluid_build.util.contract import get_primary_build + from fluid_build.util.contract import get_primary_build, is_externally_run_build contract = self._anchored(contract) build = get_primary_build(contract) + if build and is_externally_run_build(build): + # Flink SQL is not DuckDB's to run, and the demo fallback below + # must not stand in for the job. + self._log_info("local_build_runs_outside_forge", {"engine": build.get("engine")}) + return [{"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"}] sql_text: Optional[str] = None inputs_spec: List[Any] = [] diff --git a/fluid_build/providers/local/planner.py b/fluid_build/providers/local/planner.py index 5a94de3b..39e122de 100644 --- a/fluid_build/providers/local/planner.py +++ b/fluid_build/providers/local/planner.py @@ -50,6 +50,7 @@ get_expose_location, get_exposes, get_primary_build, + is_externally_run_build, ) @@ -217,6 +218,12 @@ def plan_actions( ) continue + if is_externally_run_build(build): # Flink SQL runs outside forge-cli + log.debug( + "local planner: build '%s' (engine=%s) runs outside forge-cli", build_id, engine + ) + continue + # Extract SQL from various formats sql_text = _extract_sql(build, contract, log) @@ -341,6 +348,8 @@ def plan_actions( for _ei in _expose_iter: expose_id = _ei["_id"] out_path = _ei["_path"] + if _produced_outside_forge(expose_id, builds, log): + continue if not out_path: # Generate default path @@ -503,6 +512,42 @@ def _extract_input_tables( return inputs +def _produced_outside_forge(expose_id: Any, builds: List[Dict[str, Any]], log: Any) -> bool: + """True when forge-cli must not land ``expose_id``: a job that runs outside + forge-cli (a Flink job) writes it, or may. + + The builds that write it are the ones whose ``outputs`` name it; when none + does, the ones that declare no ``outputs`` (a build that declares them has + said what it writes). When all of those run outside forge-cli, the expose is + theirs. When only some do, which build writes it is ambiguous, and landing + a local table under its name could publish another build's rows, so it is + not landed and the warning asks for ``outputs``. + """ + dict_builds = [b for b in builds if isinstance(b, dict)] + producers = [ + b for b in dict_builds if isinstance(b.get("outputs"), list) and expose_id in b["outputs"] + ] or [b for b in dict_builds if not isinstance(b.get("outputs"), list)] + outside = [b for b in producers if is_externally_run_build(b)] + if not outside: + return False + if len(outside) == len(producers): + log.info( + "local planner: expose '%s' is written by a job that runs outside forge-cli; " + "nothing to land", + expose_id, + ) + else: + log.warning( + "local planner: expose '%s' is not landed: build(s) %s run outside forge-cli " + "and may write it, as may %s. Declare builds[].outputs to say which build " + "writes which expose.", + expose_id, + ", ".join(repr(b.get("id")) for b in outside), + ", ".join(repr(b.get("id")) for b in producers if b not in outside), + ) + return True + + def _determine_source_table( expose: Dict[str, Any], builds: List[Dict[str, Any]], loaded_tables: Set[str], logger ) -> Optional[str]: @@ -524,6 +569,8 @@ def _determine_source_table( # job, a dbt model) produced none, and taking its absence for "no build # output" landed the first loaded input, raw, in the expose. for build in reversed(builds): + if not isinstance(build, dict): + continue output_table = build.get("output_table") or f"result_{build.get('id', '0')}" if output_table in loaded_tables: return output_table diff --git a/fluid_build/providers/snowflake/plan/planner.py b/fluid_build/providers/snowflake/plan/planner.py index e64b6a77..6bdbe9e1 100644 --- a/fluid_build/providers/snowflake/plan/planner.py +++ b/fluid_build/providers/snowflake/plan/planner.py @@ -42,6 +42,7 @@ validate_ident, validate_sql_expression_allowlist, ) +from fluid_build.util.contract import is_externally_run_build from ..util.config import resolve_env_templates as _resolve_env_templates from ..util.metadata import extract_snowflake_tags @@ -792,7 +793,8 @@ def _plan_build( # Modern builds[] support for native SQL happy-path contracts. for index, build_entry in enumerate(contract.get("builds", []) or []): - if not isinstance(build_entry, Mapping): + # Flink SQL is not Snowflake SQL: that job runs outside forge-cli. + if not isinstance(build_entry, Mapping) or is_externally_run_build(build_entry): continue properties = build_entry.get("properties", {}) @@ -969,8 +971,10 @@ def _plan_expose( if not isinstance(build_entry, Mapping): continue outputs = build_entry.get("outputs") or [] - if outputs and ( - build_entry.get("sql") or (build_entry.get("properties") or {}).get("sql") + if ( + outputs + and not is_externally_run_build(build_entry) + and (build_entry.get("sql") or (build_entry.get("properties") or {}).get("sql")) ): sql_build_targets.update(outputs) diff --git a/fluid_build/schedulers/airflow/fluid_apply.py b/fluid_build/schedulers/airflow/fluid_apply.py index c7771df1..d7218ba8 100644 --- a/fluid_build/schedulers/airflow/fluid_apply.py +++ b/fluid_build/schedulers/airflow/fluid_apply.py @@ -72,6 +72,7 @@ from fluid_build.build_runners._ids import IdentifierViolation, validate_identifier from fluid_build.providers.common.codegen_utils import py_str_literal +from fluid_build.util.contract import is_externally_run_build #: Engine used for a scheduled build when the contract names none. DEFAULT_ENGINE = "airflow" @@ -242,6 +243,10 @@ class ScheduledBuild: def _trigger_schedule(build: Dict[str, Any]) -> Optional[Dict[str, Any]]: """The build's cron trigger, or ``None`` when Airflow should not schedule it.""" + # A Flink job runs on its own platform: a DAG that runs ``fluid apply`` for + # it would run nothing and pass. + if is_externally_run_build(build): + return None execution = build.get("execution") if not isinstance(execution, dict): return None diff --git a/fluid_build/schedulers/synthesis.py b/fluid_build/schedulers/synthesis.py index 6279b8d7..3bd681f9 100644 --- a/fluid_build/schedulers/synthesis.py +++ b/fluid_build/schedulers/synthesis.py @@ -25,6 +25,8 @@ import re from typing import Any, Dict, List, Optional +from fluid_build.util.contract import is_externally_run_build + # Provider + engine → scheduler action mapping. # Keys are (provider, engine) tuples; values are the action string used # by the scheduler generators to pick the right operator. @@ -112,7 +114,13 @@ def synthesize_orchestration_from_builds( The returned dict is suitable for injection into the contract before calling a scheduler's ``validate()`` and ``generate()`` methods. """ - builds: List[Dict[str, Any]] = contract.get("builds", []) + # A Flink job runs continuously on its own platform: there is nothing to + # schedule for it, and a nightly task that "runs" it would only log its SQL. + builds: List[Dict[str, Any]] = [ + b + for b in contract.get("builds", []) or [] + if not (isinstance(b, dict) and is_externally_run_build(b)) + ] if not builds: return {} diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 2fc32fbd..3168d074 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -2878,7 +2878,8 @@ "meltano", "dlt", "kafka-connect", - "debezium" + "debezium", + "flink" ] }, { @@ -2886,7 +2887,7 @@ } ], "default": "dbt", - "description": "Build/ingestion engine. Transformation engines: dbt, sql, python, spark, glue, custom. Adapter-qualified dbt engines (dbt-, e.g. dbt-glue, dbt-snowflake, dbt-bigquery) are accepted too and route through the dbt build runner, which derives the adapter from the suffix. Ingestion engines (acquisition pattern only): duckdb (zero-infra files/JDBC), airbyte (350+ SaaS), meltano (Singer 600+ taps), dlt (Python-native), kafka-connect (streaming), debezium (CDC). AWS Glue: glue (Spark ETL jobs managed by Glue)." + "description": "Build/ingestion engine. Transformation engines: dbt, sql, python, spark, glue, custom. Adapter-qualified dbt engines (dbt-, e.g. dbt-glue, dbt-snowflake, dbt-bigquery) are accepted too and route through the dbt build runner, which derives the adapter from the suffix. Ingestion engines (acquisition pattern only): duckdb (zero-infra files/JDBC), airbyte (350+ SaaS), meltano (Singer 600+ taps), dlt (Python-native), kafka-connect (streaming), debezium (CDC). AWS Glue: glue (Spark ETL jobs managed by Glue). NEW in v0.7.6: flink (Apache Flink jobs, usually long-running streaming SQL with properties.language: flink_sql). forge-cli validates and plans a flink build but ships no runner for it: the platform that runs the job, such as the Flink Kubernetes Operator or AWS Managed Service for Apache Flink, executes it. Put deployment settings under properties.parameters, for example parallelism and upgradeMode, as the Flink Kubernetes Operator's job spec names them, and flinkConfiguration." }, "capabilities": { "type": "array", diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index 16c98646..d3b30e0a 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -127,6 +127,37 @@ def get_build_engine(build: Mapping[str, Any]) -> Optional[str]: return build.get("engine") +#: Engines and code languages whose jobs run on a platform outside forge-cli, +#: such as the Flink Kubernetes Operator or AWS Managed Service for Apache +#: Flink. forge-cli validates and plans such a build but ships no runner for +#: it, and never runs its ``properties.sql`` as its own SQL. ``engine: flink`` +#: is new in schema 0.7.6; ``properties.language: flink_sql`` is how 0.7.5 and +#: older contracts declare the same job. +EXTERNALLY_RUN_ENGINES = frozenset({"flink"}) +EXTERNALLY_RUN_LANGUAGES = frozenset({"flink_sql"}) + + +def is_externally_run_build(build: Mapping[str, Any]) -> bool: + """True when the build's job runs on a platform forge-cli does not drive: + its engine is ``flink``, or its ``properties.language`` is ``flink_sql``.""" + engine = get_build_engine(build) + if isinstance(engine, str) and engine.strip().lower() in EXTERNALLY_RUN_ENGINES: + return True + props = build.get("properties") + language = props.get("language") if isinstance(props, Mapping) else None + return isinstance(language, str) and language.strip().lower() in EXTERNALLY_RUN_LANGUAGES + + +def describe_externally_run_build(build: Mapping[str, Any]) -> str: + """What makes the build run elsewhere, for messages: ``engine 'flink'`` or + ``language 'flink_sql'``.""" + engine = get_build_engine(build) + if isinstance(engine, str) and engine.strip().lower() in EXTERNALLY_RUN_ENGINES: + return f"engine '{engine}'" + props = build.get("properties") + return f"language '{props.get('language') if isinstance(props, Mapping) else None}'" + + def get_contract_version(contract: Mapping[str, Any]) -> Optional[str]: """ Get the FLUID schema version from a contract. diff --git a/tests/test_flink_engine.py b/tests/test_flink_engine.py new file mode 100644 index 00000000..9f9ecd8c --- /dev/null +++ b/tests/test_flink_engine.py @@ -0,0 +1,521 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed 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. + +"""``engine: flink`` (schema 0.7.6): validated and planned, never run here. + +A Flink job runs on its own platform (the Flink Kubernetes Operator, AWS +Managed Service for Apache Flink). Its SQL is Flink's dialect, so it must +reach neither the local DuckDB engine, a warehouse planner nor a scheduler. +The build runner counts it on its own summary line, apart from skips. The +same holds for ``properties.language: flink_sql``, the way a 0.7.5 contract +declares the job. +""" + +from __future__ import annotations + +import argparse +import copy +import json +import logging +import os +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict + +import jsonschema +import pytest +import yaml + +from fluid_build.build_runners.base import is_embedded_sql_build, run_builds_from_args +from fluid_build.util.contract import is_externally_run_build + +SCHEMAS = Path(__file__).resolve().parent.parent / "fluid_build" / "schemas" + +FLINK_SQL = ( + "INSERT INTO service_instances_full\n" + "SELECT i.id, i.status, v.location\n" + "FROM service_instances i\n" + "JOIN inventory v ON i.id = v.instance_id\n" +) + + +def _flink_build(platform: str = "kubernetes", *, engine: str = "flink") -> Dict[str, Any]: + build: Dict[str, Any] = { + "id": "reconcile", + "engine": engine, + "pattern": "embedded-logic", + "properties": { + "language": "flink_sql", + "sql": FLINK_SQL, + "parameters": {"parallelism": 2}, + }, + } + if platform: + build["execution"] = {"runtime": {"platform": platform}} + return build + + +#: ``engine: flink`` (0.7.6), and the 0.7.5 way to declare the same job. +SHAPES = [ + pytest.param(("0.7.6", "flink"), id="engine-flink"), + pytest.param(("0.7.5", "sql"), id="language-flink-sql"), +] + + +def _contract(fluid_version: str = "0.7.6", engine: str = "flink") -> Dict[str, Any]: + contract = { + "fluidVersion": fluid_version, + "kind": "DataProduct", + "id": "network.service-instances-full", + "name": "Service instances (reconciled)", + "metadata": {"owner": {"team": "data-engineering"}}, + "consumes": [ + { + "productId": "network.service-instances", + "exposeId": "service_instances", + "versionConstraint": "^1.0", + } + ], + "builds": [_flink_build(engine=engine)], + "exposes": [ + { + "exposeId": "service_instances_full", + "kind": "topic", + "binding": { + "platform": "kafka", + "format": "kafka_topic", + "location": {"topic": "network.service-instances-full"}, + }, + "contract": { + "schema": [ + {"name": "id", "type": "string"}, + {"name": "status", "type": "string"}, + {"name": "location", "type": "string"}, + ] + }, + } + ], + } + return contract + + +def _errors(version: str, contract: Dict[str, Any]) -> list: + schema = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) + validator = jsonschema.validators.validator_for(schema)(schema) + return [e.message for e in validator.iter_errors(contract)] + + +class TestSchema: + def test_0_7_6_accepts_engine_flink(self) -> None: + assert _errors("0.7.6", _contract()) == [] + + def test_0_7_5_is_unchanged(self) -> None: + # 0.7.6 is the preview; the stable version does not move. + assert any("'flink'" in msg for msg in _errors("0.7.5", _contract("0.7.5"))) + + def test_0_7_5_declares_the_job_with_language_flink_sql(self) -> None: + assert _errors("0.7.5", _contract("0.7.5", "sql")) == [] + + def test_deployment_settings_belong_under_parameters(self) -> None: + # embeddedLogicPattern is closed, so a setting beside ``sql`` fails. + contract = _contract() + props = contract["builds"][0]["properties"] + props["parallelism"] = props.pop("parameters")["parallelism"] + assert any("parallelism" in msg for msg in _errors("0.7.6", contract)) + + +@pytest.mark.parametrize( + "build, external", + [ + pytest.param(_flink_build(), True, id="engine-flink"), + pytest.param({"engine": " FLINK "}, True, id="engine-case-and-space"), + pytest.param({"engine": "sql", "properties": {"language": "Flink_SQL"}}, True, id="lang"), + pytest.param({"engine": "sql", "properties": {"sql": "SELECT 1"}}, False, id="sql"), + pytest.param({"properties": {"language": "sql"}}, False, id="language-sql"), + # Out of scope here: other languages keep the behaviour they had. + pytest.param({"properties": {"language": "python", "sql": "--"}}, False, id="python"), + pytest.param({"engine": 7, "properties": ["flink_sql"]}, False, id="malformed"), + ], +) +def test_classification(build: Dict[str, Any], external: bool) -> None: + assert is_externally_run_build(build) is external + + +@pytest.mark.parametrize("engine", ["flink", "sql"]) +def test_a_flink_build_is_not_inline_sql(engine: str) -> None: + # ``pattern: embedded-logic`` + ``properties.sql`` made it inline SQL, + # which ran Flink SQL on DuckDB, or failed on an unknown platform. + assert is_embedded_sql_build(_flink_build(engine=engine)) is False + plain = {"pattern": "embedded-logic", "properties": {"sql": "SELECT 1"}} + assert is_embedded_sql_build(plain) is True + python = {"pattern": "embedded-logic", "properties": {"sql": "--", "language": "python"}} + assert is_embedded_sql_build(python) is True # unchanged + + +def _args(contract_path: Path, *, dry_run: bool) -> argparse.Namespace: + return argparse.Namespace( + contract=str(contract_path), + build_id=None, + dry_run=dry_run, + fail_fast=False, + delay=0, + no_output=True, + mode="amend-and-build", + env=None, + sample_rows=None, + ) + + +class TestBuildRunner: + @pytest.mark.parametrize("shape", SHAPES) + @pytest.mark.parametrize("dry_run", [True, False], ids=["dry-run", "run"]) + @pytest.mark.parametrize("platform", ["kubernetes", ""], ids=["kubernetes", "no-platform"]) + def test_a_flink_build_is_reported_not_run( + self, + tmp_path: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + dry_run: bool, + platform: str, + shape: tuple, + ) -> None: + import fluid_build.build_runners.base as base + + def _must_not_run(*_a: Any, **_k: Any) -> int: + raise AssertionError("a flink build reached the inline-SQL executor") + + monkeypatch.setattr(base, "_execute_embedded_sql_build", _must_not_run) + version, engine = shape + contract = _contract(version, engine) + contract["builds"] = [_flink_build(platform, engine=engine)] + path = tmp_path / "contract.fluid.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + + # A contract whose only build runs elsewhere is not "every build was + # skipped": nothing is missing, so the apply succeeds without + # --allow-skipped-builds. + assert run_builds_from_args(_args(path, dry_run=dry_run), logging.getLogger("t")) == 0 + out = capsys.readouterr().out + said = "engine 'flink'" if engine == "flink" else "language 'flink_sql'" + assert f"{said} runs on" in out and "outside forge-cli" in out + assert "Run outside forge-cli: 1" in out + assert "Skipped: 0" in out + assert "INSERT INTO" not in out + + def test_the_run_report_says_why_it_did_not_run( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + import fluid_build.observability.apply_run as apply_run + + recorded: list = [] + + class _Report: + def record_build(self, **kwargs: Any) -> None: + recorded.append(kwargs) + + def build_failed(self, event: str) -> None: + raise AssertionError(event) + + monkeypatch.setattr(apply_run, "current_apply_run", lambda: _Report()) + path = tmp_path / "contract.fluid.yaml" + path.write_text(yaml.safe_dump(_contract()), encoding="utf-8") + assert run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 + assert recorded == [ + {"build_id": "reconcile", "status": "skipped", "reason": "runs_outside_forge"} + ] + + +@pytest.mark.parametrize("shape", SHAPES) +class TestPlanners: + """No planner turns Flink SQL into a DuckDB, Athena, BigQuery or Snowflake action.""" + + def test_local_planner(self, shape: tuple) -> None: + from fluid_build.providers.local.planner import plan_actions + + actions = plan_actions(copy.deepcopy(_contract(*shape)), logger=logging.getLogger("t")) + assert not [a for a in actions if a.get("op") == "execute_sql"] + assert "INSERT INTO service_instances_full" not in json.dumps(actions) + + def test_local_planner_lands_the_sql_build_before_a_flink_build(self, shape: tuple) -> None: + # With the Flink build last, the expose took the first loaded input, + # raw, because the last build had produced no table. The SQL build + # names the expose; the Flink build names none. + from fluid_build.providers.local.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + clean = { + "id": "clean", + "engine": "sql", + "pattern": "embedded-logic", + "outputs": ["service_instances_full"], + "properties": { + "sql": "SELECT id FROM customers", + "parameters": {"inputs": [{"name": "customers", "path": "data/customers.csv"}]}, + }, + } + contract["builds"].insert(0, clean) + contract["exposes"][0]["binding"] = { + "platform": "local", + "format": "csv", + "location": {"path": "out/clean.csv"}, + } + actions = plan_actions(contract, logger=logging.getLogger("t")) + (materialize,) = [a for a in actions if a.get("op") == "materialize"] + assert materialize["payload"]["source_table"] == "result_clean" + + def test_local_planner_lands_nothing_for_an_expose_flink_writes(self, shape: tuple) -> None: + # With every build running elsewhere, the expose took the first loaded + # input and landed it raw under the expose's name. + from fluid_build.providers.local.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + contract["builds"][0]["properties"]["parameters"]["inputs"] = [ + {"name": "raw", "path": "data/raw.csv"} + ] + contract["exposes"][0]["binding"] = { + "platform": "local", + "format": "csv", + "location": {"path": "out/full.csv"}, + } + actions = plan_actions(contract, logger=logging.getLogger("t")) + assert not [a for a in actions if a.get("op") == "materialize"] + + @pytest.mark.parametrize("sql_outputs", [None, ["customers_full"]], ids=["none", "declared"]) + def test_an_expose_flink_may_write_never_lands_another_builds_rows( + self, shape: tuple, sql_outputs, caplog: pytest.LogCaptureFixture + ) -> None: + # A SQL build's table landed under the Flink expose's name, columns the + # expose does not declare included. + from fluid_build.providers.local.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + full = { + "id": "full", + "engine": "sql", + "pattern": "embedded-logic", + "properties": { + "sql": "SELECT id, email, status FROM customers", + "parameters": {"inputs": [{"name": "customers", "path": "data/c.csv"}]}, + }, + } + if sql_outputs: + full["outputs"] = sql_outputs + contract["builds"].insert(0, full) + contract["exposes"] = [ + { + "exposeId": name, + "binding": { + "platform": "local", + "format": "csv", + "location": {"path": f"out/{name}.csv"}, + }, + } + for name in ("customers_full", "customers_public") + ] + caplog.set_level(logging.INFO, logger="t") + actions = plan_actions(contract, logger=logging.getLogger("t")) + landed = { + a["resource_id"]: a["payload"]["source_table"] + for a in actions + if a.get("op") == "materialize" + } + if sql_outputs: + assert landed == {"customers_full": "result_full"} + else: + # Either build may write either expose: neither is guessed. + assert landed == {} + assert "Declare builds[].outputs" in caplog.text + + def test_a_null_build_does_not_crash_the_local_planner(self, shape: tuple) -> None: + from fluid_build.providers.local.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + contract["builds"].insert(0, None) + plan_actions(contract, logger=logging.getLogger("t")) + + def test_local_provider_derives_no_sql(self, tmp_path: Path, shape: tuple) -> None: + from fluid_build.providers.local.local import LocalProvider + + provider = LocalProvider(project=str(tmp_path)) + actions = provider._derive_actions_from_contract(copy.deepcopy(_contract(*shape))) + assert actions == [{"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"}] + + def test_aws_planner(self, shape: tuple) -> None: + from fluid_build.providers.aws.plan.planner import plan_actions + + actions = plan_actions(copy.deepcopy(_contract(*shape)), "123456789012", "eu-west-1") + assert not [a for a in actions if a.get("op") == "athena.execute_query"] + + def test_gcp_planner(self, shape: tuple) -> None: + from fluid_build.providers.gcp.plan.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + contract["builds"][0]["outputs"] = ["service_instances_full"] + actions = plan_actions(contract, "proj", "europe-west1", logging.getLogger("t")) + assert not [a for a in actions if a.get("op") == "bq.sql.execute"] + + def test_snowflake_planner(self, shape: tuple) -> None: + from fluid_build.providers.snowflake.plan.planner import plan_actions + + actions = plan_actions( + copy.deepcopy(_contract(*shape)), + account="acme", + warehouse="WH", + database="DB", + schema="PUBLIC", + ) + assert not [a for a in actions if a.get("op") == "sf.sql.execute"] + + def test_snowflake_replace_still_ensures_the_flink_target(self, shape: tuple) -> None: + # Replace mode skips ensure_table for an expose a SQL build recreates; + # a Flink build does not recreate it, so its table is still ensured. + from fluid_build.providers.snowflake.plan.planner import plan_actions + + contract = copy.deepcopy(_contract(*shape)) + contract["builds"][0]["outputs"] = ["full"] + contract["exposes"] = [ + { + "exposeId": "full", + "binding": { + "platform": "snowflake", + "format": "snowflake_table", + "location": {"database": "DB", "schema": "PUBLIC", "table": "FULL"}, + }, + "contract": {"schema": [{"name": "ID", "type": "STRING"}]}, + } + ] + actions = plan_actions( + contract, account="acme", warehouse="WH", database="DB", schema="PUBLIC", mode="replace" + ) + assert [a for a in actions if a.get("op") == "sf.table.ensure"] + + +@pytest.mark.parametrize("shape", SHAPES) +def test_a_scheduler_gets_no_task_for_a_flink_build(shape: tuple) -> None: + # A nightly task "running" a continuous job would only log its SQL and pass. + from fluid_build.schedulers.synthesis import synthesize_orchestration_from_builds + + contract = _contract(*shape) + assert synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} + contract["builds"].append({"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}}) + tasks = synthesize_orchestration_from_builds(contract, "airflow")["tasks"] + assert [t["taskId"] for t in tasks] == ["daily"] + + +def _run_cli(*argv: str, cwd: Path) -> "subprocess.CompletedProcess[str]": + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", *argv], + cwd=str(cwd), + capture_output=True, + text=True, + env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + ) + + +class TestGenerators: + @pytest.mark.parametrize("shape", SHAPES) + def test_generate_transformation_has_nothing_to_generate( + self, tmp_path: Path, shape: tuple + ) -> None: + # It failed with engine_not_found and told the user to write the code. + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(_contract(*shape)), encoding="utf-8" + ) + result = _run_cli( + "generate", "transformation", "contract.fluid.yaml", "-o", "out", "-q", cwd=tmp_path + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "there is no transformation code to generate" in " ".join(result.stdout.split()) + assert not (tmp_path / "out").exists() + + @pytest.mark.parametrize("shape", SHAPES) + def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path, shape: tuple) -> None: + # It wrote a nightly DAG whose task only logged the Flink SQL. + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(_contract(*shape)), encoding="utf-8" + ) + result = _run_cli( + "generate", + "schedule", + "contract.fluid.yaml", + "-o", + "sched", + "--scheduler", + "airflow", + cwd=tmp_path, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "nothing to schedule" in " ".join(result.stdout.split()) + assert not (tmp_path / "sched").exists() + + @pytest.mark.parametrize( + "variant, extra_args", + [ + pytest.param("cron-trigger", [], id="cron-trigger-no-scheduler"), + pytest.param("engine-no-tasks", [], id="orchestration-engine-no-tasks"), + ], + ) + def test_nothing_to_schedule_whatever_names_the_scheduler( + self, tmp_path: Path, variant: str, extra_args: list + ) -> None: + # A cron trigger defaulted the scheduler and then failed on the empty + # list; an orchestration.engine with no tasks wrote an empty DAG. + contract = _contract() + if variant == "cron-trigger": + contract["builds"][0]["execution"]["trigger"] = { + "type": "schedule", + "schedule": "@daily", + } + else: + contract["orchestration"] = {"engine": "airflow"} + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), encoding="utf-8") + result = _run_cli( + "generate", "schedule", "contract.fluid.yaml", "-o", "sched", *extra_args, cwd=tmp_path + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "nothing to schedule" in " ".join(result.stdout.split()) + assert not (tmp_path / "sched").exists() + + def test_an_unknown_scheduler_is_still_an_error(self, tmp_path: Path) -> None: + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_contract()), encoding="utf-8") + result = _run_cli( + "generate", "schedule", "contract.fluid.yaml", "--scheduler", "bogus", cwd=tmp_path + ) + assert result.returncode == 1 + assert "No scheduler available for 'bogus'" in result.stdout + + def test_a_cron_trigger_on_a_flink_build_gets_no_fluid_apply_dag(self) -> None: + # The DAG would run `fluid apply --build-id reconcile`, which runs + # nothing for a Flink job, and report green every time. + from fluid_build.schedulers.airflow.fluid_apply import ( + has_scheduled_builds, + scheduled_builds, + ) + + contract = _contract() + contract["builds"][0]["execution"]["trigger"] = {"type": "schedule", "schedule": "@daily"} + assert has_scheduled_builds(contract) is False + assert scheduled_builds(contract) == [] + + +def test_every_build_skipped_counts_only_builds_forge_cli_runs( + tmp_path: Path, capsys: pytest.CaptureFixture +) -> None: + contract = _contract() + contract["builds"].append({"id": "ingest", "engine": "python", "repository": "./"}) + path = tmp_path / "contract.fluid.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + assert run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 1 + assert "Every build was skipped (1/1)" in capsys.readouterr().err From 30f8c2e454135261daa9febea835952c3d4d5338 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:45:34 +0200 Subject: [PATCH 3/5] docs(changelog): engine: flink, flink_sql builds run outside forge-cli, the local source table An Added entry for the 0.7.6 engine, an Upgrade note for builds with properties.language: flink_sql (no longer run or planned as SQL), and a Fixed entry for the local planner landing the raw input. --- CHANGELOG.md | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0762fb90..19ea4172 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,9 +24,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 as its `id`. A bundle that relied on a file named after the port's `contractId` but declaring another id fails a strict import, or skips the port under `--lenient` (an input port is always skipped); make the port's `contractId` and the ODCS `id` agree. +- A build with `properties.language: flink_sql` is no longer run as SQL of forge-cli's own. + `fluid apply --mode amend-and-build` ran its Flink SQL on DuckDB, or refused an unknown + embedded-SQL platform, and the local, AWS, GCP and Snowflake planners planned it as a + DuckDB, Athena, BigQuery or Snowflake statement. It is now treated like `engine: flink` + (below): reported as run outside forge-cli, planned without SQL, and left out of + generated schedules. Declare `engine: flink` with `fluidVersion: "0.7.6"` to name the + engine too. + +### Added + +- **`engine: flink` (0.7.6 preview).** A build can declare Apache Flink as its engine, + usually a long-running Flink SQL job with `properties.language: flink_sql`; the contract + must say `fluidVersion: "0.7.6"`. forge-cli validates and plans it but ships no runner: + the job runs on the Flink Kubernetes Operator, AWS Managed Service for Apache Flink, or + whatever platform runs it. `fluid apply --mode amend-and-build` reports such a build as + run outside forge-cli and counts it on its own summary line, so a contract whose only + builds run elsewhere is not "every build was skipped"; the run report records it as + `skipped` with `reason: runs_outside_forge`. No provider planner turns its SQL into a + DuckDB, Athena, BigQuery or Snowflake action, and the local planner does not land an + expose it writes; when a contract mixes such a build with a local one and does not say + through `builds[].outputs` which writes which expose, the local planner lands neither + candidate and warns rather than guess. `fluid generate schedule` gives it no task or DAG + (a contract with no other build, and no hand-declared orchestration tasks, has nothing + to schedule and exits 0), and `fluid generate transformation` says there is no code to + generate for it. Deployment settings (parallelism, upgrade mode, Flink configuration) go + under `properties.parameters`. 0.7.5 is unchanged. ### Fixed +- **A local expose lands the last build that produced a table.** The local planner took + the output of the contract's last build, and when that build produced none here (a dbt + model, a Flink job) it fell back to the first input it had loaded and landed it raw in + the expose, columns the expose does not declare included. It now takes the last build + whose output it has. - **`fluid forge --seed-from` catches a dropped expose.** When the model left out an expose the seed declares, the ground-truth guard raised `IndexError`, logged `structural_seed_guard_failed` and kept the generated contract, so the expose was lost From 6bf0880f2bf2a9a19ea3bd134c00b8d48b7086e0 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:16:37 +0200 Subject: [PATCH 4/5] test(flink): patch the embedded-SQL executor by name, one import style for base The runner test imported fluid_build.build_runners.base twice, as a module and by name (code-quality review on #731). --- tests/test_flink_engine.py | 145 ++++++++++++++++++++++++++++--------- 1 file changed, 112 insertions(+), 33 deletions(-) diff --git a/tests/test_flink_engine.py b/tests/test_flink_engine.py index 9f9ecd8c..a5407c72 100644 --- a/tests/test_flink_engine.py +++ b/tests/test_flink_engine.py @@ -51,7 +51,9 @@ ) -def _flink_build(platform: str = "kubernetes", *, engine: str = "flink") -> Dict[str, Any]: +def _flink_build( + platform: str = "kubernetes", *, engine: str = "flink" +) -> Dict[str, Any]: build: Dict[str, Any] = { "id": "reconcile", "engine": engine, @@ -112,7 +114,9 @@ def _contract(fluid_version: str = "0.7.6", engine: str = "flink") -> Dict[str, def _errors(version: str, contract: Dict[str, Any]) -> list: - schema = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) + schema = json.loads( + (SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8") + ) validator = jsonschema.validators.validator_for(schema)(schema) return [e.message for e in validator.iter_errors(contract)] @@ -141,11 +145,17 @@ def test_deployment_settings_belong_under_parameters(self) -> None: [ pytest.param(_flink_build(), True, id="engine-flink"), pytest.param({"engine": " FLINK "}, True, id="engine-case-and-space"), - pytest.param({"engine": "sql", "properties": {"language": "Flink_SQL"}}, True, id="lang"), - pytest.param({"engine": "sql", "properties": {"sql": "SELECT 1"}}, False, id="sql"), + pytest.param( + {"engine": "sql", "properties": {"language": "Flink_SQL"}}, True, id="lang" + ), + pytest.param( + {"engine": "sql", "properties": {"sql": "SELECT 1"}}, False, id="sql" + ), pytest.param({"properties": {"language": "sql"}}, False, id="language-sql"), # Out of scope here: other languages keep the behaviour they had. - pytest.param({"properties": {"language": "python", "sql": "--"}}, False, id="python"), + pytest.param( + {"properties": {"language": "python", "sql": "--"}}, False, id="python" + ), pytest.param({"engine": 7, "properties": ["flink_sql"]}, False, id="malformed"), ], ) @@ -160,7 +170,10 @@ def test_a_flink_build_is_not_inline_sql(engine: str) -> None: assert is_embedded_sql_build(_flink_build(engine=engine)) is False plain = {"pattern": "embedded-logic", "properties": {"sql": "SELECT 1"}} assert is_embedded_sql_build(plain) is True - python = {"pattern": "embedded-logic", "properties": {"sql": "--", "language": "python"}} + python = { + "pattern": "embedded-logic", + "properties": {"sql": "--", "language": "python"}, + } assert is_embedded_sql_build(python) is True # unchanged @@ -181,7 +194,9 @@ def _args(contract_path: Path, *, dry_run: bool) -> argparse.Namespace: class TestBuildRunner: @pytest.mark.parametrize("shape", SHAPES) @pytest.mark.parametrize("dry_run", [True, False], ids=["dry-run", "run"]) - @pytest.mark.parametrize("platform", ["kubernetes", ""], ids=["kubernetes", "no-platform"]) + @pytest.mark.parametrize( + "platform", ["kubernetes", ""], ids=["kubernetes", "no-platform"] + ) def test_a_flink_build_is_reported_not_run( self, tmp_path: Path, @@ -191,12 +206,12 @@ def test_a_flink_build_is_reported_not_run( platform: str, shape: tuple, ) -> None: - import fluid_build.build_runners.base as base - def _must_not_run(*_a: Any, **_k: Any) -> int: raise AssertionError("a flink build reached the inline-SQL executor") - monkeypatch.setattr(base, "_execute_embedded_sql_build", _must_not_run) + monkeypatch.setattr( + "fluid_build.build_runners.base._execute_embedded_sql_build", _must_not_run + ) version, engine = shape contract = _contract(version, engine) contract["builds"] = [_flink_build(platform, engine=engine)] @@ -206,7 +221,10 @@ def _must_not_run(*_a: Any, **_k: Any) -> int: # A contract whose only build runs elsewhere is not "every build was # skipped": nothing is missing, so the apply succeeds without # --allow-skipped-builds. - assert run_builds_from_args(_args(path, dry_run=dry_run), logging.getLogger("t")) == 0 + assert ( + run_builds_from_args(_args(path, dry_run=dry_run), logging.getLogger("t")) + == 0 + ) out = capsys.readouterr().out said = "engine 'flink'" if engine == "flink" else "language 'flink_sql'" assert f"{said} runs on" in out and "outside forge-cli" in out @@ -231,9 +249,15 @@ def build_failed(self, event: str) -> None: monkeypatch.setattr(apply_run, "current_apply_run", lambda: _Report()) path = tmp_path / "contract.fluid.yaml" path.write_text(yaml.safe_dump(_contract()), encoding="utf-8") - assert run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 + assert ( + run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 + ) assert recorded == [ - {"build_id": "reconcile", "status": "skipped", "reason": "runs_outside_forge"} + { + "build_id": "reconcile", + "status": "skipped", + "reason": "runs_outside_forge", + } ] @@ -244,11 +268,15 @@ class TestPlanners: def test_local_planner(self, shape: tuple) -> None: from fluid_build.providers.local.planner import plan_actions - actions = plan_actions(copy.deepcopy(_contract(*shape)), logger=logging.getLogger("t")) + actions = plan_actions( + copy.deepcopy(_contract(*shape)), logger=logging.getLogger("t") + ) assert not [a for a in actions if a.get("op") == "execute_sql"] assert "INSERT INTO service_instances_full" not in json.dumps(actions) - def test_local_planner_lands_the_sql_build_before_a_flink_build(self, shape: tuple) -> None: + def test_local_planner_lands_the_sql_build_before_a_flink_build( + self, shape: tuple + ) -> None: # With the Flink build last, the expose took the first loaded input, # raw, because the last build had produced no table. The SQL build # names the expose; the Flink build names none. @@ -262,7 +290,9 @@ def test_local_planner_lands_the_sql_build_before_a_flink_build(self, shape: tup "outputs": ["service_instances_full"], "properties": { "sql": "SELECT id FROM customers", - "parameters": {"inputs": [{"name": "customers", "path": "data/customers.csv"}]}, + "parameters": { + "inputs": [{"name": "customers", "path": "data/customers.csv"}] + }, }, } contract["builds"].insert(0, clean) @@ -275,7 +305,9 @@ def test_local_planner_lands_the_sql_build_before_a_flink_build(self, shape: tup (materialize,) = [a for a in actions if a.get("op") == "materialize"] assert materialize["payload"]["source_table"] == "result_clean" - def test_local_planner_lands_nothing_for_an_expose_flink_writes(self, shape: tuple) -> None: + def test_local_planner_lands_nothing_for_an_expose_flink_writes( + self, shape: tuple + ) -> None: # With every build running elsewhere, the expose took the first loaded # input and landed it raw under the expose's name. from fluid_build.providers.local.planner import plan_actions @@ -292,7 +324,9 @@ def test_local_planner_lands_nothing_for_an_expose_flink_writes(self, shape: tup actions = plan_actions(contract, logger=logging.getLogger("t")) assert not [a for a in actions if a.get("op") == "materialize"] - @pytest.mark.parametrize("sql_outputs", [None, ["customers_full"]], ids=["none", "declared"]) + @pytest.mark.parametrize( + "sql_outputs", [None, ["customers_full"]], ids=["none", "declared"] + ) def test_an_expose_flink_may_write_never_lands_another_builds_rows( self, shape: tuple, sql_outputs, caplog: pytest.LogCaptureFixture ) -> None: @@ -349,13 +383,19 @@ def test_local_provider_derives_no_sql(self, tmp_path: Path, shape: tuple) -> No from fluid_build.providers.local.local import LocalProvider provider = LocalProvider(project=str(tmp_path)) - actions = provider._derive_actions_from_contract(copy.deepcopy(_contract(*shape))) - assert actions == [{"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"}] + actions = provider._derive_actions_from_contract( + copy.deepcopy(_contract(*shape)) + ) + assert actions == [ + {"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"} + ] def test_aws_planner(self, shape: tuple) -> None: from fluid_build.providers.aws.plan.planner import plan_actions - actions = plan_actions(copy.deepcopy(_contract(*shape)), "123456789012", "eu-west-1") + actions = plan_actions( + copy.deepcopy(_contract(*shape)), "123456789012", "eu-west-1" + ) assert not [a for a in actions if a.get("op") == "athena.execute_query"] def test_gcp_planner(self, shape: tuple) -> None: @@ -378,7 +418,9 @@ def test_snowflake_planner(self, shape: tuple) -> None: ) assert not [a for a in actions if a.get("op") == "sf.sql.execute"] - def test_snowflake_replace_still_ensures_the_flink_target(self, shape: tuple) -> None: + def test_snowflake_replace_still_ensures_the_flink_target( + self, shape: tuple + ) -> None: # Replace mode skips ensure_table for an expose a SQL build recreates; # a Flink build does not recreate it, so its table is still ensured. from fluid_build.providers.snowflake.plan.planner import plan_actions @@ -397,7 +439,12 @@ def test_snowflake_replace_still_ensures_the_flink_target(self, shape: tuple) -> } ] actions = plan_actions( - contract, account="acme", warehouse="WH", database="DB", schema="PUBLIC", mode="replace" + contract, + account="acme", + warehouse="WH", + database="DB", + schema="PUBLIC", + mode="replace", ) assert [a for a in actions if a.get("op") == "sf.table.ensure"] @@ -408,8 +455,12 @@ def test_a_scheduler_gets_no_task_for_a_flink_build(shape: tuple) -> None: from fluid_build.schedulers.synthesis import synthesize_orchestration_from_builds contract = _contract(*shape) - assert synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} - contract["builds"].append({"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}}) + assert ( + synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} + ) + contract["builds"].append( + {"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}} + ) tasks = synthesize_orchestration_from_builds(contract, "airflow")["tasks"] assert [t["taskId"] for t in tasks] == ["daily"] @@ -434,14 +485,24 @@ def test_generate_transformation_has_nothing_to_generate( yaml.safe_dump(_contract(*shape)), encoding="utf-8" ) result = _run_cli( - "generate", "transformation", "contract.fluid.yaml", "-o", "out", "-q", cwd=tmp_path + "generate", + "transformation", + "contract.fluid.yaml", + "-o", + "out", + "-q", + cwd=tmp_path, ) assert result.returncode == 0, result.stdout + result.stderr - assert "there is no transformation code to generate" in " ".join(result.stdout.split()) + assert "there is no transformation code to generate" in " ".join( + result.stdout.split() + ) assert not (tmp_path / "out").exists() @pytest.mark.parametrize("shape", SHAPES) - def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path, shape: tuple) -> None: + def test_generate_schedule_has_nothing_to_schedule( + self, tmp_path: Path, shape: tuple + ) -> None: # It wrote a nightly DAG whose task only logged the Flink SQL. (tmp_path / "contract.fluid.yaml").write_text( yaml.safe_dump(_contract(*shape)), encoding="utf-8" @@ -480,18 +541,33 @@ def test_nothing_to_schedule_whatever_names_the_scheduler( } else: contract["orchestration"] = {"engine": "airflow"} - (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), encoding="utf-8") + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(contract), encoding="utf-8" + ) result = _run_cli( - "generate", "schedule", "contract.fluid.yaml", "-o", "sched", *extra_args, cwd=tmp_path + "generate", + "schedule", + "contract.fluid.yaml", + "-o", + "sched", + *extra_args, + cwd=tmp_path, ) assert result.returncode == 0, result.stdout + result.stderr assert "nothing to schedule" in " ".join(result.stdout.split()) assert not (tmp_path / "sched").exists() def test_an_unknown_scheduler_is_still_an_error(self, tmp_path: Path) -> None: - (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_contract()), encoding="utf-8") + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(_contract()), encoding="utf-8" + ) result = _run_cli( - "generate", "schedule", "contract.fluid.yaml", "--scheduler", "bogus", cwd=tmp_path + "generate", + "schedule", + "contract.fluid.yaml", + "--scheduler", + "bogus", + cwd=tmp_path, ) assert result.returncode == 1 assert "No scheduler available for 'bogus'" in result.stdout @@ -505,7 +581,10 @@ def test_a_cron_trigger_on_a_flink_build_gets_no_fluid_apply_dag(self) -> None: ) contract = _contract() - contract["builds"][0]["execution"]["trigger"] = {"type": "schedule", "schedule": "@daily"} + contract["builds"][0]["execution"]["trigger"] = { + "type": "schedule", + "schedule": "@daily", + } assert has_scheduled_builds(contract) is False assert scheduled_builds(contract) == [] From f395d9b3dd473e4cdab42f6f7d8c1a31de0c9c83 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:20:53 +0200 Subject: [PATCH 5/5] style(tests): black 24.10.0 at the repo's line length --- tests/test_flink_engine.py | 93 ++++++++++---------------------------- 1 file changed, 23 insertions(+), 70 deletions(-) diff --git a/tests/test_flink_engine.py b/tests/test_flink_engine.py index a5407c72..9158c89e 100644 --- a/tests/test_flink_engine.py +++ b/tests/test_flink_engine.py @@ -51,9 +51,7 @@ ) -def _flink_build( - platform: str = "kubernetes", *, engine: str = "flink" -) -> Dict[str, Any]: +def _flink_build(platform: str = "kubernetes", *, engine: str = "flink") -> Dict[str, Any]: build: Dict[str, Any] = { "id": "reconcile", "engine": engine, @@ -114,9 +112,7 @@ def _contract(fluid_version: str = "0.7.6", engine: str = "flink") -> Dict[str, def _errors(version: str, contract: Dict[str, Any]) -> list: - schema = json.loads( - (SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8") - ) + schema = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) validator = jsonschema.validators.validator_for(schema)(schema) return [e.message for e in validator.iter_errors(contract)] @@ -145,17 +141,11 @@ def test_deployment_settings_belong_under_parameters(self) -> None: [ pytest.param(_flink_build(), True, id="engine-flink"), pytest.param({"engine": " FLINK "}, True, id="engine-case-and-space"), - pytest.param( - {"engine": "sql", "properties": {"language": "Flink_SQL"}}, True, id="lang" - ), - pytest.param( - {"engine": "sql", "properties": {"sql": "SELECT 1"}}, False, id="sql" - ), + pytest.param({"engine": "sql", "properties": {"language": "Flink_SQL"}}, True, id="lang"), + pytest.param({"engine": "sql", "properties": {"sql": "SELECT 1"}}, False, id="sql"), pytest.param({"properties": {"language": "sql"}}, False, id="language-sql"), # Out of scope here: other languages keep the behaviour they had. - pytest.param( - {"properties": {"language": "python", "sql": "--"}}, False, id="python" - ), + pytest.param({"properties": {"language": "python", "sql": "--"}}, False, id="python"), pytest.param({"engine": 7, "properties": ["flink_sql"]}, False, id="malformed"), ], ) @@ -194,9 +184,7 @@ def _args(contract_path: Path, *, dry_run: bool) -> argparse.Namespace: class TestBuildRunner: @pytest.mark.parametrize("shape", SHAPES) @pytest.mark.parametrize("dry_run", [True, False], ids=["dry-run", "run"]) - @pytest.mark.parametrize( - "platform", ["kubernetes", ""], ids=["kubernetes", "no-platform"] - ) + @pytest.mark.parametrize("platform", ["kubernetes", ""], ids=["kubernetes", "no-platform"]) def test_a_flink_build_is_reported_not_run( self, tmp_path: Path, @@ -221,10 +209,7 @@ def _must_not_run(*_a: Any, **_k: Any) -> int: # A contract whose only build runs elsewhere is not "every build was # skipped": nothing is missing, so the apply succeeds without # --allow-skipped-builds. - assert ( - run_builds_from_args(_args(path, dry_run=dry_run), logging.getLogger("t")) - == 0 - ) + assert run_builds_from_args(_args(path, dry_run=dry_run), logging.getLogger("t")) == 0 out = capsys.readouterr().out said = "engine 'flink'" if engine == "flink" else "language 'flink_sql'" assert f"{said} runs on" in out and "outside forge-cli" in out @@ -249,9 +234,7 @@ def build_failed(self, event: str) -> None: monkeypatch.setattr(apply_run, "current_apply_run", lambda: _Report()) path = tmp_path / "contract.fluid.yaml" path.write_text(yaml.safe_dump(_contract()), encoding="utf-8") - assert ( - run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 - ) + assert run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 assert recorded == [ { "build_id": "reconcile", @@ -268,15 +251,11 @@ class TestPlanners: def test_local_planner(self, shape: tuple) -> None: from fluid_build.providers.local.planner import plan_actions - actions = plan_actions( - copy.deepcopy(_contract(*shape)), logger=logging.getLogger("t") - ) + actions = plan_actions(copy.deepcopy(_contract(*shape)), logger=logging.getLogger("t")) assert not [a for a in actions if a.get("op") == "execute_sql"] assert "INSERT INTO service_instances_full" not in json.dumps(actions) - def test_local_planner_lands_the_sql_build_before_a_flink_build( - self, shape: tuple - ) -> None: + def test_local_planner_lands_the_sql_build_before_a_flink_build(self, shape: tuple) -> None: # With the Flink build last, the expose took the first loaded input, # raw, because the last build had produced no table. The SQL build # names the expose; the Flink build names none. @@ -290,9 +269,7 @@ def test_local_planner_lands_the_sql_build_before_a_flink_build( "outputs": ["service_instances_full"], "properties": { "sql": "SELECT id FROM customers", - "parameters": { - "inputs": [{"name": "customers", "path": "data/customers.csv"}] - }, + "parameters": {"inputs": [{"name": "customers", "path": "data/customers.csv"}]}, }, } contract["builds"].insert(0, clean) @@ -305,9 +282,7 @@ def test_local_planner_lands_the_sql_build_before_a_flink_build( (materialize,) = [a for a in actions if a.get("op") == "materialize"] assert materialize["payload"]["source_table"] == "result_clean" - def test_local_planner_lands_nothing_for_an_expose_flink_writes( - self, shape: tuple - ) -> None: + def test_local_planner_lands_nothing_for_an_expose_flink_writes(self, shape: tuple) -> None: # With every build running elsewhere, the expose took the first loaded # input and landed it raw under the expose's name. from fluid_build.providers.local.planner import plan_actions @@ -324,9 +299,7 @@ def test_local_planner_lands_nothing_for_an_expose_flink_writes( actions = plan_actions(contract, logger=logging.getLogger("t")) assert not [a for a in actions if a.get("op") == "materialize"] - @pytest.mark.parametrize( - "sql_outputs", [None, ["customers_full"]], ids=["none", "declared"] - ) + @pytest.mark.parametrize("sql_outputs", [None, ["customers_full"]], ids=["none", "declared"]) def test_an_expose_flink_may_write_never_lands_another_builds_rows( self, shape: tuple, sql_outputs, caplog: pytest.LogCaptureFixture ) -> None: @@ -383,19 +356,13 @@ def test_local_provider_derives_no_sql(self, tmp_path: Path, shape: tuple) -> No from fluid_build.providers.local.local import LocalProvider provider = LocalProvider(project=str(tmp_path)) - actions = provider._derive_actions_from_contract( - copy.deepcopy(_contract(*shape)) - ) - assert actions == [ - {"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"} - ] + actions = provider._derive_actions_from_contract(copy.deepcopy(_contract(*shape))) + assert actions == [{"op": "noop", "skipped": True, "reason": "build_runs_outside_forge"}] def test_aws_planner(self, shape: tuple) -> None: from fluid_build.providers.aws.plan.planner import plan_actions - actions = plan_actions( - copy.deepcopy(_contract(*shape)), "123456789012", "eu-west-1" - ) + actions = plan_actions(copy.deepcopy(_contract(*shape)), "123456789012", "eu-west-1") assert not [a for a in actions if a.get("op") == "athena.execute_query"] def test_gcp_planner(self, shape: tuple) -> None: @@ -418,9 +385,7 @@ def test_snowflake_planner(self, shape: tuple) -> None: ) assert not [a for a in actions if a.get("op") == "sf.sql.execute"] - def test_snowflake_replace_still_ensures_the_flink_target( - self, shape: tuple - ) -> None: + def test_snowflake_replace_still_ensures_the_flink_target(self, shape: tuple) -> None: # Replace mode skips ensure_table for an expose a SQL build recreates; # a Flink build does not recreate it, so its table is still ensured. from fluid_build.providers.snowflake.plan.planner import plan_actions @@ -455,12 +420,8 @@ def test_a_scheduler_gets_no_task_for_a_flink_build(shape: tuple) -> None: from fluid_build.schedulers.synthesis import synthesize_orchestration_from_builds contract = _contract(*shape) - assert ( - synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} - ) - contract["builds"].append( - {"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}} - ) + assert synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} + contract["builds"].append({"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}}) tasks = synthesize_orchestration_from_builds(contract, "airflow")["tasks"] assert [t["taskId"] for t in tasks] == ["daily"] @@ -494,15 +455,11 @@ def test_generate_transformation_has_nothing_to_generate( cwd=tmp_path, ) assert result.returncode == 0, result.stdout + result.stderr - assert "there is no transformation code to generate" in " ".join( - result.stdout.split() - ) + assert "there is no transformation code to generate" in " ".join(result.stdout.split()) assert not (tmp_path / "out").exists() @pytest.mark.parametrize("shape", SHAPES) - def test_generate_schedule_has_nothing_to_schedule( - self, tmp_path: Path, shape: tuple - ) -> None: + def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path, shape: tuple) -> None: # It wrote a nightly DAG whose task only logged the Flink SQL. (tmp_path / "contract.fluid.yaml").write_text( yaml.safe_dump(_contract(*shape)), encoding="utf-8" @@ -541,9 +498,7 @@ def test_nothing_to_schedule_whatever_names_the_scheduler( } else: contract["orchestration"] = {"engine": "airflow"} - (tmp_path / "contract.fluid.yaml").write_text( - yaml.safe_dump(contract), encoding="utf-8" - ) + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), encoding="utf-8") result = _run_cli( "generate", "schedule", @@ -558,9 +513,7 @@ def test_nothing_to_schedule_whatever_names_the_scheduler( assert not (tmp_path / "sched").exists() def test_an_unknown_scheduler_is_still_an_error(self, tmp_path: Path) -> None: - (tmp_path / "contract.fluid.yaml").write_text( - yaml.safe_dump(_contract()), encoding="utf-8" - ) + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_contract()), encoding="utf-8") result = _run_cli( "generate", "schedule",