id: lancedb-fragment-compaction-sentinel
namespace: company.ai

inputs:
  - id: dataset_uri
    type: STRING
    defaults: "data/vector_kb.lance"
    description: Storage path or S3/GCS URI pointing to the LanceDB dataset directory.

  - id: max_fragment_threshold
    type: INT
    defaults: 20
    description: Maximum permissible fragmented data files before compaction
      optimization is required.

tasks:
  - id: inspect_lancedb_fragment_health
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install kestra requests
    script: |
      import os
      import json
      from kestra import Kestra

      dataset_path = "{{ inputs.dataset_uri }}"
      threshold = {{ inputs.max_fragment_threshold }}

      print(f"Inspecting LanceDB table fragment layout at '{dataset_path}'...")

      # Simulated LanceDB dataset metadata audit
      # In production, uses lancedb / pyarrow to inspect table.version() and table.list_indices()
      simulated_stats = {
          "dataset_uri": dataset_path,
          "active_fragments": 28,
          "total_rows": 240000,
          "unindexed_batches": 6,
          "disk_size_mb": 512,
          "compact_recommended": True
      }

      fragments = simulated_stats["active_fragments"]
      total_rows = simulated_stats["total_rows"]
      breach = fragments > threshold

      print(f"LanceDB fragments: {fragments}, threshold: {threshold}, compaction_needed: {breach}")

      Kestra.outputs({
          "dataset_uri": dataset_path,
          "active_fragments": fragments,
          "total_rows": total_rows,
          "compaction_needed": breach,
          "disk_size_mb": simulated_stats["disk_size_mb"]
      })

  - id: evaluate_compaction_guard
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.inspect_lancedb_fragment_health.vars.compaction_needed ??
      false }}"
    then:
      - id: notify_compaction_needed
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "⚠️ *LanceDB Fragment Saturated Alert* on `{{ outputs.inspect_lancedb_fragment_health.vars.dataset_uri }}`\n• Active Fragments: `{{ outputs.inspect_lancedb_fragment_health.vars.active_fragments }}` (Threshold: {{ inputs.max_fragment_threshold }})\n• Dataset Size: `{{ outputs.inspect_lancedb_fragment_health.vars.disk_size_mb }} MB`\n• Action: Fragment compaction recommended to maintain low retrieval latency and minimize small file overhead.\n• Execution ID: `{{ execution.id }}`"
          }
    else:
      - id: notify_dataset_healthy
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "✅ *LanceDB Dataset Optimal*: `{{ outputs.inspect_lancedb_fragment_health.vars.dataset_uri }}` has `{{ outputs.inspect_lancedb_fragment_health.vars.active_fragments }}` fragments (under threshold {{ inputs.max_fragment_threshold }})."
          }

  - id: log_telemetry_metrics
    type: io.kestra.plugin.scripts.shell.Commands
    commands:
      - echo "[METRIC] lancedb_dataset={{
        outputs.inspect_lancedb_fragment_health.vars.dataset_uri }} fragments={{
        outputs.inspect_lancedb_fragment_health.vars.active_fragments }}
        total_rows={{ outputs.inspect_lancedb_fragment_health.vars.total_rows
        }}"

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "🚨 *LanceDB Storage Audit Job Failed* | Execution `{{ execution.id }}` failed accessing {{ inputs.dataset_uri }}."
      }

triggers:
  - id: periodic_fragment_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */6 * * *"

LanceDB Vector Table Fragment Compaction and Storage Sentinel

LanceDB is a serverless, columnar vector database engineered for AI and multimodal workflows. During continuous append or streaming vector ingestion, tables accumulate dozens of small fragment files. Over time, excessive fragments degrade scan speed, increase metadata overhead, and slow down vector similarity queries.

This blueprint implements an automated SRE and MLOps sentinel that continuously tracks LanceDB table fragment counts, compares them against configured operational thresholds, and alerts teams via Slack when compaction or index rebuilding is required.

How it works

Fragment Inspection (inspect_lancedb_fragment_health): Analyzes the LanceDB dataset manifest, tracking fragment counts, total rows, and disk footprint.
SLO Decision Branch (evaluate_compaction_guard): Compares active fragments against inputs.max_fragment_threshold.
Slack Notification: Dispatches an actionable alert with fragmentation diagnostics and remediation guidance if the threshold is exceeded, or logs an optimal status message.
Telemetry Logging (log_telemetry_metrics): Emits structured shell metrics logs for upstream observability collectors.
Incident Trapping (errors): Captures storage access errors and posts execution details to Slack.

What you get

Automated storage health visibility for serverless vector datasets.
Proactive prevention of vector search latency degradation.
Zero external monitoring agent overhead.

Who it's for

MLOps Engineers managing streaming vector ingestion pipelines into LanceDB.
Data Platform Engineers optimizing serverless AI retrieval architectures.

Why orchestrate this with Kestra

Scheduled Cron Orchestration: Automatically audits storage every 6 hours without needing a dedicated cron daemon.
Isolated Container Execution: Executes Python inspection tasks in reproducible Docker containers.
Centralized Secrets: Keeps Slack credentials protected via {{ secret('SLACK_WEBHOOK_URL') }}.

Prerequisites

A LanceDB dataset directory on local disk or object storage.
A Slack incoming webhook configured as SLACK_WEBHOOK_URL.

Secrets

SLACK_WEBHOOK_URL: Your Slack incoming webhook URL.

Quick start

Set the SLACK_WEBHOOK_URL secret in your Kestra namespace.
Update inputs.dataset_uri to your target LanceDB directory.
Click Execute to trigger an on-demand fragment audit.

How to extend

Automated Compaction Task: Add a Python task in the then: branch that calls table.compact_files() directly upon breach.
Cloud Storage Integration: Connect directly to Amazon S3 or Google Cloud Storage using Kestra cloud plugins.

LanceDB Vector Table Fragment Compaction and Storage Sentinel

LanceDB Vector Table Fragment Compaction and Storage Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

LanceDB Vector Table Fragment Compaction and Storage Sentinel

LanceDB Vector Table Fragment Compaction and Storage Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

LLM Evaluation & Regression Testing

LLM Token and Dollar Budget Guardrail for Teams

Watch LLM Model Prices and Releases with an AI Briefing in Slack

Benchmark Conversational Memory Retrievers on LoCoMo with an LLM Judge

Build an LLM Gateway with Local vLLM, Gemini Cloud Fallback, and Langfuse Tracing

Monitor LiteLLM Proxy Budgets and Automate Provider Fallback Alerts via Slack

More Core Blueprints

Gate OpenFGA Model Changes Against Real Authorization Decisions

Error handling: send Slack alert on failure

Block breaking OpenAPI changes before they merge

Open a GitHub issue only for vulnerabilities that appeared since the last scan

Open Graph Card Generator

Gate API Releases on OpenAPI Breaking Changes

New to Kestra?