SQL transform examples

Basic usage

Identity query


SELECT * FROM logs

Return all log events unchanged.

Input:


{
  "id": "0H19GZK97FTKS",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "State backend is set to heap memory",
  "tags": {
    "service": ["grepr-query"],
    "host": ["ip-10-12-4-129.ec2.internal"]
  },
  "attributes": {
    "process": {
      "thread": {
        "name": "thread-0"
      }
    }
  }
}

Output:


{
  "id": "0H19GZK97FTKS",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "State backend is set to heap memory",
  "tags": {
    "service": ["grepr-query"],
    "host": ["ip-10-12-4-129.ec2.internal"]
  },
  "attributes": {
    "process": {
      "thread": {
        "name": "thread-0"
      }
    }
  }
}

Simple filtering


SELECT * FROM logs WHERE severity < 5

Filter events to include only those with severity less than 5 (TRACE level).

Input:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 1,
    "message": "Debug trace message",
    "tags": {"service": ["api"]},
    "attributes": {}
  },
  {
    "id": "evt2",
    "eventtimestamp": 1724214074063,
    "receivedtimestamp": 1724214074189,
    "severity": 9,
    "message": "Info level message",
    "tags": {"service": ["api"]},
    "attributes": {}
  }
]

Output:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 1,
    "message": "Debug trace message",
    "tags": {"service": ["api"]},
    "attributes": {}
  }
]

Message transformation


SELECT UPPER(message) as message, * FROM logs

Convert all log messages to uppercase.

When transforming existing fields to override the original value, the transformed column must come before the * wildcard in your SELECT clause. For example: SELECT UPPER(message) as message, * FROM logs

Adding computed fields


SELECT *,
  UPPER(message) as upper_message,
  CHAR_LENGTH(message) as message_length,
  severity * 10 as severity_scaled
FROM logs

Add three new computed fields to the event attributes.

Input:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Processing request",
  "tags": {"service": ["api"]},
  "attributes": {
    "userId": "12345"
  }
}

Output:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Processing request",
  "tags": {"service": ["api"]},
  "attributes": {
    "userId": "12345",
    "upper_message": "PROCESSING REQUEST",
    "message_length": 18,
    "severity_scaled": 90
  }
}

Multi-statement examples

Basic multi-statement transform

The following example shows how to create a multi-statement configuration that filters and transforms data. This example:

Creates a VIEW that selects logs with severity >= 13.
Produces an OUTPUT that transforms the filtered logs by uppercasing messages and adding a processing status.


{
  "type": "sql-operation",
  "inputs": {
    "logs": "LOG_EVENT"
  },
  "statements": [
    {
      "type": "sql_view",
      "tableName": "high_severity_logs",
      "sqlQuery": "SELECT * FROM logs WHERE severity >= 13"
    },
    {
      "type": "sql_output",
      "outputName": "processed_logs",
      "outputType": "LOG_EVENT",
      "sqlQuery": "SELECT *, UPPER(message) as message, 'processed' as processing_status FROM high_severity_logs"
    }
  ],
  "watermarkDelay": "PT5S",
  "globalStateTtl": "PT2M"
}

Complex multi-statement pipeline

This example:

Filters logs that have user IDs and extracts user and request IDs.
Creates a summary of user activity counts.
Enriches original logs with user activity summaries.
Creates synthetic log events summarizing user activity.


{
  "type": "sql-operation",
  "inputs": {
    "logs": "LOG_EVENT"
  },
  "statements": [
    {
      "type": "sql_view",
      "tableName": "parsed_logs",
      "sqlQuery": "SELECT *, VARIANT_VALUE(attributes, '$.userId', 'BIGINT') as user_id, VARIANT_VALUE(attributes, '$.requestId', 'BIGINT') as request_id FROM logs WHERE VARIANT_VALUE_CONTAINS(attributes, 'userId', false)"
    },
    {
      "type": "sql_view",
      "tableName": "user_activity_summary",
      "sqlQuery": "SELECT user_id, COUNT(*) as activity_count, MAX(eventtimestamp) as last_activity FROM parsed_logs GROUP BY user_id"
    },
    {
      "type": "sql_output",
      "outputName": "enriched_logs",
      "outputType": "LOG_EVENT",
      "sqlQuery": "SELECT p.*, s.activity_count, s.last_activity FROM parsed_logs p LEFT JOIN user_activity_summary s ON p.user_id = s.user_id"
    },
    {
      "type": "sql_output",
      "outputName": "user_summaries",
      "outputType": "LOG_EVENT",
      "sqlQuery": "SELECT user_id as id, last_activity as eventtimestamp, CURRENT_TIMESTAMP as receivedtimestamp, CONCAT('User ', user_id, ' had ', CAST(activity_count AS STRING), ' activities') as message, 9 as severity, 'user_activity' as `tags.summary_type`, activity_count FROM user_activity_summary"
    }
  ]
}

