Roadmap¶

TRY
    SET result = HTTP_GET('https://api.example.com/data')
    SET parsed = JSON_PARSE(result)
CATCH http_error
    PRINT 'HTTP call failed: ' || error['message']
    CALL log_error(error)
CATCH parse_error
    PRINT 'JSON parsing failed'
    SET parsed = {}
CATCH
    -- Catch-all for any other errors
    PRINT 'Unexpected error: ' || error['message']
FINALLY
    -- Always runs (cleanup)
    CALL close_connections()
END TRY

-- THROW/RAISE with error codes
THROW 'Resource not found' WITH CODE 'HTTP_404';
RAISE error_msg WITH CODE error_code;  -- Expressions supported

-- Define a function that returns a value
CREATE FUNCTION calculate_severity(error_count NUMBER, warn_count NUMBER) 
RETURNS STRING AS
BEGIN
    DECLARE score NUMBER = error_count * 10 + warn_count

    IF score > 100 THEN
        RETURN 'critical'
    ELSIF score > 50 THEN
        RETURN 'high'
    ELSIF score > 20 THEN
        RETURN 'medium'
    ELSE
        RETURN 'low'
    END IF
END FUNCTION

-- Usage: Functions can be used in expressions
SET severity = calculate_severity(errors, warnings)
SET message = 'Status: ' || calculate_severity(5, 10)

-- Build query dynamically based on conditions
DECLARE query STRING = 'FROM logs-*'

