openapi: 3.0.3 info: title: Feldera Input Connectors Pipeline CRUD API description: "\nWith Feldera, users create data pipelines out of SQL programs.\nA SQL program comprises tables and views, and includes as well the definition of\ninput and output connectors for each respectively. A connector defines a data\nsource or data sink to feed input data into tables or receive output data\ncomputed by the views respectively.\n\n## Pipeline\n\nThe API is centered around the **pipeline**, which most importantly consists\nout of the SQL program, but also has accompanying metadata and configuration parameters\n(e.g., compilation profile, number of workers, etc.).\n\n* A pipeline is identified and referred to by its user-provided unique name.\n* The pipeline program is asynchronously compiled when the pipeline is first created or\n when its program is subsequently updated.\n* Pipeline deployment start is only able to proceed to provisioning once the program is successfully\n compiled.\n* A pipeline cannot be updated while it is deployed.\n\n## Concurrency\n\nEach pipeline has a version, which is incremented each time its core fields are updated.\nThe version is monotonically increasing. There is additionally a program version which covers\nonly the program-related core fields, and is used by the compiler to discern when to recompile.\n\n## Client request handling\n\n### Request outcome expectations\n\nThe outcome of a request is that it either fails (e.g., DNS lookup failed) without any response\n(no status code nor body), or it succeeds and gets back a response status code and body.\n\nIn case of a response, usually it is the Feldera endpoint that generated it:\n- If it is success (2xx), it will return whichever body belongs to the success response.\n- Otherwise, if it is an error (4xx, 5xx), it will return a Feldera error response JSON body\n which will have an application-level `error_code`.\n\nHowever, there are two notable exceptions when the response is not generated by the Feldera\nendpoint:\n- If the HTTP server, to which the endpoint belongs, encountered an issue, it might return\n 4xx (e.g., for an unknown endpoint) or 5xx error codes by itself (e.g., when it is initializing).\n- If the Feldera API server is behind a (reverse) proxy, the proxy can return error codes by itself,\n for example BAD GATEWAY (502) or GATEWAY TIMEOUT (504).\n\nAs such, it is not guaranteed that the (4xx, 5xx) will have a Feldera error response JSON body\nin these latter cases.\n\n### Error handling and retrying\n\nThe error type returned by the client should distinguish between the error responses generated\nby Feldera endpoints themselves (which have a Feldera error response body) and those that are\ngenerated by other sources.\n\nIn order for a client operation (e.g., `pipeline.resume()`) to be robust (i.e., not fail due to\na single HTTP request not succeeding) the client should use a retry mechanism if the operation\nis idempotent. The retry mechanism must however have a time limit, after which it times out.\nThis guarantees that the client operation is eventually responsive, which enables the script\nit is a part of to not hang indefinitely on Feldera operations and instead be able to decide\nby itself whether and how to proceed. If no response is returned, the mechanism should generally\nretry. When a response is returned, the decision whether to retry can generally depend on the status\ncode: especially the status codes 408, 502, 503 and 504 should be considered as transient errors.\nFiner grained retry decisions should be made by taking into account the application-level\n`error_code` if the response body was indeed a Feldera error response body.\n\n## Feldera client errors (4xx)\n\n_Client behavior:_ clients should generally return with an error when they get back a 4xx status\ncode, as it usually means the request will likely not succeed even if it is sent again. Certain\nrequests might make use of a timed retry mechanism when the client error is transient without\nrequiring any user intervention to overcome, for instance a transaction already being in progress\nleading to a temporary CONFLICT (409) error.\n\n- **BAD REQUEST (400)**: invalid user request (general).\n - _Example:_ the new pipeline name `example1@~` contains invalid characters.\n\n- **UNAUTHORIZED (401)**: the user is not authorized to issue the request.\n - _Example:_ an invalid API key is provided.\n\n- **NOT FOUND (404)**: a resource required to exist in order to process the request was not found.\n - _Example:_ a pipeline named `example` does not exist when trying to update it.\n\n- **CONFLICT (409)**: there is a conflict between the request and a relevant resource.\n - _Example:_ a pipeline named `example` already exists.\n - _Example:_ another transaction is already in process.\n\n## Feldera server errors (5xx)\n\n- **INTERNAL SERVER ERROR (500)**: the server is unexpectedly unable to process the request\n (general).\n - _Example:_ unable to reach the database.\n - _Client behavior:_ immediately return with an error.\n\n- **NOT IMPLEMENTED (501)**: the server does not implement functionality required to process the\n request.\n - _Example:_ making a request to an enterprise-only endpoint in the OSS edition.\n - _Client behavior:_ immediately return with an error.\n\n- **SERVICE UNAVAILABLE (503)**: the server is not (yet) able to process the request.\n - _Example:_ pausing a pipeline which is still provisioning.\n - _Client behavior:_ depending on the type of request, client may use a timed retry mechanism.\n\n## Feldera error response body\n\nWhen the Feldera API returns an HTTP error status code (4xx, 5xx), the body will contain the\nfollowing JSON object:\n\n```json\n{\n \"message\": \"Human-readable explanation.\",\n \"error_code\": \"CodeSpecifyingError\",\n \"details\": {\n\n }\n}\n```\n\nIt contains the following fields:\n- **message (string)**: human-readable explanation of the error that occurred and potentially\n hinting what can be done about it.\n- **error_code (string)**: application-level code about the error that occurred, written in CamelCase.\n For example: `UnknownPipelineName`, `DuplicateName`, `PauseWhileNotProvisioned`, ... .\n- **details (object)**: JSON object corresponding to the `error_code` with fields that provide\n details relevant to it. For example: if a name is unknown, a field with the unknown name in\n question.\n" contact: name: Feldera Team email: dev@feldera.com license: name: MIT OR Apache-2.0 version: 0.323.0 tags: - name: Pipeline CRUD paths: /v0/pipelines: get: tags: - Pipeline CRUD summary: List Pipelines description: 'Retrieve the list of pipelines. Configure which fields are included using the `selector` query parameter.' operationId: list_pipelines parameters: - name: selector in: query description: 'The `selector` parameter limits which fields are returned for a pipeline. Limiting which fields is particularly handy for instance when frequently monitoring over low bandwidth connections while being only interested in pipeline status.' required: false schema: $ref: '#/components/schemas/PipelineFieldSelector' responses: '200': description: List of pipelines retrieved successfully content: application/json: schema: type: array items: $ref: '#/components/schemas/PipelineSelectedInfo' example: - id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null - id: 67e55044-10b1-426f-9247-bb680e5fe0c9 name: example2 description: Description of the pipeline example2 tags: [] created_at: '1970-01-01T00:00:00Z' version: 1 platform_version: v0 runtime_config: workers: 10 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: false tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 100000 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: 1000 memory_mb_max: null storage_mb_max: 10000 storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 100000 pin_cpus: [] provisioning_timeout_secs: 1200 max_parallel_connector_init: 10 init_containers: null checkpoint_during_suspend: false http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table2 ( col2 VARCHAR ); udf_rust: '' udf_toml: '' program_config: profile: unoptimized cache: true runtime_version: null use_platform_compiler: false program_version: 1 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 1 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] post: tags: - Pipeline CRUD summary: Create Pipeline description: Create a new pipeline with the provided configuration. operationId: post_pipeline requestBody: content: application/json: schema: $ref: '#/components/schemas/PostPutPipeline' example: name: example1 description: Description of the pipeline example1 tags: [] runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: null udf_toml: null program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false required: true responses: '201': description: Pipeline successfully created content: application/json: schema: $ref: '#/components/schemas/PipelineInfo' example: id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '400': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' examples: Name does not match pattern: value: message: 'Name ''name-with-invalid-char-#'' should be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore (pattern: ''^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$'')' error_code: NameDoesNotMatchPattern details: name: name-with-invalid-char-# pattern: ^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$ pattern_description: be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore '409': description: Cannot create pipeline as the name already exists content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: An entity with this name already exists error_code: DuplicateName details: null '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] /v0/pipelines/{pipeline_name}: get: tags: - Pipeline CRUD summary: Get Pipeline description: 'Retrieve a pipeline. Configure which fields are included using the `selector` query parameter.' operationId: get_pipeline parameters: - name: pipeline_name in: path description: Unique pipeline name required: true schema: type: string - name: selector in: query description: 'The `selector` parameter limits which fields are returned for a pipeline. Limiting which fields is particularly handy for instance when frequently monitoring over low bandwidth connections while being only interested in pipeline status.' required: false schema: $ref: '#/components/schemas/PipelineFieldSelector' responses: '200': description: Pipeline retrieved successfully content: application/json: schema: $ref: '#/components/schemas/PipelineSelectedInfo' example: id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '404': description: Pipeline with that name does not exist content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: Unknown pipeline name 'non-existent-pipeline' error_code: UnknownPipelineName details: pipeline_name: non-existent-pipeline '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] put: tags: - Pipeline CRUD summary: Upsert Pipeline description: Fully update a pipeline if it already exists, otherwise create a new pipeline. operationId: put_pipeline parameters: - name: pipeline_name in: path description: Unique pipeline name required: true schema: type: string requestBody: content: application/json: schema: $ref: '#/components/schemas/PostPutPipeline' example: name: example1 description: Description of the pipeline example1 tags: [] runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: null udf_toml: null program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false required: true responses: '200': description: Pipeline successfully updated content: application/json: schema: $ref: '#/components/schemas/PipelineInfo' example: id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '201': description: Pipeline successfully created content: application/json: schema: $ref: '#/components/schemas/PipelineInfo' example: id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '400': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' examples: Cannot update non-stopped pipeline: value: message: Pipeline can only be updated while stopped. Stop it first by invoking '/stop'. error_code: UpdateRestrictedToStopped details: null Name does not match pattern: value: message: 'Name ''name-with-invalid-char-#'' should be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore (pattern: ''^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$'')' error_code: NameDoesNotMatchPattern details: name: name-with-invalid-char-# pattern: ^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$ pattern_description: be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore '409': description: Cannot rename pipeline as the new name already exists content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: An entity with this name already exists error_code: DuplicateName details: null '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] delete: tags: - Pipeline CRUD summary: Delete Pipeline description: Delete an existing pipeline by name. operationId: delete_pipeline parameters: - name: pipeline_name in: path description: Unique pipeline name required: true schema: type: string responses: '200': description: Pipeline successfully deleted '400': description: Pipeline must be fully stopped and cleared to be deleted content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: Cannot delete a pipeline which is not fully stopped. Stop the pipeline first fully by invoking the '/stop' endpoint. error_code: DeleteRestrictedToFullyStopped details: null '404': description: Pipeline with that name does not exist content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: Unknown pipeline name 'non-existent-pipeline' error_code: UnknownPipelineName details: pipeline_name: non-existent-pipeline '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] patch: tags: - Pipeline CRUD summary: Patch Pipeline description: Partially update a pipeline. operationId: patch_pipeline parameters: - name: pipeline_name in: path description: Unique pipeline name required: true schema: type: string requestBody: content: application/json: schema: $ref: '#/components/schemas/PatchPipeline' example: name: null description: This is a new description runtime_config: null program_code: CREATE TABLE table3 ( col3 INT ); udf_rust: null udf_toml: null program_config: null required: true responses: '200': description: Pipeline successfully updated content: application/json: schema: $ref: '#/components/schemas/PipelineInfo' example: id: 67e55044-10b1-426f-9247-bb680e5fe0c8 name: example1 description: Description of the pipeline example1 tags: [] created_at: '1970-01-01T00:00:00Z' version: 4 platform_version: v0 runtime_config: workers: 16 max_rss_mb: null datafusion_memory_mb: null hosts: 1 storage: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null fault_tolerance: model: none checkpoint_interval_secs: 60 cpu_profiler: true tracing: false tracing_endpoint_jaeger: '' min_batch_size_records: 0 max_buffering_delay_usecs: 0 resources: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null clock_resolution_usecs: 1000000 pin_cpus: [] provisioning_timeout_secs: null max_parallel_connector_init: null init_containers: null checkpoint_during_suspend: true http_workers: null io_workers: null env: {} dev_tweaks: {} logging: null pipeline_template_configmap: null program_code: CREATE TABLE table1 ( col1 INT ); udf_rust: '' udf_toml: '' program_config: profile: optimized cache: true runtime_version: null use_platform_compiler: false program_version: 2 program_status: Pending program_status_since: '1970-01-01T00:00:00Z' program_error: sql_compilation: null rust_compilation: null system_error: null program_info: null deployment_error: null refresh_version: 4 storage_status: Cleared storage_status_details: null deployment_id: null deployment_initial: null deployment_status: Stopped deployment_status_since: '1970-01-01T00:00:00Z' deployment_desired_status: Stopped deployment_desired_status_since: '1970-01-01T00:00:00Z' deployment_resources_status: Stopped deployment_resources_status_details: null deployment_resources_status_since: '1970-01-01T00:00:00Z' deployment_resources_desired_status: Stopped deployment_resources_desired_status_since: '1970-01-01T00:00:00Z' deployment_runtime_status: null deployment_runtime_status_details: null deployment_runtime_status_since: null deployment_runtime_desired_status: null deployment_runtime_desired_status_since: null '400': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' examples: Cannot update non-stopped pipeline: value: message: Pipeline can only be updated while stopped. Stop it first by invoking '/stop'. error_code: UpdateRestrictedToStopped details: null Name does not match pattern: value: message: 'Name ''name-with-invalid-char-#'' should be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore (pattern: ''^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$'')' error_code: NameDoesNotMatchPattern details: name: name-with-invalid-char-# pattern: ^([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9]$ pattern_description: be non-empty and only contain lowercase (a-z), uppercase (A-Z), number (0-9), underscore (_) or hyphen (-) characters, but cannot start or end with hyphen or underscore '404': description: Pipeline with that name does not exist content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: Unknown pipeline name 'non-existent-pipeline' error_code: UnknownPipelineName details: pipeline_name: non-existent-pipeline '409': description: Cannot rename pipeline as the name already exists content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' example: message: An entity with this name already exists error_code: DuplicateName details: null '500': description: '' content: application/json: schema: $ref: '#/components/schemas/ErrorResponse' security: - JSON web token (JWT) or API key: [] components: schemas: FormatConfig: type: object description: 'Data format specification used to parse raw data received from the endpoint or to encode data sent to the endpoint.' required: - name properties: config: type: object description: Format-specific parser or encoder configuration. name: type: string description: Format name, e.g., "csv", "json", "bincode", etc. FileInputConfig: type: object description: Configuration for reading data from a file with `FileInputTransport` required: - path properties: buffer_size_bytes: type: integer description: 'Read buffer size. Default: when this parameter is not specified, a platform-specific default is used.' nullable: true minimum: 0 follow: type: boolean description: 'Enable file following. When `false`, the endpoint outputs an `InputConsumer::eoi` message and stops upon reaching the end of file. When `true`, the endpoint will keep watching the file and outputting any new content appended to it.' path: type: string description: 'File path. This may be a file name or a `file://` URL with an absolute path.' SyncConfig: type: object required: - bucket properties: access_key: type: string description: 'The access key used to authenticate with the storage provider. If not provided, rclone will fall back to environment-based credentials, such as `RCLONE_S3_ACCESS_KEY_ID`. In Kubernetes environments using IRSA (IAM Roles for Service Accounts), this can be left empty to allow automatic authentication via the pod''s service account.' nullable: true bucket: type: string description: 'The name of the storage bucket. This may include a path to a folder inside the bucket (e.g., `my-bucket/data`).' checkers: type: integer format: int32 description: 'The number of checkers to run in parallel. Default: 20' nullable: true minimum: 0 endpoint: type: string description: 'The endpoint URL for the storage service. This is typically required for custom or local S3-compatible storage providers like MinIO. Example: `http://localhost:9000` Relevant rclone config key: [`endpoint`](https://rclone.org/s3/#s3-endpoint)' nullable: true fail_if_no_checkpoint: type: boolean description: 'When true, the pipeline will fail to initialize if fetching the specified checkpoint fails (missing, download error). When false, the pipeline will start from scratch instead. False by default.' default: false flags: type: array items: type: string description: 'Extra flags to pass to `rclone`. WARNING: Supplying incorrect or conflicting flags can break `rclone`. Use with caution. Refer to the docs to see the supported flags: - [Global flags](https://rclone.org/flags/) - [S3 specific flags](https://rclone.org/s3/)' nullable: true ignore_checksum: type: boolean description: 'Set to skip post copy check of checksums, and only check the file sizes. This can significantly improve the throughput. Defualt: false' nullable: true multi_thread_cutoff: type: string description: 'Use multi-thread download for files above this size. Format: `[size][Suffix]` (Example: 1G, 500M) Supported suffixes: k|M|G|T Default: 100M' nullable: true multi_thread_streams: type: integer format: int32 description: 'Number of streams to use for multi-thread downloads. Default: 10' nullable: true minimum: 0 optimize_download_resources: type: boolean description: 'When true, checkpoint downloads use the maximum resources available on the host: `transfers` and `checkers` are scaled to the number of CPUs, and the download buffer is allowed to grow up to most of the available memory. This maximizes download throughput at the cost of higher CPU and memory usage during a pull. When false, downloads use the values configured via `transfers`, `checkers`, and the rclone defaults instead. Default: true' default: true provider: type: string description: 'The name of the cloud storage provider (e.g., `"AWS"`, `"Minio"`). Used for provider-specific behavior in rclone. If omitted, defaults to `"Other"`. See [rclone S3 provider documentation](https://rclone.org/s3/#s3-provider)' nullable: true pull_interval: type: integer format: int64 description: 'The interval (in seconds) between each attempt to fetch the latest checkpoint from object store while in standby mode. Applies only when `start_from_checkpoint` is set to `latest`. Default: 10 seconds' default: 10 minimum: 0 push_interval: type: integer format: int64 description: 'The interval (in seconds) between each push of checkpoints to object store. Default: disabled (no periodic push).' nullable: true minimum: 0 read_bucket: type: string description: 'A read-only bucket used as a fallback checkpoint source. When the pipeline has no local checkpoint and `bucket` contains no checkpoint either, it will attempt to fetch the checkpoint from this location instead. All connection settings (`endpoint`, `region`, `provider`, `access_key`, `secret_key`) are shared with `bucket`. The pipeline **never writes** to `read_bucket`. Must point to a different location than `bucket`.' nullable: true region: type: string description: 'The region that this bucket is in. Leave empty for Minio or the default region (`us-east-1` for AWS).' nullable: true retention_min_age: type: integer format: int32 description: 'The minimum age (in days) a checkpoint must reach before it becomes eligible for deletion. All younger checkpoints will be preserved. Default: 30' default: 30 minimum: 0 retention_min_count: type: integer format: int32 description: 'The minimum number of checkpoints to retain in object store. No checkpoints will be deleted if the total count is below this threshold. Default: 10' default: 10 minimum: 0 secret_key: type: string description: 'The secret key used together with the access key for authentication. If not provided, rclone will fall back to environment-based credentials, such as `RCLONE_S3_SECRET_ACCESS_KEY`. In Kubernetes environments using IRSA (IAM Roles for Service Accounts), this can be left empty to allow automatic authentication via the pod''s service account.' nullable: true standby: type: boolean description: '**Deprecated.** Use `initial=standby` when starting the pipeline instead.' default: false deprecated: true start_from_checkpoint: allOf: - $ref: '#/components/schemas/StartFromCheckpoint' nullable: true transfers: type: integer format: int32 description: 'The number of file transfers to run in parallel. Default: 20' nullable: true minimum: 0 upload_concurrency: type: integer format: int32 description: 'The number of chunks of the same file that are uploaded for multipart uploads. Default: 10' nullable: true minimum: 0 Version: type: integer format: int64 description: Version number. SqlCompilationInfo: type: object description: SQL compilation information. required: - exit_code - messages properties: exit_code: type: integer format: int32 description: Exit code of the SQL compiler. messages: type: array items: $ref: '#/components/schemas/SqlCompilerMessage' description: Messages (warnings and errors) generated by the SQL compiler. PipelineId: type: string format: uuid description: Pipeline identifier. ProgramStatus: type: string description: Program compilation status. enum: - Pending - CompilingSql - SqlCompiled - CompilingRust - Success - SqlError - RustError - SystemError HttpInputConfig: type: object description: 'Configuration for data input via HTTP. HTTP input adapters cannot be usefully configured as part of pipeline configuration. Instead, instantiate them through the REST API as `/pipelines/{pipeline_name}/ingress/{table_name}`.' required: - name properties: name: type: string description: Autogenerated name. DatagenInputConfig: type: object description: Configuration for generating random data for a table. properties: plan: type: array items: $ref: '#/components/schemas/GenerationPlan' description: 'The sequence of generations to perform. If not set, the generator will produce a single sequence with default settings. If set, the generator will produce the specified sequences in sequential order. Note that if one of the sequences before the last one generates an unlimited number of rows the following sequences will not be executed.' default: - rate: null limit: null worker_chunk_size: null fields: {} seed: type: integer format: int64 description: 'Optional seed for the random generator. Setting this to a fixed value will make the generator produce the same sequence of records every time the pipeline is run. # Notes - To ensure the set of generated input records is deterministic across multiple runs, apart from setting a seed, `workers` also needs to remain unchanged. - The input will arrive in non-deterministic order if `workers > 1`.' default: null nullable: true minimum: 0 transaction_size: type: integer description: 'By default, the data generator does not request [transactions]. Set this to a nonzero value for the data generator to automatically orchestrate transactions of approximately the specified number of rows. [transactions]: https://docs.feldera.com/pipelines/transactions' default: null nullable: true minimum: 0 workers: type: integer description: Number of workers to use for generating data. default: 1 minimum: 0 additionalProperties: false StorageCompression: type: string description: Storage compression algorithm. enum: - default - none - snappy DynamoDBWriteMode: type: string description: DynamoDB write API used by the output connector. enum: - batch - transactional SqlType: type: string description: The available SQL column type names. Each value is the platform's wire encoding of the type (e.g. `BIGINT`, `INTEGER`, `INTERVAL_DAY`), not valid SQL type syntax. enum: - BOOLEAN - TINYINT - SMALLINT - INTEGER - BIGINT - UTINYINT - USMALLINT - UINTEGER - UBIGINT - REAL - DOUBLE - DECIMAL - CHAR - VARCHAR - BINARY - VARBINARY - TIME - DATE - TIMESTAMP - TIMESTAMP_TZ - INTERVAL_DAY - INTERVAL_DAY_HOUR - INTERVAL_DAY_MINUTE - INTERVAL_DAY_SECOND - INTERVAL_HOUR - INTERVAL_HOUR_MINUTE - INTERVAL_HOUR_SECOND - INTERVAL_MINUTE - INTERVAL_MINUTE_SECOND - INTERVAL_MONTH - INTERVAL_SECOND - INTERVAL_YEAR - INTERVAL_YEAR_MONTH - ARRAY - STRUCT - MAP - 'NULL' - UUID - VARIANT S3InputConfig: type: object description: Configuration for reading data from AWS S3. required: - region - bucket_name properties: aws_access_key_id: type: string description: AWS Access Key id. This property must be specified unless `no_sign_request` is set to `true`. nullable: true aws_secret_access_key: type: string description: Secret Access Key. This property must be specified unless `no_sign_request` is set to `true`. nullable: true bucket_name: type: string description: S3 bucket name to access. endpoint_url: type: string description: 'The endpoint URL used to communicate with this service. Can be used to make this connector talk to non-AWS services with an S3 API.' nullable: true key: type: string description: Read a single object specified by a key. nullable: true max_concurrent_fetches: type: integer format: int32 description: 'Controls the number of S3 objects fetched in parallel. Increasing this value can improve throughput by enabling greater concurrency. However, higher concurrency may lead to timeouts or increased memory usage due to in-memory buffering. Recommended range: 1–10. Default: 8.' minimum: 0 max_retries: type: integer format: int32 description: 'Retry `max_retries` times with exponential backoff. If the object changes during a retry attempt, the remaining part of the object will not be processed. Recommended range: 3-10. Default: 5.' minimum: 0 no_sign_request: type: boolean description: Do not sign requests. This is equivalent to the `--no-sign-request` flag in the AWS CLI. prefix: type: string description: Read all objects whose keys match a prefix. Set to an empty string to read all objects in the bucket. nullable: true region: type: string description: AWS region. ObjectStorageConfig: type: object required: - url properties: url: type: string description: 'URL. The following URL schemes are supported: * S3: - `s3:///` - `s3a:///` - `https://s3..amazonaws.com/` - `https://.s3..amazonaws.com` - `https://ACCOUNT_ID.r2.cloudflarestorage.com/bucket` * Google Cloud Storage: - `gs:///` * Microsoft Azure Blob Storage: - `abfs[s]:///` (according to [fsspec](https://github.com/fsspec/adlfs)) - `abfs[s]://@.dfs.core.windows.net/` - `abfs[s]://@.dfs.fabric.microsoft.com/` - `az:///` (according to [fsspec](https://github.com/fsspec/adlfs)) - `adl:///` (according to [fsspec](https://github.com/fsspec/adlfs)) - `azure:///` (custom) - `https://.dfs.core.windows.net` - `https://.blob.core.windows.net` - `https://.blob.core.windows.net/` - `https://.dfs.fabric.microsoft.com` - `https://.dfs.fabric.microsoft.com/` - `https://.blob.fabric.microsoft.com` - `https://.blob.fabric.microsoft.com/` Settings derived from the URL will override other settings.' additionalProperties: type: string description: 'Additional options as key-value pairs. The following keys are supported: * S3: - `access_key_id`: AWS Access Key. - `secret_access_key`: AWS Secret Access Key. - `region`: Region. - `default_region`: Default region. - `endpoint`: Custom endpoint for communicating with S3, e.g. `https://localhost:4566` for testing against a localstack instance. - `token`: Token to use for requests (passed to underlying provider). - [Other keys](https://docs.rs/object_store/latest/object_store/aws/enum.AmazonS3ConfigKey.html#variants). * Google Cloud Storage: - `service_account`: Path to the service account file. - `service_account_key`: The serialized service account key. - `google_application_credentials`: Application credentials path. - [Other keys](https://docs.rs/object_store/latest/object_store/gcp/enum.GoogleConfigKey.html). * Microsoft Azure Blob Storage: - `access_key`: Azure Access Key. - `container_name`: Azure Container Name. - `account`: Azure Account. - `bearer_token_authorization`: Static bearer token for authorizing requests. - `client_id`: Client ID for use in client secret or Kubernetes federated credential flow. - `client_secret`: Client secret for use in client secret flow. - `tenant_id`: Tenant ID for use in client secret or Kubernetes federated credential flow. - `endpoint`: Override the endpoint for communicating with blob storage. - [Other keys](https://docs.rs/object_store/latest/object_store/azure/enum.AzureConfigKey.html#variants). Options set through the URL take precedence over those set with these options.' Credentials: oneOf: - type: object required: - FromString properties: FromString: type: string - type: object required: - FromFile properties: FromFile: type: string example: /path/to/credentials.json RuntimeStatusDetails: type: object description: 'Details about the current runtime status. The fields in this struct should all be **optional** and set only by a runtime status when they are known. Otherwise, they can just be set `None`.' properties: approval_diff: description: 'The diff which is awaiting approval. Specifically useful for: `AwaitingApproval`.' nullable: true connector_stats: allOf: - $ref: '#/components/schemas/ConnectorStats' nullable: true reason: type: string description: 'Free form text giving an explanation why it is currently in this runtime status. Specifically useful for: `Unavailable`, `Initializing`.' nullable: true PropertyValue: type: object required: - value - key_position - value_position properties: key_position: $ref: '#/components/schemas/SourcePosition' value: type: string value_position: $ref: '#/components/schemas/SourcePosition' GlueCatalogConfig: type: object description: AWS Glue catalog config. properties: glue.access-key-id: type: string description: Access key id used to access the Glue catalog. nullable: true glue.endpoint: type: string description: 'Configure an alternative endpoint of the Glue service for Glue catalog to access. Example: `"https://glue.us-east-1.amazonaws.com"`' nullable: true glue.id: type: string description: The 12-digit ID of the Glue catalog. nullable: true glue.profile-name: type: string description: Profile used to access the Glue catalog. nullable: true glue.region: type: string description: Region of the Glue catalog. nullable: true glue.secret-access-key: type: string description: Secret access key used to access the Glue catalog. nullable: true glue.session-token: type: string nullable: true glue.warehouse: type: string description: 'Location for table metadata. Example: `"s3://my-data-warehouse/tables/"`' nullable: true DatagenStrategy: type: string description: Strategy used to generate values. enum: - increment - uniform - zipf - word - words - sentence - sentences - paragraph - paragraphs - first_name - last_name - title - suffix - name - name_with_title - domain_suffix - email - username - password - field - position - seniority - job_title - ipv4 - ipv6 - ip - mac_address - user_agent - rfc_status_code - valid_status_code - company_suffix - company_name - buzzword - buzzword_middle - buzzword_tail - catch_phrase - bs_verb - bs_adj - bs_noun - bs - profession - industry - currency_code - currency_name - currency_symbol - credit_card_number - city_prefix - city_suffix - city_name - country_name - country_code - street_suffix - street_name - time_zone - state_name - state_abbr - secondary_address_type - secondary_address - zip_code - post_code - building_number - latitude - longitude - isbn - isbn13 - isbn10 - phone_number - cell_number - file_path - file_name - file_extension - dir_path CheckpointMetadata: type: object description: 'Holds meta-data about a checkpoint that was taken for persistent storage and recovery of a circuit''s state.' required: - uuid - fingerprint properties: fingerprint: type: integer format: int64 description: Fingerprint of the circuit at the time of the checkpoint. minimum: 0 identifier: type: string description: An optional name for the checkpoint. nullable: true processed_records: type: integer format: int64 description: Total number of records processed. nullable: true minimum: 0 size: type: integer format: int64 description: Total size of the checkpoint files in bytes. nullable: true minimum: 0 steps: type: integer format: int64 description: Total number of steps made. nullable: true minimum: 0 uuid: type: string format: uuid description: 'A unique identifier for the given checkpoint. This is used to identify the checkpoint in the file-system hierarchy.' KafkaHeader: type: object description: Kafka message header. required: - key properties: key: type: string value: allOf: - $ref: '#/components/schemas/KafkaHeaderValue' nullable: true ProgramError: type: object description: Log, warning and error information about the program compilation. properties: rust_compilation: allOf: - $ref: '#/components/schemas/RustCompilationInfo' nullable: true sql_compilation: allOf: - $ref: '#/components/schemas/SqlCompilationInfo' nullable: true system_error: type: string description: 'System error that occurred. - Set `Some(...)` upon transition to `SystemError` - Set `None` upon transition to `Pending`' nullable: true UrlInputConfig: type: object description: 'Configuration for reading data from an HTTP or HTTPS URL with `UrlInputTransport`.' required: - path properties: path: type: string description: URL. pause_timeout: type: integer format: int32 description: 'Timeout before disconnection when paused, in seconds. If the pipeline is paused, or if the input adapter reads data faster than the pipeline can process it, then the controller will pause the input adapter. If the input adapter stays paused longer than this timeout, it will drop the network connection to the server. It will automatically reconnect when the input adapter starts running again.' minimum: 0 IcebergIngestMode: type: string description: 'Iceberg table read mode. Three options are available: * `snapshot` - read a snapshot of the table and stop. * `follow` - continuously ingest changes to the table, starting from a specified snapshot or timestamp. * `snapshot_and_follow` - read a snapshot of the table before switching to continuous ingestion mode.' enum: - snapshot - follow - snapshot_and_follow TransportConfig: oneOf: - type: object required: - name - config properties: config: $ref: '#/components/schemas/FileInputConfig' name: type: string enum: - file_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/FileOutputConfig' name: type: string enum: - file_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/NatsInputConfig' name: type: string enum: - nats_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/KafkaInputConfig' name: type: string enum: - kafka_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/KafkaOutputConfig' name: type: string enum: - kafka_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/PubSubInputConfig' name: type: string enum: - pub_sub_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/UrlInputConfig' name: type: string enum: - url_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/S3InputConfig' name: type: string enum: - s3_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/DeltaTableReaderConfig' name: type: string enum: - delta_table_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/DeltaTableWriterConfig' name: type: string enum: - delta_table_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/DynamoDBWriterConfig' name: type: string enum: - dynamodb_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/RedisOutputConfig' name: type: string enum: - redis_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/IcebergReaderConfig' name: type: string enum: - iceberg_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/PostgresReaderConfig' name: type: string enum: - postgres_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/PostgresCdcReaderConfig' name: type: string enum: - postgres_cdc_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/PostgresWriterConfig' name: type: string enum: - postgres_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/DatagenInputConfig' name: type: string enum: - datagen - type: object required: - name - config properties: config: $ref: '#/components/schemas/NexmarkInputConfig' name: type: string enum: - nexmark - type: object required: - name - config properties: config: $ref: '#/components/schemas/HttpInputConfig' name: type: string enum: - http_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/HttpOutputConfig' name: type: string enum: - http_output - type: object required: - name - config properties: config: $ref: '#/components/schemas/AdHocInputConfig' name: type: string enum: - ad_hoc_input - type: object required: - name - config properties: config: $ref: '#/components/schemas/ClockConfig' name: type: string enum: - clock_input - type: object required: - name properties: name: type: string enum: - null_output - type: object required: - name properties: name: type: string enum: - empty_input description: 'Transport-specific endpoint configuration passed to `crate::OutputTransport::new_endpoint` and `crate::InputTransport::new_endpoint`.' discriminator: propertyName: name CombinedStatus: type: string enum: - Stopped - Provisioning - Unavailable - Coordination - Standby - AwaitingApproval - Initializing - Bootstrapping - ConcurrentBootstrapping - Synchronizing - Replaying - Paused - Running - Suspended - Stopping StorageStatus: type: string description: 'Storage status. The storage status can only transition when the resources status is `Stopped`. ```text Cleared ───┐ ▲ │ /clear │ │ │ │ Clearing │ ▲ │ │ │ InUse ◄───┘ ```' enum: - Cleared - InUse - Clearing ResourceConfig: type: object properties: cpu_cores_max: type: number format: double description: 'The maximum number of CPU cores to reserve for an instance of this pipeline' default: null nullable: true cpu_cores_min: type: number format: double description: 'The minimum number of CPU cores to reserve for an instance of this pipeline' default: null nullable: true memory_mb_max: type: integer format: int64 description: 'The maximum memory in Megabytes to reserve for an instance of this pipeline' default: null nullable: true minimum: 0 memory_mb_min: type: integer format: int64 description: 'The minimum memory in Megabytes to reserve for an instance of this pipeline' default: null nullable: true minimum: 0 namespace: type: string description: 'Kubernetes namespace to use for an instance of this pipeline. The namespace determines the scope of names for resources created for the pipeline. If not set, the pipeline will be deployed in the same namespace as the control-plane.' default: null nullable: true service_account_name: type: string description: 'Kubernetes service account name to use for an instance of this pipeline. The account determines permissions and access controls.' default: null nullable: true storage_class: type: string description: 'Storage class to use for an instance of this pipeline. The class determines storage performance such as IOPS and throughput.' default: null nullable: true storage_mb_max: type: integer format: int64 description: 'The total storage in Megabytes to reserve for an instance of this pipeline' default: null nullable: true minimum: 0 FileBackendConfig: type: object description: Configuration for local file system access. properties: async_threads: type: boolean description: 'Whether to use background threads for file I/O. Background threads should improve performance, but they can reduce performance if too few cores are available. This is provided for debugging and fine-tuning and should ordinarily be left unset.' default: null nullable: true ioop_delay: type: integer format: int64 description: 'Per-I/O operation sleep duration, in milliseconds. This is for simulating slow storage devices. Do not use this in production.' default: null nullable: true minimum: 0 sync: allOf: - $ref: '#/components/schemas/SyncConfig' default: null nullable: true PipelineInfo: allOf: - $ref: '#/components/schemas/ClientMetadata' - type: object required: - id - name - created_at - version - platform_version - runtime_config - program_code - udf_rust - udf_toml - program_config - program_version - program_status - program_status_since - program_error - refresh_version - storage_status - deployment_status - deployment_status_since - deployment_desired_status - deployment_desired_status_since - deployment_resources_status - deployment_resources_status_since - deployment_resources_desired_status - deployment_resources_desired_status_since properties: created_at: type: string format: date-time deployment_desired_status: $ref: '#/components/schemas/CombinedDesiredStatus' deployment_desired_status_since: type: string format: date-time deployment_error: allOf: - $ref: '#/components/schemas/ErrorResponse' nullable: true deployment_id: type: string format: uuid nullable: true deployment_initial: allOf: - $ref: '#/components/schemas/RuntimeDesiredStatus' nullable: true deployment_resources_desired_status: $ref: '#/components/schemas/ResourcesDesiredStatus' deployment_resources_desired_status_since: type: string format: date-time deployment_resources_status: $ref: '#/components/schemas/ResourcesStatus' deployment_resources_status_details: nullable: true deployment_resources_status_since: type: string format: date-time deployment_runtime_desired_status: allOf: - $ref: '#/components/schemas/RuntimeDesiredStatus' nullable: true deployment_runtime_desired_status_since: type: string format: date-time nullable: true deployment_runtime_status: allOf: - $ref: '#/components/schemas/RuntimeStatus' nullable: true deployment_runtime_status_details: allOf: - $ref: '#/components/schemas/RuntimeStatusDetails' nullable: true deployment_runtime_status_since: type: string format: date-time nullable: true deployment_status: $ref: '#/components/schemas/CombinedStatus' deployment_status_since: type: string format: date-time id: $ref: '#/components/schemas/PipelineId' name: type: string platform_version: type: string program_code: type: string program_config: $ref: '#/components/schemas/ProgramConfig' program_error: $ref: '#/components/schemas/ProgramError' program_info: allOf: - $ref: '#/components/schemas/PartialProgramInfo' nullable: true program_status: $ref: '#/components/schemas/ProgramStatus' program_status_since: type: string format: date-time program_version: $ref: '#/components/schemas/Version' refresh_version: $ref: '#/components/schemas/Version' runtime_config: $ref: '#/components/schemas/RuntimeConfig' storage_status: $ref: '#/components/schemas/StorageStatus' storage_status_details: allOf: - $ref: '#/components/schemas/StorageStatusDetails' nullable: true udf_rust: type: string udf_toml: type: string version: $ref: '#/components/schemas/Version' description: 'Pipeline information. It both includes fields which are user-provided and system-generated.' RuntimeConfig: type: object description: 'Global pipeline configuration settings. This is the publicly exposed type for users to configure pipelines.' properties: checkpoint_during_suspend: type: boolean description: 'Deprecated: setting this true or false does not have an effect anymore.' default: true clock_resolution_usecs: type: integer format: int64 description: 'Real-time clock resolution in microseconds. This parameter controls the execution of queries that use the `NOW()` function. The output of such queries depends on the real-time clock and can change over time without any external inputs. If the query uses `NOW()`, the pipeline will update the clock value and trigger incremental recomputation at most each `clock_resolution_usecs` microseconds. If the query does not use `NOW()`, then clock value updates are suppressed and the pipeline ignores this setting. It is set to 1 second (1,000,000 microseconds) by default.' default: 1000000 nullable: true minimum: 0 cpu_profiler: type: boolean description: 'Enable CPU profiler. The default value is `true`.' default: true datafusion_memory_mb: type: integer format: int64 description: 'DataFusion memory pool size, in MB, shared by the ad-hoc query engine and the Delta Lake / Iceberg connectors. Carved out of `max_rss_mb` (falling back to `resources.memory_mb_max`); the remainder goes to the DBSP circuit, so the two do not double-book RAM. Unset: defaults to 5% of the effective budget, capped at 2 GB. Pipelines that don''t run heavy ad-hoc / Delta / Iceberg workloads can leave this unset. Sort/aggregate-heavy ad-hoc queries (especially at high `workers` counts) should set this explicitly. An under-sized pool surfaces as `ResourcesExhausted` on the failing query — the pipeline keeps running and only that query fails. No pool limit applied if no overall budget is configured. See [documentation on the pipeline''s memory usage](https://docs.feldera.com/operations/memory) for more details.' default: null nullable: true minimum: 0 dev_tweaks: allOf: - $ref: '#/components/schemas/DevTweaks' default: {} env: type: object description: 'Environment variables for the pipeline process. These are key-value pairs injected into the pipeline process environment. Some variable names are reserved by the platform and cannot be overridden (for example `RUST_LOG`, and variables in the `FELDERA_`, `KUBERNETES_`, and `TOKIO_` namespaces).' default: {} additionalProperties: type: string fault_tolerance: allOf: - $ref: '#/components/schemas/FtConfig' default: model: none checkpoint_interval_secs: 60 hosts: type: integer description: 'Number of DBSP hosts. The worker threads are evenly divided among the hosts. For single-host deployments, this should be 1 (the default). Multihost pipelines are an enterprise-only preview feature.' default: 1 minimum: 0 http_workers: type: integer format: int64 description: 'Sets the number of available runtime threads for the http server. In most cases, this does not need to be set explicitly and the default is sufficient. Can be increased in case the pipeline HTTP API operations are a bottleneck. If not specified, the default is set to `workers`.' default: null nullable: true minimum: 0 init_containers: description: Specification of additional (sidecar) containers. nullable: true io_workers: type: integer format: int64 description: 'Sets the number of available runtime threads for async IO tasks. This affects some networking and file I/O operations especially adapters and ad-hoc queries. In most cases, this does not need to be set explicitly and the default is sufficient. Can be increased in case ingress, egress or ad-hoc queries are a bottleneck. If not specified, the default is set to `workers`.' default: null nullable: true minimum: 0 logging: type: string description: 'Log filtering directives. If set to a valid [tracing-subscriber] filter, this controls the log messages emitted by the pipeline process. Otherwise, or if the filter has invalid syntax, messages at "info" severity and higher are written to the log and all others are discarded. [tracing-subscriber]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html#directives' default: null nullable: true max_buffering_delay_usecs: type: integer format: int64 description: 'Maximal delay in microseconds to wait for `min_batch_size_records` to get buffered by the controller, defaults to 0.' default: 0 minimum: 0 max_parallel_connector_init: type: integer format: int64 description: 'The maximum number of connectors initialized in parallel during pipeline startup. At startup, the pipeline must initialize all of its input and output connectors. Depending on the number and types of connectors, this can take a long time. To accelerate the process, multiple connectors are initialized concurrently. This option controls the maximum number of connectors that can be initialized in parallel. The default is 10.' default: null nullable: true minimum: 0 max_rss_mb: type: integer format: int64 description: 'The maximum amount of memory, in Megabytes, that the pipeline is allowed to use on each host. Setting this property activates memory pressure monitoring and backpressure mechanisms. The pipeline will track the amount of remaining memory and report the memory pressure level via the `memory_pressure` metric. As the memory pressure increases, the system will apply increasing backpressure to push state cached in memory to storage, preventing the pipeline from running out of memory at the cost of some performance degradation. It is strongly recommended to set this property to prevent the pipeline from running out of memory. The setting should not exceed the memory limit of the pipeline instance. When `max_rss_mb` is not specified but `resources.memory_mb_max` is set, the latter is used as the effective memory cap for the pipeline. See [documentation on the pipeline''s memory usage](https://docs.feldera.com/operations/memory) for more details.' default: null nullable: true minimum: 0 min_batch_size_records: type: integer format: int64 description: 'Minimal input batch size. The controller delays pushing input records to the circuit until at least `min_batch_size_records` records have been received (total across all endpoints) or `max_buffering_delay_usecs` microseconds have passed since at least one input records has been buffered. Defaults to 0.' default: 0 minimum: 0 pin_cpus: type: array items: type: integer minimum: 0 description: 'Optionally, a list of CPU numbers for CPUs to which the pipeline may pin its worker threads. Specify at least twice as many CPU numbers as workers. CPUs are generally numbered starting from 0. The pipeline might not be able to honor CPU pinning requests. CPU pinning can make pipelines run faster and perform more consistently, as long as different pipelines running on the same machine are pinned to different CPUs.' default: [] pipeline_template_configmap: allOf: - $ref: '#/components/schemas/PipelineTemplateConfig' default: null nullable: true provisioning_timeout_secs: type: integer format: int64 description: 'Timeout in seconds for the `Provisioning` phase of the pipeline. Setting this value will override the default of the runner.' default: null nullable: true minimum: 0 resources: allOf: - $ref: '#/components/schemas/ResourceConfig' default: cpu_cores_min: null cpu_cores_max: null memory_mb_min: null memory_mb_max: null storage_mb_max: null storage_class: null service_account_name: null namespace: null storage: allOf: - $ref: '#/components/schemas/StorageOptions' default: backend: name: default min_storage_bytes: null min_step_storage_bytes: null compression: default cache_mib: null nullable: true tracing: type: boolean description: Enable pipeline tracing. default: false tracing_endpoint_jaeger: type: string description: Jaeger tracing endpoint to send tracing information to. default: 127.0.0.1:6831 workers: type: integer format: int32 description: 'Number of DBSP worker threads. Each DBSP "foreground" worker thread is paired with a "background" thread for LSM merging, making the total number of threads twice the specified number. The typical sweet spot for the number of workers is between 4 and 16. Each worker increases overall memory consumption for data structures used during a step.' default: 8 minimum: 0 NatsInputConfig: type: object required: - connection_config - stream_name - consumer_config properties: connection_config: $ref: '#/components/schemas/ConnectOptions' consumer_config: $ref: '#/components/schemas/ConsumerConfig' inactivity_timeout_secs: type: integer format: int64 description: 'Maximum time in seconds to wait for the next message before running a stream/server health check. Must be at least 1.' minimum: 0 retry_interval_secs: type: integer format: int64 description: 'Delay between automatic reconnect attempts while in retry mode. Must be at least 1.' minimum: 0 stream_name: type: string PatchClientMetadata: type: object description: 'Client-generated metadata as supplied in a `PATCH` request body: the field-by-field patch form of [`ClientMetadata`]. Each field is optional. A `Some` value overwrites the stored field; a `None` (absent) field leaves it unchanged. An empty string or empty list is a value in its own right, not a request to unset the field.' properties: description: type: string description: Human-readable description of the pipeline. nullable: true tags: type: array items: type: string description: Free-form labels used to organize, group, and filter pipelines. nullable: true ConnectOptions: type: object description: Options for connecting to a NATS server. required: - server_url properties: auth: $ref: '#/components/schemas/Auth' connection_timeout_secs: type: integer format: int64 description: 'Connection timeout How long to wait when establishing the initial connection to the NATS server.' minimum: 0 request_timeout_secs: type: integer format: int64 description: 'Request timeout in seconds. How long to wait for responses to requests.' minimum: 0 server_url: type: string description: NATS server URL (e.g., "nats://localhost:4222"). ProgramConfig: type: object description: Program configuration. properties: cache: type: boolean description: 'If `true` (default), when a prior compilation with the same checksum already exists, the output of that (i.e., binary) is used. Set `false` to always trigger a new compilation, which might take longer and as well can result in overriding an existing binary.' default: true profile: allOf: - $ref: '#/components/schemas/CompilationProfile' default: null nullable: true runtime_version: type: string description: 'Override runtime version of the pipeline being executed. Warning: This setting is experimental and may change in the future. Requires the platform to run with the unstable feature `runtime_version` enabled. Should only be used for testing purposes, and requires network access. A runtime version can be specified in the form of a version or SHA taken from the `feldera/feldera` repository main branch. Examples: `v0.96.0` or `f4dcac0989ca0fda7d2eb93602a49d007cb3b0ae` A platform of version `0.x.y` may be capable of running future and past runtimes with versions `>=0.x.y` and `<=0.x.y` until breaking API changes happen, the exact bounds for each platform version are unspecified until we reach a stable version. Compatibility is only guaranteed if platform and runtime version are exact matches. Note that any enterprise features are currently considered to be part of the platform. If not set (null), the runtime version will be the same as the platform version.' default: null nullable: true use_platform_compiler: type: boolean description: 'Use the platform SQL compiler when a non-platform `runtime_version` is specified. Warning: This setting is experimental and may change in the future. Requires the platform to run with the unstable feature `runtime_version` enabled. When `false` (default), the SQL compiler matching the `runtime_version` is downloaded and used. When `true`, the platform''s SQL compiler is used instead. Setting this to `true` avoids downloading the runtime-version-specific SQL compiler JAR (e.g., when network access is unavailable or slow), at the cost of potentially using a mismatched SQL compiler. The Rust runtime sources are still checked out and compiled from the requested `runtime_version`. Has no effect when `runtime_version` is not set or the platform does not have the unstable feature `runtime_version` enabled.' default: false DeltaTableWriteMode: type: string description: 'Delta table write mode. Determines how the Delta table connector handles an existing table at the target location.' enum: - append - truncate - error_if_exists ProgramSchema: type: object description: 'A struct containing the tables (inputs) and views for a program. Parse from the JSON data-type of the DDL generated by the SQL compiler.' required: - inputs - outputs properties: inputs: type: array items: $ref: '#/components/schemas/Relation' outputs: type: array items: $ref: '#/components/schemas/Relation' DeliverPolicy: oneOf: - type: string enum: - All - type: string enum: - Last - type: string enum: - New - type: object required: - ByStartSequence properties: ByStartSequence: type: object required: - start_sequence properties: start_sequence: type: integer format: int64 minimum: 0 - type: object required: - ByStartTime properties: ByStartTime: type: object required: - start_time properties: start_time: type: string format: date-time example: '2023-01-15T09:30:00Z' - type: string enum: - LastPerSubject AdHocInputConfig: type: object description: 'Configuration for inserting data with ad-hoc queries An ad-hoc input adapters cannot be usefully configured as part of pipeline configuration. Instead, use ad-hoc queries through the UI, the REST API, or the `fda` command-line tool.' required: - name properties: name: type: string description: Autogenerated name. PostgresReaderConfig: allOf: - type: object description: TLS/SSL configuration for PostgreSQL connectors. properties: ssl_ca_location: type: string description: Path to a file containing a sequence of CA certificates in PEM format. nullable: true ssl_ca_pem: type: string description: A sequence of CA certificates in PEM format. nullable: true ssl_certificate_chain_location: type: string description: 'The path to the certificate chain file. The file must contain a sequence of PEM-formatted certificates, the first being the leaf certificate, and the remainder forming the chain of certificates up to and including the trusted root certificate.' nullable: true ssl_client_key: type: string description: The client certificate key in PEM format. nullable: true ssl_client_key_location: type: string description: Path to the client certificate key. nullable: true ssl_client_location: type: string description: Path to the client certificate. nullable: true ssl_client_pem: type: string description: The client certificate in PEM format. nullable: true verify_hostname: type: boolean description: True to enable hostname verification when using TLS. True by default. nullable: true - type: object required: - uri - query properties: query: type: string description: Query that specifies what data to fetch from postgres. uri: type: string description: 'Postgres URI. See: ' description: Postgres input connector configuration. SqlCompilerMessage: type: object description: 'A SQL compiler error. The SQL compiler returns a list of errors in the following JSON format if it''s invoked with the `-je` option. ```ignore [ { "start_line_number" : 2, "start_column" : 4, "end_line_number" : 2, "end_column" : 8, "warning" : false, "error_type" : "PRIMARY KEY cannot be nullable", "message" : "PRIMARY KEY column ''C'' has type INTEGER, which is nullable", "snippet" : " 2| c INT PRIMARY KEY\n ^^^^^\n 3|);\n" } ] ```' required: - start_line_number - start_column - end_line_number - end_column - warning - error_type - message properties: end_column: type: integer minimum: 0 end_line_number: type: integer minimum: 0 error_type: type: string message: type: string snippet: type: string nullable: true start_column: type: integer minimum: 0 start_line_number: type: integer minimum: 0 warning: type: boolean KafkaOutputConfig: type: object description: Configuration for writing data to a Kafka topic with `OutputTransport`. required: - topic properties: fault_tolerance: allOf: - $ref: '#/components/schemas/KafkaOutputFtConfig' nullable: true headers: type: array items: $ref: '#/components/schemas/KafkaHeader' description: Kafka headers to be added to each message produced by this connector. initialization_timeout_secs: type: integer format: int32 description: 'Maximum timeout in seconds to wait for the endpoint to connect to a Kafka broker. Defaults to 60.' minimum: 0 kafka_service: type: string description: If specified, this service is used to provide defaults for the Kafka options. nullable: true log_level: allOf: - $ref: '#/components/schemas/KafkaLogLevel' nullable: true region: type: string description: The AWS region to use while connecting to AWS Managed Streaming for Kafka (MSK). nullable: true topic: type: string description: Topic to write to. additionalProperties: type: string description: 'Options passed directly to `rdkafka`. See [`librdkafka` options](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md) used to configure the Kafka producer.' ColumnType: type: object description: 'A SQL column type description. Matches the Calcite JSON format.' required: - nullable properties: component: allOf: - $ref: '#/components/schemas/ColumnType' nullable: true fields: type: array items: $ref: '#/components/schemas/Field' description: 'The fields of the type (if available). For example this would specify the fields of a `CREATE TYPE` construct. ```sql CREATE TYPE person_typ AS ( firstname VARCHAR(30), lastname VARCHAR(30), address ADDRESS_TYP ); ``` Would lead to the following `fields` value: ```sql [ ColumnType { name: "firstname, ... }, ColumnType { name: "lastname", ... }, ColumnType { name: "address", fields: [ ... ] } ] ```' nullable: true key: allOf: - $ref: '#/components/schemas/ColumnType' nullable: true nullable: type: boolean description: Does the type accept NULL values? precision: type: integer format: int64 description: 'Precision of the type. # Examples - `VARCHAR` sets precision to `-1`. - `VARCHAR(255)` sets precision to `255`. - `BIGINT`, `DATE`, `FLOAT`, `DOUBLE`, `GEOMETRY`, etc. sets precision to None - `TIME`, `TIMESTAMP` set precision to `0`.' nullable: true scale: type: integer format: int64 description: 'The scale of the type. # Example - `DECIMAL(1,2)` sets scale to `2`.' nullable: true type: $ref: '#/components/schemas/SqlType' value: allOf: - $ref: '#/components/schemas/ColumnType' nullable: true StorageStatusDetails: type: object description: 'Details about pipeline storage, which are returned as part of the regular runtime status polling by the runner.' required: - checkpoints properties: checkpoints: type: array items: $ref: '#/components/schemas/CheckpointMetadata' description: Present checkpoints. MergerType: type: string description: Which merger to use. enum: - push_merger - list_merger IcebergCatalogType: type: string enum: - rest - glue - s3tables FtConfig: type: object description: 'Fault-tolerance configuration. The default [FtConfig] (via [FtConfig::default]) disables fault tolerance, which is the configuration that one gets if [RuntimeConfig] omits fault tolerance configuration. The default value for [FtConfig::model] enables fault tolerance, as `Some(FtModel::default())`. This is the configuration that one gets if [RuntimeConfig] includes a fault tolerance configuration but does not specify a particular model.' properties: checkpoint_interval_secs: type: integer format: int64 description: 'Interval between automatic checkpoints, in seconds. The default is 60 seconds. Values less than 1 or greater than 3600 will be forced into that range.' nullable: true minimum: 0 model: oneOf: - $ref: '#/components/schemas/FtModel' - type: string enum: - none default: exactly_once DeltaTableWriterConfig: type: object description: Delta table output connector configuration. required: - uri properties: checkpoint_interval: type: integer format: int32 description: 'Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables. The option is only available when creating the Delta table (`mode = append` and there is no existing table at the target location or `mode = truncate`). It configures the `checkpointInterval` table property, which determines the number of commits after which a new checkpoint should be created. 0 means no checkpoints are created. Default: 10.' nullable: true minimum: 0 enable_expired_log_cleanup: type: boolean description: 'Whether to clean up expired log entries when a checkpoint is written. Configures the `delta.enableExpiredLogCleanup` table property. When set to `false`, transaction log entries are retained indefinitely regardless of `log_retention_duration`. The option is only available when creating the Delta table (`mode = append` and there is no existing table at the target location, or `mode = truncate`). Default: `true` (Delta Lake default).' nullable: true log_retention_duration: type: string description: 'Log retention duration for newly created Delta tables. Configures the `delta.logRetentionDuration` table property, which controls how long the transaction log history of the table is kept. Each time a checkpoint is written, Delta Lake automatically cleans up log entries older than this interval (subject to `enable_expired_log_cleanup`). The option is only available when creating the Delta table (`mode = append` and there is no existing table at the target location, or `mode = truncate`). The value follows the Delta Lake interval syntax: `"interval "`, where `` is one of `nanosecond[s]`, `microsecond[s]`, `millisecond[s]`, `second[s]`, `minute[s]`, `hour[s]`, `day[s]`, or `week[s]`. Examples: `"interval 30 days"`, `"interval 6 hours"`. Default: `"interval 30 days"` (Delta Lake default).' nullable: true max_retries: type: integer format: int32 description: 'Maximum number of retries for failed operations. The connector performs retries on several levels: individual S3 operations, Delta Lake transaction commits, and overall operation retries. This setting controls the overall operation retries. When a write to the table fails, because of an S3 timeout or any other reason that was not resolved by lower-level retries, the connector will retry the entire operation. When not specified, the connector performs infinite retries. When set to 0, the connector doesn''t retry failed operations.' nullable: true minimum: 0 mode: $ref: '#/components/schemas/DeltaTableWriteMode' threads: type: integer description: 'Number of parallel threads used by the connector. Increasing this value can improve Delta Lake write throughput by enabling concurrent writes. Default: 1.' nullable: true minimum: 1 uri: type: string description: Table URI. additionalProperties: type: string description: 'Storage options for configuring backend object store. For specific options available for different storage backends, see: * [Azure options](https://docs.rs/object_store/latest/object_store/azure/enum.AzureConfigKey.html) * [Amazon S3 options](https://docs.rs/object_store/latest/object_store/aws/enum.AmazonS3ConfigKey.html) * [Google Cloud Storage options](https://docs.rs/object_store/latest/object_store/gcp/enum.GoogleConfigKey.html)' ConsumerConfig: type: object required: - deliver_policy properties: deliver_policy: $ref: '#/components/schemas/DeliverPolicy' description: type: string nullable: true filter_subjects: type: array items: type: string max_batch: type: integer format: int64 nullable: true max_bytes: type: integer format: int64 nullable: true max_expires: type: string nullable: true max_waiting: type: integer format: int64 metadata: type: object additionalProperties: type: string name: type: string nullable: true rate_limit: type: integer format: int64 minimum: 0 replay_policy: $ref: '#/components/schemas/ReplayPolicy' StorageOptions: type: object description: Storage configuration for a pipeline. properties: backend: allOf: - $ref: '#/components/schemas/StorageBackendConfig' default: name: default cache_mib: type: integer description: 'The maximum size of the in-memory storage cache, in MiB. If set, the specified cache size is spread across all the foreground and background threads. If unset, each foreground or background thread cache is limited to 256 MiB.' default: null nullable: true minimum: 0 compression: allOf: - $ref: '#/components/schemas/StorageCompression' default: default min_step_storage_bytes: type: integer description: 'For a batch of data passed through the pipeline during a single step, the minimum estimated number of bytes to write it to storage. This is provided for debugging and fine-tuning and should ordinarily be left unset. A value of 0 will write even empty batches to storage, and nonzero values provide a threshold. `usize::MAX`, the default, effectively disables storage for such batches. If it is set to another value, it should ordinarily be greater than or equal to `min_storage_bytes`.' default: null nullable: true minimum: 0 min_storage_bytes: type: integer description: 'For a batch of data maintained as part of a persistent index during a pipeline run, the minimum estimated number of bytes to write it to storage. This is provided for debugging and fine-tuning and should ordinarily be left unset. A value of 0 will write even empty batches to storage, and nonzero values provide a threshold. `usize::MAX` would effectively disable storage for such batches. The default is 10,048,576 (10 MiB).' default: null nullable: true minimum: 0 BufferCacheAllocationStrategy: type: string description: Controls how caches are shared across a foreground/background worker pair. enum: - shared_per_worker_pair - per_thread - global StartFromCheckpoint: oneOf: - type: string enum: - latest - type: string format: uuid nullable: true PatchPipeline: allOf: - $ref: '#/components/schemas/PatchClientMetadata' - type: object properties: name: type: string nullable: true program_code: type: string nullable: true program_config: allOf: - $ref: '#/components/schemas/ProgramConfig' nullable: true runtime_config: allOf: - $ref: '#/components/schemas/RuntimeConfig' nullable: true udf_rust: type: string nullable: true udf_toml: type: string nullable: true description: 'Partially update the pipeline (PATCH). Note that the patching only applies to the main fields, not subfields. For instance, it is not possible to update only the number of workers; it is required to again pass the whole runtime configuration with the change.' PartialProgramInfo: type: object description: Program information is the result of the SQL compilation. required: - schema - udf_stubs - input_connectors - output_connectors properties: input_connectors: type: object description: Input connectors derived from the schema. additionalProperties: $ref: '#/components/schemas/InputEndpointConfig' output_connectors: type: object description: Output connectors derived from the schema. additionalProperties: $ref: '#/components/schemas/OutputEndpointConfig' schema: $ref: '#/components/schemas/ProgramSchema' udf_stubs: type: string description: 'Generated user defined function (UDF) stubs Rust code: stubs.rs' RedisOutputConfig: type: object description: Redis output connector configuration. required: - connection_string properties: connection_string: type: string description: 'The URL format: `redis://[][:@][:port][/[][?protocol=]]` This is parsed by the [redis](https://docs.rs/redis/latest/redis/#connection-parameters) crate.' key_separator: type: string description: 'Separator used to join multiple components into a single key. ":" by default.' UserAndPassword: type: object required: - user - password properties: password: type: string user: type: string PostgresWriteMode: type: string description: 'PostgreSQL write mode. Determines how the PostgreSQL output connector writes data to the target table.' enum: - materialized - cdc Auth: type: object properties: credentials: allOf: - $ref: '#/components/schemas/Credentials' nullable: true jwt: type: string nullable: true nkey: type: string nullable: true token: type: string nullable: true user_and_password: allOf: - $ref: '#/components/schemas/UserAndPassword' nullable: true PostgresWriterConfig: allOf: - type: object description: TLS/SSL configuration for PostgreSQL connectors. properties: ssl_ca_location: type: string description: Path to a file containing a sequence of CA certificates in PEM format. nullable: true ssl_ca_pem: type: string description: A sequence of CA certificates in PEM format. nullable: true ssl_certificate_chain_location: type: string description: 'The path to the certificate chain file. The file must contain a sequence of PEM-formatted certificates, the first being the leaf certificate, and the remainder forming the chain of certificates up to and including the trusted root certificate.' nullable: true ssl_client_key: type: string description: The client certificate key in PEM format. nullable: true ssl_client_key_location: type: string description: Path to the client certificate key. nullable: true ssl_client_location: type: string description: Path to the client certificate. nullable: true ssl_client_pem: type: string description: The client certificate in PEM format. nullable: true verify_hostname: type: boolean description: True to enable hostname verification when using TLS. True by default. nullable: true - type: object required: - uri - table properties: cdc_op_column: type: string description: 'Name of the operation metadata column in CDC mode. Only used when `mode = "cdc"`. This column will contain: - `"i"` for insert operations - `"u"` for upsert operations - `"d"` for delete operations Default: `"__feldera_op"`' default: __feldera_op cdc_ts_column: type: string description: 'Name of the timestamp metadata column in CDC mode. Only used when `mode = "cdc"`. This column will contain the timestamp (in RFC 3339 format) when the batch of updates was output by the pipeline. Default: `"__feldera_ts"`' default: __feldera_ts extra_columns: type: array items: type: string description: 'The names of the extra columns in the Postgres table that are not part of the view schema. These connector can write user-defined values, configured using the `set_extra_columns` connector command, to these columns.' max_buffer_size_bytes: type: integer description: 'The maximum buffer size in for a single operation. Note that the buffers of `INSERT`, `UPDATE` and `DELETE` queries are separate. Default: 1 MiB' default: 1048576 minimum: 0 max_records_in_buffer: type: integer description: The maximum number of records in a single buffer. nullable: true minimum: 0 mode: allOf: - $ref: '#/components/schemas/PostgresWriteMode' default: materialized on_conflict_do_nothing: type: boolean description: 'Specifies how the connector handles conflicts when executing an `INSERT` into a table with a primary key. By default, an existing row with the same key is overwritten. Setting this flag to `true` preserves the existing row and ignores the new insert. This setting does not affect `UPDATE` statements, which always replace the value associated with the key. This setting is not supported when `mode = "cdc"`, since all operations are performed as append-only `INSERT`s into the target table. Any conflict in CDC mode will result in an error. Default: `false`' table: type: string description: The table to write the output to. threads: type: integer description: 'The number of threads to use during encoding. Default: 1' default: 1 minimum: 0 uri: type: string description: 'Postgres URI. See: ' description: Postgres output connector configuration. KafkaHeaderValue: type: string format: binary description: Kafka header value encoded as a UTF-8 string or a byte array. PostprocessorConfig: type: object description: Configuration for describing a postprocessor required: - name - config properties: config: description: 'Arbitrary additional configuration expected by the postprocessor encoded as a JSON Value.' name: type: string description: 'Name of the postprocessor. All postprocessors with the same name will perform the same task.' CombinedDesiredStatus: type: string enum: - Stopped - Unavailable - Standby - Paused - Running - Suspended PipelineFieldSelector: type: string enum: - all - status - status_with_connectors DeltaTableReaderConfig: type: object description: Delta table input connector configuration. required: - uri - mode properties: cdc_delete_filter: type: string description: 'A predicate that determines whether the record represents a deletion. This setting is only valid in the `cdc` mode. It specifies a predicate applied to each row in the Delta table to determine whether the row represents a deletion event. Its value must be a valid Boolean SQL expression that can be used in a query of the form `SELECT * from WHERE `.' nullable: true cdc_order_by: type: string description: 'An expression that determines the ordering of updates in the Delta table. This setting is only valid in the `cdc` mode. It specifies a predicate applied to each row in the Delta table to determine the order in which updates in the table should be applied. Its value must be a valid SQL expression that can be used in a query of the form `SELECT * from
ORDER BY `.' nullable: true datetime: type: string description: 'Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector finds and opens the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data). In `snapshot` and `snapshot_and_follow` modes, it retrieves the snapshot of this version of the table. In `follow`, `snapshot_and_follow`, and `cdc` modes, it follows transaction log records **after** this version. Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.' nullable: true end_version: type: integer format: int64 description: 'Optional final table version. Valid only when the connector is configured in `follow`, `snapshot_and_follow`, or `cdc` mode. When set, the connector will stop scanning the table’s transaction log after reaching this version or any greater version. This bound is inclusive: if the specified version appears in the log, it will be processed before signaling end-of-input.' nullable: true filter: type: string description: 'Optional row filter. When specified, only rows that satisfy the filter condition are read from the delta table. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from my_table where ...` query.' nullable: true max_concurrent_readers: type: integer format: int32 description: 'Maximum number of concurrent object store reads performed by all Delta Lake connectors. This setting is used to limit the number of concurrent reads of the object store in a pipeline with a large number of Delta Lake connectors. When multiple connectors are simultaneously reading from the object store, this can lead to transport timeouts. When enabled, this setting limits the number of concurrent reads across all connectors. This is a global setting that affects all Delta Lake connectors, and not just the connector where it is specified. It should therefore be used at most once in a pipeline. If multiple connectors specify this setting, they must all use the same value. The default value is 6.' nullable: true minimum: 0 max_retries: type: integer format: int32 description: 'Maximum number of retries for failed object store operations. Controls how many times the connector retries high-level storage operations, such as reading a Delta log entry or a Parquet file. This is in addition to lower-level retries (e.g., individual S3 operation retries governed by storage options like `retry_timeout`). If those retries are exhausted or the failure is otherwise unrecoverable at the storage layer, the connector retries the entire operation. Defaults to unlimited retries. Set to 0 to disable retries.' nullable: true minimum: 0 mode: $ref: '#/components/schemas/DeltaTableIngestMode' num_parsers: type: integer format: int32 description: 'The number of parallel parsing tasks the connector uses to process data read from the table. Increasing this value can enhance performance by allowing more concurrent processing. Recommended range: 1–10. The default is 4.' minimum: 0 skip_unused_columns: type: boolean description: 'Don''t read unused columns from the Delta table. When set to `true`, this option instructs the connector to avoid reading columns from the Delta table that are not used in any view definitions. To be skipped, the columns must be either nullable or have default values. This can improve ingestion performance, especially for wide tables. Note: The simplest way to exclude unused columns is to omit them from the Feldera SQL table declaration. The connector never reads columns that aren''t declared in the SQL schema. Additionally, the SQL compiler emits warnings for declared but unused columns—use these as a guide to optimize your schema.' snapshot_filter: type: string description: 'Optional snapshot filter. This option is only valid when `mode` is set to `snapshot` or `snapshot_and_follow`. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ...` query. Unlike the `filter` option, which applies to all records retrieved from the table, this filter only applies to rows in the initial snapshot of the table. For instance, it can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP ''2005-01-01 00:00:00'' AND TIMESTAMP ''2010-12-31 23:59:59''`. This option can be used together with the `filter` option. During the initial snapshot, only rows that satisfy both `filter` and `snapshot_filter` are retrieved from the Delta table. When subsequently following changes in the the transaction log (`mode = snapshot_and_follow`), all rows that meet the `filter` condition are ingested, regardless of `snapshot_filter`.' nullable: true timestamp_column: type: string description: 'Table column that serves as an event timestamp. When this option is specified, and `mode` is one of `snapshot` or `snapshot_and_follow`, table rows are ingested in the timestamp order, respecting the [`LATENESS`](https://docs.feldera.com/sql/streaming#lateness-expressions) property of the column: each ingested row has a timestamp no more than `LATENESS` time units earlier than the most recent timestamp of any previously ingested row. The ingestion is performed by partitioning the table into timestamp ranges of width `LATENESS`. Each range is processed sequentially, in increasing timestamp order. # Example Consider a table with timestamp column of type `TIMESTAMP` and lateness attribute `INTERVAL 1 DAY`. Assuming that the oldest timestamp in the table is `2024-01-01T00:00:00``, the connector will fetch all records with timestamps from `2024-01-01`, then all records for `2024-01-02`, `2024-01-03`, etc., until all records in the table have been ingested. # Requirements * The timestamp column must be of a supported type: integer, `DATE`, or `TIMESTAMP`. * The timestamp column must be declared with non-zero `LATENESS`. * For efficient ingest, the table must be optimized for timestamp-based queries using partitioning, Z-ordering, or liquid clustering.' nullable: true transaction_mode: $ref: '#/components/schemas/DeltaTableTransactionMode' uri: type: string description: 'Table URI. Example: "s3://feldera-fraud-detection-data/demographics_train"' verbose: type: integer format: int32 description: 'Enable verbose logging. When enabled, the connector will log detailed information at INFO level. Supported values: * 0 - no verbose logging * 1 - log all Delta log entries in follow and cdc modes. * >1 - reserved for future use' minimum: 0 version: type: integer format: int64 description: 'Optional table version. When this option is set, the connector finds and opens the specified version of the table. In `snapshot` and `snapshot_and_follow` modes, it retrieves the snapshot of this version of the table. In `follow`, `snapshot_and_follow`, and `cdc` modes, it follows transaction log records **after** this version. Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.' nullable: true additionalProperties: type: string description: 'Storage options for configuring backend object store. For specific options available for different storage backends, see: * [Azure options](https://docs.rs/object_store/latest/object_store/azure/enum.AzureConfigKey.html) * [Amazon S3 options](https://docs.rs/object_store/latest/object_store/aws/enum.AmazonS3ConfigKey.html) * [Google Cloud Storage options](https://docs.rs/object_store/latest/object_store/gcp/enum.GoogleConfigKey.html)' RestCatalogConfig: type: object description: Iceberg REST catalog config. properties: rest.audience: type: string description: Logical name of target resource or service. nullable: true rest.credential: type: string description: 'Credential to use for OAuth2 credential flow when initializing the catalog. A key and secret pair separated by ":" (key is optional).' nullable: true rest.headers: type: array items: type: array items: allOf: - type: string - type: string description: Additional HTTP request headers added to each catalog REST API call. nullable: true rest.oauth2-server-uri: type: string description: 'Authentication URL to use for client credentials authentication (default: uri + ''v1/oauth/tokens'')' nullable: true rest.prefix: type: string description: 'Customize table storage paths. When combined with the `warehouse` property, the prefix determines how table data is organized within the storage.' nullable: true rest.resource: type: string description: URI for the target resource or service. nullable: true rest.scope: type: string nullable: true rest.token: type: string description: Bearer token value to use for `Authorization` header. nullable: true rest.uri: type: string description: URI identifying the REST catalog server. nullable: true rest.warehouse: type: string description: The default location for managed tables created by the catalog. nullable: true StorageBackendConfig: oneOf: - type: object required: - name properties: name: type: string enum: - default - type: object required: - name - config properties: config: $ref: '#/components/schemas/FileBackendConfig' name: type: string enum: - file - type: object required: - name - config properties: config: $ref: '#/components/schemas/ObjectStorageConfig' name: type: string enum: - object description: Backend storage configuration. discriminator: propertyName: name PipelineSelectedInfo: allOf: - $ref: '#/components/schemas/ClientMetadata' - type: object required: - id - name - created_at - version - platform_version - program_version - program_status - program_status_since - refresh_version - storage_status - deployment_status - deployment_status_since - deployment_desired_status - deployment_desired_status_since - deployment_resources_status - deployment_resources_status_since - deployment_resources_desired_status - deployment_resources_desired_status_since properties: connectors: allOf: - $ref: '#/components/schemas/ConnectorStats' nullable: true created_at: type: string format: date-time deployment_desired_status: $ref: '#/components/schemas/CombinedDesiredStatus' deployment_desired_status_since: type: string format: date-time deployment_error: allOf: - $ref: '#/components/schemas/ErrorResponse' nullable: true deployment_id: type: string format: uuid nullable: true deployment_initial: allOf: - $ref: '#/components/schemas/RuntimeDesiredStatus' nullable: true deployment_resources_desired_status: $ref: '#/components/schemas/ResourcesDesiredStatus' deployment_resources_desired_status_since: type: string format: date-time deployment_resources_status: $ref: '#/components/schemas/ResourcesStatus' deployment_resources_status_details: nullable: true deployment_resources_status_since: type: string format: date-time deployment_runtime_desired_status: allOf: - $ref: '#/components/schemas/RuntimeDesiredStatus' nullable: true deployment_runtime_desired_status_since: type: string format: date-time nullable: true deployment_runtime_status: allOf: - $ref: '#/components/schemas/RuntimeStatus' nullable: true deployment_runtime_status_details: allOf: - $ref: '#/components/schemas/RuntimeStatusDetails' nullable: true deployment_runtime_status_since: type: string format: date-time nullable: true deployment_status: $ref: '#/components/schemas/CombinedStatus' deployment_status_since: type: string format: date-time id: $ref: '#/components/schemas/PipelineId' name: type: string platform_version: type: string program_code: type: string nullable: true program_config: allOf: - $ref: '#/components/schemas/ProgramConfig' nullable: true program_error: allOf: - $ref: '#/components/schemas/ProgramError' nullable: true program_info: allOf: - $ref: '#/components/schemas/PartialProgramInfo' nullable: true program_status: $ref: '#/components/schemas/ProgramStatus' program_status_since: type: string format: date-time program_version: $ref: '#/components/schemas/Version' refresh_version: $ref: '#/components/schemas/Version' runtime_config: allOf: - $ref: '#/components/schemas/RuntimeConfig' nullable: true storage_status: $ref: '#/components/schemas/StorageStatus' storage_status_details: allOf: - $ref: '#/components/schemas/StorageStatusDetails' nullable: true udf_rust: type: string nullable: true udf_toml: type: string nullable: true version: $ref: '#/components/schemas/Version' description: 'Pipeline information which has a selected subset of optional fields. It both includes fields which are user-provided and system-generated. If an optional field is not selected (i.e., is `None`), it will not be serialized.' DynamoDBWriterConfig: type: object description: DynamoDB output connector configuration. required: - table - region properties: aws_access_key_id: type: string description: 'AWS access key ID. If both `aws_access_key_id` and `aws_secret_access_key` are specified, the connector uses these static credentials. Otherwise it uses the default AWS credential provider chain, including IAM Roles for Service Accounts (IRSA) in EKS. Static credentials are treated as long-lived IAM keys: no session token is sent, so STS-issued temporary credentials are not supported via these fields. To use temporary credentials, rely on the default provider chain instead (leave these unset).' nullable: true aws_secret_access_key: type: string description: AWS secret access key. nullable: true batch_size: type: integer description: 'Maximum number of write requests in one DynamoDB write call. DynamoDB supports at most 100 for `TransactWriteItems` and at most 25 for `BatchWriteItem`. If omitted, the connector uses the maximum for the selected `write_mode`.' nullable: true minimum: 0 delete_condition_expression: type: string description: '[Condition expression] evaluated before each delete. When the condition is false, the delete is skipped without failing the connector. This option requires `transactional` [`write_mode`]. The expression cannot use `ExpressionAttributeNames` or `ExpressionAttributeValues` placeholders. [Condition expression]: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Expressions.ConditionExpressions.html [`write_mode`]: Self::write_mode' nullable: true endpoint_url: type: string description: 'Optional endpoint URL, for example when using a local DynamoDB-compatible service.' nullable: true max_buffer_size_bytes: type: integer description: 'Maximum number of bytes buffered by each worker before flushing writes. This is an approximate size based on encoded DynamoDB attributes.' default: 1048576 minimum: 0 max_concurrent_requests: type: integer description: 'Maximum number of DynamoDB write requests in flight per worker thread. The total in-flight request count across the connector is `threads × max_concurrent_requests`. Size this accordingly when tuning against a provisioned-throughput table to avoid excessive throttling.' default: 64 minimum: 0 max_retries: type: integer format: int32 description: 'Maximum number of retries for a failed or partially-applied DynamoDB write chunk. For `batch` writes, `BatchWriteItem` may return some items as "unprocessed" in a successful 200 response; those are re-submitted and counted as retries. For `transactional` writes, a failed `TransactWriteItems` call is retried in full. Transient errors (throttling, network failures) are first handled transparently by the AWS SDK. Only attempts that reach this connector''s retry loop count against this limit. Each retry waits longer than the previous one (exponential backoff), up to a ceiling of roughly 13 seconds. Set to `null` to retry indefinitely. After the backoff ceiling is reached the connector keeps retrying at that interval, providing backpressure until DynamoDB recovers. Defaults to `10`.' default: 10 nullable: true minimum: 0 put_condition_expression: type: string description: '[Condition expression] evaluated before each put (insert or upsert). When the condition is false, the record is skipped without failing the connector. This option requires `transactional` [`write_mode`]. The condition gates every value write, and the connector cannot tell a replayed insert from a legitimate update: both arrive as puts. A guard such as `attribute_not_exists(id)` therefore also suppresses updates to existing keys, not just replayed duplicates. Choose the expression accordingly. The connector does not currently support `ExpressionAttributeNames` or `ExpressionAttributeValues`, so the expression cannot use placeholders. [Condition expression]: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Expressions.ConditionExpressions.html [`write_mode`]: Self::write_mode' nullable: true region: type: string description: AWS region. table: type: string description: Name of the DynamoDB table to write to. threads: type: integer description: Number of worker threads used to encode and write disjoint key ranges. default: 1 minimum: 0 write_mode: allOf: - $ref: '#/components/schemas/DynamoDBWriteMode' default: batch ErrorResponse: type: object description: Information returned by REST API endpoints on error. required: - message - error_code - details properties: details: description: 'Detailed error metadata. The contents of this field is determined by `error_code`.' error_code: type: string description: Error code is a string that specifies this error type. example: CodeSpecifyingErrorType message: type: string description: Human-readable error message. example: Explanation of the error that occurred. IcebergReaderConfig: allOf: - $ref: '#/components/schemas/GlueCatalogConfig' - $ref: '#/components/schemas/RestCatalogConfig' - $ref: '#/components/schemas/S3TablesCatalogConfig' - type: object required: - mode properties: catalog_type: allOf: - $ref: '#/components/schemas/IcebergCatalogType' nullable: true datetime: type: string description: 'Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector finds and opens the snapshot of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data). In `snapshot` and `snapshot_and_follow` modes, it retrieves this snapshot. In `follow` and `snapshot_and_follow` modes, it follows transaction log records **after** this snapshot. Note: at most one of `snapshot_id` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.' nullable: true metadata_location: type: string description: 'Location of the table metadata JSON file. This propery is used to access an Iceberg table without a catalog. It is mutually exclusive with the `catalog_type` property.' nullable: true mode: $ref: '#/components/schemas/IcebergIngestMode' snapshot_filter: type: string description: 'Optional row filter. This option is only valid when `mode` is set to `snapshot` or `snapshot_and_follow`. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ...` query. This option can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN ''2005-01-01 00:00:00'' AND ''2010-12-31 23:59:59''`.' nullable: true snapshot_id: type: integer format: int64 description: 'Optional snapshot id. When this option is set, the connector finds the specified snapshot of the table. In `snapshot` and `snapshot_and_follow` modes, it loads this snapshot. In `follow` and `snapshot_and_follow` modes, it follows table updates **after** this snapshot. Note: at most one of `snapshot_id` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.' nullable: true table_name: type: string description: 'Specifies the Iceberg table name in the "namespace.table" format. This option is applicable when an Iceberg catalog is configured using the `catalog_type` property.' nullable: true timestamp_column: type: string description: 'Table column that serves as an event timestamp. When this option is specified, and `mode` is one of `snapshot` or `snapshot_and_follow`, table rows are ingested in the timestamp order, respecting the [`LATENESS`](https://docs.feldera.com/sql/streaming#lateness-expressions) property of the column: each ingested row has a timestamp no more than `LATENESS` time units earlier than the most recent timestamp of any previously ingested row. The ingestion is performed by partitioning the table into timestamp ranges of width `LATENESS`. Each range is processed sequentially, in increasing timestamp order. # Example Consider a table with timestamp column of type `TIMESTAMP` and lateness attribute `INTERVAL 1 DAY`. Assuming that the oldest timestamp in the table is `2024-01-01T00:00:00``, the connector will fetch all records with timestamps from `2024-01-01`, then all records for `2024-01-02`, `2024-01-03`, etc., until all records in the table have been ingested. # Requirements * The timestamp column must be of a supported type: integer, `DATE`, or `TIMESTAMP`. * The timestamp column must be declared with non-zero `LATENESS`. * For efficient ingest, the table must be optimized for timestamp-based queries using partitioning, Z-ordering, or liquid clustering.' nullable: true additionalProperties: type: string description: 'Storage options for configuring backend object store. See the [list of available options in PyIceberg documentation](https://py.iceberg.apache.org/configuration/#fileio).' description: Iceberg input connector configuration. PipelineTemplateConfig: type: object description: 'Configuration for supplying a custom pipeline StatefulSet template via a Kubernetes ConfigMap. Operators can provide a custom StatefulSet YAML that the Kubernetes runner will use when creating pipeline StatefulSets for a pipeline. The custom template must be stored as the value of a key in a ConfigMap in the same namespace as the pipeline; set `name` to the ConfigMap name and `key` to the entry that contains the template. Recommendations and requirements: - **Start from the default template and modify it as needed.** The default template is present in ConfigMap named as `-pipeline-template`, with key `pipelineTemplate` in the release namespace and should be used as a reference. - The template must contain a valid Kubernetes `StatefulSet` manifest in YAML form. The runner substitutes variables in the template before parsing; therefore the final YAML must be syntactically valid. - The runner performs simple string substitution for the following placeholders. Please ensure these placeholders are placed at appropriate location for their semantics: - `{id}`: pipeline Kubernetes name (used for object names and labels) - `{namespace}`: Kubernetes namespace where the pipeline runs - `{pipeline_executor_image}`: container image used to run the pipeline executor - `{binary_ref}`: program binary reference passed as an argument - `{program_info_ref}`: program info reference passed as an argument - `{pipeline_storage_path}`: mount path for persistent pipeline storage - `{storage_class_name}`: storage class name to use for PVCs (if applicable) - `{deployment_id}`: UUID identifying the deployment instance - `{deployment_initial}`: initial desired runtime status (e.g., `provisioning`) - `{bootstrap_policy}`: bootstrap policy value when applicable' required: - name properties: key: type: string description: 'Key in the ConfigMap containing the pipeline template. If not set, defaults to `pipelineTemplate`.' default: pipelineTemplate name: type: string description: Name of the ConfigMap containing the pipeline template. ReplayPolicy: type: string enum: - Instant - Original OutputBufferConfig: type: object properties: enable_output_buffer: type: boolean description: 'Enable output buffering. The output buffering mechanism allows decoupling the rate at which the pipeline pushes changes to the output transport from the rate of input changes. By default, output updates produced by the pipeline are pushed directly to the output transport. Some destinations may prefer to receive updates in fewer bigger batches. For instance, when writing Parquet files, producing one bigger file every few minutes is usually better than creating small files every few milliseconds. To achieve such input/output decoupling, users can enable output buffering by setting the `enable_output_buffer` flag to `true`. When buffering is enabled, output updates produced by the pipeline are consolidated in an internal buffer and are pushed to the output transport when one of several conditions is satisfied: * data has been accumulated in the buffer for more than `max_output_buffer_time_millis` milliseconds. * buffer size exceeds `max_output_buffer_size_records` records. This flag is `false` by default.' default: false max_output_buffer_size_records: type: integer description: 'Maximum number of updates to be kept in the output buffer. This parameter bounds the maximal size of the buffer. Note that the size of the buffer is not always equal to the total number of updates output by the pipeline. Updates to the same record can overwrite or cancel previous updates. The default is 10,000,000. NOTE: this configuration option requires the `enable_output_buffer` flag to be set.' default: 10000000 minimum: 0 max_output_buffer_time_millis: type: integer description: 'Maximum time in milliseconds data is kept in the output buffer. By default, data is kept in the buffer indefinitely until one of the other output conditions is satisfied. When this option is set the buffer will be flushed at most every `max_output_buffer_time_millis` milliseconds. NOTE: this configuration option requires the `enable_output_buffer` flag to be set.' default: 18446744073709551615 minimum: 0 Field: allOf: - $ref: '#/components/schemas/SqlIdentifier' - type: object required: - columntype - unused properties: columntype: $ref: '#/components/schemas/ColumnType' default: type: string nullable: true lateness: type: string nullable: true unused: type: boolean watermark: type: string nullable: true description: 'A SQL field. Matches the SQL compiler JSON format.' DevTweaks: type: object description: 'Optional settings for tweaking Feldera internals. These settings reflect experiments that may come and go and change from version to version. Users should not consider them to be stable.' properties: adaptive_joins: type: boolean description: 'Enable adaptive joins. Adaptive joins dynamically change their partitioning policy to avoid skew. Adaptive joins are disabled by default.' default: null nullable: true balancer_balance_tax: type: number format: double description: 'Factor that discourages the use of the Balance policy in a perfectly balanced collection. Assuming a perfectly balanced key distribution, the Balance policy is slightly less efficient than Shard, since it requires computing the hash of the entire key/value pair. This factor discourages the use of this policy if the skew is `` is accepted (years `0001` through `9999`), in the past or future relative to wall clock. This is a testing knob for queries that depend on `NOW()`. On resume the clock continues from the last journaled `NOW()`; `now_offset`''s value is honored only on a fresh start: | Initial run | Resume from checkpoint | Post-replay `NOW()` | |---|---|---| | no offset | no offset | wall clock (unchanged) | | offset | offset | wall-clock pace from the last journaled value; the new offset value is ignored | | offset | no offset | jumps to wall clock (explicit opt-out of the anchor) | | no offset | offset | wall-clock pace from the last journaled value; the new offset value is ignored |' default: null nullable: true optimize_input_during_commit: type: boolean description: Optimize input operators during transaction commit. default: null nullable: true splitter_chunk_size_records: type: integer format: int64 description: 'Controls the maximal number of records output by splitter operators (joins, distinct, aggregation, rolling window and group operators) at each step. The default value is 10,000 records.' default: null nullable: true minimum: 0 stack_overflow_backtrace: type: boolean description: 'Attempt to print a stack trace on stack overflow. To be used for debugging only; do not enable in production.' default: null nullable: true storage_mb_max: type: integer format: int64 description: 'If set, the maximum amount of storage, in MiB, for the POSIX backend to allow to be in use before failing all writes with `StorageFull`. This is useful for testing on top of storage that does not implement its own quota mechanism.' default: null nullable: true minimum: 0 streaming_exchange: type: boolean description: Enable streaming exchange. default: null nullable: true additionalProperties: description: 'Options not understood by this particular version. This allows the pipeline manager to take options that a custom or old runtime version accepts.' default: {} HttpOutputConfig: type: object description: 'Configuration for data output via HTTP. HTTP output adapters cannot be usefully configured as part of pipeline configuration. Instead, instantiate them through the REST API as `/pipelines/{pipeline_name}/egress`.' properties: backpressure: type: boolean description: 'Apply backpressure on the pipeline when the HTTP client cannot receive data fast enough. When this flag is set to false (the default), the HTTP connector drops data chunks if the client is not keeping up with its output. This prevents a slow HTTP client from slowing down the entire pipeline. When the flag is set to true, the connector waits for the client to receive each chunk and blocks the pipeline if the client cannot keep up.' default: false RngFieldSettings: type: object description: Configuration for generating random data for a field of a table. properties: e: type: integer format: int64 description: 'The frequency rank exponent for the Zipf distribution. - This value is only used if the strategy is set to `Zipf`. - The default value is 1.0.' default: 1 fields: type: object description: Specifies the values that the generator should produce in case the field is a struct type. default: null additionalProperties: $ref: '#/components/schemas/RngFieldSettings' nullable: true key: allOf: - $ref: '#/components/schemas/RngFieldSettings' default: null nullable: true null_percentage: type: integer description: 'Percentage of records where this field should be set to NULL. If not set, the generator will produce only records with non-NULL values. If set to `1..=100`, the generator will produce records with NULL values with the specified percentage.' default: null nullable: true minimum: 0 range: type: object description: 'An optional, exclusive range [a, b) to limit the range of values the generator should produce. - For integer/floating point types specifies min/max values as an integer. If not set, the generator will produce values for the entire range of the type for number types. - For string/binary types specifies min/max length as an integer, values are required to be >=0. If not set, a range of [0, 25) is used by default. - For timestamp types specifies the min/max as two strings in the RFC 3339 format (e.g., ["2021-01-01T00:00:00Z", "2022-01-02T00:00:00Z"]). Alternatively, the range values can be specified as a number of non-leap milliseconds since January 1, 1970 0:00:00.000 UTC (aka “UNIX timestamp”). If not set, a range of ["1970-01-01T00:00:00Z", "2100-01-01T00:00:00Z") or [0, 4102444800000) is used by default. - For time types specifies the min/max as two strings in the "HH:MM:SS" format. Alternatively, the range values can be specified in milliseconds as two positive integers. If not set, the range is 24h. - For date types, the min/max range is specified as two strings in the "YYYY-MM-DD" format. Alternatively, two integers that represent number of days since January 1, 1970 can be used. If not set, a range of ["1970-01-01", "2100-01-01") or [0, 54787) is used by default. - For array types specifies the min/max number of elements as an integer. If not set, a range of [0, 5) is used by default. Range values are required to be >=0. - For map types specifies the min/max number of key-value pairs as an integer. If not set, a range of [0, 5) is used by default. - For struct/boolean/null types `range` is ignored.' scale: type: integer format: int64 description: 'A scale factor to apply a multiplier to the generated value. - For integer/floating point types, the value is multiplied by the scale factor. - For timestamp types, the generated value (milliseconds) is multiplied by the scale factor. - For time types, the generated value (milliseconds) is multiplied by the scale factor. - For date types, the generated value (days) is multiplied by the scale factor. - For string/binary/array/map/struct/boolean/null types, the scale factor is ignored. - If `values` is specified, the scale factor is ignored. - If `range` is specified and the range is required to be positive (struct, map, array etc.) the scale factor is required to be positive too. The default scale factor is 1.' default: 1 strategy: allOf: - $ref: '#/components/schemas/DatagenStrategy' default: increment value: allOf: - $ref: '#/components/schemas/RngFieldSettings' default: null nullable: true values: type: array items: type: object description: 'An optional set of values the generator will pick from. If set, the generator will pick values from the specified set. If not set, the generator will produce values according to the specified range. If set to an empty set, the generator will produce NULL values. If set to a single value, the generator will produce only that value. Note that `range` is ignored if `values` is set.' default: null nullable: true additionalProperties: false CompilationProfile: type: string description: 'Enumeration of possible compilation profiles that can be passed to the Rust compiler as an argument via `cargo build --profile <>`. A compilation profile affects among other things the compilation speed (how long till the program is ready to be run) and runtime speed (the performance while running).' enum: - dev - unoptimized - optimized - optimized_symbols ConnectorStats: type: object description: Statistics across all connectors. required: - num_errors properties: num_errors: type: integer format: int64 description: 'Total number of errors across all connectors. - `num_transport_errors` from all input connectors - `num_parse_errors` from all input connectors - `num_encode_errors` from all output connectors - `num_transport_errors` from all output connectors' minimum: 0 RustCompilationInfo: type: object description: Rust compilation information. required: - exit_code - stdout - stderr properties: exit_code: type: integer format: int32 description: Exit code of the `cargo` compilation command. stderr: type: string description: Output printed to stderr by the `cargo` compilation command. stdout: type: string description: Output printed to stdout by the `cargo` compilation command. DeltaTableTransactionMode: type: string description: 'Delta table transaction mode. Determines how the connector breaks up its input into transactions. * `none` - the connector does not break up its input into transactions. * `snapshot` - ingest the initial snapshot of the table in one or several transactions. If the connector is configured in the `snapshot_and_follow` mode, it will ingest the transaction log of the table without transactions. * `catchup` - batch the initial snapshot and transaction-log catch-up into Feldera transactions (see below). * `always` - all updates generated by the connector are broken up into transactions, both during the initial snapshot and when following the transaction log. # How table snapshot is ingested using transactions If the connector is configured in the `snapshot` or `snapshot_and_follow` mode, and its `transaction_mode` is set to `snapshot`, `catchup`, or `always`, it will ingest the snapshot of the table in one or several transactions. The exact behavior depends on the value of the `timestamp_column` option. If `timestamp_column` is not set, `catchup` and `always` ingest the snapshot in one Feldera transaction; `snapshot` mode uses the same one-transaction behavior when `timestamp_column` is not set. If `timestamp_column` is set, the connector will ingest the snapshot of the table in a series of batches, one for each timestamp range of width equal to the `LATENESS` attribute of the `timestamp_column`. Each range will be ingested in a separate transaction. See `timestamp_column` documentation for more details. # How transaction log is ingested using transactions If the connector is configured in the `follow`, `snapshot_and_follow`, or `cdc` mode, and its `transaction_mode` is set to `catchup`, it ingests the transaction log in batches. When it starts a Feldera transaction, it reads the latest available version of the Delta table (capped by `end_version` if set) and ingests all log entries up to and including that version in a single Feldera transaction before committing. It then repeats for subsequent versions as they appear in the log. The `catchup` mode offers the most efficient way for the connector to process the entire contents of the table during backfill and keep up with table changes in steady state. If `transaction_mode` is set to `always`, the connector ingests the transaction log in a series of transactions, generating exactly one Feldera transaction for each entry in the table''s transaction log. Feldera transaction boundaries then match the transaction boundaries of the Delta Lake table: as long as the table keeps changing, each log entry initiates a new Feldera transaction. Use `always` when that one-to-one alignment is required.' enum: - none - snapshot - always - catchup OutputEndpointConfig: allOf: - $ref: '#/components/schemas/ConnectorConfig' - type: object required: - stream properties: stream: type: string description: 'The name of the output stream of the circuit that this endpoint is connected to.' description: Describes an output connector configuration PubSubInputConfig: type: object description: Google Pub/Sub input connector configuration. required: - subscription properties: connect_timeout_seconds: type: integer format: int32 description: gRPC connection timeout. nullable: true minimum: 0 credentials: type: string description: 'The content of a Google Cloud credentials JSON file. When this option is specified, the connector will use the provided credentials for authentication. Otherwise, it will use Application Default Credentials (ADC) configured in the environment where the Feldera service is running. See [Google Cloud documentation](https://cloud.google.com/docs/authentication/provide-credentials-adc) for information on configuring application default credentials. When running Feldera in an environment where ADC are not configured, e.g., a Docker container, use this option to ship Google Cloud credentials from another environment. For example, if you use the [`gcloud auth application-default login`](https://cloud.google.com/pubsub/docs/authentication#client-libs) command for authentication in your local development environment, ADC are stored in the `.config/gcloud/application_default_credentials.json` file in your home directory.' nullable: true emulator: type: string description: 'Set in order to use a Pub/Sub [emulator](https://cloud.google.com/pubsub/docs/emulator) instead of the production service, e.g., ''localhost:8681''.' nullable: true endpoint: type: string description: Override the default service endpoint 'pubsub.googleapis.com' nullable: true pool_size: type: integer format: int32 description: gRPC channel pool size. nullable: true minimum: 0 project_id: type: string description: 'Google Cloud project_id. When not specified, the connector will use the project id associated with the authenticated account.' nullable: true snapshot: type: string description: 'Reset subscription''s backlog to a given snapshot on startup, using the Pub/Sub `Seek` API. This option is mutually exclusive with the `timestamp` option.' nullable: true subscription: type: string description: Subscription name. timeout_seconds: type: integer format: int32 description: gRPC request timeout. nullable: true minimum: 0 timestamp: type: string description: 'Reset subscription''s backlog to a given timestamp on startup, using the Pub/Sub `Seek` API. The value of this option is an ISO 8601-encoded UTC time, e.g., "2024-08-17T16:39:57-08:00". This option is mutually exclusive with the `snapshot` option.' nullable: true ClockConfig: type: object required: - clock_resolution_usecs properties: clock_resolution_usecs: type: integer format: int64 minimum: 0 http_driven: type: boolean description: 'If `true`, the clock does not advance on wall-clock cadence. `NOW()` is held at its current value and only advances when an external caller invokes the pipeline''s `POST /clock/advance` endpoint. Populated from `DevTweaks::now_http_driven`.' now_offset_ms: type: integer format: int64 description: 'Target value for `NOW()` at the worker''s first emitted tick, in milliseconds since the Unix epoch. Populated verbatim from `DevTweaks::now_offset` at endpoint construction; the wall-clock delta is computed inside the connector''s worker task from a single `SystemTime::now()` reading, so there is no drift between config construction and the first emitted tick. `None` means no shift is applied.' nullable: true HeaderFilter: oneOf: - type: object required: - header properties: header: $ref: '#/components/schemas/HeaderMatch' - type: object required: - and properties: and: type: array items: $ref: '#/components/schemas/HeaderFilter' description: 'Conjunction: matches when every nested filter matches. Must have at least one operand.' - type: object required: - or properties: or: type: array items: $ref: '#/components/schemas/HeaderFilter' description: 'Disjunction: matches when at least one nested filter matches. Must have at least one operand.' - type: object required: - not properties: not: $ref: '#/components/schemas/HeaderFilter' description: 'A boolean filter over the headers of a Kafka message. The Kafka input connector uses this to drop messages whose headers do not satisfy a predicate. It is a tree of boolean operators (`and`, `or`, `not`) whose leaves are regular expression tests on individual header values. It serializes as an externally tagged JSON object, for example: ```json { "and": [ { "header": { "name": "event-type", "pattern": "created|updated" } }, { "not": { "header": { "name": "source", "pattern": "test-.*" } } } ] } ``` This admits a message only if it has an `event-type` header valued exactly `created` or `updated` and does not have a `source` header whose value starts with `test-`.' PostgresCdcReaderConfig: allOf: - type: object description: TLS/SSL configuration for PostgreSQL connectors. properties: ssl_ca_location: type: string description: Path to a file containing a sequence of CA certificates in PEM format. nullable: true ssl_ca_pem: type: string description: A sequence of CA certificates in PEM format. nullable: true ssl_certificate_chain_location: type: string description: 'The path to the certificate chain file. The file must contain a sequence of PEM-formatted certificates, the first being the leaf certificate, and the remainder forming the chain of certificates up to and including the trusted root certificate.' nullable: true ssl_client_key: type: string description: The client certificate key in PEM format. nullable: true ssl_client_key_location: type: string description: Path to the client certificate key. nullable: true ssl_client_location: type: string description: Path to the client certificate. nullable: true ssl_client_pem: type: string description: The client certificate in PEM format. nullable: true verify_hostname: type: boolean description: True to enable hostname verification when using TLS. True by default. nullable: true - type: object required: - uri - publication - source_table properties: publication: type: string description: Name of the pre-created Postgres publication to replicate from. source_table: type: string description: 'Postgres table to replicate (e.g. "public.orders"). Must be included in the publication.' uri: type: string description: 'Postgres connection URI. The user must have REPLICATION privilege. See: ' description: 'Postgres CDC input connector configuration. Uses logical replication to capture ongoing changes from a Postgres database. Requires a pre-created publication and a user with REPLICATION privilege. Tables must have primary keys and `REPLICA IDENTITY FULL` is recommended for UPDATE/DELETE support.' NexmarkInputConfig: type: object description: 'Configuration for generating Nexmark input data. This connector must be used exactly three times in a pipeline if it is used at all, once for each [`NexmarkTable`].' required: - table properties: options: allOf: - $ref: '#/components/schemas/NexmarkInputOptions' nullable: true table: $ref: '#/components/schemas/NexmarkTable' PreprocessorConfig: type: object description: Configuration for describing a preprocessor required: - name - message_oriented - config properties: config: description: Arbitrary additional configuration. message_oriented: type: boolean description: 'True if the preprocessor is message-oriented: true if each preprocessor output record corresponds to a whole number of of parser records.' name: type: string description: 'Name of the preprocessor. All preprocessors with the same name will perform the same task.' ConnectorConfig: allOf: - $ref: '#/components/schemas/OutputBufferConfig' - type: object required: - transport properties: format: allOf: - $ref: '#/components/schemas/FormatConfig' nullable: true index: type: string description: 'Name of the index that the connector is attached to. This property is valid for output connectors only. It is used with data transports and formats that expect output updates in the form of key/value pairs, where the key typically represents a unique id associated with the table or view. To support such output formats, an output connector can be attached to an index created using the SQL CREATE INDEX statement. An index of a table or view contains the same updates as the table or view itself, indexed by one or more key columns. See individual connector documentation for details on how they work with indexes.' nullable: true labels: type: array items: type: string description: 'Arbitrary user-defined text labels associated with the connector. These labels can be used in conjunction with the `start_after` property to control the start order of connectors.' max_batch_size: type: integer format: int64 description: 'Maximum number of records from this connector to process in a single batch. When set, this caps how many records are taken from the connector’s input buffer and pushed through the circuit at once. This is typically configured lower than `max_queued_records` to allow the connector time to restart and refill its buffer while a batch is being processed. Not all input adapters honor this limit. If this is not set, the batch size is derived from `max_worker_batch_size`.' nullable: true minimum: 0 max_queued_bytes: type: integer format: int64 description: 'Backpressure threshold, in bytes. Maximal number of bytes queued by the endpoint before the endpoint is paused by the backpressure mechanism. For input endpoints, this setting bounds the number of bytes that have been received from the input transport but haven''t yet been consumed by the circuit since the circuit, since the circuit is still busy processing previous inputs. This setting is not yet implemented for output endpoints. Note that this is not a hard bound: there can be a small delay between the backpressure mechanism is triggered and the endpoint is paused, during which more data may be queued. When this is unspecified, it defaults to `1000 * max_queued_records`.' nullable: true minimum: 0 max_queued_records: type: integer format: int64 description: 'Backpressure threshold, in records. Maximal number of records queued by the endpoint before the endpoint is paused by the backpressure mechanism. For input endpoints, this setting bounds the number of records that have been received from the input transport but haven''t yet been consumed by the circuit, since the circuit is still busy processing previous inputs. For output endpoints, this setting bounds the number of records that have been produced by the circuit but not yet sent via the output transport endpoint nor stored in the output buffer (see `enable_output_buffer`). Note that this is not a hard bound: there can be a small delay between the backpressure mechanism is triggered and the endpoint is paused, during which more data may be queued. The default is 1 million.' minimum: 0 max_worker_batch_size: type: integer format: int64 description: 'Maximum number of records processed per batch, per worker thread. When `max_batch_size` is not set, this setting is used to cap the number of records that can be taken from the connector’s input buffer and pushed through the circuit at once. The effective batch size is computed as: `max_worker_batch_size × workers`. This provides an alternative to `max_batch_size` that automatically adjusts batch size as the number of worker threads changes to maintain constant amount of work per worker per batch. Defaults to 10,000 records per worker.' nullable: true minimum: 0 paused: type: boolean description: 'Create connector in paused state. The default is `false`.' postprocessor: type: array items: $ref: '#/components/schemas/PostprocessorConfig' description: Optional postprocessor configuration nullable: true preprocessor: type: array items: $ref: '#/components/schemas/PreprocessorConfig' description: Optional preprocessor configuration nullable: true send_snapshot: type: boolean description: 'Send a full snapshot of a materialized view when the connector first starts. Valid for output connectors only. When `true`, the pipeline emits the current contents of the view as the initial batch the first time the connector runs. The view must be materialized (declared with `CREATE MATERIALIZED VIEW`). The snapshot is sent exactly once per connector lifetime: it does not fire again when the pipeline resumes from a checkpoint. Modifying the connector configuration or invoking the reset API triggers a fresh snapshot when the connector supports reset (e.g., Delta Lake in `truncate` mode and Postgres).' start_after: type: array items: type: string description: 'Start the connector after all connectors with specified labels. This property is used to control the start order of connectors. The connector will not start until all connectors with the specified labels have finished processing all inputs.' nullable: true transport: $ref: '#/components/schemas/TransportConfig' S3TablesCatalogConfig: type: object description: Amazon S3 Tables catalog config. properties: s3tables.access-key-id: type: string description: Access key id used to access the S3 Tables catalog. nullable: true s3tables.endpoint: type: string description: 'Custom endpoint URL for the S3 Tables service. Primarily used to target a local or mock S3 Tables implementation for testing. When omitted, the default regional endpoint is used.' nullable: true s3tables.profile-name: type: string description: Profile used to access the S3 Tables catalog. nullable: true s3tables.region: type: string description: Region of the S3 Tables catalog. nullable: true s3tables.secret-access-key: type: string description: Secret access key used to access the S3 Tables catalog. nullable: true s3tables.session-token: type: string description: Static session token used to access the S3 Tables catalog. nullable: true s3tables.table-bucket-arn: type: string description: 'ARN of the S3 table bucket that contains the table. Note that this is the ARN of the table *bucket*, not of an individual table, e.g., `"arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket"`.' nullable: true ClientMetadata: type: object description: 'Client-generated data stored alongside a pipeline. The fields are stored together as a single JSON object in the `client_metadata` text column, so adding a field needs no migration. Deserialization is lenient: missing keys take their default and unknown keys are ignored. [`PatchClientMetadata`] is the optional, per-field form used by `PATCH` request bodies.' properties: description: type: string description: Human-readable description of the pipeline. Default is an empty string. tags: type: array items: type: string description: 'Free-form labels used to organize, group, and filter pipelines. Default is no tags (empty vector).' ResourcesStatus: type: string description: 'Pipeline resources status. ```text /start (early start or provision check failed) ┌───┐ │ ▼ Stopped ◄────────── Stopping /start │ ▲ │ │ /stop?force=true │ │ OR: timeout (from Provisioning) ▼ │ OR: fatal runtime or resource error ⌛Provisioning ────────────│ OR: runtime status is Suspended │ │ │ │ ▼ │ Provisioned ─────────────┘ ``` ### Desired and actual status We use the desired state model to manage the lifecycle of a pipeline. In this model, the pipeline has two status attributes associated with it: the **desired** status, which represents what the user would like the pipeline to do, and the **current** status, which represents the actual (last observed) status of the pipeline. The pipeline runner service continuously monitors the desired status field to decide where to steer the pipeline towards. There are two desired statuses: - `Provisioned` (set by invoking `/start`) - `Stopped` (set by invoking `/stop?force=true`) The user can monitor the current status of the pipeline via the `GET /v0/pipelines/{name}` endpoint. In a typical scenario, the user first sets the desired status, e.g., by invoking the `/start` endpoint, and then polls the `GET /v0/pipelines/{name}` endpoint to monitor the actual status of the pipeline until its `deployment_resources_status` attribute changes to `Provisioned` indicating that the pipeline has been successfully provisioned, or `Stopped` with `deployment_error` being set.' enum: - Stopped - Provisioning - Provisioned - Stopping FtModel: type: string description: 'Fault tolerance model. The ordering is significant: we consider [Self::ExactlyOnce] to be a "higher level" of fault tolerance than [Self::AtLeastOnce].' enum: - at_least_once - exactly_once SourcePosition: type: object required: - start_line_number - start_column - end_line_number - end_column properties: end_column: type: integer minimum: 0 end_line_number: type: integer minimum: 0 start_column: type: integer minimum: 0 start_line_number: type: integer minimum: 0 InputEndpointConfig: allOf: - $ref: '#/components/schemas/ConnectorConfig' - type: object required: - stream properties: stream: type: string description: 'The name of the input stream of the circuit that this endpoint is connected to.' description: Describes an input connector configuration KafkaInputConfig: type: object description: Configuration for reading data from Kafka topics with `InputTransport`. required: - topic - resume_earliest_if_data_expires properties: group_join_timeout_secs: type: integer format: int32 description: 'Maximum timeout in seconds to wait for the endpoint to join the Kafka consumer group during initialization.' minimum: 0 header_filter: allOf: - $ref: '#/components/schemas/HeaderFilter' nullable: true include_headers: type: boolean description: 'Whether to include Kafka headers in the record metadata. When `true`, Kafka message headers are available via the `CONNECTOR_METADATA()` function. See for details.' nullable: true include_offset: type: boolean description: 'Whether to include Kafka message offset in the record metadata. When `true`, Kafka message offset is available via the `CONNECTOR_METADATA()` function. See for details.' nullable: true include_partition: type: boolean description: 'Whether to include Kafka partition in the record metadata. When `true`, Kafka partition from which the message was read is available via the `CONNECTOR_METADATA()` function. See for details.' nullable: true include_timestamp: type: boolean description: 'Whether to include Kafka message timestamp in the record metadata. When `true`, Kafka message timestamp is available via the `CONNECTOR_METADATA()` function. See for details.' nullable: true include_topic: type: boolean description: 'Whether to include Kafka topic in the record metadata. When `true`, Kafka topic from which the message was read is available via the `CONNECTOR_METADATA()` function. See for details.' nullable: true log_level: allOf: - $ref: '#/components/schemas/KafkaLogLevel' nullable: true partitions: type: array items: type: integer format: int32 description: 'The list of Kafka partitions to read from. Only the specified partitions will be consumed. If this field is not set, the connector will consume from all available partitions. If `start_from` is set to `offsets` and this field is provided, the number of partitions must exactly match the number of offsets, and the order of partitions must correspond to the order of offsets. If offsets are provided for all partitions, this field can be omitted.' nullable: true poller_threads: type: integer description: 'Set to 1 or more to fix the number of threads used to poll `rdkafka`. Multiple threads can increase performance with small Kafka messages; for large messages, one thread is enough. In either case, too many threads can harm performance. If unset, the default is 3, which helps with small messages but will not harm performance with large messagee' nullable: true minimum: 0 region: type: string description: The AWS region to use while connecting to AWS Managed Streaming for Kafka (MSK). nullable: true resume_earliest_if_data_expires: type: boolean description: 'By default, if the input connector resumes from a checkpoint and the data where it needs to resume has expired from the Kafka topic, the input connector fails initialization and the pipeline will fail to start. Set this to true to change the behavior so that, if data is not available on resume, the input connector starts from the earliest offsets that are now available.' start_from: $ref: '#/components/schemas/KafkaStartFromConfig' synchronize_partitions: type: boolean description: 'When lateness is enabled on a Feldera table, Feldera only produces correct output if input arrives approximately in order within the bounds of the lateness. The Feldera Kafka input connector can reorder input when there are multiple partitions: - If partitions start at different times, then reading all the partitions in parallel will naturally consume data out of order. - Even if they start at the same time, partitions might contain events at different rates. - Even if the partitions start at the same time and have the same number of events per unit time, if partitions are spread across brokers, different brokers may fetch data at different rates. - Even if all of the partitions are on a single broker, one cannot expect all of the partitions to naturally remain exactly in sync forever. Setting this option to `true` addresses the issue by synchronizing ingestion across partitions, ingesting records in order of their Kafka event timestamps. Pitfalls of this solution include: - Kafka event timestamps are not necessarily monotonically increasing even within a single partition. If timestamps jump backward beyond the lateness, then this can also cause correctness problems. (This can be avoided by keeping clocks on Kafka producers and brokers synchronized.) - If an event with a timestamp far in the future is added to a partition, that event, and all those that follow it, will never be processed. - If one or a few partitions have timestamps far behind the others, only those partitions will be processed until all the old events are processed. (This is the flip side of the previous pitfall.) - One or more empty partitions will prevent any data from being processed at all, because there is no way to know the timestamp for the first event that will be added to that partition. - In a topic with `N` nonempty partitions, at least `N - 1` events will always be left unprocessed (one in each of `N - 1` partitions), because there is no way to know the timestamp for the next event to be added to the partition whose events have been completely processed.' topic: type: string description: Topic to subscribe to. additionalProperties: type: string description: 'Options passed directly to `rdkafka`. [`librdkafka` options](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md) used to configure the Kafka consumer. This input connector does not use consumer groups, so options related to consumer groups are rejected, including: * `group.id`, if present, is ignored. * `auto.offset.reset` (use `start_from` instead). * "enable.auto.commit", if present, must be set to "false". * "enable.auto.offset.store", if present, must be set to "false".' KafkaLogLevel: type: string description: Kafka logging levels. enum: - emerg - alert - critical - error - warning - notice - info - debug HeaderMatch: type: object description: 'A leaf of a [`HeaderFilter`]: a regular expression tested against the value of a named Kafka header.' required: - name - pattern properties: name: type: string description: Name of the header to test, matched exactly against the header key. pattern: type: string description: 'Regular expression ([Rust `regex` crate syntax](https://docs.rs/regex/latest/regex/#syntax)) tested against the header value. The value is matched as raw bytes, so non-UTF-8 values and byte patterns work. The pattern must match the *entire* value: it is anchored automatically, so `^`/`$` are unnecessary (though harmless). A header present with a null value is matched as an empty byte sequence; a header that appears more than once matches if any of its values match.' KafkaStartFromConfig: oneOf: - type: string description: Start from the beginning of the topic. enum: - earliest - type: string description: 'Start from the current end of the topic. This will only read any data that is added to the topic after the connector initializes.' enum: - latest - type: object required: - offsets properties: offsets: type: array items: type: integer format: int64 description: 'Start from particular offsets in the topic. The number of offsets must match the number of partitions in the topic.' - type: object required: - timestamp properties: timestamp: type: integer format: int64 description: 'Start from a particular timestamp in the topic. Kafka timestamps are in milliseconds since the epoch.' description: Where to begin reading a Kafka topic. BufferCacheStrategy: type: string description: Selects which eviction strategy backs a cache instance. enum: - s3_fifo - lru ResourcesDesiredStatus: type: string enum: - Stopped - Provisioned SqlIdentifier: type: object description: 'An SQL identifier. This struct is used to represent SQL identifiers in a canonical form. We store table names or field names as identifiers in the schema.' required: - name - case_sensitive properties: case_sensitive: type: boolean name: type: string PostPutPipeline: allOf: - $ref: '#/components/schemas/ClientMetadata' - type: object required: - name - program_code properties: name: type: string program_code: type: string program_config: allOf: - $ref: '#/components/schemas/ProgramConfig' nullable: true runtime_config: allOf: - $ref: '#/components/schemas/RuntimeConfig' nullable: true udf_rust: type: string nullable: true udf_toml: type: string nullable: true description: 'Create a new pipeline (POST), or fully update an existing pipeline (PUT). Fields which are optional and not provided will be set to their empty type value (for strings: an empty string `""`, for objects: an empty dictionary `{}`).' GenerationPlan: type: object description: A random generation plan for a table that generates either a limited amount of rows or runs continuously. properties: fields: type: object description: Specifies the values that the generator should produce. default: {} additionalProperties: $ref: '#/components/schemas/RngFieldSettings' limit: type: integer description: 'Total number of new rows to generate. If not set, the generator will produce new/unique records as long as the pipeline is running. If set to 0, the table will always remain empty. If set, the generator will produce new records until the specified limit is reached. Note that if the table has one or more primary keys that don''t use the `increment` strategy to generate the key there is a potential that an update is generated instead of an insert. In this case it''s possible the total number of records is less than the specified limit.' default: null nullable: true minimum: 0 rate: type: integer format: int32 description: 'Non-zero number of rows to generate per second. If not set, the generator will produce rows as fast as possible.' default: null nullable: true minimum: 0 worker_chunk_size: type: integer description: 'When multiple workers are used, each worker will pick a consecutive "chunk" of records to generate. By default, if not specified, the generator will use the formula `min(rate, 10_000)` to determine it. This works well in most situations. However, if you''re running tests with lateness and many workers you can e.g., reduce the chunk size to make sure a smaller range of records is being ingested in parallel. This also controls the sizes of input batches. If, for example, `rate` and `worker_chunk_size` are both 1000, with a single worker, the generator will output 1000 records once a second. But if we reduce `worker_chunk_size` to 100 without changing `rate`, the generator will instead output 100 records 10 times per second. # Example Assume you generate a total of 125 records with 4 workers and a chunk size of 25. In this case, worker A will generate records 0..25, worker B will generate records 25..50, etc. A, B, C, and D will generate records in parallel. The first worker to finish its chunk will pick up the last chunk of records (100..125) to generate.' default: null nullable: true minimum: 0 additionalProperties: false Relation: allOf: - $ref: '#/components/schemas/SqlIdentifier' - type: object required: - fields properties: fields: type: array items: $ref: '#/components/schemas/Field' materialized: type: boolean primary_key: type: array items: type: string nullable: true properties: type: object additionalProperties: $ref: '#/components/schemas/PropertyValue' description: 'A SQL table or view. It has a name and a list of fields. Matches the Calcite JSON format.' NexmarkInputOptions: type: object description: Configuration for generating Nexmark input data. properties: batch_size_per_thread: type: integer format: int64 description: 'Number of events to generate and submit together, per thread. Each thread generates this many records, which are then combined with the records generated by the other threads, to form combined input batches of size `threads × batch_size_per_thread`.' default: 1000 minimum: 0 events: type: integer format: int64 description: Number of events to generate. default: 100000000 minimum: 0 max_step_size_per_thread: type: integer format: int64 description: 'Maximum number of events to submit in a single step, per thread. This should really be per worker thread, not per generator thread, but the connector does not know how many worker threads there are. This stands in for `max_batch_size` from the connector configuration because it must be a constant across all three of the nexmark tables.' default: 10000 minimum: 0 threads: type: integer description: 'Number of event generator threads. It''s reasonable to choose the same number of generator threads as worker threads.' default: 4 minimum: 0 RuntimeStatus: type: string description: 'Runtime status of the pipeline. Of the statuses, only `Unavailable` is determined by the runner. All other statuses are determined by the pipeline and taken over by the runner.' enum: - Unavailable - Coordination - Standby - Initializing - AwaitingApproval - Bootstrapping - Replaying - Paused - Running - Suspended - ConcurrentBootstrapping - Synchronizing RuntimeDesiredStatus: type: string enum: - Unavailable - Coordination - Standby - Paused - Running - Suspended DeltaTableIngestMode: type: string description: 'Delta table read mode. Three options are available: * `snapshot` - read a snapshot of the table and stop. * `follow` - continuously ingest changes to the table, starting from a specified version or timestamp. * `snapshot_and_follow` - read a snapshot of the table before switching to continuous ingestion mode.' enum: - snapshot - follow - snapshot_and_follow - cdc NexmarkTable: type: string description: Table in Nexmark. enum: - bid - auction - person FileOutputConfig: type: object description: Configuration for writing data to a file with `FileOutputTransport`. required: - path properties: path: type: string description: File path. KafkaOutputFtConfig: type: object description: Fault tolerance configuration for Kafka output connector. properties: consumer_options: type: object description: 'Options passed to `rdkafka` for consumers only, as documented at [`librdkafka` options](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). These options override `kafka_options` for consumers, and may be empty.' default: {} additionalProperties: type: string producer_options: type: object description: 'Options passed to `rdkafka` for producers only, as documented at [`librdkafka` options](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). These options override `kafka_options` for producers, and may be empty.' default: {} additionalProperties: type: string securitySchemes: JSON web token (JWT) or API key: type: http scheme: bearer bearerFormat: JWT description: "Use a JWT token obtained via an OAuth2/OIDC\n login workflow or an API key obtained via\n the `/v0/api-keys` endpoint."