Advanced examples

Content-based filtering and enrichment


SELECT *,  -- Select all original fields
  -- Add a category based on message content
  CASE
    WHEN message LIKE '%ERROR%' THEN 'error_detected'  -- If message contains "ERROR"
    WHEN message LIKE '%WARN%' THEN 'warning_detected'  -- If message contains "WARN"
    ELSE 'normal'  -- For all other messages
  END as log_category,
  -- Replace any IP-like address with placeholder text
  REGEXP_REPLACE(message, '\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}', '<redacted>') as sanitized_message
FROM logs
WHERE severity >= 13  -- Only process WARN level and above (13-24)

Input:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 17,
    "message": "ERROR: Database connection to IP 192.1.23.32 failed",
    "tags": {"service": ["database"]},
    "attributes": {}
  },
  {
    "id": "evt2",
    "eventtimestamp": 1724214074063,
    "receivedtimestamp": 1724214074189,
    "severity": 9,
    "message": "INFO: Processing complete",
    "tags": {"service": ["api"]},
    "attributes": {}
  }
]

Output:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 17,
    "message": "ERROR: Database connection to IP 192.1.23.32 failed",
    "tags": {"service": ["database"]},
    "attributes": {
      "log_category": "error_detected",
      "sanitized_message": "ERROR: Database connection to ip redacted failed",
    }
  }
]

Process log event attributes


SELECT *,  -- Select all original fields
  -- Extract thread name from a nested path
  VARIANT_VALUE(attributes, '$.process.thread.name', 'STRING') as thread_name,
  -- Extract status field from attributes
  VARIANT_VALUE(attributes, '$.status', 'STRING') as status_code
FROM logs
-- Only process logs that have a 'process' field in
WHERE
  VARIANT_VALUE_CONTAINS(attributes, 'process', true)

Input:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 9,
    "message": "Request processed",
    "tags": {"service": ["api"]},
    "attributes": {
      "process": {
        "thread": {
          "name": "worker-1"
        }
      },
      "status": "success"
    }
  },
  {
    "id": "evt2",
    "eventtimestamp": 1724214074063,
    "receivedtimestamp": 1724214074189,
    "severity": 9,
    "message": "Simple log message",
    "tags": {"service": ["web"]},
    "attributes": {
      "simple": true
    }
  }
]

Output:


[
  {
    "id": "evt1",
    "eventtimestamp": 1724214074062,
    "receivedtimestamp": 1724214074188,
    "severity": 9,
    "message": "Request processed",
    "tags": {"service": ["api"]},
    "attributes": {
      "process": {"thread": {"name": "worker-1"}},
      "status": "success",
      "thread_name": "worker-1",
      "status_code": "success"
    }
  }
]

Time-Based Analysis


SELECT *,  -- Select all original fields
  -- Extract the hour (0-23) from the event timestamp
  EXTRACT(HOUR FROM eventtimestamp) as event_hour,
  -- Extract the day of the month (1-31) from the event timestamp
  EXTRACT(DAY FROM eventtimestamp) as event_day
FROM logs

Input:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Processing request",
  "tags": {"service": ["api"]},
  "attributes": {}
}

Output:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Processing request",
  "tags": {"service": ["api"]},
  "attributes": {
    "event_hour": 4,
    "event_day": 21
  }
}

Multi-field transformations


