pipeline-manager: reduce dependency of platform on the runtime - #6288

This looks pretty simple

The direction is clearly right — reducing the manager↔runtime coupling on ProgramSchema is a long-overdue layering fix, and ProgramSchemaPropertiesOnly is a tidy way to peel off only the names+properties the manager actually needs. The two commits split cleanly. Posting COMMENT rather than APPROVE because there are no new tests covering the new manager behavior (parsing only the properties subset, and surviving a runtime schema field the manager doesn't know about) — given how much logic is shifting at the layer boundary, even one round-trip test where the JSON contains an unknown field nested inside fields[].columntype would make me much happier that future schema drift won't silently break.

Inline findings below — all soft, none blocking.

Thanks for the quick turnaround — cf187f7c07 addresses most of what I raised:

• (1) Default is now removed from ProgramInfo, so the Value::Null footgun is gone.
• (2) InvalidProgramSchema now returns an all-zero SourcePosition explicitly. Functional, but reading line 0:0 in a SqlCompilerMessage will still confuse anyone debugging a manager-side parse failure. Nit, not a blocker.
• (3) The new validate_program_info parses via ProgramSchemaPropertiesOnly::deserialize(&program_schema) (borrow), nice. The deep clone() is still present at the pipeline_diff.rs site though.
• (4) Doc wording fixed — clear now.
• (5) pipeline_diff.rs now has an explicit TODO: consider parsing only the necessary subset — acknowledging the drift problem isn't solved at the diff site. Fair as a follow-up.

The one thing I still wish were in this PR is a manager-layer roundtrip test: a JSON blob with an extra nested field inside fields[].columntype (or a wholly new top-level field on Relation) that ProgramSchemaPropertiesOnly swallows cleanly while a ProgramSchema parse on the same blob would fail. That's the contract this layering buys us, and without a test it's only true by inspection. Happy to switch to APPROVE once that's covered (or with a tracking issue if you'd rather defer).

Looks good, it will need an downstream PR as well when including my proposed edits in PR #6294

LGTM, upgrading to APPROVE.

The two remaining items from my prior COMMENT review are addressed in 65a8830:

• ProgramSchemaPropertiesOnly is now exercised in unit tests in sql_compiler.rs::test::compile_basic_program (asserts name + properties match the full ProgramSchema) and in compile_program_with_connector_properties (asserts the inline-connector properties survive subset parsing). Direct coverage of the manager-layer subset-parse path I asked about \u2014 thanks.
• generate_pipeline_config is simplified: program_info is no longer threaded through, and the pipeline_diff.rs site has a TODO acknowledging the same compatibility risk for that consumer. Fine as a follow-up.

