feldera · xiasongh · Jul 9, 2026
diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs
@@ -128,6 +128,7 @@ pub fn input_transport_config_to_endpoint(
         | TransportConfig::HttpOutput(_)
         | TransportConfig::RedisOutput(_)
         | TransportConfig::IcebergInput(_)
+        | TransportConfig::Unknown { .. }
         | TransportConfig::NullOutput => return Ok(None),
     };
     Ok(Some(endpoint))

diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs
@@ -1885,6 +1885,14 @@ pub enum TransportConfig {
     NullOutput,
     /// Input connector that produces no data.
     EmptyInput,
+    /// A transport unknown to this Feldera version, preserved for a newer runtime.
+    #[serde(untagged)]
+    Unknown {
+        name: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        #[schema(value_type = Object)]
+        config: Option<JsonValue>,
+    },
 }
 
 impl TransportConfig {
@@ -1914,6 +1922,7 @@ impl TransportConfig {
             TransportConfig::ClockInput(_) => "clock".to_string(),
             TransportConfig::NullOutput => "null_output".to_string(),
             TransportConfig::EmptyInput => "empty_input".to_string(),
+            TransportConfig::Unknown { name, .. } => name.clone(),
         }
     }
 

diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs
@@ -35,6 +35,7 @@ use tokio::sync::watch;
 use tokio::sync::{Mutex, RwLock};
 use tracing::{error, info, Level};
 use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme};
+use utoipa::openapi::{RefOr, Schema};
 use utoipa::{Modify, OpenApi};
 use utoipa_swagger_ui::SwaggerUi;
 
