feat: engine: flink (0.7.6 preview) runs outside forge-cli; Flink SQL stays off the inline-SQL paths

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

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

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":

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"

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,

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",

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):

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] = []
 

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]:
@@ -511,7 +556,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 +565,13 @@ 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):
+        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