From fb16b2a1965b8822d6224187d8ac11c0e4c5039c Mon Sep 17 00:00:00 2001 From: Song Huang Date: Thu, 9 Jul 2026 01:05:24 -0400 Subject: [PATCH] Preserve unknown transport configurations --- crates/adapters/src/transport.rs | 1 + crates/feldera-types/src/config.rs | 9 ++ crates/pipeline-manager/src/api/main.rs | 39 +++++- .../pipeline-manager/src/db/types/program.rs | 6 +- .../platform/test_custom_runtime_transport.py | 55 ++++++++ scripts/test_custom_runtime_transport.sh | 119 ++++++++++++++++++ 6 files changed, 226 insertions(+), 3 deletions(-) create mode 100644 python/tests/platform/test_custom_runtime_transport.py create mode 100755 scripts/test_custom_runtime_transport.sh diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index f634e86c0e4..f9ed7dfffc9 100644 --- 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 index e881be1693e..c1f3101fbaa 100644 --- 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, + }, } 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 index 328ba13c7b3..b552034c59b 100644 --- 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 index 4057794b639..aaf8b5f711d 100644 --- 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 new file mode 100644 index 00000000000..61d96ed03f7 --- /dev/null +++ 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 new file mode 100755 index 00000000000..7687b21a3f7 --- /dev/null +++ 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 'test\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