SELECT
  -- Explicitly select these core fields
  id,
  eventtimestamp,
  receivedtimestamp,
  tags,
  attributes,
  -- Transform the message by adding a suffix
  CONCAT(message, ' [PROCESSED]') as message,
  -- Increase severity by 1 (e.g., INFO becomes WARN)
  severity + 1 as severity,
  -- Store the original message in uppercase as a computed field
  UPPER(message) as original_uppercase,
  -- Store the original severity scaled by 100 as a computed field
  severity * 100 as original_severity_scaled
FROM logs

Input:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "user authenticated",
  "tags": {"service": ["auth"]},
  "attributes": {}
}

Output:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 10,
  "message": "user authenticated [PROCESSED]",
  "tags": {"service": ["auth"]},
  "attributes": {
    "original_uppercase": "USER AUTHENTICATED",
    "original_severity_scaled": 900
  }
}

Add tags to a log event

The SQL transform provides a special column-naming syntax to add tags to output log events. The values you assign to a tag can be either scalar or ARRAY types.

Adding a tag that matches an existing tag causes the existing tag to be overwritten. Use ARRAY_APPEND to extend an existing set of tag values as shown in the example below.

To learn more, see Add tags to a LOG_EVENT using a special column syntax.

Add scalar tag values


SELECT *,
  'production' as `tags.environment`,
  tags['service'][1] as `tags.service`,
  CASE WHEN severity >= 17 THEN 'high' ELSE 'normal' END as `tags.priority`
FROM logs

Input:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 17,
  "message": "Database connection failed",
  "tags": {"service": ["database", "mysql"]},
  "attributes": {}
}

Output:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 17,
  "message": "Database connection failed",
  "tags": {
    "service": ["database"],
    "environment": ["production"],
    "priority": ["high"]
  },
  "attributes": {}
}

Add multiple tag values with arrays

This example adds multiple values to a new tag, adds additional values to an existing tag, and adds a new tag using a team name extracted from the input event:


SELECT *,
  ARRAY['web', 'api', 'frontend'] as `tags.components`,
  -- extends the existing 'service' tag with an additional value
  ARRAY_APPEND(tags['service'], 'all_services') as `tags.service`
  REGEXP_EXTRACT(message, 'team=([a-z]+)', 1) as `tags.team`
FROM logs
WHERE message LIKE '%team=%'

Remove tags and attributes from log events

Use MAP_REMOVE_KEYS to remove log event tag keys and VARIANT_REMOVE_KEYS to remove fields from log event attributes. In a LOG_EVENT output, you must place the replacement tags and attributes columns before *. If a replacement column comes after *, the original column from * takes precedence and the removal has no effect.


SELECT
  MAP_REMOVE_KEYS(tags, 'container_id', 'pod_uid') AS tags,
  VARIANT_REMOVE_KEYS(
    attributes,
    '$.kubernetes.pod.uid',
    '$["datadog.trace_payload.hostname"]'
  ) AS attributes,
  *
FROM logs

Input:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Request processed",
  "tags": {
    "service": ["api"],
    "container_id": ["container-123"],
    "pod_uid": ["pod-123"]
  },
  "attributes": {
    "kubernetes": {
      "namespace": "payments",
      "pod": {
        "name": "checkout-0",
        "uid": "pod-123"
      }
    },
    "datadog.trace_payload.hostname": "worker-node-1"
  }
}

Output:


{
  "id": "evt1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "Request processed",
  "tags": {
    "service": ["api"]
  },
  "attributes": {
    "kubernetes": {
      "namespace": "payments",
      "pod": {
        "name": "checkout-0"
      }
    }
  }
}

Examples from production

Security log processing


SELECT *,
  CASE
    WHEN message LIKE '%authentication failed%' THEN 'auth_failure'
    WHEN message LIKE '%unauthorized access%' THEN 'unauthorized'
    WHEN message LIKE '%suspicious activity%' THEN 'suspicious'
    ELSE 'normal'
  END as security_category,
  REGEXP_REPLACE(message, 'user=[^\\\\s]+', 'user=REDACTED') as sanitized_message
FROM logs
WHERE severity >= 17 -- ERROR level and above

Input:


{
  "id": "sec1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 17,
  "message": "authentication failed for user=johndoe from IP 192.168.1.100",
  "tags": {"service": ["auth"], "env": ["prod"]},
  "attributes": {
    "ip": "192.168.1.100",
    "attempt": 3
  }
}

Output:


{
  "id": "sec1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 17,
  "message": "authentication failed for user=johndoe from IP 192.168.1.100",
  "tags": {"service": ["auth"], "env": ["prod"]},
  "attributes": {
    "ip": "192.168.1.100",
    "attempt": 3,
    "security_category": "auth_failure",
    "sanitized_message": "authentication failed for user=REDACTED from IP 192.168.1.100"
  }
}

Application performance monitoring


SELECT *,
  CAST(REGEXP_EXTRACT(message, 'response_time=(\\\\d+)ms', 1) AS BIGINT) as response_time_ms,
  CASE
    WHEN message LIKE '%response_time=%ms' AND CAST(REGEXP_EXTRACT(message, 'response_time=(\\\\d+)ms', 1) AS BIGINT) > 1000 THEN 'slow'
    ELSE 'normal'
  END as performance_category
FROM logs
WHERE message LIKE '%response_time=%'

Input:


{
  "id": "perf1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "API request completed with response_time=1250ms for endpoint /api/users",
  "tags": {"service": ["api"], "endpoint": ["/api/users"]},
  "attributes": {
    "method": "GET",
    "status_code": 200,
    "user_id": "12345"
  }
}

Output:


{
  "id": "perf1",
  "eventtimestamp": 1724214074062,
  "receivedtimestamp": 1724214074188,
  "severity": 9,
  "message": "API request completed with response_time=1250ms for endpoint /api/users",
  "tags": {"service": ["api"], "endpoint": ["/api/users"]},
  "attributes": {
    "method": "GET",
    "status_code": 200,
    "user_id": "12345",
    "response_time_ms": 1250,
    "performance_category": "slow"
  }
}

COMPLETE_SPAN transform examples

COMPLETE_SPAN transforms apply to trace data from Datadog Agent and OTLP sources.

Change the Datadog hostname for spans

This example sets datadog.trace_payload.hostname to a custom value so you can rename hosts in the Datadog UI, while keeping the original host.name intact and saving it to span.attributes['grepr.host'].

**SQL — Modify resource attributes and span attributes **


SELECT *,
  -- Updates an existing resource attribute
  CASE
    WHEN VARIANT_VALUE(resource.attributes, '$["host.name"]', 'STRING') IS NOT NULL
    THEN VARIANT_BUILD(
      '$["datadog.trace_payload.hostname"]', 'apm-host',
      '$["host.name"]', 'dev-apm-host'
    )
    ELSE VARIANT_BUILD(
      '$["datadog.trace_payload.hostname"]', 'apm-host'
    )
  END AS `resource.attributes`,
  -- Adds a new attribute to the span
  VARIANT_BUILD(
    '$["grepr.host"]',
    COALESCE(
      VARIANT_VALUE(resource.attributes, '$["host.name"]', 'STRING'),
      VARIANT_VALUE(resource.attributes, '$["datadog.trace_payload.hostname"]', 'STRING')
    )
  ) AS `span.attributes`
FROM spans

Replace <new-hostname> with your normalized hostname value.

Omit nullable span metadata fields

If a COMPLETE_SPAN query omits nullable top-level metadata fields, Grepr sets the missing output fields to NULL. This allows SQL transforms written before new nullable metadata fields were added to keep working. When authoring new transforms, prefer SELECT * or include all current fields unless you intentionally want to clear the omitted metadata.


SELECT
  receivedtimestamp,
  eventtimestamp,
  servicename,
  resource,
  instrumentationscope,
  span
FROM spans

Do not use this shortcut for nested resource, instrumentationscope, or span rows that you rebuild with ROW(...). If you replace one of those nested rows, include every nested field in the current order. Grepr validates nested row shape before the pipeline starts.

Remove resource and span attributes

Use VARIANT_REMOVE_KEYS to remove fields from COMPLETE_SPAN resource attributes and span attributes. When you replace nested COMPLETE_SPAN fields, explicitly select each output column so that the replacement resource and span fields keep their expected names.


