feat(horizondb): support Azure HorizonDB — SQL queries and pg_durable durable workflow management

[fdelbrayelle]

Summary

Azure HorizonDB is Microsoft's fully managed, cloud-native PostgreSQL service engineered for AI-ready, mission-critical workloads at scale, featuring a disaggregated compute-and-storage architecture and a built-in durable execution engine (pg_durable). This plugin adds Kestra tasks and triggers to run SQL queries against HorizonDB — including the pg_durable SQL DSL — so teams can submit, monitor, signal, and react to long-running durable workflows entirely from within a Kestra flow, without external orchestrators or custom Python glue scripts.

Motivation

Teams using Azure HorizonDB currently run durable workflows (ETL jobs, approval flows, scheduled maintenance) by cobbling together Azure Functions, cron jobs, or external orchestrators that must reach back into the database. With this plugin, every step — submitting a durable function, waiting for completion, sending an external signal, reacting to a state change — lives in the same Kestra flow as upstream data preparation, notifications, and downstream integrations. This removes the operational burden of a separate orchestration service for workloads whose work is primarily database-side.

Context

Azure HorizonDB is in public preview (announced at Build 2026) and is available in select Azure regions (Central US, West US 2, West US 3, Sweden Central, Australia East). The plugin should be added as a new sub-package io.kestra.plugin.azure.horizondb within plugin-azure, following the same structural conventions as io.kestra.plugin.azure.storage.cosmosdb and io.kestra.plugin.azure.monitoring. The existing plugin-jdbc (PostgreSQL) plugin can serve as a reference for JDBC-based query tasks; the durable.* tasks have no prior counterpart and should be implemented fresh.

Official documentation:

• Overview: https://learn.microsoft.com/en-us/azure/horizondb/overview
• Durable Functions (pg_durable): https://learn.microsoft.com/en-us/azure/horizondb/development/durable-functions

API Reference

• Official docs: https://learn.microsoft.com/en-us/azure/horizondb/
• Authentication: PostgreSQL password auth or Azure Entra ID (via com.azure.identity.extensions.jdbc.postgresql.AzurePostgresqlAuthenticationPlugin passed as authenticationPluginClassName in the JDBC URL)
• Base URL pattern: jdbc:postgresql://<host>:5432/<database>
• SDK / client library: Standard PostgreSQL JDBC driver — no proprietary HorizonDB SDK. Entra ID integration uses the Azure Identity Extensions library.

Gradle Dependencies

Add to build.gradle:

// PostgreSQL JDBC driver
implementation "org.postgresql:postgresql:42.7.5"

// Azure Identity JDBC extensions (Entra ID / Managed Identity auth for PostgreSQL)
implementation "com.azure:azure-identity-extensions:1.2.0"

Use the latest stable versions available on Maven Central.

Kestra framework inclusions (do not add as dependencies):
Kestra's internal HTTP client (io.kestra.core.http.client) and Jackson serializers are provided by the framework. Only list an explicit HTTP dependency when an official SDK is required — never list OkHttp, Apache HttpComponents, or java.net.http wrappers.

Plugin Structure

• Repository: plugin-azure
• Namespace: io.kestra.plugin.azure.horizondb
• Sub-plugins: horizondb (SQL tasks), horizondb.durable (pg_durable workflow management)
• Categories: CLOUD
  (Each sub-package must have its own package-info.java annotated with @PluginSubGroup(category = PluginSubGroup.PluginCategory.CLOUD).)

Task class naming: task class names must not repeat the plugin or package name as a prefix or suffix. Use concise action or resource names: Query, Queries, Start, Cancel, Signal, GetStatus, ListInstances, Trigger — not HorizonDbQuery, DurableStart, etc.

Suggested Tasks

1. AbstractHorizonDb — abstract base class with shared connection properties (host, port, database, username, password, useEntraId); opens and closes a JDBC Connection; shared by all SQL and durable tasks.
2. Query — execute a single SQL statement; support fetchType (FETCH_ONE, FETCH, STORE) for result handling; emit row count and (optionally) an internal storage URI.
3. Queries — execute multiple SQL statements in a single connection; return a list of per-statement outputs.
4. durable.Start — submit a pg_durable function body via SELECT df.start(...) and return the instance ID.
5. durable.Cancel — cancel a running instance by ID via SELECT df.cancel(...).
6. durable.Signal — send an external signal to a waiting instance via SELECT df.signal(...).
7. durable.GetStatus — return the current status and result of a durable instance via df.status() / df.result().
8. durable.ListInstances — query df.list_instances() with optional status filter; support FETCH and STORE fetch types.
9. durable.Trigger — polling trigger that queries df.list_instances() on an interval and fires an execution when one or more instances reach a target state (e.g. Completed, Failed).
10. Write unit tests (using Testcontainers with the postgres image plus the pg_durable extension, if a public Docker image is available; otherwise use Wiremock to simulate JDBC responses).
11. Add package-info.java per sub-package with @PluginSubGroup(category = PluginSubGroup.PluginCategory.CLOUD).
12. Add metadata/horizondb.yaml and an io.kestra.plugin.azure.horizondb.svg icon.
13. Add YAML examples and @Plugin(examples = ...) entries; update AGENTS.md.

