Weaviate Autoschema Datatype Drift Repair Guard with Kestra

id: weaviate-autoschema-datatype-drift-repair-guard
namespace: company.team

inputs:
  - id: weaviate_url
    type: STRING
    defaults: http://localhost:8080
    description: Weaviate endpoint reachable by the Kestra worker.

  - id: anonymous
    type: BOOL
    defaults: true
    description: Use true for an anonymous local cluster. Set false to send
      WEAVIATE_API_KEY.

  - id: degraded_class
    type: STRING
    defaults: WvasProbe
    description: Existing class to inspect for autoschema drift.

  - id: repaired_class
    type: STRING
    defaults: WvasProbeRepair
    description: Versioned class created when drift is approved for repair.

  - id: probe_title
    type: STRING
    defaults: "Autoschema drift probe"
    description: Title in the bounded probe batch re-ingested into the repaired class.

  - id: probe_date
    type: STRING
    defaults: "2026-10-17T12:00:00Z"
    description: ISO date value re-ingested into the repaired date property. The
      degraded fixture may contain a non-ISO source string.

  - id: probe_tenant
    type: STRING
    defaults: platform
    description: Tenant value used by the filtered verification query and cleanup filter.

  - id: dry_run
    type: BOOL
    defaults: true
    description: Report the live schema diff without writing when true.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: Allow versioned class creation and probe re-ingestion only with
      dry_run false and approval.