IF severity_filter IS NOT NULL THEN
    SET query = query || ' | WHERE level = ''' || severity_filter || ''''
END IF

IF service_filter IS NOT NULL THEN
    SET query = query || ' | WHERE service = ''' || service_filter || ''''
END IF

SET query = query || ' | LIMIT ' || max_results

-- Execute the dynamic query
EXECUTE IMMEDIATE query INTO results

-- With bind variables (SQL injection safe)
EXECUTE IMMEDIATE 
    'FROM logs-* | WHERE service = :svc AND level = :lvl | LIMIT :lim'
    USING service_name, 'ERROR', 100
    INTO results

-- MAP literal syntax
DECLARE config MAP = MAP { 'host' => 'localhost', 'port' => 9200 };

-- Create empty map
DECLARE counts MAP = MAP {};

-- Add/update entries with MAP_PUT (returns new map)
SET counts = MAP_PUT(counts, 'api-service', 42);
SET counts = MAP_PUT(counts, 'db-service', 
    MAP_GET_OR_DEFAULT(counts, 'db-service', 0) + 1);

-- Get values
DECLARE val = MAP_GET(config, 'host');
DECLARE port = MAP_GET_OR_DEFAULT(config, 'timeout', 30);

-- Check existence
IF MAP_CONTAINS_KEY(counts, 'api-service') THEN
    PRINT 'API has ' || MAP_GET(counts, 'api-service') || ' errors';
END IF

-- Get keys and values as arrays
DECLARE keys ARRAY = MAP_KEYS(counts);
DECLARE values ARRAY = MAP_VALUES(counts);

-- Iterate over keys
DECLARE i NUMBER;
FOR i IN 1..ARRAY_LENGTH(keys) LOOP
    DECLARE k STRING = keys[i];
    IF MAP_GET(counts, k) > 10 THEN
        CALL alert_team(k, MAP_GET(counts, k));
    END IF
END LOOP

-- Merge maps
SET config = MAP_MERGE(defaults, overrides);

-- Create from arrays
SET my_map = MAP_FROM_ARRAYS(['a', 'b', 'c'], [1, 2, 3]);

-- Explicit cursor for large datasets
DECLARE CURSOR log_cursor FOR 
    FROM logs-* 
    | WHERE @timestamp > NOW() - 1 HOUR 
    | LIMIT 100000

OPEN log_cursor

-- Process in batches
DECLARE batch ARRAY<DOCUMENT>
DECLARE processed NUMBER = 0

WHILE FETCH log_cursor LIMIT 1000 INTO batch LOOP
    -- Process batch
    FOR doc IN batch LOOP
        CALL process_log(doc)
    END LOOP

    SET processed = processed + ARRAY_LENGTH(batch)
    PRINT 'Processed: ' || processed || ' documents'

    -- Optional: yield control for long-running operations
    IF processed % 10000 = 0 THEN
        COMMIT WORK  -- Checkpoint progress
    END IF
END LOOP

CLOSE log_cursor

-- Bulk collect from query
DECLARE logs ARRAY<DOCUMENT>
BULK COLLECT INTO logs
    FROM logs-*
    | WHERE level = 'ERROR'
    | LIMIT 5000

-- Bulk process with FORALL
FORALL log IN logs
    CALL process_and_archive(log)
    SAVE EXCEPTIONS  -- Continue on individual failures

-- Check for errors
IF @bulk_errors.COUNT > 0 THEN
    PRINT @bulk_errors.COUNT || ' documents failed processing'
    FOR err IN @bulk_errors LOOP
        PRINT 'Index ' || err.index || ': ' || err.message
    END LOOP
END IF

-- Bulk index with retry
FORALL doc IN transformed_docs
    INDEX_DOCUMENT('output-index', doc)
    ON_FAIL RETRY 3 THEN SKIP

-- Create a recurring job
CREATE JOB daily_log_cleanup
SCHEDULE '0 2 * * *'  -- 2 AM daily (cron syntax)
TIMEZONE 'UTC'
ENABLED true
AS
BEGIN
    PRINT 'Starting daily cleanup at ' || CURRENT_TIMESTAMP

    -- Archive logs older than 30 days
    CALL archive_old_logs(30)

    -- Clean up temporary indices
    CALL cleanup_temp_indices()

    -- Send daily report
    CALL generate_and_send_report()

    PRINT 'Cleanup completed'
END JOB

-- Job management
ALTER JOB daily_log_cleanup DISABLE
ALTER JOB daily_log_cleanup SCHEDULE '0 3 * * *'  -- Change to 3 AM
DROP JOB daily_log_cleanup

-- View job history
SELECT * FROM @job_runs 
    WHERE job_name = 'daily_log_cleanup' 
    ORDER BY start_time DESC 
    LIMIT 10

-- Trigger on new documents
CREATE TRIGGER on_critical_error
WHEN DOCUMENT INSERTED INTO logs-*
WHERE level = 'ERROR' AND service IN ('payment', 'auth', 'checkout')
BEGIN
    -- @document contains the new document
    DECLARE doc DOCUMENT = @document

    -- Immediate alerting for critical services
    CALL SLACK_SEND(
        '#critical-alerts',
        '🚨 Critical Error in ' || doc.service || ': ' || doc.message
    )

    -- Check if this is a pattern
    DECLARE recent_count NUMBER
    SET recent_count = ESQL_QUERY(
        'FROM logs-* 
         | WHERE service = ''' || doc.service || ''' 
         AND level = ''ERROR'' 
         AND @timestamp > NOW() - 5 MINUTES
         | STATS count = COUNT(*)'
    )[0].count

    IF recent_count > 10 THEN
        CALL PAGERDUTY_TRIGGER(
            'Error storm in ' || doc.service,
            'critical',
            {'service': doc.service, 'count': recent_count}
        )
    END IF
END TRIGGER

-- Trigger on alert firing (Elasticsearch Alerting integration)
CREATE TRIGGER on_alert_fire
WHEN ALERT 'high-error-rate' FIRES
BEGIN
    -- @alert contains alert context
    CALL escalate_to_oncall(@alert)
END TRIGGER

-- Trigger on index lifecycle events
CREATE TRIGGER on_index_rollover
WHEN INDEX ROLLED OVER IN logs-*
BEGIN
    -- @old_index, @new_index available
    PRINT 'Index rolled over: ' || @old_index || ' -> ' || @new_index
    CALL archive_to_s3(@old_index)
END TRIGGER

-- Trigger management
ALTER TRIGGER on_critical_error DISABLE
DROP TRIGGER on_critical_error
SHOW TRIGGERS

-- Package specification (public interface)
CREATE PACKAGE incident_response AS
    -- Public procedures
    PROCEDURE handle_incident(incident_id STRING)
    PROCEDURE escalate(incident_id STRING, level NUMBER)
    PROCEDURE resolve(incident_id STRING, resolution STRING)

    -- Public functions
    FUNCTION get_severity(incident_id STRING) RETURNS STRING
    FUNCTION get_oncall() RETURNS STRING

    -- Package constants
    CONSTANT DEFAULT_TIMEOUT NUMBER = 300
    CONSTANT ESCALATION_LEVELS ARRAY = ['low', 'medium', 'high', 'critical']
END PACKAGE

-- Package body (implementation)
CREATE PACKAGE BODY incident_response AS
    -- Private state (per-session)
    DECLARE active_incidents MAP<STRING, DOCUMENT> = {}

    -- Private helper (not visible outside package)
    PROCEDURE internal_notify(channel STRING, message STRING) AS
    BEGIN
        CALL SLACK_SEND(channel, message)
    END

    -- Public procedure implementation
    PROCEDURE handle_incident(incident_id STRING) AS
    BEGIN
        CALL internal_notify('#incidents', 'Handling: ' || incident_id)
        SET active_incidents[incident_id] = {'status': 'in_progress'}
    END

    -- Public function implementation
    FUNCTION get_severity(incident_id STRING) RETURNS STRING AS
    BEGIN
        DECLARE incident DOCUMENT
        SET incident = active_incidents[incident_id]
        RETURN incident.severity ?? 'unknown'
    END
END PACKAGE BODY

-- Usage
CALL incident_response.handle_incident('INC-001')
SET sev = incident_response.get_severity('INC-001')
PRINT incident_response.DEFAULT_TIMEOUT

-- Grant execute permission
GRANT EXECUTE ON PROCEDURE analyze_logs TO ROLE 'analyst'
GRANT EXECUTE ON PACKAGE incident_response TO ROLE 'sre'

-- Revoke permission
REVOKE EXECUTE ON PROCEDURE delete_old_data FROM ROLE 'analyst'

-- Invoker vs definer rights
CREATE PROCEDURE admin_cleanup()
AUTHID DEFINER  -- Runs with procedure owner's privileges
AS
BEGIN
    -- Can perform admin operations even if caller is limited user
    CALL delete_old_indices()
    CALL vacuum_data()
END

CREATE PROCEDURE user_report()
AUTHID CURRENT_USER  -- Runs with caller's privileges (default)
AS
BEGIN
    -- Limited to what the calling user can access
    CALL generate_report()
END

-- Secure credential reference (no plaintext secrets)
CALL HTTP_POST(
    'https://api.pagerduty.com/incidents',
    headers = {'Authorization': CREDENTIAL('pagerduty_api_key')},
    body = incident_data
)

-- Enable profiling for session
SET PROFILING ON

-- Run procedure
CALL complex_data_pipeline()

-- View execution profile
SHOW PROFILE

-- Output:
-- ┌─────────────────────────────────────────────────────────────┐
-- │ Line │ Statement                        │ Time    │ Calls  │
-- ├──────┼──────────────────────────────────┼─────────┼────────┤
-- │ 10   │ SET results = ESQL_QUERY(...)    │ 2.345s  │ 1      │ <-- Bottleneck
-- │ 15   │ FOR doc IN results LOOP          │ 0.523s  │ 1000   │
-- │ 20   │   CALL process_document(doc)     │ 0.412s  │ 1000   │
-- │ 25   │   CALL HTTP_POST(...)            │ 1.890s  │ 1000   │ <-- Bottleneck
-- ├──────┼──────────────────────────────────┼─────────┼────────┤
-- │      │ TOTAL                            │ 5.170s  │        │
-- └─────────────────────────────────────────────────────────────┘

-- Assertions for testing
ASSERT result > 0, 'Result should be positive'
ASSERT response.status = 200, 'HTTP call should succeed'
ASSERT ARRAY_LENGTH(items) <= 100, 'Too many items returned'

-- Debug logging
SET DEBUG ON
-- Shows variable assignments, function calls, branch decisions

-- Semantic search with embeddings
DECLARE similar_docs ARRAY<DOCUMENT>
VECTOR_SEARCH INTO similar_docs
    FROM knowledge-base
    QUERY_VECTOR LLM_EMBED(user_question)
    FIELD 'embedding'
    K 10
    NUM_CANDIDATES 100

-- RAG (Retrieval Augmented Generation) pattern
DECLARE context STRING = ARRAY_JOIN(
    ARRAY_MAP(similar_docs, d => d.content),
    '\n---\n'
)

SET answer = LLM_COMPLETE(
    'Answer based on context:\n' || context || '\n\nQuestion: ' || user_question
)

-- Programmatic ILM
CREATE PROCEDURE smart_retention(pattern STRING, hot_days NUMBER, warm_days NUMBER) AS
BEGIN
    FOR idx IN (SHOW INDICES pattern) LOOP
        DECLARE age_days NUMBER = DATE_DIFF(NOW(), idx.creation_date, 'days')

        IF age_days > hot_days + warm_days THEN
            CALL archive_to_s3(idx.name)
            CALL delete_index(idx.name)
        ELSIF age_days > hot_days THEN
            CALL move_to_warm_tier(idx.name)
        END IF
    END LOOP
END

-- Query remote clusters
DECLARE remote_errors ARRAY<DOCUMENT>
FROM cluster:us-west/logs-* | WHERE level = 'ERROR' INTO remote_errors

-- Aggregate across clusters
DECLARE global_stats DOCUMENT
AGGREGATE INTO global_stats
    FROM cluster:*/logs-*
    | STATS total = COUNT(*), errors = COUNT(*) WHERE level = 'ERROR'
    BY cluster

-- Create agent using Elastic Agent Builder
CREATE AGENT log_analyst
    USING AGENT_BUILDER {
        MODEL 'azure-openai-gpt4'
        CAPABILITIES ['query_logs', 'identify_patterns', 'summarize']
        PROMPT "You are an expert at analyzing log data and identifying anomalies"
        TOOLS [
            PROCEDURE analyze_errors,
            PROCEDURE summarize_trends,
            FUNCTION ESQL_QUERY
        ]
    }

-- Invoke agent
DECLARE analysis = AGENT log_analyst 
    TASK "Analyze payment errors in the last hour and identify patterns"

-- Multi-agent orchestration
CREATE WORKFLOW incident_investigation 
    USING ONE_WORKFLOW AS
BEGIN
    SET analysis = AGENT log_analyst TASK "Investigate errors"
    IF analysis.severity = 'critical' THEN
        AGENT incident_responder TASK "Create P1 incident"
    END IF
END WORKFLOW

-- Direct natural language execution
EXECUTE "Find all payment errors in the last hour and alert the team"

-- Generate procedures from description
GENERATE PROCEDURE "Monitor payment service, alert on 5+ errors in 5 minutes"
    SAVE AS payment_monitor

-- Semantic procedure discovery
DISCOVER PROCEDURES LIKE "handle incidents"
-- Returns: incident_response, alert_handler, on_call_escalation

-- AI-powered recommendations
RECOMMEND PROCEDURES FOR "I need to set up monitoring for a new microservice"

-- Persistent context across sessions
REMEMBER "The payment service has been unstable since deployment v2.3.1"

-- Recall in procedures
CREATE PROCEDURE smart_alert()
BEGIN
    DECLARE context = RECALL "recent incidents for payment service"
    IF context.has_ongoing_incident THEN
        PRINT "Suppressing alert - ongoing incident exists"
    ELSE
        CALL create_incident()
    END IF
END PROCEDURE

-- Minimal syntax, sensible defaults
CREATE JOB cleanup SCHEDULE '@daily' AS
    CALL delete_old_logs()
END JOB
-- Defaults: ENABLED true, TIMEZONE UTC

-- One-liners for common tasks
ALERT ON (FROM logs-* | WHERE level = 'ERROR' | STATS count) > 10 
    SEND SLACK '#alerts'

-- Quick monitoring
MONITOR 'payment-service' EVERY 5 MINUTES 
    ALERT IF error_rate > 0.01

-- Level 1: Simple inline
FROM logs-* | WHERE level = 'ERROR' | STATS count

-- Level 2: Named procedure
CREATE PROCEDURE count_errors()
    RETURN FROM logs-* | WHERE level = 'ERROR' | STATS count
END PROCEDURE

-- Level 3: Parameterized with defaults
CREATE PROCEDURE count_errors(time_range STRING DEFAULT '24h')
    RETURN FROM logs-* 
        | WHERE level = 'ERROR' AND @timestamp > NOW() - @time_range
        | STATS count
END PROCEDURE

-- Level 4: Full-featured with security, observability
@description "Counts errors with alerting"
CREATE PROCEDURE count_errors(time_range STRING, alert_threshold NUMBER)
WITH { TRACING ON, MAX_EXECUTION_TIME = 30 SECONDS }
BEGIN
    TRY
        DECLARE count = FROM logs-* | WHERE ... | STATS count
        IF count > alert_threshold THEN CALL alert_team(count) END IF
        RETURN count
    CATCH
        CALL log_error(@error)
        RAISE
    END TRY
END PROCEDURE

-- Inline help
HELP SLACK_SEND
-- Shows: signature, parameters, examples

-- Interactive examples
EXAMPLE "send slack notification"
-- Returns runnable example code

-- Smart error messages
> CALL SLCK_SEND('#alerts', 'test')
-- Error: Unknown function 'SLCK_SEND'. Did you mean 'SLACK_SEND'?

-- ES|QL is used as-is
DECLARE errors = FROM logs-* | WHERE level = 'ERROR' | LIMIT 100

-- elastic-script ADDS commands that work with ES|QL results
FROM logs-* 
| WHERE level = 'ERROR' 
| INTO my_results                    -- NEW: Store results

FROM logs-* 
| PROCESS WITH analyze_error         -- NEW: Call procedure per row

-- ES|QL in expressions (ES|QL unchanged, elastic-script wraps)
IF (FROM metrics-* | STATS AVG(cpu)) > 80 THEN
    CALL alert_high_cpu()
END IF

-- Build queries programmatically (generates valid ES|QL)
CREATE FUNCTION build_log_query(
    indices STRING DEFAULT 'logs-*',
    level STRING DEFAULT NULL,
    service STRING DEFAULT NULL
) RETURNS QUERY AS
BEGIN
    DECLARE q = QUERY FROM @indices
    IF level IS NOT NULL THEN
        SET q = q | WHERE level = @level
    END IF
    IF service IS NOT NULL THEN
        SET q = q | WHERE service = @service
    END IF
    RETURN q
END FUNCTION

-- Execute composed query
DECLARE results = EXECUTE build_log_query(level := 'ERROR') | LIMIT 100

-- Introspect index mappings (uses ES _mapping API)
DECLARE schema = SCHEMA FOR 'logs-*'
PRINT schema.fields.level.type  -- 'keyword'

-- Validate data against schema
VALIDATE document AGAINST SCHEMA 'logs-*'

-- Continuous query (uses ES async search / PIT)
CREATE STREAM error_monitor AS
    FROM logs-* 
    | WHERE level = 'ERROR'
    | WINDOW TUMBLING 5 MINUTES
    | STATS count BY service
    | EMIT TO PROCEDURE handle_high_errors

-- Uses ES Cluster APIs
CLUSTER HEALTH
CLUSTER SETTINGS SET 'cluster.routing.allocation.enable' = 'all'

-- Uses ES Index APIs  
CREATE INDEX 'my-index' WITH MAPPINGS { ... }
REINDEX FROM 'source-*' TO 'dest'

-- Uses ES Alias APIs
CREATE ALIAS 'current' FOR 'logs-2026.01'
SWAP ALIAS 'current' FROM 'logs-2026.01' TO 'logs-2026.02'

-- Uses ES ILM APIs
CREATE ILM POLICY 'logs-policy' AS {
    HOT { ROLLOVER MAX_SIZE '50GB' MAX_AGE '1d' }
    WARM { MIN_AGE '7d', SHRINK NUMBER_OF_SHARDS 1 }
    DELETE { MIN_AGE '90d' }
}

APPLY ILM POLICY 'logs-policy' TO INDEX TEMPLATE 'logs-template'

-- Uses ES Alerting / Watcher APIs
CREATE ALERT high_error_rate
    TRIGGER SCHEDULE EVERY 5 MINUTES
    INPUT (FROM logs-* | WHERE level = 'ERROR' | STATS count)
    CONDITION result.count > 100
    ACTIONS {
        SLACK '#alerts' MESSAGE 'High error rate: {{count}}'
    }

-- Uses ES ML APIs
CREATE ML JOB error_anomaly
    ANALYSIS_CONFIG { DETECTORS [{ FUNCTION 'count' }], BUCKET_SPAN '15m' }
    DATAFEED (FROM logs-* | WHERE level = 'ERROR')

-- Uses ES Inference APIs
DECLARE sentiment = INFER 'sentiment-model' WITH { text: message }

-- Uses ES Ingest APIs
CREATE INGEST PIPELINE 'enrich-logs' AS {
    GROK FIELD 'message' PATTERNS ['%{TIMESTAMP:ts} %{LOGLEVEL:level}']
    ENRICH POLICY 'geo-lookup' FIELD 'ip' TARGET 'geo'
}

INDEX document INTO 'logs' PIPELINE 'enrich-logs'

-- Uses ES async capabilities
WEBSOCKET NOTIFY '#channel' ON 
    FROM logs-* | WHERE level = 'ERROR' | STATS count > 10

-- Event publishing (to Kafka, etc.)
PUBLISH EVENT 'order.created' TO 'events-topic' WITH order_data

-- Uses K8s API
DECLARE pods = K8S_GET_PODS(namespace := 'production')

FOR pod IN pods LOOP
    IF (FROM logs-* | WHERE k8s.pod.name = pod.name | WHERE level = 'ERROR' | STATS count) > 10 THEN
        CALL K8S_RESTART_POD(pod.name)
    END IF
END LOOP

-- Serverless execution model
CREATE FUNCTION process_webhook(event DOCUMENT)
SERVERLESS
TRIGGER HTTP POST '/webhook/github'
AS
BEGIN
    CALL handle_github_event(event.body)
END FUNCTION

-- Procedures stored as files, deployed via CI/CD
-- File: procedures/incident_response.escript

-- Deploy command
DEPLOY FROM GIT 'main'
PREVIEW DEPLOY 'procedures/*.escript'

-- Automatic OTEL instrumentation
CREATE PROCEDURE my_operation()
WITH OTEL { SERVICE_NAME 'my-service' }
AS
BEGIN
    -- Traces automatically created and sent to APM
END PROCEDURE

-- Custom spans
OTEL_SPAN 'process-batch' BEGIN
    FOR item IN batch LOOP
        OTEL_METRIC 'items.processed' INCREMENT 1
    END LOOP
END OTEL_SPAN

-- HTTP
DECLARE response = HTTP_GET('https://api.example.com/data')

-- gRPC  
DECLARE response = GRPC_CALL('orders.OrderService/GetOrder', { id: '123' })

-- GraphQL
DECLARE response = GRAPHQL_QUERY('https://api/graphql', 
    'query { user(id: "123") { name } }')

-- Message Queues
KAFKA_PRODUCE('topic', message)
RABBITMQ_PUBLISH('exchange', 'routing.key', message)

-- Parquet (for data interchange)
EXPORT (FROM logs-* | LIMIT 10000) TO PARQUET 's3://bucket/logs.parquet'
IMPORT PARQUET 's3://bucket/data.parquet' INTO 'imported-data'

-- CSV, JSON, YAML, XML
DECLARE records = PARSE_CSV(csv_string)
DECLARE config = PARSE_YAML(yaml_string)

-- AWS
DECLARE secret = AWS_SECRETS_MANAGER_GET('my-secret')
CALL AWS_LAMBDA_INVOKE('my-function', payload)

-- GCP
CALL GCP_PUBSUB_PUBLISH('projects/.../topics/my-topic', message)

-- Azure
DECLARE secret = AZURE_KEYVAULT_GET('my-secret')

-- Register custom function (Java)
REGISTER FUNCTION my_custom_function
    CLASS 'com.mycompany.escript.MyFunction'
    JAR 's3://plugins/my-functions.jar'

-- Register from external service
REGISTER FUNCTION external_calc
    HTTP POST 'https://calc.service/compute'

-- Connect to function registry
CONNECT REGISTRY 'https://registry.company.com/escript-functions'

-- Install functions from registry
INSTALL FUNCTIONS FROM 'company/data-quality' VERSION '^2.0'

-- Define middleware
CREATE MIDDLEWARE audit_all
BEFORE EXECUTE ANY PROCEDURE
AS
BEGIN
    CALL log_execution_start(@procedure, @args, @user)
END MIDDLEWARE

AFTER EXECUTE ANY PROCEDURE
AS
BEGIN
    CALL log_execution_end(@procedure, @result, @duration)
END MIDDLEWARE

-- Apply middleware
APPLY MIDDLEWARE audit_all TO ALL PROCEDURES

-- Extend grammar with domain-specific syntax
EXTEND GRAMMAR WITH {
    'MONITOR' service:STRING 'FOR' duration:DURATION 'ALERT' 'IF' condition:EXPRESSION
    => CREATE TRIGGER ... (generates standard elastic-script)
}

-- Use new syntax
MONITOR 'payment-api' FOR 5 MINUTES ALERT IF error_rate > 0.05

CREATE PROCEDURE untrusted_operation()
WITH {
    -- Execution limits
    MAX_EXECUTION_TIME = 60 SECONDS
    MAX_MEMORY = 256MB
    MAX_ES_QUERIES = 100
    MAX_RESULT_SIZE = 10MB

    -- Data access restrictions  
    ALLOWED_INDICES = ['logs-*', 'metrics-*']
    DENIED_INDICES = ['security-*', '.kibana*']

    -- Network restrictions
    NETWORK = DENY
    -- OR
    ALLOWED_HOSTS = ['api.slack.com', 'api.pagerduty.com']

    -- Function restrictions
    DENIED_FUNCTIONS = ['ES_DELETE', 'DROP_INDEX']
}
BEGIN
    -- Runs in sandboxed environment
END PROCEDURE

CREATE POLICY production_safety AS
BEGIN
    REQUIRE MAX_EXECUTION_TIME <= 300 SECONDS
    DENY FUNCTION 'ES_DELETE' UNLESS ROLE IN ('admin')
    DENY INDEX_PATTERN '.security-*'
    REQUIRE APPROVAL FOR FUNCTION 'DROP_INDEX'
END POLICY

APPLY POLICY production_safety TO ROLE 'developer'

-- Reference secrets by name (uses ES keystore or external vault)
CALL HTTP_POST(
    'https://api.pagerduty.com/incidents',
    headers = {'Authorization': SECRET('pagerduty_key')}
)

-- Audit logging
SHOW AUDIT LOG FOR PROCEDURE sensitive_operation
    WHERE user = 'john' AND @timestamp > NOW() - 7 DAYS

-- Role-based access
GRANT EXECUTE ON PROCEDURE analyze_logs TO ROLE 'analyst'
REVOKE EXECUTE ON PROCEDURE delete_data FROM ROLE 'developer'

-- Rich output in Jupyter notebooks
DISPLAY TABLE errors WITH { TITLE 'Errors by Service', CHART 'bar' }
DISPLAY CHART { TYPE 'timeseries', DATA query_results }

SHOW PROCEDURE HISTORY my_procedure
DIFF PROCEDURE my_procedure VERSION 3 WITH VERSION 5
ROLLBACK PROCEDURE my_procedure TO VERSION 3

CREATE PROCEDURE process_orders()
WITH TRACING ON
BEGIN
    -- Traces automatically created and sent to APM
END PROCEDURE

SHOW PROCEDURE METRICS my_procedure
-- Shows: executions, success rate, P50/P99 duration

-- ✅ Proposed: First-class commands
INDEX document INTO 'my-index';
DELETE FROM 'my-index' WHERE _id = '123';
SEARCH 'my-index' QUERY { "match": { "title": "elastic" } };
REFRESH 'my-index';
CREATE INDEX 'new-index' WITH { "mappings": {...} };

-- ❌ Current: Function calls (will be deprecated)
INDEX_DOCUMENT('my-index', document);
ES_DELETE('my-index', '123');

-- ✅ Proposed: Type-namespaced functions
DECLARE len = ARRAY.LENGTH(my_array);
DECLARE keys = DOCUMENT.KEYS(my_doc);
DECLARE upper_name = STRING.UPPER(name);
DECLARE tomorrow = DATE.ADD(today, 1, 'DAY');

-- Extensions follow same pattern
DECLARE pods = K8S.GET_PODS('default');
DECLARE result = OPENAI.COMPLETE(prompt);
DECLARE msg = SLACK.POST_MESSAGE(channel, text);

-- ❌ Current: SNAKE_CASE functions (will be deprecated)
ARRAY_LENGTH(my_array);
DOCUMENT_KEYS(my_doc);
STRING_UPPER(name);

-- ═══════════════════════════════════════════════════════════════
-- CURSOR: Multiple rows → Iterate one at a time
-- ═══════════════════════════════════════════════════════════════
DECLARE logs CURSOR FOR 
    FROM logs-* | WHERE level = 'ERROR' | LIMIT 100;

FOR log IN logs LOOP
    PRINT log.message;
END LOOP

-- ═══════════════════════════════════════════════════════════════
-- ARRAY: Multiple rows → Capture all rows at once
-- ═══════════════════════════════════════════════════════════════
DECLARE logs ARRAY FROM logs-* | WHERE level = 'ERROR' | LIMIT 100;
PRINT 'Found ' || ARRAY.LENGTH(logs) || ' errors';

-- ═══════════════════════════════════════════════════════════════
-- DOCUMENT: Single row with multiple fields → One document
-- ═══════════════════════════════════════════════════════════════
DECLARE stats DOCUMENT FROM logs-* 
    | STATS count = COUNT(*), errors = COUNT(*) WHERE level = 'ERROR';

PRINT 'Total: ' || stats.count || ', Errors: ' || stats.errors;

-- ═══════════════════════════════════════════════════════════════
-- SCALAR: Single row, single column → Direct value
-- ═══════════════════════════════════════════════════════════════
DECLARE total_count NUMBER FROM logs-* | STATS c = COUNT(*);
PRINT 'Total logs: ' || total_count;  -- Directly a number, not {c: 42}

DECLARE latest DATE FROM logs-* | STATS latest = MAX(@timestamp);
PRINT 'Latest: ' || latest;

DECLARE service STRING FROM services 
    | WHERE id = 'svc-001' | KEEP name | LIMIT 1;
PRINT 'Service: ' || service;

-- ERROR: Expected scalar but got multiple rows
DECLARE count NUMBER FROM logs-* | LIMIT 10;  
-- → RuntimeError: Query returned 10 rows, expected 1 for NUMBER

-- ERROR: Expected scalar but got multiple columns  
DECLARE count NUMBER FROM logs-* | STATS a = COUNT(*), b = SUM(size);
-- → RuntimeError: Query returned 2 columns, expected 1 for NUMBER

CREATE APP incident_response
    ROUTE '/incidents'                    -- URL path
    TITLE 'Incident Response Center'      -- Browser title
    ICON 'alert'                          -- EUI icon
    AUTH REQUIRED                         -- Requires login
    ROLES ['sre', 'oncall']               -- Who can access
AS
BEGIN
    -- App state (reactive)
    STATE selected_incident DOCUMENT DEFAULT NULL
    STATE severity_filter STRING DEFAULT 'all'

    -- UI components
    RENDER HEADER 'Incident Response' WITH {
        ACTIONS [
            BUTTON 'Create' ON_CLICK => MODAL create_modal
        ]
    }

    RENDER STATS { ... }
    RENDER TABLE { ... }
    RENDER CHART { ... }
    RENDER FORM { ... }
    RENDER DASHBOARD 'service-health' { ... }

END APP