feat(assets): create assets for dbt models

[fdelbrayelle]

⚠️ TO BE MERGED BETWEEN 1.3 JAR AND DOCKER IMAGE (CHANGE VERSION TO 1.3.0 IN gradle.properties)

closes https://github.com/kestra-io/kestra-ee/issues/6473

What changes are being made and why?

Upsert assets as tables by parsing the manifest file and handle lineages through the "depends on".

---

How the changes have been QAed?

Checked basic assets with this flow:

id: dbt_cli_assets_kv_test2
namespace: company.team

inputs:
  - id: kv_namespace
    type: STRING
    defaults: "{{ flow.namespace }}"
  - id: kv_key
    type: STRING
    defaults: "manifest.json"

tasks:
  - id: wd
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      # 1) Create a minimal dbt project in the working directory
      - id: create_project
        type: io.kestra.plugin.core.storage.LocalFiles
        inputs:
          dbt_project.yml: |
            name: 'my_dbt_project'
            version: '1.0.0'
            config-version: 2

            profile: 'my_dbt_project'

            model-paths: ["models"]
            analysis-paths: ["analyses"]
            test-paths: ["tests"]
            seed-paths: ["seeds"]
            macro-paths: ["macros"]
            snapshot-paths: ["snapshots"]

            target-path: "target"
            clean-targets:
              - "target"
              - "dbt_packages"

            models:
              my_dbt_project:
                +materialized: table
          packages.yml: |
            packages: []
          models/example.sql: |
            select 1 as id
          models/example_two.sql: |
            select 2 as id
          models/example_three.sql: |
            select 3 as id

      # 2) Run #1: dbt deps + dbt build (generates manifest + run_results)
      #    and store the manifest in the KV store
      - id: dbt_build_and_store_manifest
        type: io.kestra.plugin.dbt.cli.DbtCLI
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          delete: true
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
        assets:
          enableAuto: true

        # IMPORTANT: this test creates the manifest, parses it (with assets), and pushes it to KV
        storeManifest:
          namespace: "{{ inputs.kv_namespace }}"
          key: "{{ inputs.kv_key }}"

        commands:
          - dbt deps
          - dbt build --no-partial-parse
          - ls -lah
          - ls -lah target || true
          - find . -maxdepth 3 -type f -name "manifest.json" -o -name "run_results.json"
          - head -n 5 target/manifest.json || true

        profiles: |
          my_dbt_project:
            outputs:
              dev:
                type: duckdb
                path: ":memory:"
                threads: 4
            target: dev

      # 3) Run #2: simulate a "defer/state" run (or just a run) by reloading
      #    the manifest from KV before running dbt again
      - id: dbt_build_with_loaded_manifest
        type: io.kestra.plugin.dbt.cli.DbtCLI
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          delete: true
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
        assets:
          enableAuto: true

        loadManifest:
          namespace: "{{ inputs.kv_namespace }}"
          key: "{{ inputs.kv_key }}"

        # (optional) shows it works with flags too; adapt as needed
        commands:
          - dbt build --no-partial-parse

        profiles: |
          my_dbt_project:
            outputs:
              dev:
                type: duckdb
                path: ":memory:"
                threads: 4
            target: dev

outputs:
  - id: first_manifest_uri
    type: STRING
    value: "{{ outputs.dbt_build_and_store_manifest.outputFiles['manifest.json'] }}"
  - id: first_run_results_uri
    type: STRING
    value: "{{ outputs.dbt_build_and_store_manifest.outputFiles['run_results.json'] }}"
  - id: second_manifest_uri
    type: STRING
    value: "{{ outputs.dbt_build_with_loaded_manifest.outputFiles['manifest.json'] }}"
  - id: second_run_results_uri
    type: STRING
    value: "{{ outputs.dbt_build_with_loaded_manifest.outputFiles['run_results.json'] }}"

Results:

Which leads to 3 assets in the assets table of Kestra:

[
  {
    "key": "fde_memory.main.example_three",
    "value": {
      "id": "memory.main.example_three",
      "name": "example_three",
      "type": "io.kestra.plugin.ee.assets.Table",
      "schema": "main",
      "system": "duckdb",
      "created": "2026-02-19T10:35:25.323642Z",
      "deleted": false,
      "updated": "2026-02-19T10:35:25.424067Z",
      "database": "memory",
      "metadata": {
        "name": "example_three",
        "schema": "main",
        "system": "duckdb",
        "database": "memory"
      },
      "tenantId": "fde"
    },
    "tenant_id": "fde",
    "namespace": null,
    "id": "memory.main.example_three",
    "type": "io.kestra.plugin.ee.assets.Table",
    "display_name": null,
    "created": "2026-02-19 10:35:25.323642 +00:00",
    "updated": "2026-02-19 10:35:25.424067 +00:00",
    "deleted": false,
    "fulltext": null
  },
  {
    "key": "fde_memory.main.example",
    "value": {
      "id": "memory.main.example",
      "name": "example",
      "type": "io.kestra.plugin.ee.assets.Table",
      "schema": "main",
      "system": "duckdb",
      "created": "2026-02-19T10:35:25.323646Z",
      "deleted": false,
      "updated": "2026-02-19T10:35:25.436137Z",
      "database": "memory",
      "metadata": {
        "name": "example",
        "schema": "main",
        "system": "duckdb",
        "database": "memory"
      },
      "tenantId": "fde"
    },
    "tenant_id": "fde",
    "namespace": null,
    "id": "memory.main.example",
    "type": "io.kestra.plugin.ee.assets.Table",
    "display_name": null,
    "created": "2026-02-19 10:35:25.323646 +00:00",
    "updated": "2026-02-19 10:35:25.436137 +00:00",
    "deleted": false,
    "fulltext": null
  },
  {
    "key": "fde_memory.main.example_two",
    "value": {
      "id": "memory.main.example_two",
      "name": "example_two",
      "type": "io.kestra.plugin.ee.assets.Table",
      "schema": "main",
      "system": "duckdb",
      "created": "2026-02-19T10:35:25.323648Z",
      "deleted": false,
      "updated": "2026-02-19T10:35:25.446084Z",
      "database": "memory",
      "metadata": {
        "name": "example_two",
        "schema": "main",
        "system": "duckdb",
        "database": "memory"
      },
      "tenantId": "fde"
    },
    "tenant_id": "fde",
    "namespace": null,
    "id": "memory.main.example_two",
    "type": "io.kestra.plugin.ee.assets.Table",
    "display_name": null,
    "created": "2026-02-19 10:35:25.323648 +00:00",
    "updated": "2026-02-19 10:35:25.446084 +00:00",
    "deleted": false,
    "fulltext": null
  }
]

And also to outputs in the asset_lineage_events table:

[
  {
    "key": "4zzQTn4VudZXXZLEfQod1P",
    "value": {
      "uid": "4zzQTn4VudZXXZLEfQod1P",
      "state": "SUCCESS",
      "flowId": "dbt_cli_assets_kv_test2",
      "inputs": [],
      "taskId": "dbt_build_and_store_manifest",
      "created": "2026-02-19T10:35:19.685076Z",
      "endDate": "2026-02-19T10:35:19.580537Z",
      "outputs": [
        {
          "id": "memory.main.example_three",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example_two",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        }
      ],
      "tenantId": "fde",
      "namespace": "company.team",
      "startDate": "2026-02-19T10:34:56.987494Z",
      "taskRunId": "5hFXq7GEsGkQy4frkuyUgy",
      "executionId": "1bhCojxnR7XK3s6dnttrsM",
      "flowRevision": 1
    },
    "tenant_id": "fde",
    "namespace": "company.team",
    "flow_id": "dbt_cli_assets_kv_test2",
    "flow_revision": 1,
    "execution_id": "1bhCojxnR7XK3s6dnttrsM",
    "task_id": "dbt_build_and_store_manifest",
    "task_run_id": "5hFXq7GEsGkQy4frkuyUgy",
    "created": "2026-02-19 10:35:19.685076 +00:00"
  },
  {
    "key": "6wjWH4mOKQXfpUvJKUvBtC",
    "value": {
      "uid": "6wjWH4mOKQXfpUvJKUvBtC",
      "state": "SUCCESS",
      "flowId": "dbt_cli_assets_kv_test2",
      "inputs": [],
      "taskId": "dbt_build_with_loaded_manifest",
      "created": "2026-02-19T10:35:25.353971Z",
      "endDate": "2026-02-19T10:35:25.341924Z",
      "outputs": [
        {
          "id": "memory.main.example_three",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example_two",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        }
      ],
      "tenantId": "fde",
      "namespace": "company.team",
      "startDate": "2026-02-19T10:35:19.590976Z",
      "taskRunId": "2hYP2qamOR3OwdLGjr8tcU",
      "executionId": "1bhCojxnR7XK3s6dnttrsM",
      "flowRevision": 1
    },
    "tenant_id": "fde",
    "namespace": "company.team",
    "flow_id": "dbt_cli_assets_kv_test2",
    "flow_revision": 1,
    "execution_id": "1bhCojxnR7XK3s6dnttrsM",
    "task_id": "dbt_build_with_loaded_manifest",
    "task_run_id": "2hYP2qamOR3OwdLGjr8tcU",
    "created": "2026-02-19 10:35:25.353971 +00:00"
  }
]

And finally we see the 3 sql examples in the lineage:

Second example flow to check the lineage:

id: dbt_cli_assets_lineage_test
namespace: company.team

tasks:
  - id: wd
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: create_project
        type: io.kestra.plugin.core.storage.LocalFiles
        inputs:
          dbt_project.yml: |
            name: 'my_dbt_project'
            version: '1.0.0'
            config-version: 2
            profile: 'my_dbt_project'

            model-paths: ["models"]
            macro-paths: ["macros"]

            target-path: "target"
            clean-targets:
              - "target"
              - "dbt_packages"

            models:
              my_dbt_project:
                +materialized: table

          packages.yml: |
            packages: []

          models/example_base.sql: |
            select 1 as id

          models/example_two.sql: |
            select * from {% raw %}{{ ref('example_base') }}{% endraw %}

          models/example_three.sql: |
            select count(*) as cnt from {% raw %}{{ ref('example_two') }}{% endraw %}

      - id: dbt_build
        type: io.kestra.plugin.dbt.cli.DbtCLI
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          delete: true
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest

        assets:
          enableAuto: true

        commands:
          - dbt deps
          - dbt build --no-partial-parse

        profiles: |
          my_dbt_project:
            outputs:
              dev:
                type: duckdb
                path: ":memory:"
                threads: 4
            target: dev

Results:

Which leads to 3 assets with lineage in the assets_lineage_events table of Kestra:

[
  {
    "key": "1PSncPPjkJEcpmTImYQanT",
    "value": {
      "uid": "1PSncPPjkJEcpmTImYQanT",
      "state": "SUCCESS",
      "flowId": "dbt_cli_assets_lineage_test",
      "inputs": [],
      "taskId": "dbt_build",
      "created": "2026-02-02T15:52:14.041424Z",
      "endDate": "2026-02-02T15:52:13.942231Z",
      "outputs": [
        {
          "id": "memory.main.example_base",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example_three",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        },
        {
          "id": "memory.main.example_two",
          "type": "io.kestra.plugin.ee.assets.Table",
          "tenantId": "fde"
        }
      ],
      "tenantId": "fde",
      "namespace": "company.team",
      "startDate": "2026-02-02T15:52:02.957498Z",
      "taskRunId": "2cKEhiYR8dVdWtvHlKy3Rx",
      "executionId": "67XPdrJYENHYxOk2B3ljHW",
      "flowRevision": 2
    },
    "tenant_id": "fde",
    "namespace": "company.team",
    "flow_id": "dbt_cli_assets_lineage_test",
    "flow_revision": 2,
    "execution_id": "67XPdrJYENHYxOk2B3ljHW",
    "task_id": "dbt_build",
    "task_run_id": "2cKEhiYR8dVdWtvHlKy3Rx",
    "created": "2026-02-02 15:52:14.041424 +00:00"
  }
]

---

Setup Instructions

---

Contributor Checklist ✅

• I have read and followed the plugin contribution guidelines

[github-actions]

🧪 Java Unit Tests

	Tests	Passed ✅	Skipped ⚠️	Failed	Time ⏱
Java Tests Report	24 ran	23 ✅	1 ⚠️	0 ❌	6m 27s 137ms

📦 Artifacts

Name	Size	Updated	Expiration
	207.51 KB	Mar 2, 26, 2:53:47 PM UTC	Mar 9, 26, 2:53:46 PM UTC

🔁 Unreleased Commits

✅ No unreleased commits found.

[Pradumnasaraf]

Hey @fdelbrayelle,

After running the above flow, the assets are not showing up, am I doing something wrong?
https://jam.dev/c/7c5d59cf-02b8-4847-bfab-0661dca44782

[Pradumnasaraf]

Hey @fdelbrayelle It's working now, after the no-plugins approach. strike. And it looks good!