tasks:
  - id: read_live_schema
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.weaviate_url }}/v1/schema/{{ inputs.degraded_class }}"
    method: GET
    headers:
      Authorization: "{{ inputs.anonymous ? '' : 'Bearer ' ~ secret('WEAVIATE_API_KEY') }}"

  - id: schema_diff
    type: io.kestra.plugin.core.log.Log
    message: >-
      Live schema diff for {{ inputs.degraded_class }}: {{
      fromJson(outputs.read_live_schema.body) | jq('{ class: .class, vectorizer:
      (.vectorizer // "none"), properties: [.properties[] | {name, dataType,
      tokenization: (.tokenization // null), indexFilterable: (.indexFilterable
      // null), indexSearchable: (.indexSearchable // null)}] }') | first |
      toJson }}.

  - id: detect_drift
    type: io.kestra.plugin.core.flow.If
    description: Detects the committed probe contract mismatch, including datatype,
      tokenization, index flags and vectorizer.
    condition: >-
      {{ (fromJson(outputs.read_live_schema.body) | jq('( (.vectorizer //
      "none") != "none" or ([.properties[] | select(.name == "title") |
      .dataType[0]] | first) != "text" or ([.properties[] | select(.name ==
      "title") | .tokenization] | first) != "word" or ([.properties[] |
      select(.name == "title") | .indexFilterable] | first) != true or
      ([.properties[] | select(.name == "title") | .indexSearchable] | first) !=
      true or ([.properties[] | select(.name == "eventDate") | .dataType[0]] |
      first) != "date" or ([.properties[] | select(.name == "eventDate") |
      .indexFilterable] | first) != true or ([.properties[] | select(.name ==
      "eventDate") | .indexSearchable] | first) != false or ([.properties[] |
      select(.name == "tenant") | .dataType[0]] | first) != "text" or
      ([.properties[] | select(.name == "tenant") | .tokenization] | first) !=
      "word" or ([.properties[] | select(.name == "tenant") | .indexFilterable]
      | first) != true or ([.properties[] | select(.name == "tenant") |
      .indexSearchable] | first) != true )') | first) }}
    then:
      - id: drift_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text":"Weaviate schema drift detected in {{ inputs.degraded_class }}. The live datatype/index/vectorizer diff is recorded in execution {{ execution.id }}. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."}

      - id: remediation_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approve_schema_repair
            type: io.kestra.plugin.core.flow.Pause
            description: Review the live schema diff before creating the versioned class and
              re-ingesting the probe batch.
            timeout: PT4H
            onResume:
              - id: approved
                type: BOOL
              - id: note
                type: STRING

          - id: approval_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve_schema_repair.onResume.approved ?? false }}"
            then:
              - id: reread_before_write
                type: io.kestra.plugin.core.http.Request
                uri: "{{ inputs.weaviate_url }}/v1/schema/{{ inputs.degraded_class }}"
                method: GET
                headers:
                  Authorization: "{{ inputs.anonymous ? '' : 'Bearer ' ~
                    secret('WEAVIATE_API_KEY') }}"

              - id: stale_plan_gate
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.reread_before_write.body != outputs.read_live_schema.body
                  }}"
                then:
                  - id: stale_plan_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: The live schema changed while approval was pending. No class was
                      created and no objects were written; run a fresh
                      detection.
                else:
                  - id: read_existing_repaired_schema
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.weaviate_url }}/v1/schema/{{ inputs.repaired_class }}"
                    method: GET
                    options:
                      allowFailed: true
                    headers:
                      Authorization: "{{ inputs.anonymous ? '' : 'Bearer ' ~
                        secret('WEAVIATE_API_KEY') }}"

                  - id: create_versioned_schema
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ outputs.read_existing_repaired_schema.code != 200 }}"
                    then:
                      - id: schema_create
                        type: io.kestra.plugin.weaviate.SchemaCreate
                        url: "{{ inputs.weaviate_url }}"
                        apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                        # SchemaCreate's className is not rendered by the current plugin, so this committed version name is literal.
                        className: WvasProbeRepair
                        fields:
                          title:
                            - text
                          eventDate:
                            - date
                          tenant:
                            - text
                    else:
                      - id: log_existing_schema
                        type: io.kestra.plugin.core.log.Log
                        message: "Versioned class {{ inputs.repaired_class }} already exists;
                          reconciling it instead of issuing a duplicate
                          SchemaCreate."

                  - id: reingest_probe
                    type: io.kestra.plugin.weaviate.BatchCreate
                    url: "{{ inputs.weaviate_url }}"
                    apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                    className: "{{ inputs.repaired_class }}"
                    objects:
                      - title: "{{ inputs.probe_title }}"
                        eventDate: "{{ inputs.probe_date }}"
                        tenant: "{{ inputs.probe_tenant }}"

                  - id: read_repaired_schema
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.weaviate_url }}/v1/schema/{{ inputs.repaired_class }}"
                    method: GET
                    headers:
                      Authorization: "{{ inputs.anonymous ? '' : 'Bearer ' ~
                        secret('WEAVIATE_API_KEY') }}"

                  - id: verify_schema
                    type: io.kestra.plugin.core.flow.If
                    condition: >-
                      {{ (fromJson(outputs.read_repaired_schema.body) | jq('(
                      (.vectorizer // "none") == "none" and ([.properties[] |
                      select(.name == "title") | .dataType[0]] | first) ==
                      "text" and ([.properties[] | select(.name == "title") |
                      .tokenization] | first) == "word" and ([.properties[] |
                      select(.name == "title") | .indexFilterable] | first) ==
                      true and ([.properties[] | select(.name == "title") |
                      .indexSearchable] | first) == true and ([.properties[] |
                      select(.name == "eventDate") | .dataType[0]] | first) ==
                      "date" and ([.properties[] | select(.name == "eventDate")
                      | .indexFilterable] | first) == true and ([.properties[] |
                      select(.name == "eventDate") | .indexSearchable] | first)
                      == false and ([.properties[] | select(.name == "tenant") |
                      .dataType[0]] | first) == "text" and ([.properties[] |
                      select(.name == "tenant") | .tokenization] | first) ==
                      "word" and ([.properties[] | select(.name == "tenant") |
                      .indexFilterable] | first) == true and ([.properties[] |
                      select(.name == "tenant") | .indexSearchable] | first) ==
                      true )') | first) }}
                    then:
                      - id: filtered_probe_query
                        type: io.kestra.plugin.weaviate.Query
                        url: "{{ inputs.weaviate_url }}"
                        apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                        fetchType: FETCH
                        query: |
                          {
                            Get {
                              {{ inputs.repaired_class }}(
                                limit: 1,
                                where: {
                                  operator: And,
                                  operands: [
                                    {path: ["eventDate"], operator: GreaterThanEqual, valueDate: "2026-01-01T00:00:00Z"},
                                    {path: ["tenant"], operator: Equal, valueText: "{{ inputs.probe_tenant }}"}
                                  ]
                                }
                              ) {
                                title
                                eventDate
                                tenant
                              }
                            }
                          }

                      - id: verify_filtered_query
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ outputs.filtered_probe_query.size > 0 }}"
                        then:
                          - id: delete_probe
                            type: io.kestra.plugin.weaviate.Delete
                            url: "{{ inputs.weaviate_url }}"
                            apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                            className: "{{ inputs.repaired_class }}"
                            filter:
                              tenant: "{{ inputs.probe_tenant }}"
                          - id: log_repaired
                            type: io.kestra.plugin.core.log.Log
                            message: "Schema repaired in {{ inputs.repaired_class }} and the filtered
                              date-range probe returned {{
                              outputs.filtered_probe_query.size }} object(s).
                              The probe was deleted after verification."
                        else:
                          - id: delete_failed_probe
                            type: io.kestra.plugin.weaviate.Delete
                            url: "{{ inputs.weaviate_url }}"
                            apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                            className: "{{ inputs.repaired_class }}"
                            filter:
                              tenant: "{{ inputs.probe_tenant }}"
                          - id: filtered_query_failed
                            type: io.kestra.plugin.core.execution.Fail
                            errorMessage: The repaired class schema matched but the date-range plus tenant
                              query did not return the probe object; the probe
                              cleanup was attempted.
                    else:
                      - id: delete_bad_schema_probe
                        type: io.kestra.plugin.weaviate.Delete
                        url: "{{ inputs.weaviate_url }}"
                        apiKey: "{{ inputs.anonymous ? '' : secret('WEAVIATE_API_KEY') }}"
                        className: "{{ inputs.repaired_class }}"
                        filter:
                          tenant: "{{ inputs.probe_tenant }}"
                      - id: repaired_schema_failed
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: SchemaCreate did not produce the committed
                          datatype/index/tokenization/vectorizer contract; the
                          probe cleanup was attempted.
            else:
              - id: log_declined
                type: io.kestra.plugin.core.log.Log
                message: "Schema repair was not approved ({{
                  outputs.approve_schema_repair.onResume.note ?? 'no approval
                  note' }}); no class or object was written."
        else:
          - id: log_dry_run
            type: io.kestra.plugin.core.log.Log
            message: "Drift found in {{ inputs.degraded_class }}. No write was attempted
              because auto_remediate={{ inputs.auto_remediate }} and dry_run={{
              inputs.dry_run }}."
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.degraded_class }} matches the committed probe contract (date
          datatype, tokenization, index flags and vectorizer). No repair was
          needed."