@@ -52,7 +53,7 @@ macro_rules! log_with_level {
 
 #[derive(OpenApi)]
 #[openapi(
-    modifiers(&SecurityAddon),
+    modifiers(&SecurityAddon, &HideUnknownTransport),
     info(
         title = "Feldera API",
         description = r#"
@@ -739,6 +740,24 @@ impl Modify for SecurityAddon {
     }
 }
 
+struct HideUnknownTransport;
+
+// Utoipa cannot skip an enum variant only in the schema yet:
+// https://github.com/juhaku/utoipa/pull/1513
+impl Modify for HideUnknownTransport {
+    fn modify(&self, openapi: &mut utoipa::openapi::OpenApi) {
+        let Some(RefOr::T(Schema::OneOf(schema))) = openapi
+            .components
+            .as_mut()
+            .and_then(|components| components.schemas.get_mut("TransportConfig"))
+        else {
+            return;
+        };
+
+        schema.items.pop();
+    }
+}
+
 // The below types and methods are used for running the api-server
 
 pub(crate) struct ServerState {
@@ -1041,6 +1060,24 @@ mod tests {
     use actix_web::http::Method;
     use actix_web::test;
 
+    #[actix_web::test]
+    async fn openapi_hides_unknown_transport_fallback() {
+        let transport = ApiDoc::openapi()
+            .components
+            .unwrap()
+            .schemas
+            .remove("TransportConfig")
+            .unwrap();
+        let RefOr::T(Schema::OneOf(transport)) = transport else {
+            panic!("TransportConfig must be a oneOf schema");
+        };
+
+        assert_eq!(transport.items.len(), 24);
+        assert!(!serde_json::to_string(&transport)
+            .unwrap()
+            .contains("\"unknown\""));
+    }
+
     /// Content-hashed bundle paths get year-long immutable caching plus
     /// `ACAO: *` so SvelteKit's `crossorigin` script loads can be reused
     /// from cache.

diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs
@@ -731,7 +731,8 @@ pub fn generate_program_info(
                 | TransportConfig::IcebergInput(_)
                 | TransportConfig::Datagen(_)
                 | TransportConfig::Nexmark(_)
-                | TransportConfig::EmptyInput => {}
+                | TransportConfig::EmptyInput
+                | TransportConfig::Unknown { .. } => {}
                 _ => {
                     return Err(ConnectorGenerationError::ExpectedInputConnector {
                         position: origin_value.value_position,
@@ -780,7 +781,8 @@ pub fn generate_program_info(
                 | TransportConfig::DeltaTableOutput(_)
                 | TransportConfig::DynamoDBOutput(_)
                 | TransportConfig::RedisOutput(_)
-                | TransportConfig::NullOutput => {}
+                | TransportConfig::NullOutput
+                | TransportConfig::Unknown { .. } => {}
                 _ => {
                     return Err(ConnectorGenerationError::ExpectedOutputConnector {
                         position: origin_value.value_position,

diff --git a/python/tests/platform/test_custom_runtime_transport.py b/python/tests/platform/test_custom_runtime_transport.py
@@ -0,0 +1,55 @@
+"""Cross-version test for transports provided only by a selected runtime."""
+
+import os
+
+import pytest
+from feldera import PipelineBuilder
+from feldera.enums import CompilationProfile
+from tests import TEST_CLIENT
+
+
+RUNTIME_VERSION = os.environ.get("FELDERA_CUSTOM_RUNTIME_VERSION")
+
+pytestmark = pytest.mark.skipif(
+    RUNTIME_VERSION is None,
+    reason="requires a runtime prepared by scripts/test_custom_runtime_transport.sh",
+)
+
+
+def test_custom_runtime_transport(pipeline_name):
+    sql = """
+CREATE TABLE t (id BIGINT NOT NULL) WITH (
+    'connectors' = '[{
+        "name": "runtime-only-datagen",
+        "transport": {
+            "name": "runtime_only_datagen",
+            "config": {"plan": [{"limit": 3}]}
+        }
+    }]'
+);
+
+CREATE MATERIALIZED VIEW row_count AS SELECT COUNT(*) AS c FROM t;
+"""
+
+    builder = PipelineBuilder(
+        TEST_CLIENT,
+        pipeline_name,
+        sql,
+        compilation_profile=CompilationProfile.DEV,
+        runtime_version=RUNTIME_VERSION,
+    )
+    builder.use_platform_compiler = True
+    pipeline = builder.create_or_replace()
+
+    messages = pipeline.program_error()["sql_compilation"]["messages"]
+    assert any(
+        message["warning"]
+        and message["error_type"] == "Unknown format"
+        and '"runtime_only_datagen" is not known' in message["message"]
+        for message in messages
+    )
+
+    pipeline.start()
+    pipeline.wait_for_completion(timeout_s=300)
+
+    assert list(pipeline.query("SELECT c FROM row_count")) == [{"c": 3}]
diff --git a/scripts/test_custom_runtime_transport.sh b/scripts/test_custom_runtime_transport.sh
@@ -0,0 +1,119 @@
+#!/usr/bin/env bash
+
+# Verify that an older pipeline manager can forward a transport implemented by a
+# newer custom runtime.
+#
+# 1. Create a runtime-only Datagen alias in a local clone.
+# 2. Start the unmodified pipeline manager.
+# 3. Compile a pipeline against the cloned runtime revision.
+# 4. Verify that the pipeline processes data successfully.
+
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${ROOT_DIR}"
+
+SQL_COMPILER_PATH="sql-to-dbsp-compiler/SQL-compiler/target/sql2dbsp-jar-with-dependencies.jar"
+API_PORT="${FELDERA_TEST_API_PORT:-18080}"
+COMPILER_PORT="${FELDERA_TEST_COMPILER_PORT:-18085}"
+RUNNER_PORT="${FELDERA_TEST_RUNNER_PORT:-18089}"
+TEST_DIR="$(mktemp -d "${TMPDIR:-/tmp}/feldera-custom-runtime-test.XXXXXX")"
+COMPILER_DIR="${TEST_DIR}/compiler"
+RUNTIME_REPO="${COMPILER_DIR}/feldera-checkout"
+MANAGER_LOG="${TEST_DIR}/pipeline-manager.log"
+MANAGER_PID=""
+
+cleanup() {
+    status=$?
+    trap - EXIT INT TERM
+    if [[ -n "${MANAGER_PID}" ]]; then
+        kill "${MANAGER_PID}" 2>/dev/null || true
+        wait "${MANAGER_PID}" 2>/dev/null || true
+    fi
+    if [[ ${status} -ne 0 ]]; then
+        tail -n 200 "${MANAGER_LOG}" 2>/dev/null || true
+    fi
+    rm -rf "${TEST_DIR}"
+    exit "${status}"
+}
+trap cleanup EXIT INT TERM
+
+if [[ -z "${JAVA_HOME:-}" && -d /opt/homebrew/opt/openjdk/libexec/openjdk.jdk/Contents/Home ]]; then
+    export JAVA_HOME=/opt/homebrew/opt/openjdk/libexec/openjdk.jdk/Contents/Home
+fi
+if [[ -n "${JAVA_HOME:-}" ]]; then
+    export PATH="${JAVA_HOME}/bin:${PATH}"
+fi
+if ! java -version >/dev/null 2>&1; then
+    echo "A working Java installation is required to run the SQL compiler." >&2
+    exit 1
+fi
+
+if [[ ! -f "${SQL_COMPILER_PATH}" ]]; then
+    (cd sql-to-dbsp-compiler && ./build.sh)
+fi
+
+if pgrep -f '[p]ipeline-manager' >/dev/null; then
+    echo "A pipeline-manager process is already running; stop it before this test." >&2
+    exit 1
+fi
+
+mkdir -p "${COMPILER_DIR}"
+git clone --quiet --shared . "${RUNTIME_REPO}"
+
+uv run --project python --frozen python - "${RUNTIME_REPO}/crates/feldera-types/src/config.rs" <<'PY'
+from pathlib import Path
+import sys
+
+path = Path(sys.argv[1])
+source = path.read_text()
+needle = "    Datagen(DatagenInputConfig),"
+if source.count(needle) != 1:
+    raise SystemExit(f"expected one Datagen transport variant in {path}")
+path.write_text(source.replace(
+    needle,
+    '    #[serde(alias = "runtime_only_datagen")]\n' + needle,
+))
+PY
+
+git -C "${RUNTIME_REPO}" config user.name "Feldera Test"
+git -C "${RUNTIME_REPO}" config user.email "test@feldera.com"
+git -C "${RUNTIME_REPO}" add crates/feldera-types/src/config.rs
+git -C "${RUNTIME_REPO}" commit --quiet -m "Add runtime-only Datagen alias"
+RUNTIME_VERSION="$(git -C "${RUNTIME_REPO}" rev-parse HEAD)"
+
+mkdir -p "${TEST_DIR}/web-console"
+printf '<!doctype html><title>test</title>\n' >"${TEST_DIR}/web-console/index.html"
+WEBCONSOLE_BUILD_DIR="${TEST_DIR}/web-console" \
+    cargo build -p pipeline-manager
+
+FELDERA_UNSTABLE_FEATURES=runtime_version \
+    target/debug/pipeline-manager \
+    --api-port "${API_PORT}" \
+    --compiler-port "${COMPILER_PORT}" \
+    --runner-port "${RUNNER_PORT}" \
+    --pg-embed-working-directory "${TEST_DIR}/postgres" \
+    --compiler-working-directory "${COMPILER_DIR}" \
+    --runner-working-directory "${TEST_DIR}/runner" \
+    --sql-compiler-path "${SQL_COMPILER_PATH}" \
+    --dev-mode \
+    >"${MANAGER_LOG}" 2>&1 &
+MANAGER_PID=$!
+
+for _ in {1..120}; do
+    if curl --fail --silent "http://127.0.0.1:${API_PORT}/healthz" >/dev/null; then
+        break
+    fi
+    if ! kill -0 "${MANAGER_PID}" 2>/dev/null; then
+        echo "pipeline-manager exited before becoming healthy" >&2
+        exit 1
+    fi
+    sleep 1
+done
+curl --fail --silent "http://127.0.0.1:${API_PORT}/healthz" >/dev/null
+
+cd python
+FELDERA_HOST="http://127.0.0.1:${API_PORT}" \
+FELDERA_CUSTOM_RUNTIME_VERSION="${RUNTIME_VERSION}" \
+PYTHONPATH=. \
+    uv run --frozen pytest tests/platform/test_custom_runtime_transport.py -vv