YAML Examples

Example 1 — Execute a SQL query against HorizonDB

id: horizondb_query
namespace: company.team

inputs:
  - id: host
    type: STRING

tasks:
  - id: query
    type: io.kestra.plugin.azure.horizondb.Query
    host: "{{ inputs.host }}"
    port: 5432
    database: mydb
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    sql: |
      SELECT id, total, status FROM orders WHERE loaded_at > now() - INTERVAL '1 day'
    fetchType: STORE

Example 2 — Submit a pg_durable ETL workflow and wait for it to complete

id: horizondb_durable_etl
namespace: company.team

inputs:
  - id: host
    type: STRING

tasks:
  - id: start_etl
    type: io.kestra.plugin.azure.horizondb.durable.Start
    host: "{{ inputs.host }}"
    port: 5432
    database: mydb
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    functionBody: |
      'DELETE FROM target WHERE loaded_at < now() - INTERVAL ''1 day'''
      ~> 'INSERT INTO target SELECT * FROM staging'
      ~> 'REINDEX TABLE target'
      ~> 'INSERT INTO etl_log (job, finished_at) VALUES (''nightly'', now())'
    label: nightly-etl

  - id: poll_status
    type: io.kestra.plugin.azure.horizondb.durable.GetStatus
    host: "{{ inputs.host }}"
    port: 5432
    database: mydb
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    instanceId: "{{ outputs.start_etl.instanceId }}"

Example 3 — Trigger a Kestra flow when a durable function instance completes

id: horizondb_durable_on_completion
namespace: company.team

triggers:
  - id: on_durable_complete
    type: io.kestra.plugin.azure.horizondb.durable.Trigger
    host: "{{ secret('HORIZONDB_HOST') }}"
    port: 5432
    database: mydb
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    targetStatus: Completed
    interval: PT30S

tasks:
  - id: log_completion
    type: io.kestra.plugin.core.log.Log
    message: "Durable instance {{ trigger.instanceId }} completed with status {{ trigger.status }}"

Acceptance Criteria

Functional

• Abstract connection base class with password and Entra ID authentication
• Query and Queries tasks execute SQL and return results with fetchType support
• durable.Start, durable.Cancel, durable.Signal, durable.GetStatus, durable.ListInstances tasks implemented
• durable.Trigger polling trigger fires executions on instance state transitions
• Unit + integration tests pass (./gradlew test)
• Build passes (./gradlew build)

Kestra Plugin Coding Standards

• All new properties use Property<T> — no legacy @PluginProperty(dynamic = true) on new code
• password and any credential properties annotated with @PluginProperty(secret = true)
• Every property and output has a @Schema annotation
• Task classes carry the five mandatory Lombok annotations (@SuperBuilder, @ToString, @EqualsAndHashCode, @Getter, @NoArgsConstructor)
• Logging via runContext.logger() only
• JSON serialization uses Jackson mappers from io.kestra.core.serializers
• All Property<T> fields support Kestra expression language

Documentation & Structure

• @Plugin(examples = ...) entries each set full = true with a complete runnable flow (id + namespace + tasks/triggers)
• Sensitive values in examples use {{ secret('SECRET_NAME') }}
• package-info.java per sub-package with @PluginSubGroup(category = PluginSubGroup.PluginCategory.CLOUD)
• metadata/horizondb.yaml and io.kestra.plugin.azure.horizondb.svg icon present
• AGENTS.md updated with new task class references

---

Repository Setup Checklist

Step 1 (Scaffold) is omitted — plugin-azure already exists.

1. Add to Sanity check page

Add this plugin to the Sanity check Notion page.

2. Run scoped Terraform apply

Run the following from infra/terraform/github:

terraform apply \
    -target='github_repository.repo["plugin-azure"]' \
    -target='github_issue_labels.plugins["plugin-azure"]' \
    -target='github_repository_ruleset.branch["plugin-azure"]'

[nancysangani]

Hi @fdelbrayelle, I would like to work on this issue. Could you please assign it to me?

[fdelbrayelle]

Hi @nancysangani, yes sure, thanks!

[nancysangani]

Hi @fdelbrayelle, PR #316 is ready for review. Please take a look. Thanks!