errors:
  - id: alert_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text":"Weaviate autoschema datatype drift guard FAILED for {{ inputs.degraded_class }} in execution {{ execution.id }}. No unverified repair should be assumed."}

triggers:
  - id: daily_scan
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

Safety

An undeclared Weaviate class can accept a successful batch while auto-schema stores a date as text. The retrieval pipeline then reports ingestion success but date filters return no objects. This flow protects the existing class and writes only to a new versioned class after approval.

Guard	What it stops
Live schema diff	A count-only canary passing a wrongly typed class
Dry-run and auto-remediate gates	An accidental class creation or re-ingestion
Pause plus post-pause schema read	Acting on a stale approval snapshot
SchemaCreate under a versioned name	Mutating or deleting the existing class
Schema read-back and filtered Query	Treating a successful write as proof of a usable schema
Probe Delete	Leaving verification data in the repaired class

This blueprint was created by zkasuran.

How it works

read_live_schema (io.kestra.plugin.core.http.Request) reads /v1/schema/<class> from the target Weaviate instance. schema_diff records the datatype, tokenization, index flags and vectorizer values from that live response.
detect_drift compares the live response with the committed probe contract: title and tenant are tokenized text, eventDate is filterable date plus the class vectorizer is none.
A finding is reported to Slack. The flow stays read-only unless auto_remediate=true and dry_run=false; the disabled schedule always forces those safe values.
approve_schema_repair (io.kestra.plugin.core.flow.Pause) presents the diff for approval. reread_before_write reads the old class again and stale_plan_gate refuses the write if its bytes changed while approval was pending.
create_versioned_schema (io.kestra.plugin.weaviate.SchemaCreate) creates the typed class and reingest_probe (io.kestra.plugin.weaviate.BatchCreate) writes the bounded probe batch into it.
read_repaired_schema verifies every committed property setting. filtered_probe_query (io.kestra.plugin.weaviate.Query) requires both a date range and tenant where clause to return the probe. The run fails if either check misses plus the probe is deleted on both success and failed verification paths.
The errors block sends a Slack alert for an API, schema or verification failure.