The new AutomatonCannotConstructProgramInfoUrl early-return when program_info_integrity_checksum is missing is consistent with the new mandatory-program_info_url shape (matches #6294, now merged).

Note: Default derive is gone from ProgramInfo, so callers that previously used ProgramInfo::default() have been removed cleanly in db/test.rs. Good.

Re-approve. Prior concerns addressed: ConfigMap-size workaround in generate_pipeline_config is gone; manager now parses only ProgramSchemaPropertiesOnly in generate_program_info (db/types/program.rs); program_info_url is required end-to-end with a dedicated error variant; sql_compiler.rs adds a roundtrip test that decodes the same payload as both ProgramSchema and ProgramSchemaPropertiesOnly and asserts name+properties match. The deep-clone in pipeline_diff.rs is still there but carries a TODO and is acknowledged as the remaining drift site — fair as a follow-up. LGTM.

Re-review on 2aa51ea7de (force-pushed squash; PR renamed from "Relax pipeline manager → program schema dependency" to "pipeline-manager: reduce dependency of platform on the runtime").

The new commit is much broader than what I last approved at 6ba37743ed. A single squashed commit now bundles at least five logically distinct changes:

1. RuntimeConfig::datafusion_memory_mb — new config field, 5% default capped at 2 GB, plumbing into circuit_config (controller.rs:4866) and the DataFusion runtime env (adapterlib/src/utils/datafusion.rs), plus ConfigError::DatafusionMemoryExceedsBudget and an adhoc-sort pool-size warning.
2. BootstrapConfig wrapper (in feldera-types/src/runtime_status.rs) introducing a new silent_bootstrap flag on the start endpoint, plumbed through the DB column (backward-compatible string parse) and the local runner CLI as --silent-bootstrap.
3. New PipelineExecutor::can_provision() trait method called during Stopped to fail-fast on bad runner config; new RunnerError::RunnerConfigError.
4. runtime_config: &serde_json::Value threaded through every trait method (provision, is_provisioned, check, stop, clear), plus discover_namespace_from_runtime_config_json utility — clearly motivated by the Kubernetes runner needing the namespace without re-deserializing RuntimeConfig (and the new comment on ResourceConfig::namespace asking future PRs not to move/retype that field is good).
5. State-machine refactor: new Action::RemainStoppedUpdateError replacing several TransitionToStopping paths in pipeline_automata.rs (unsupported platform version, failed compilation, AutomatonInvalidRuntimeConfig, AutomatonMissingProgramInfo, AutomatonInvalidProgramInfo, AutomatonFailedToSerializeDeploymentConfig). db/test.rs updated to assert the new semantics (transit_deployment_resources_status_to_stopped now errors with InvalidResourcesStatusNotRemain for an already-stopped pipeline; refresh_version stays at 21 instead of bumping to 22).

The headline finding from the original PR — replacing the typed schema with serde_json::Value in ProgramInfo and parsing only ProgramSchemaPropertiesOnly at the connector-generation site — is still here, and the test in sql_compiler.rs for "parse subset, tolerate unknown nested fields" still covers it. The decoupling goal is unchanged from my prior APPROVE.

This is too much for one squash. Reviewers downstream (and git blame readers) cannot easily separate "the platform/runtime decoupling we approved" from "datafusion memory budget split", "silent bootstrap", "can_provision trait extension", and "state-machine refactor". Each of items 1, 2, 3+5 deserves its own PR with its own title/description and its own bisection point if it regresses. Even keeping them as separate commits inside this PR (no squash, with the original titles preserved) would help — the merge-queue squash will still flatten on merge, but review can proceed on isolated diffs. Strongly suggest splitting before merging.

A few concrete concerns within the new scope:

BootstrapConfig::active_bootstrap_policy() (runtime_status.rs:255) panics via expect("bootstrap policy must be set for an active deployment") if the inner bootstrap_policy: Option<BootstrapPolicy> is None. The invariant on the active-deployment path should hold: pipeline_management.rs always constructs the value with bootstrap_policy: Some(..), and parse_string_as_bootstrap_config for the legacy string formats produces Some too. But pipeline_automata.rs:1319 constructs it via Some(pipeline.bootstrap_policy.unwrap_or_default()), and the Default for BootstrapConfig is bootstrap_policy: None, silent_bootstrap: false — i.e. invalid for any consumer of active_bootstrap_policy(). If pipeline.bootstrap_policy is ever None here (e.g. a stored row that hasn't been written yet, or a future code path that forgets to set it), the local runner will panic at local_runner.rs:600 rather than fail-stop the pipeline with a clear error. Two reasonable fixes: drop Default from BootstrapConfig and use pipeline.bootstrap_policy.expect(..) so the panic point is explicit, or make the inner bootstrap_policy non-optional and default to BootstrapPolicy::default(). Today's shape is a footgun.

datafusion_memory_mb defaults. The 5% / 2 GB ceiling is sensible for typical pipelines, but resolved_datafusion_memory_mb does plain integer arithmetic effective * 5 / 100. For tiny budgets (say max_rss_mb: Some(10)) you get 0 MB; controller.rs then accepts it (0 < 10) and DataFusion is built with FairSpillPool::new(0). Worth a small floor so a misconfigured tiny budget gives a clearer error than first-allocation ResourcesExhausted. Non-blocking, but the docstring already calls out that under-sized pools surface as ResourcesExhausted — bound the obvious zero case at construction.

python/feldera/_helpers.py. Defaulting an unknown column shape to "struct" silently is a future bug magnet:

  if "type" in column["columntype"]:
      column_type = column["columntype"]["type"]
  else:
      # default column type
      column_type = "struct"

If a new top-level column-type shape lands in the schema, every Python client will silently treat it as a struct and likely produce wrong rows. At minimum log a warning; better, raise on unknown shape and require explicit handling.

pipeline_diff.rs still deep-clones the program schema on every diff (the TODO from my original review). Not regressed, just unchanged; acceptable as follow-up.

Net: the decoupling work is good and the can_provision + state-machine refactor are sensible improvements, but I'd like to see this split. If splitting is genuinely off the table, please at least (a) fix the active_bootstrap_policy() invariant so we cannot construct an invalid BootstrapConfig and then panic later, and (b) update the PR description to enumerate the five themes so future archaeology has a chance.