SELECT
  receivedtimestamp,
  eventtimestamp,
  servicename,
  ROW(
    VARIANT_REMOVE_KEYS(
      resource.attributes,
      '$["cloud.account.id"]',
      '$.kubernetes.pod.uid'
    ),
    resource.dropped_attributes_count,
    resource.entity_refs
  ) AS resource,
  instrumentationscope,
  ROW(
    span.traceIdHigh,
    span.traceIdLow,
    span.spanId,
    span.traceState,
    span.parentSpanId,
    span.operationName,
    span.spanKind,
    span.startTimeNanos,
    span.endTimeNanos,
    VARIANT_REMOVE_KEYS(
      span.attributes,
      '$["http.request.header.authorization"]',
      '$["db.statement"]'
    ),
    span.droppedAttributesCount,
    span.events,
    span.droppedEventsCount,
    span.links,
    span.droppedLinksCount,
    span.status,
    span.flags,
    span.spanKindByte
  ) AS span,
  tracesignature,
  duration,
  haserror,
  percentilebucket,
  samplingtype,
  crit_path_ns,
  crit_path_pct
FROM spans

Frequently Asked Questions

What does the identity query do and when is it useful?

The identity query SELECT * FROM logs returns all log events unchanged. It's useful as a base template for building more complex transforms, for testing pipeline configuration, or as a placeholder before adding transformation logic.

How do I filter logs by severity in SQL?

Use a WHERE clause with severity comparisons. For example, SELECT * FROM logs WHERE severity < 5 filters for TRACE level logs, or SELECT * FROM logs WHERE severity >= 13 filters for WARN and above. Severity follows OpenTelemetry convention with values 1-24.

What is the warning about transforming fields like message or severity?

When you want to override an original field (like transforming message to uppercase), the transformed column must appear BEFORE the * wildcard in your SELECT clause. For example: SELECT UPPER(message) as message, * FROM logs. Otherwise, the original value takes precedence.

How do I add new computed fields to log events?

Use SELECT * with additional computed columns. For example: SELECT *, UPPER(message) as upper_message, CHAR_LENGTH(message) as msg_length FROM logs. The computed fields are automatically added to the event's attributes map in the output.

What is a multi-statement SQL transform?

Multi-statement transforms allow you to create intermediate views and multiple outputs. You define sql_view statements to create temporary tables, then sql_output statements to transform and output results. This enables complex multi-step transformations with intermediate processing.

How do I extract and transform JSON from attributes?

Use JSON_VALUE to extract scalar values and JSON_QUERY for objects. For example: JSON_VALUE(attributes, '$.userId' RETURNING STRING) as user_id extracts a user ID, or JSON_QUERY(attributes, '$.process') as process_info extracts nested objects. Use JSON_EXISTS to check if paths exist.

How can I add tags to log events in SQL?

Use the special column-naming syntax with backticks: SELECT *, 'production' as `tags.environment` FROM logs adds a scalar tag. Use ARRAY for multiple values: ARRAY['web', 'api'] as `tags.components`. Use ARRAY_APPEND to extend existing tags: ARRAY_APPEND(tags['service'], 'billing') as `tags.service`.

How do I sanitize or redact sensitive data from logs?

Use REGEXP_REPLACE to remove or mask sensitive information. For example: REGEXP_REPLACE(message, 'user=[^\s]+', 'user=REDACTED') redacts usernames, or REGEXP_REPLACE(message, '\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '<redacted>') masks IP addresses.

How do I extract numeric values from log messages?

Use REGEXP_EXTRACT to find patterns in messages, then CAST to convert to numbers. For example: CAST(REGEXP_EXTRACT(message, 'response_time=(\d+)ms', 1) AS BIGINT) as response_time_ms extracts a response time value and converts to a long integer for calculations.

How do I process logs based on time windows in SQL?

Use windowing table-valued functions: TUMBLE for fixed-size windows, HOP for overlapping windows, SESSION for activity-based grouping, or CUMULATE for expanding windows. For example: FROM TABLE(TUMBLE(TABLE logs, DESCRIPTOR(eventtimestamp), INTERVAL '5' MINUTES)) creates 5-minute windows. Always include window_time in GROUP BY.