What you get

A live schema diff that distinguishes a populated but incorrectly typed class from a healthy class.
A dry-run report and explicit approval before creating a versioned repair class.
A real SchemaCreate plus BatchCreate repair path, with schema read-back and a date-range/tenant query as the recovery proof.
Probe cleanup and Slack alerts for failed runs.

Who it's for

Platform and data teams operating Weaviate ingestion pipelines where a successful batch response must also mean that filtering and retrieval semantics remain usable.

Why orchestrate this with Kestra

Weaviate stores schema and data but does not coordinate a reviewable drift plan, an approval pause, a versioned repair plus an independent query proof. Kestra provides those steps as an auditable execution with safe trigger inputs and an error notification path.

Tested end to end

Tested on Kestra 2.0.5 with Weaviate 1.25.0. The test fixture reproduces autoschema by batch-inserting eventDate=17/10/2026 into an undeclared class, which creates eventDate as text; a GraphQL date-range query returns no probe. Remediation re-ingests the same probe with the normalized ISO value 2026-10-17T12:00:00Z into the typed versioned class. Counts and query results are read from Weaviate's REST and GraphQL APIs.

Case	Execution	Result
Healthy control	`2qMzrsehOYkQ0ftEvGEIvU`	Healthy class spared
Degraded dry run	`2zLjqIm5iKvDbRhmwirapt`	Drift reported with no write
Declined approval	`3wEyoJXmlA9HcPNBpVeMsw`	No class or object write
Remediation	`19D2peSAghuloZUI6riDZn`	Typed class and filtered probe verified
Repeat	`5SEIitSJM09aTpUxEsVBEY`	Existing version reconciled
Ineffective write	`4EaYZjvU7dvEFL9F0Ezjgg`	Failed schema verification
Race during approval	`3ZAkgXzwWsJN0jx8Tergor`	Stale plan refused

Prerequisites

Kestra 2.0.5 with the Weaviate plugin and a Weaviate 1.25+ instance reachable from the worker.
Permission to read /v1/schema, create a class, batch-create objects, query GraphQL and delete the probe objects.

Secrets

WEAVIATE_API_KEY: bearer credential when anonymous=false.
SLACK_WEBHOOK_URL: incoming webhook for findings and failures.

Inputs

Input	Purpose
`weaviate_url`	Weaviate endpoint
`anonymous`	Whether to omit the API key
`degraded_class`	Class whose live schema is diffed
`repaired_class`	Versioned class created by remediation
`probe_title`, `probe_date`, `probe_tenant`	Bounded re-ingestion and verification values
`dry_run`	Report only when true
`auto_remediate`	Arm the approval and write path

Quick start

Start Weaviate and configure the two secrets; set anonymous=true for a local fixture when appropriate.
Batch-insert a probe into an undeclared degraded_class so autoschema creates the class, then run with the defaults to inspect the diff.
Set auto_remediate=true and dry_run=false, approve the displayed diff plus inspect the schema read-back and filtered query.
Confirm the probe was deleted from the versioned class and keep the original class for a separately planned data migration.

How to extend

Replace the bounded probe object with the production batch produced by the upstream source and keep a stable versioned class name per schema revision.
Add additional committed properties to the SchemaCreate fields and both schema diff expressions before enabling the schedule.
Put a source-specific re-ingestion task before BatchCreate when the original class contains more than the canary batch.

Limits: SchemaCreate exposes property datatypes but not arbitrary Weaviate module configuration. This blueprint therefore commits and verifies the server defaults for tokenization, index flags and the none vectorizer; use a dedicated schema-management plugin when a different vectorizer or non-default index setting is required. It repairs into a versioned class and proves the supplied probe batch, not a full historical backfill.

Repair Weaviate autoschema datatype drift behind a verified canary

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Repair Weaviate autoschema datatype drift behind a verified canary

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

Airbyte Connection Sync Failure Sentinel

Autonomous Mathematical Research and Scientific Computation Pipeline

Run a Scheduled ETL Sync with Airbyte Cloud

New to Kestra?