local-schema.sql

-- This file is generated by scripts/generate-clickhouse-schema-sql.ts
-- Do not edit manually. Run `bun run clickhouse:schema` to regenerate.
-- projectRevision: 2cc5757c35a332dd84455e14103a33295893e68fdca89de9bf4ccdbbf33dbf8c
-- localSchemaVersion: 36

CREATE TABLE IF NOT EXISTS ai_crawler_requests (
    OrgId LowCardinality(String),
    Timestamp DateTime64(9),
    TraceId String,
    ServiceName LowCardinality(String),
    Crawler LowCardinality(String),
    Host LowCardinality(String),
    Path String,
    HttpStatus UInt16
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, Timestamp, TraceId)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS ai_trace_index (
    OrgId LowCardinality(String),
    Timestamp DateTime64(9),
    TraceId String,
    SessionId String,
    VendorId LowCardinality(String),
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    Model LowCardinality(String),
    AgentName LowCardinality(String),
    ToolName LowCardinality(String),
    SpanId String,
    ParentSpanId String,
    Duration UInt64,
    IsError UInt8,
    IsLlmCall UInt8,
    IsToolCall UInt8,
    Tokens Float64,
    Cost Float64,
    ResponseId String,
    VendorVersion LowCardinality(String),
    InputTokens Float64,
    CacheReadTokens Float64,
    CacheWriteTokens Float64,
    OutputTokens Float64,
    ReasoningTokens Float64,
    ErrorType LowCardinality(String),
    StatusMessage String,
    ToolDescription String,
    FailedToolCallResult String,
    ErrorFingerprint UInt64
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, Timestamp, TraceId)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS alert_checks (
    OrgId LowCardinality(String),
    RuleId String,
    GroupKey String,
    Timestamp DateTime64(3),
    Status LowCardinality(String),
    SignalType LowCardinality(String),
    Comparator LowCardinality(String),
    Threshold Float64,
    ObservedValue Nullable(Float64),
    SampleCount UInt32,
    WindowMinutes UInt16,
    WindowStart DateTime64(3),
    WindowEnd DateTime64(3),
    ConsecutiveBreaches UInt16,
    ConsecutiveHealthy UInt16,
    IncidentId Nullable(String),
    IncidentTransition LowCardinality(String),
    EvaluationDurationMs UInt32,
    ErrorMessage Nullable(String),
    ErrorCategory LowCardinality(String),
    SkipReason LowCardinality(String) DEFAULT '',
    Level LowCardinality(String) DEFAULT ''
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, RuleId, GroupKey, Timestamp)
TTL toDate(Timestamp) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS attribute_keys_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    AttributeKey LowCardinality(String),
    AttributeScope LowCardinality(String),
    UsageCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, AttributeScope, Hour, AttributeKey)
TTL toDate(Hour) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS attribute_values_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    AttributeKey LowCardinality(String),
    AttributeValue String,
    AttributeScope LowCardinality(String),
    UsageCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, AttributeScope, AttributeKey, Hour, AttributeValue)
TTL toDate(Hour) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS audit_log (
    OrgId LowCardinality(String),
    Id String,
    OccurredAt DateTime64(3),
    RecordedAt DateTime64(3),
    ActorType LowCardinality(String),
    UserId String,
    ApiKeyId String,
    ActorId String,
    ActorLabel String,
    AffectedUserId String,
    Source LowCardinality(String),
    Action LowCardinality(String),
    Outcome LowCardinality(String),
    DenialReason String,
    ResourceType LowCardinality(String),
    ResourceId String,
    ChangedFields Array(String),
    Changes String,
    Metadata String,
    RequestId String,
    OriginIp String,
    OriginCountry LowCardinality(String)
)
ENGINE = ReplacingMergeTree
PARTITION BY toYYYYMM(OccurredAt)
ORDER BY (OrgId, OccurredAt, Id)
TTL toDate(OccurredAt) + INTERVAL 2190 DAY;

CREATE TABLE IF NOT EXISTS error_events (
    OrgId LowCardinality(String),
    Timestamp DateTime,
    TraceId String,
    SpanId String,
    ParentSpanId String DEFAULT '__unset__',
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    ExceptionType LowCardinality(String),
    ExceptionMessage String,
    ExceptionStacktrace String,
    TopFrame String,
    FingerprintHash UInt64,
    StatusMessage String,
    Duration UInt64,
    ErrorLabel String,
    ServiceVersion LowCardinality(String)
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, FingerprintHash, Timestamp)
TTL toDate(Timestamp) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS error_events_by_time (
    OrgId LowCardinality(String),
    Timestamp DateTime,
    TraceId String,
    SpanId String,
    ParentSpanId String DEFAULT '__unset__',
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    ExceptionType LowCardinality(String),
    ExceptionMessage String,
    ExceptionStacktrace String,
    TopFrame String,
    FingerprintHash UInt64,
    StatusMessage String,
    Duration UInt64,
    ErrorLabel String,
    ServiceVersion LowCardinality(String)
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, Timestamp, FingerprintHash)
TTL toDate(Timestamp) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS error_fingerprints_minutely (
    OrgId LowCardinality(String),
    Minute DateTime,
    FingerprintHash UInt64,
    ServiceName SimpleAggregateFunction(anyLast, String),
    ExceptionType SimpleAggregateFunction(anyLast, String),
    ExceptionMessage SimpleAggregateFunction(anyLast, String),
    ErrorLabel SimpleAggregateFunction(anyLast, String),
    TopFrame SimpleAggregateFunction(anyLast, String),
    OccurrenceCount SimpleAggregateFunction(sum, UInt64),
    FirstSeen SimpleAggregateFunction(min, DateTime),
    LastSeen SimpleAggregateFunction(max, DateTime),
    ServiceVersions SimpleAggregateFunction(groupUniqArrayArray, Array(String))
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(Minute)
ORDER BY (OrgId, Minute, FingerprintHash)
TTL toDate(Minute) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS identity_links (
    OrgId LowCardinality(String),
    VisitorId String,
    UserId String,
    FirstSeen SimpleAggregateFunction(min, DateTime64(9))
)
ENGINE = AggregatingMergeTree
PARTITION BY tuple()
ORDER BY (OrgId, VisitorId, UserId)
TTL toDate(FirstSeen) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS logs (
    OrgId LowCardinality(String),
    Timestamp DateTime64(9),
    TimestampTime DateTime,
    TraceId String,
    SpanId String,
    TraceFlags UInt8,
    SeverityText LowCardinality(String),
    SeverityNumber UInt8,
    ServiceName LowCardinality(String),
    Body String,
    ResourceSchemaUrl String,
    ResourceAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    LogAttributes Map(LowCardinality(String), String),
    ResourceAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(ResourceAttributes), mapValues(ResourceAttributes)),
    ScopeAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(ScopeAttributes), mapValues(ScopeAttributes)),
    LogAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(LogAttributes), mapValues(LogAttributes)),
    INDEX idx_trace_id TraceId TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_resource_attr_keys mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_resource_attr_vals mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_scope_attr_keys mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_scope_attr_vals mapValues(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_log_attr_keys mapKeys(LogAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_log_attr_vals mapValues(LogAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_lower_body lower(Body) TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 8
)
ENGINE = MergeTree
PARTITION BY toDate(TimestampTime)
ORDER BY (OrgId, toStartOfFiveMinutes(Timestamp), ServiceName, Timestamp)
TTL toDate(TimestampTime) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS logs_aggregates_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    SeverityText LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    Count SimpleAggregateFunction(sum, UInt64),
    SizeBytes SimpleAggregateFunction(sum, UInt64),
    ServiceNamespace LowCardinality(String),
    INDEX idx_service_namespace ServiceNamespace TYPE set(1000) GRANULARITY 4
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, ServiceName, SeverityText, DeploymentEnv, ServiceNamespace)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS metric_catalog (
    OrgId LowCardinality(String),
    Hour DateTime,
    MetricType LowCardinality(String),
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    MetricDescription SimpleAggregateFunction(anyLast, String),
    MetricUnit SimpleAggregateFunction(anyLast, String),
    IsMonotonic SimpleAggregateFunction(anyLast, UInt8),
    DataPointCount SimpleAggregateFunction(sum, UInt64),
    FirstSeen SimpleAggregateFunction(min, DateTime),
    LastSeen SimpleAggregateFunction(max, DateTime)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, MetricType, ServiceName, MetricName, Hour)
TTL toDate(Hour) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS metrics_exponential_histogram (
    OrgId LowCardinality(String),
    ResourceAttributes Map(LowCardinality(String), String),
    ResourceSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    MetricDescription LowCardinality(String),
    MetricUnit LowCardinality(String),
    Attributes Map(LowCardinality(String), String),
    StartTimeUnix DateTime64(9),
    TimeUnix DateTime64(9),
    Count UInt64,
    Sum Float64,
    Scale Int32,
    ZeroCount UInt64,
    PositiveOffset Int32,
    PositiveBucketCounts Array(UInt64),
    NegativeOffset Int32,
    NegativeBucketCounts Array(UInt64),
    ExemplarsTraceId Array(String),
    ExemplarsSpanId Array(String),
    ExemplarsTimestamp Array(DateTime64(9)),
    ExemplarsValue Array(Float64),
    ExemplarsFilteredAttributes Array(Map(LowCardinality(String), String)),
    Flags UInt32,
    Min Nullable(Float64),
    Max Nullable(Float64),
    AggregationTemporality Int32
)
ENGINE = MergeTree
PARTITION BY toDate(TimeUnix)
ORDER BY (OrgId, ServiceName, MetricName, Attributes, toUnixTimestamp64Nano(TimeUnix))
TTL toDate(TimeUnix) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS metrics_gauge (
    OrgId LowCardinality(String),
    ResourceAttributes Map(LowCardinality(String), String),
    ResourceSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    MetricDescription LowCardinality(String),
    MetricUnit LowCardinality(String),
    Attributes Map(LowCardinality(String), String),
    StartTimeUnix DateTime64(9),
    TimeUnix DateTime64(9),
    Value Float64,
    Flags UInt32,
    ExemplarsTraceId Array(String),
    ExemplarsSpanId Array(String),
    ExemplarsTimestamp Array(DateTime64(9)),
    ExemplarsValue Array(Float64),
    ExemplarsFilteredAttributes Array(Map(LowCardinality(String), String))
)
ENGINE = MergeTree
PARTITION BY toDate(TimeUnix)
ORDER BY (OrgId, ServiceName, MetricName, Attributes, toUnixTimestamp64Nano(TimeUnix))
TTL toDate(TimeUnix) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS metrics_histogram (
    OrgId LowCardinality(String),
    ResourceAttributes Map(LowCardinality(String), String),
    ResourceSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    MetricDescription LowCardinality(String),
    MetricUnit LowCardinality(String),
    Attributes Map(LowCardinality(String), String),
    StartTimeUnix DateTime64(9),
    TimeUnix DateTime64(9),
    Count UInt64,
    Sum Float64,
    BucketCounts Array(UInt64),
    ExplicitBounds Array(Float64),
    ExemplarsTraceId Array(String),
    ExemplarsSpanId Array(String),
    ExemplarsTimestamp Array(DateTime64(9)),
    ExemplarsValue Array(Float64),
    ExemplarsFilteredAttributes Array(Map(LowCardinality(String), String)),
    Flags UInt32,
    Min Nullable(Float64),
    Max Nullable(Float64),
    AggregationTemporality Int32
)
ENGINE = MergeTree
PARTITION BY toDate(TimeUnix)
ORDER BY (OrgId, ServiceName, MetricName, Attributes, toUnixTimestamp64Nano(TimeUnix))
TTL toDate(TimeUnix) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS metrics_sum (
    OrgId LowCardinality(String),
    ResourceAttributes Map(LowCardinality(String), String),
    ResourceSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    MetricDescription LowCardinality(String),
    MetricUnit LowCardinality(String),
    Attributes Map(LowCardinality(String), String),
    StartTimeUnix DateTime64(9),
    TimeUnix DateTime64(9),
    Value Float64,
    Flags UInt32,
    ExemplarsTraceId Array(String),
    ExemplarsSpanId Array(String),
    ExemplarsTimestamp Array(DateTime64(9)),
    ExemplarsValue Array(Float64),
    ExemplarsFilteredAttributes Array(Map(LowCardinality(String), String)),
    AggregationTemporality Int32,
    IsMonotonic Bool
)
ENGINE = MergeTree
PARTITION BY toDate(TimeUnix)
ORDER BY (OrgId, ServiceName, MetricName, Attributes, toUnixTimestamp64Nano(TimeUnix))
TTL toDate(TimeUnix) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS product_events (
    OrgId LowCardinality(String),
    Timestamp DateTime64(9),
    Source LowCardinality(String) DEFAULT 'browser',
    SessionId String DEFAULT '',
    Seq UInt32 DEFAULT 0,
    VisitorId String DEFAULT '',
    UserId String DEFAULT '',
    GroupId String DEFAULT '',
    Kind LowCardinality(String),
    EventName String,
    Host LowCardinality(String) DEFAULT '',
    PagePath String DEFAULT '',
    Url String DEFAULT '',
    ServiceName LowCardinality(String) DEFAULT '',
    Attributes Map(String, String) DEFAULT map(),
    TraceId String DEFAULT '',
    SpanId String DEFAULT '',
    INDEX idx_event_name EventName TYPE set(64) GRANULARITY 4,
    INDEX idx_user_id UserId TYPE bloom_filter GRANULARITY 4,
    INDEX idx_trace_id TraceId TYPE bloom_filter GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, Timestamp, VisitorId, SessionId, Seq)
TTL toDate(Timestamp) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_address_resolutions_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    SourceService LowCardinality(String),
    ParentServerAddress String,
    ResolvedTargetService LowCardinality(String),
    DeploymentEnv LowCardinality(String)
)
ENGINE = ReplacingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, DeploymentEnv, SourceService, ParentServerAddress, ResolvedTargetService)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_external_edges_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    TargetType LowCardinality(String),
    TargetSystem LowCardinality(String),
    TargetName String,
    DeploymentEnv LowCardinality(String),
    CallCount SimpleAggregateFunction(sum, UInt64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    DurationSumMs SimpleAggregateFunction(sum, Float64),
    MaxDurationMs SimpleAggregateFunction(max, Float64),
    SampleRateSum SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigestWeighted(0.5, 0.95), UInt64, UInt32)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, DeploymentEnv, ServiceName, TargetType, TargetSystem, TargetName)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_map_children (
    OrgId LowCardinality(String),
    Timestamp DateTime,
    TraceId String,
    ParentSpanId String,
    ServiceName LowCardinality(String),
    SpanKind LowCardinality(String),
    Duration UInt64,
    StatusCode LowCardinality(String),
    TraceState String,
    DeploymentEnv LowCardinality(String)
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, TraceId, ParentSpanId, Timestamp)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS service_map_db_edges_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    DbSystem LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    CallCount SimpleAggregateFunction(sum, UInt64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    DurationSumMs SimpleAggregateFunction(sum, Float64),
    MaxDurationMs SimpleAggregateFunction(max, Float64),
    SampledSpanCount SimpleAggregateFunction(sum, UInt64),
    UnsampledSpanCount SimpleAggregateFunction(sum, UInt64),
    SampleRateSum SimpleAggregateFunction(sum, Float64),
    DbNamespace LowCardinality(String),
    DurationQuantiles AggregateFunction(quantilesTDigestWeighted(0.5, 0.95), UInt64, UInt32)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, DeploymentEnv, ServiceName, DbSystem, DbNamespace)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_map_db_query_shapes_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    DbSystem LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    QueryKey String,
    QueryLabel SimpleAggregateFunction(any, String),
    SampleStatement SimpleAggregateFunction(any, String),
    CallCount SimpleAggregateFunction(sum, UInt64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    EstimatedCount SimpleAggregateFunction(sum, Float64),
    EstimatedErrorCount SimpleAggregateFunction(sum, Float64),
    WeightedDurationSumMs SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigestWeighted(0.5, 0.95), UInt64, UInt32),
    DbNamespace LowCardinality(String)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, DeploymentEnv, ServiceName, DbSystem, DbNamespace, QueryKey)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_map_edges_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    SourceService LowCardinality(String),
    TargetService String,
    DeploymentEnv LowCardinality(String),
    CallCount SimpleAggregateFunction(sum, UInt64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    DurationSumMs SimpleAggregateFunction(sum, Float64),
    MaxDurationMs SimpleAggregateFunction(max, Float64),
    SampledSpanCount SimpleAggregateFunction(sum, UInt64),
    UnsampledSpanCount SimpleAggregateFunction(sum, UInt64),
    SampleRateSum SimpleAggregateFunction(sum, Float64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, DeploymentEnv, SourceService, TargetService)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_map_edges_hourly_ingest (
    OrgId LowCardinality(String),
    Hour DateTime,
    SourceService LowCardinality(String),
    TargetService String,
    DeploymentEnv LowCardinality(String),
    CallCount UInt64,
    ErrorCount UInt64,
    DurationSumMs Float64,
    MaxDurationMs Float64,
    SampledSpanCount UInt64,
    UnsampledSpanCount UInt64,
    SampleRateSum Float64
)
ENGINE = Null;

CREATE TABLE IF NOT EXISTS service_map_spans (
    OrgId LowCardinality(String),
    Timestamp DateTime,
    TraceId String,
    SpanId String,
    ParentSpanId String,
    ServiceName LowCardinality(String),
    SpanKind LowCardinality(String),
    Duration UInt64,
    StatusCode LowCardinality(String),
    TraceState String,
    DeploymentEnv LowCardinality(String)
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, TraceId, SpanId, Timestamp)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS service_operations_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    SpanName String,
    SpanCount SimpleAggregateFunction(sum, UInt64),
    EstimatedSpanCount SimpleAggregateFunction(sum, Float64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    EstimatedErrorCount SimpleAggregateFunction(sum, Float64),
    DurationSum SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64),
    ClassifiedSpanCount SimpleAggregateFunction(sum, UInt64),
    ServerSpanCount SimpleAggregateFunction(sum, UInt64),
    RoutedSpanCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(Hour)
ORDER BY (OrgId, ServiceName, DeploymentEnv, Hour, SpanName)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_operations_minutely (
    OrgId LowCardinality(String),
    Minute DateTime,
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    SpanName String,
    SpanCount SimpleAggregateFunction(sum, UInt64),
    EstimatedSpanCount SimpleAggregateFunction(sum, Float64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    EstimatedErrorCount SimpleAggregateFunction(sum, Float64),
    DurationSum SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64),
    ClassifiedSpanCount SimpleAggregateFunction(sum, UInt64),
    ServerSpanCount SimpleAggregateFunction(sum, UInt64),
    RoutedSpanCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Minute)
ORDER BY (OrgId, ServiceName, DeploymentEnv, Minute, SpanName)
TTL toDate(Minute) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS service_overview_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    ServiceNamespace LowCardinality(String),
    CommitSha LowCardinality(String),
    SpanCount SimpleAggregateFunction(sum, UInt64),
    EstimatedSpanCount SimpleAggregateFunction(sum, Float64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    EstimatedErrorCount SimpleAggregateFunction(sum, Float64),
    DurationSum SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95, 0.99), UInt64),
    FirstSeen SimpleAggregateFunction(min, DateTime),
    ApdexSatisfiedCount SimpleAggregateFunction(sum, UInt64),
    ApdexToleratingCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(Hour)
ORDER BY (OrgId, ServiceName, Hour, DeploymentEnv, ServiceNamespace, CommitSha)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_overview_minutely (
    OrgId LowCardinality(String),
    Minute DateTime,
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    ServiceNamespace LowCardinality(String),
    CommitSha LowCardinality(String),
    SpanCount SimpleAggregateFunction(sum, UInt64),
    EstimatedSpanCount SimpleAggregateFunction(sum, Float64),
    ErrorCount SimpleAggregateFunction(sum, UInt64),
    EstimatedErrorCount SimpleAggregateFunction(sum, Float64),
    DurationSum SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95, 0.99), UInt64),
    FirstSeen SimpleAggregateFunction(min, DateTime),
    ApdexSatisfiedCount SimpleAggregateFunction(sum, UInt64),
    ApdexToleratingCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Minute)
ORDER BY (OrgId, ServiceName, Minute, DeploymentEnv, ServiceNamespace, CommitSha)
TTL toDate(Minute) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS service_overview_spans (
    OrgId LowCardinality(String),
    Timestamp DateTime,
    ServiceName LowCardinality(String),
    Duration UInt64,
    StatusCode LowCardinality(String),
    TraceState String,
    DeploymentEnv LowCardinality(String),
    CommitSha LowCardinality(String),
    SampleRate Float64 DEFAULT 1,
    ServiceNamespace LowCardinality(String),
    SpanKind LowCardinality(String) DEFAULT '',
    IsRoot UInt8 DEFAULT 0,
    INDEX idx_service_namespace ServiceNamespace TYPE set(1000) GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, ServiceName, Timestamp)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS service_platforms_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    K8sCluster SimpleAggregateFunction(max, String),
    K8sPodName SimpleAggregateFunction(max, String),
    K8sDeploymentName SimpleAggregateFunction(max, String),
    K8sStatefulSetName SimpleAggregateFunction(max, String),
    K8sDaemonSetName SimpleAggregateFunction(max, String),
    K8sNamespaceName SimpleAggregateFunction(max, String),
    CloudPlatform SimpleAggregateFunction(max, String),
    CloudProvider SimpleAggregateFunction(max, String),
    FaasName SimpleAggregateFunction(max, String),
    MapleSdkType SimpleAggregateFunction(max, String),
    ProcessRuntimeName SimpleAggregateFunction(max, String),
    SpanCount SimpleAggregateFunction(sum, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, ServiceName, DeploymentEnv)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS service_usage (
    OrgId LowCardinality(String),
    ServiceName LowCardinality(String),
    Hour DateTime,
    LogCount UInt64,
    LogSizeBytes UInt64,
    TraceCount UInt64,
    TraceSizeBytes UInt64,
    SumMetricCount UInt64,
    SumMetricSizeBytes UInt64,
    GaugeMetricCount UInt64,
    GaugeMetricSizeBytes UInt64,
    HistogramMetricCount UInt64,
    HistogramMetricSizeBytes UInt64,
    ExpHistogramMetricCount UInt64,
    ExpHistogramMetricSizeBytes UInt64
)
ENGINE = SummingMergeTree
ORDER BY (OrgId, ServiceName, Hour)
TTL Hour + INTERVAL 365 DAY;

CREATE TABLE IF NOT EXISTS session_events (
    OrgId LowCardinality(String),
    SessionId String,
    Timestamp DateTime64(9),
    Seq UInt32 DEFAULT 0,
    Type LowCardinality(String),
    Url String DEFAULT '',
    TraceId String DEFAULT '',
    Level LowCardinality(String) DEFAULT '',
    Message String DEFAULT '',
    TargetSelector String DEFAULT '',
    TargetText String DEFAULT '',
    NetMethod LowCardinality(String) DEFAULT '',
    NetUrl String DEFAULT '',
    NetStatus UInt16 DEFAULT 0,
    NetDurationMs UInt32 DEFAULT 0,
    ErrorStack String DEFAULT '',
    Attributes Map(String, String),
    VisitorId String DEFAULT '',
    UserId String DEFAULT '',
    GroupId String DEFAULT '',
    INDEX idx_type Type TYPE set(16) GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, SessionId, Timestamp, Seq)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS session_replay_events (
    OrgId LowCardinality(String),
    SessionId String,
    ChunkSeq UInt32,
    Timestamp DateTime64(9),
    DurationMs UInt32 DEFAULT 0,
    EventCount UInt32 DEFAULT 0,
    ByteSize UInt32 DEFAULT 0,
    Events String,
    IsCheckpoint UInt8 DEFAULT 0
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, SessionId, ChunkSeq)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS session_replays (
    OrgId LowCardinality(String),
    SessionId String,
    StartTime DateTime64(9),
    EndTime Nullable(DateTime64(9)),
    DurationMs Nullable(UInt32),
    Status LowCardinality(String),
    UserId String,
    UrlInitial String,
    UserAgent String,
    BrowserName LowCardinality(String),
    OsName LowCardinality(String),
    DeviceType LowCardinality(String),
    Country LowCardinality(String) DEFAULT '',
    ServiceName LowCardinality(String),
    PageViews UInt32 DEFAULT 0,
    ClickCount UInt32 DEFAULT 0,
    ErrorCount UInt32 DEFAULT 0,
    TraceIds Array(String) DEFAULT [],
    ResourceAttributes Map(LowCardinality(String), String),
    Version UInt32,
    VisitorId String DEFAULT '',
    VisitorIsNew UInt8 DEFAULT 0,
    UserEmail String DEFAULT '',
    UserName String DEFAULT '',
    GroupId String DEFAULT '',
    GroupName String DEFAULT '',
    UserTraits Map(String, String) DEFAULT map(),
    Referrer String DEFAULT '',
    ReferrerHost LowCardinality(String) DEFAULT '',
    UtmSource LowCardinality(String) DEFAULT '',
    UtmMedium LowCardinality(String) DEFAULT '',
    UtmCampaign LowCardinality(String) DEFAULT '',
    UtmTerm String DEFAULT '',
    UtmContent String DEFAULT '',
    Host LowCardinality(String) DEFAULT '',
    EntryPath String DEFAULT '',
    ExitPath String DEFAULT '',
    Language LowCardinality(String) DEFAULT '',
    LastActivityAt Nullable(DateTime64(9))
)
ENGINE = ReplacingMergeTree
PARTITION BY toDate(StartTime)
ORDER BY (OrgId, SessionId)
TTL toDate(StartTime) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS span_metrics_calls_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    MetricName LowCardinality(String),
    SpanKind LowCardinality(String),
    AttrFingerprint UInt64,
    ResourceFingerprint UInt64,
    StartTimeUnix DateTime64(9),
    LastValue AggregateFunction(argMax, Float64, DateTime64(9))
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, ServiceName, MetricName, SpanKind, AttrFingerprint, ResourceFingerprint, StartTimeUnix)
TTL toDate(Hour) + INTERVAL 90 DAY;

CREATE TABLE IF NOT EXISTS trace_facets_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    SpanName String,
    HttpMethod LowCardinality(String),
    HttpStatusCode LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    ServiceNamespace LowCardinality(String),
    HasError UInt8,
    TraceCount SimpleAggregateFunction(sum, UInt64),
    DurationMin SimpleAggregateFunction(min, UInt64),
    DurationMax SimpleAggregateFunction(max, UInt64),
    DurationQuantiles AggregateFunction(quantilesTDigest(0.5, 0.95), UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, ServiceName, SpanName, HttpMethod, HttpStatusCode, DeploymentEnv, ServiceNamespace, HasError)
TTL toDate(Hour) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS trace_index (
    OrgId LowCardinality(String),
    TraceId String,
    ServiceName LowCardinality(String),
    Bucket DateTime,
    MinTimestamp SimpleAggregateFunction(min, DateTime64(9)),
    MaxTimestamp SimpleAggregateFunction(max, DateTime64(9))
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Bucket)
ORDER BY (OrgId, TraceId, ServiceName, Bucket)
TTL toDate(Bucket) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS trace_list_mv (
    OrgId LowCardinality(String),
    TraceId String,
    Timestamp DateTime,
    ServiceName LowCardinality(String),
    SpanName String,
    SpanKind LowCardinality(String),
    Duration UInt64,
    StatusCode LowCardinality(String),
    HttpMethod LowCardinality(String),
    HttpRoute String,
    HttpStatusCode LowCardinality(String),
    DeploymentEnv LowCardinality(String),
    HasError UInt8,
    TraceState String,
    ServiceNamespace LowCardinality(String),
    INDEX idx_service_namespace ServiceNamespace TYPE set(1000) GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, Timestamp, TraceId)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS traces (
    OrgId LowCardinality(String),
    Timestamp DateTime64(9),
    TraceId String,
    SpanId String,
    ParentSpanId String,
    TraceState String,
    SpanName LowCardinality(String),
    SpanKind LowCardinality(String),
    ServiceName LowCardinality(String),
    ResourceSchemaUrl String,
    ResourceAttributes Map(LowCardinality(String), String),
    ScopeSchemaUrl String,
    ScopeName String,
    ScopeVersion String,
    ScopeAttributes Map(LowCardinality(String), String),
    Duration UInt64 DEFAULT 0,
    StatusCode LowCardinality(String),
    StatusMessage String,
    SpanAttributes Map(LowCardinality(String), String),
    EventsTimestamp Array(DateTime64(9)),
    EventsName Array(LowCardinality(String)),
    EventsAttributes Array(Map(LowCardinality(String), String)),
    LinksTraceId Array(String),
    LinksSpanId Array(String),
    LinksTraceState Array(String),
    LinksAttributes Array(Map(LowCardinality(String), String)),
    SampleRate Float64 DEFAULT multiIf(SpanAttributes['SampleRate'] != '' AND toFloat64OrZero(SpanAttributes['SampleRate']) >= 1.0, toFloat64OrZero(SpanAttributes['SampleRate']), match(TraceState, 'th:[0-9a-f]+'), 1.0 / greatest(1.0 - reinterpretAsUInt64(reverse(unhex(rightPad(extract(TraceState, 'th:([0-9a-f]+)'), 16, '0')))) / pow(2.0, 64), 0.0001), 1.0),
    IsEntryPoint UInt8 DEFAULT if(SpanKind IN ('Server', 'Consumer') OR ParentSpanId = '', 1, 0),
    ResourceAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(ResourceAttributes), mapValues(ResourceAttributes)),
    ScopeAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(ScopeAttributes), mapValues(ScopeAttributes)),
    SpanAttributeItems Array(String) DEFAULT arrayMap((k, v) -> concat(k, char(31), v), mapKeys(SpanAttributes), mapValues(SpanAttributes)),
    INDEX idx_trace_id TraceId TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_name SpanName TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_keys mapKeys(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_vals mapValues(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_resource_attr_keys mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_resource_attr_vals mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_scope_attr_keys mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_scope_attr_vals mapValues(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (OrgId, ServiceName, toStartOfFiveMinutes(Timestamp), TraceId)
TTL toDate(Timestamp) + INTERVAL 30 DAY;

CREATE TABLE IF NOT EXISTS traces_aggregates_hourly (
    OrgId LowCardinality(String),
    Hour DateTime,
    ServiceName LowCardinality(String),
    SpanName LowCardinality(String),
    SpanKind LowCardinality(String),
    StatusCode LowCardinality(String),
    IsEntryPoint UInt8,
    DeploymentEnv LowCardinality(String),
    WeightedCount SimpleAggregateFunction(sum, Float64),
    WeightedDurationSum SimpleAggregateFunction(sum, Float64),
    WeightedErrorCount SimpleAggregateFunction(sum, Float64),
    DurationQuantiles AggregateFunction(quantilesTDigestWeighted(0.5, 0.95, 0.99), UInt64, UInt32),
    DurationMin SimpleAggregateFunction(min, UInt64),
    DurationMax SimpleAggregateFunction(max, UInt64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(Hour)
ORDER BY (OrgId, Hour, ServiceName, SpanName, SpanKind, StatusCode, IsEntryPoint, DeploymentEnv)
TTL toDate(Hour) + INTERVAL 365 DAY;

CREATE MATERIALIZED VIEW IF NOT EXISTS ai_crawler_requests_mv TO ai_crawler_requests AS
SELECT
          OrgId,
          Timestamp,
          TraceId,
          ServiceName,
          arrayElement(['GPTBot', 'OAI-SearchBot', 'ChatGPT-User', 'ClaudeBot', 'anthropic-ai', 'Claude-SearchBot', 'Claude-User', 'PerplexityBot', 'Perplexity-User', 'Meta-ExternalAgent', 'Meta-WebIndexer', 'Meta-ExternalFetcher', 'Bytespider', 'MistralAI-User', 'CCBot', 'Amazonbot', 'DuckAssistBot', 'cohere-training-data-crawler', 'cohere-ai'], multiSearchFirstIndexCaseInsensitive(coalesce(nullIf(SpanAttributes['user_agent.original'], ''), nullIf(SpanAttributes['http.user_agent'], ''), ''), ['GPTBot', 'OAI-SearchBot', 'ChatGPT-User', 'ClaudeBot', 'anthropic-ai', 'Claude-SearchBot', 'Claude-User', 'PerplexityBot', 'Perplexity-User', 'meta-externalagent', 'meta-webindexer', 'meta-externalfetcher', 'Bytespider', 'MistralAI-User', 'CCBot', 'Amazonbot', 'DuckAssistBot', 'cohere-training-data-crawler', 'cohere-ai'])) AS Crawler,
          lower(replaceRegexpOne(coalesce(nullIf(SpanAttributes['server.address'], ''), nullIf(SpanAttributes['http.host'], ''), nullIf(SpanAttributes['net.host.name'], ''), nullIf(domain(SpanAttributes['url.full']), ''), nullIf(domain(SpanAttributes['http.url']), ''), ''), ':[0-9]+$', '')) AS Host,
          leftUTF8(coalesce(nullIf(SpanAttributes['url.path'], ''), nullIf(replaceRegexpOne(SpanAttributes['http.target'], '[?#].*$', ''), ''), nullIf(path(SpanAttributes['url.full']), ''), nullIf(path(SpanAttributes['http.url']), ''), ''), 512) AS Path,
          toUInt16OrZero(coalesce(nullIf(SpanAttributes['http.response.status_code'], ''), nullIf(SpanAttributes['http.status_code'], ''), '')) AS HttpStatus
        FROM traces
        WHERE SpanKind = 'Server'
          AND multiSearchFirstIndexCaseInsensitive(coalesce(nullIf(SpanAttributes['user_agent.original'], ''), nullIf(SpanAttributes['http.user_agent'], ''), ''), ['GPTBot', 'OAI-SearchBot', 'ChatGPT-User', 'ClaudeBot', 'anthropic-ai', 'Claude-SearchBot', 'Claude-User', 'PerplexityBot', 'Perplexity-User', 'meta-externalagent', 'meta-webindexer', 'meta-externalfetcher', 'Bytespider', 'MistralAI-User', 'CCBot', 'Amazonbot', 'DuckAssistBot', 'cohere-training-data-crawler', 'cohere-ai']) > 0
          AND leftUTF8(coalesce(nullIf(SpanAttributes['url.path'], ''), nullIf(replaceRegexpOne(SpanAttributes['http.target'], '[?#].*$', ''), ''), nullIf(path(SpanAttributes['url.full']), ''), nullIf(path(SpanAttributes['http.url']), ''), ''), 512) != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS ai_trace_index_mv TO ai_trace_index AS
SELECT
          OrgId,
          Timestamp,
          TraceId,
          SpanAttributes['maple_ai.session.id'] AS SessionId,
          SpanAttributes['maple_ai.vendor.id'] AS VendorId,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          SpanAttributes['maple_ai.model'] AS Model,
          SpanAttributes['maple_ai.agent.name'] AS AgentName,
          SpanAttributes['maple_ai.tool.name'] AS ToolName,
          SpanId,
          ParentSpanId,
          Duration,
          toUInt8(SpanAttributes['maple_ai.error'] = '1') AS IsError,
          toUInt8(SpanAttributes['maple_ai.llm_call'] = '1') AS IsLlmCall,
          toUInt8(SpanAttributes['maple_ai.tool_call'] = '1') AS IsToolCall,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.input_tokens']) + toFloat64OrZero(SpanAttributes['maple_ai.usage.cache_read_tokens']) + toFloat64OrZero(SpanAttributes['maple_ai.usage.cache_write_tokens']) + toFloat64OrZero(SpanAttributes['maple_ai.usage.output_tokens']) + toFloat64OrZero(SpanAttributes['maple_ai.usage.reasoning_tokens']) AS Tokens,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.cost']) AS Cost,
          SpanAttributes['maple_ai.response.id'] AS ResponseId,
          SpanAttributes['maple_ai.vendor.version'] AS VendorVersion,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.input_tokens']) AS InputTokens,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.cache_read_tokens']) AS CacheReadTokens,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.cache_write_tokens']) AS CacheWriteTokens,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.output_tokens']) AS OutputTokens,
          toFloat64OrZero(SpanAttributes['maple_ai.usage.reasoning_tokens']) AS ReasoningTokens,
          SpanAttributes['error.type'] AS ErrorType,
          leftUTF8(StatusMessage, 400) AS StatusMessage,
          SpanAttributes['maple_ai.tool.description'] AS ToolDescription,
          SpanAttributes['maple_ai.tool.error_result'] AS FailedToolCallResult,
          if(SpanAttributes['maple_ai.error'] = '1', cityHash64(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(leftUTF8(coalesce(nullIf(SpanAttributes['maple_ai.tool.error_result'], ''), leftUTF8(StatusMessage, 400)), 400), '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}', 'EMAIL'), 'https?://[^/ )"]+', ''), '/(Users|home)/[^/ ]+', '/~'), '[?][A-Za-z0-9_]+=[^ )"]*', '?#'), '\'[^\' ]*/[^\' ]*\'|\'[^\' ]{25,}\'', '\'#\''), '"[^" ]*/[^" ]*"|"[^" ]{25,}"', '"#"'), '`[^` ]*/[^` ]*`|`[^` ]{25,}`', '`#`'), '[0-9a-fA-F-]{6,}|[0-9]+', '#')), 0) AS ErrorFingerprint
        FROM traces
        WHERE SpanAttributes['maple_ai.vendor.id'] != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS error_events_by_time_mv TO error_events_by_time AS
WITH
          arrayFirstIndex(n -> n = 'exception', EventsName) AS _ei,
          -- Only fill the old Unknown Error bucket. Event values (including
          -- empty fields) and spans with StatusMessage keep every hash input.
          _ei = 0 AND StatusMessage = '' AS _useAttrs,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.type'],
            if(_useAttrs, coalesce(nullIf(SpanAttributes['exception.type'], ''), SpanAttributes['error.type']), '')
          ) AS _exType,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.message'],
            if(_useAttrs, coalesce(nullIf(SpanAttributes['exception.message'], ''), SpanAttributes['error.message']), StatusMessage)
          ) AS _exMsg,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.stacktrace'],
            if(_useAttrs, SpanAttributes['exception.stacktrace'], '')
          ) AS _exStack,
          if(_useAttrs, _exMsg, StatusMessage) AS _msgText,
          -- JSON detection for the message signature below.
          isValidJSON(_msgText) AS _isJson,
          _isJson AND JSONType(_msgText) = 'Object' AS _isJsonObj,
          -- General, KEY-NAME-AGNOSTIC canonical signature: iterate ALL top-level
          -- keys, redact volatile tokens (long hex / numbers) in each raw value, then
          -- sort by "key=value" so key order & whitespace don't matter. No assumption
          -- about which keys exist — works for any producer's JSON shape. (Nested
          -- objects are hashed as their raw substring; only top-level is canonicalized.)
          arrayStringConcat(
            arraySort(
              arrayMap(
                kv -> concat(kv.1, '=', replaceRegexpAll(kv.2, '[0-9a-fA-F-]{6,}|[0-9]+', '#')),
                JSONExtractKeysAndValuesRaw(_msgText)
              )
            ),
            '|'
          ) AS _jsonSig,
          -- The message signature is folded in ALWAYS, not only when there are no
          -- frames. Bundled runtimes minify every module into one file, so the top
          -- three frames of a Worker error are `toDatabaseError (worker.js)` for
          -- every failing query alike: on frames alone, 25 distinct DatabaseError
          -- bugs (316k occurrences) collapse into a single issue. The signature
          -- restores that discrimination, and it cannot reinflate cardinality the
          -- way a raw prefix would because everything variable is redacted first:
          -- emails, URL origins, home directories, query strings, quoted values,
          -- then ids and every digit run. See ERROR_MSG_REDACTIONS for the order,
          -- what is deliberately kept, and the one residual it cannot reach.
          multiIf(
            _isJsonObj, _jsonSig,
            substringUTF8(
              replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(substringUTF8(_msgText, 1, 400), '\\x1b\\[[0-9;]*m', ''), '^((\\[Error\\]|[\\s\\S]*?DB::Exception:)[\\s\\S]*?)( [Ii]n scope | [Ww]hile processing )[\\s\\S]*', '\\1'), '(^|[ \\t\\r\\n\'"(`=])~/[^ \'"`:,)]*', '\\1PATH'), '(^|[ \\t\\r\\n\'"(`=])/(tmp|var|srv|private|opt|mnt|data|Volumes|System|app|work|run|Users/[^/ \\t\\r\\n\'"`:,)]+|home/[^/ \\t\\r\\n\'"`:,)]+)([/.][^ \\t\\r\\n\'"`:,)]*)?([ \\t\\r\\n\'"`:,)]|$)', '\\1PATH\\4'), '(^|[ \\t\\r\\n\'"(`=])/(tmp|var|srv|private|opt|mnt|data|Volumes|System|app|work|run|Users/[^/ \\t\\r\\n\'"`:,)]+|home/[^/ \\t\\r\\n\'"`:,)]+)([/.][^ \\t\\r\\n\'"`:,)]*)?([ \\t\\r\\n\'"`:,)]|$)', '\\1PATH\\4'), '\\b[a-z]{2,8}_[a-z]*[A-Z0-9][A-Za-z0-9]{8,}\\b', 'ID'), '\\b[0-9A-HJKMNP-TV-Z]{26}\\b', 'ID'), '\\b(HTTP|[Ss]tatus|returned|code|failed|failed with)(:? ?)([1-4][0-9][0-9]|500)\\b', '\\1\\2§\\3'), '\\b(HTTP|[Ss]tatus|returned|code|failed|failed with)(:? ?)5(0[1-9]|[1-9][0-9])\\b', '\\1\\2§5xx'), '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}', 'EMAIL'), 'https?://[^/ )"]+', ''), '/(Users|home)/[^/ ]+', '/~'), '[?][A-Za-z0-9_]+=[^ )"]*', '?#'), '\'[^\' ]*/[^\' ]*\'|\'[^\' ]{25,}\'', '\'#\''), '"[^" ]*/[^" ]*"|"[^" ]{25,}"', '"#"'), '`[^` ]*/[^` ]*`|`[^` ]{25,}`', '`#`'), '(§[1-4][0-9][0-9]|§500|§5xx)|[0-9a-fA-F-]{6,}|[0-9]+', '\\1#'), '§([1-4][0-9][0-9]|500|5xx)#', '\\1'),
              1, 120
            )
          ) AS _msgSig,
          -- Frame lines are matched by SHAPE, not by "contains :NUMBER". The old
          -- rule accepted any line with a colon-digit, which let non-frame lines
          -- in: Drizzle's `params: <row values>` line, and the `Type: message`
          -- header (`Code: 62`, `position 1628`, embedded timestamps). Row values
          -- and message text then entered the hash and split one bug into
          -- thousands of issues — 23,035 fingerprints for six real
          -- AnomalyPersistenceError call sites, 15,051 for thirteen DatabaseError
          -- ones.
          --
          -- The pattern is rendered from FRAME_LINE_PATTERN in fingerprint.ts,
          -- as is every redaction below. They used to be hand-copied here, which
          -- let the reference implementation the tests exercise drift away from
          -- the SQL that actually runs, silently.
          arraySlice(
            arrayFilter(
              line -> match(line, '^[ \\t]*at |^[ \\t]*File "|^[ \\t]+from [^ ]+:[0-9]+|^[^ \\t@]+@[^ \\t]*:[0-9]+|^[ \\t]+[^ \\t]+\\.(go|rs):[0-9]+|^[0-9]+ +\\S.* +0x[0-9a-fA-F]+')
                AND position(line, '(definition)') = 0
                AND NOT match(line, '^[ \\t]*at \\[')
                AND NOT match(line, '^[ \\t]*at ([^(:]|:[^0-9])*:?$'),
              splitByChar('\n', _exStack)
            ),
            -- A namespaced type plus its message names one failure: keep only the
            -- throw site. With no message the frames are all that is left; keep 3.
            1, if(position(_exType, '/') > 0 AND _msgSig != '', 1, 3)
          ) AS _rawFrames,
          -- Redact every volatile token a frame line can carry: the URL origin
          -- (so preview hosts share one fingerprint), Vite's 8-char bundle
          -- content hash (so a deploy does not re-split every triaged browser and
          -- Worker issue), then line numbers, hex pointers and long id runs. See
          -- FRAME_REDACTIONS for the order and the reasoning.
          arrayMap(
            line -> replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(line, '@[0-9]+\\.[0-9]+\\.[0-9]+[^/ )]*', '@#'), '\\(index [0-9]+\\)', '(index #)'), 'file:///bundle/', ''), 'https?://[^/ )]+', ''), '(-[A-Za-z0-9_-]{8})+\\.js', '.js'), '(-[A-Za-z0-9_-]{8})+\\.css', '.css'), '(/_astro/[^/ ).]+)\\.[A-Za-z0-9_-]{8}\\.js', '\\1.js'), ':[0-9]+|line [0-9]+|0x[0-9a-fA-F]+|[0-9a-fA-F]{8,}|[0-9]{6,}', ''),
            _rawFrames
          ) AS _topFrames,
          if(length(_topFrames) > 0, _topFrames[1], '') AS _topFrame,
          arrayStringConcat(_topFrames, '\n') AS _fpFrames,
          -- Display-only, best-effort human label (decoupled from the fingerprint:
          -- many labels may map to one hash). The broad key list here is a DISPLAY
          -- heuristic only; the fingerprint above makes no key-name assumption.
          multiIf(
            JSONExtractString(_msgText, 'title')   != '', JSONExtractString(_msgText, 'title'),
            JSONExtractString(_msgText, 'message') != '', JSONExtractString(_msgText, 'message'),
            JSONExtractString(_msgText, 'error')   != '', JSONExtractString(_msgText, 'error'),
            JSONExtractString(_msgText, '_tag')    != '', JSONExtractString(_msgText, '_tag'),
            JSONExtractString(_msgText, 'reason')  != '', JSONExtractString(_msgText, 'reason'),
            JSONExtractString(_msgText, 'name')    != '', JSONExtractString(_msgText, 'name'),
            JSONExtractString(_msgText, 'type')    != '', extract(JSONExtractString(_msgText, 'type'), '([^/]+)$'),
            'JSON error'
          ) AS _jsonLabel,
          multiIf(
            _msgText = '', 'Unknown Error',
            position(_msgText, '{ readonly') = 1 OR position(_msgText, '└─') > 0,
              if(
                extract(_msgText, 'readonly (\\w+)') != '',
                concat('Schema parse error: ', extract(_msgText, 'readonly (\\w+)')),
                'Schema parse error'
              ),
            _isJsonObj OR position(_msgText, '[') = 1, _jsonLabel,
            left(_msgText, multiIf(
              position(_msgText, ': ')  > 3, toInt64(position(_msgText, ': '))  - 1,
              position(_msgText, ' (')  > 3, toInt64(position(_msgText, ' (')) - 1,
              position(_msgText, '\n') > 3, toInt64(position(_msgText, '\n')) - 1,
              least(toInt64(length(_msgText)), 150)
            ))
          ) AS _statusLabel,
          if(_exType != '', _exType, _statusLabel) AS _errorLabel,
          -- Both semconv spellings; the current key wins when both are present.
          toUInt16OrZero(
            if(
              SpanAttributes['http.response.status_code'] != '',
              SpanAttributes['http.response.status_code'],
              SpanAttributes['http.status_code']
            )
          ) AS _httpStatus
        SELECT
          OrgId,
          toDateTime(Timestamp) AS Timestamp,
          TraceId,
          SpanId,
          ParentSpanId,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          _exType AS ExceptionType,
          _exMsg AS ExceptionMessage,
          _exStack AS ExceptionStacktrace,
          _topFrame AS TopFrame,
          cityHash64(OrgId, ServiceName, _exType, _fpFrames, _msgSig, 'fp3') AS FingerprintHash,
          StatusMessage,
          Duration,
          _errorLabel AS ErrorLabel,
          ResourceAttributes['service.version'] AS ServiceVersion
        FROM traces
        WHERE StatusCode = 'Error'
          -- Client-side runtimes (notably the native Cloudflare Workers
          -- observability) mark ANY non-2xx fetch span as Error, so 404s from bot
          -- traffic arrived here as unlabelled "Unknown Error" issues. Drop a
          -- span only when all hold: 4xx, no exception event, no exception.type
          -- attribute, and no error.type beyond the status code itself (HTTP
          -- semconv sets error.type to the bare status on a non-2xx response,
          -- which carries no exception). 5xx and anything carrying a real
          -- exception still count, and SpanKind is deliberately not consulted —
          -- these are Client spans.
          AND NOT (
            _httpStatus >= 400 AND _httpStatus < 500
            AND _ei = 0
            AND SpanAttributes['exception.type'] = ''
            AND (SpanAttributes['error.type'] = '' OR SpanAttributes['error.type'] = toString(_httpStatus))
          );

CREATE MATERIALIZED VIEW IF NOT EXISTS error_events_mv TO error_events AS
WITH
          arrayFirstIndex(n -> n = 'exception', EventsName) AS _ei,
          -- Only fill the old Unknown Error bucket. Event values (including
          -- empty fields) and spans with StatusMessage keep every hash input.
          _ei = 0 AND StatusMessage = '' AS _useAttrs,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.type'],
            if(_useAttrs, coalesce(nullIf(SpanAttributes['exception.type'], ''), SpanAttributes['error.type']), '')
          ) AS _exType,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.message'],
            if(_useAttrs, coalesce(nullIf(SpanAttributes['exception.message'], ''), SpanAttributes['error.message']), StatusMessage)
          ) AS _exMsg,
          if(
            _ei > 0, EventsAttributes[_ei]['exception.stacktrace'],
            if(_useAttrs, SpanAttributes['exception.stacktrace'], '')
          ) AS _exStack,
          if(_useAttrs, _exMsg, StatusMessage) AS _msgText,
          -- JSON detection for the message signature below.
          isValidJSON(_msgText) AS _isJson,
          _isJson AND JSONType(_msgText) = 'Object' AS _isJsonObj,
          -- General, KEY-NAME-AGNOSTIC canonical signature: iterate ALL top-level
          -- keys, redact volatile tokens (long hex / numbers) in each raw value, then
          -- sort by "key=value" so key order & whitespace don't matter. No assumption
          -- about which keys exist — works for any producer's JSON shape. (Nested
          -- objects are hashed as their raw substring; only top-level is canonicalized.)
          arrayStringConcat(
            arraySort(
              arrayMap(
                kv -> concat(kv.1, '=', replaceRegexpAll(kv.2, '[0-9a-fA-F-]{6,}|[0-9]+', '#')),
                JSONExtractKeysAndValuesRaw(_msgText)
              )
            ),
            '|'
          ) AS _jsonSig,
          -- The message signature is folded in ALWAYS, not only when there are no
          -- frames. Bundled runtimes minify every module into one file, so the top
          -- three frames of a Worker error are `toDatabaseError (worker.js)` for
          -- every failing query alike: on frames alone, 25 distinct DatabaseError
          -- bugs (316k occurrences) collapse into a single issue. The signature
          -- restores that discrimination, and it cannot reinflate cardinality the
          -- way a raw prefix would because everything variable is redacted first:
          -- emails, URL origins, home directories, query strings, quoted values,
          -- then ids and every digit run. See ERROR_MSG_REDACTIONS for the order,
          -- what is deliberately kept, and the one residual it cannot reach.
          multiIf(
            _isJsonObj, _jsonSig,
            substringUTF8(
              replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(substringUTF8(_msgText, 1, 400), '\\x1b\\[[0-9;]*m', ''), '^((\\[Error\\]|[\\s\\S]*?DB::Exception:)[\\s\\S]*?)( [Ii]n scope | [Ww]hile processing )[\\s\\S]*', '\\1'), '(^|[ \\t\\r\\n\'"(`=])~/[^ \'"`:,)]*', '\\1PATH'), '(^|[ \\t\\r\\n\'"(`=])/(tmp|var|srv|private|opt|mnt|data|Volumes|System|app|work|run|Users/[^/ \\t\\r\\n\'"`:,)]+|home/[^/ \\t\\r\\n\'"`:,)]+)([/.][^ \\t\\r\\n\'"`:,)]*)?([ \\t\\r\\n\'"`:,)]|$)', '\\1PATH\\4'), '(^|[ \\t\\r\\n\'"(`=])/(tmp|var|srv|private|opt|mnt|data|Volumes|System|app|work|run|Users/[^/ \\t\\r\\n\'"`:,)]+|home/[^/ \\t\\r\\n\'"`:,)]+)([/.][^ \\t\\r\\n\'"`:,)]*)?([ \\t\\r\\n\'"`:,)]|$)', '\\1PATH\\4'), '\\b[a-z]{2,8}_[a-z]*[A-Z0-9][A-Za-z0-9]{8,}\\b', 'ID'), '\\b[0-9A-HJKMNP-TV-Z]{26}\\b', 'ID'), '\\b(HTTP|[Ss]tatus|returned|code|failed|failed with)(:? ?)([1-4][0-9][0-9]|500)\\b', '\\1\\2§\\3'), '\\b(HTTP|[Ss]tatus|returned|code|failed|failed with)(:? ?)5(0[1-9]|[1-9][0-9])\\b', '\\1\\2§5xx'), '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}', 'EMAIL'), 'https?://[^/ )"]+', ''), '/(Users|home)/[^/ ]+', '/~'), '[?][A-Za-z0-9_]+=[^ )"]*', '?#'), '\'[^\' ]*/[^\' ]*\'|\'[^\' ]{25,}\'', '\'#\''), '"[^" ]*/[^" ]*"|"[^" ]{25,}"', '"#"'), '`[^` ]*/[^` ]*`|`[^` ]{25,}`', '`#`'), '(§[1-4][0-9][0-9]|§500|§5xx)|[0-9a-fA-F-]{6,}|[0-9]+', '\\1#'), '§([1-4][0-9][0-9]|500|5xx)#', '\\1'),
              1, 120
            )
          ) AS _msgSig,
          -- Frame lines are matched by SHAPE, not by "contains :NUMBER". The old
          -- rule accepted any line with a colon-digit, which let non-frame lines
          -- in: Drizzle's `params: <row values>` line, and the `Type: message`
          -- header (`Code: 62`, `position 1628`, embedded timestamps). Row values
          -- and message text then entered the hash and split one bug into
          -- thousands of issues — 23,035 fingerprints for six real
          -- AnomalyPersistenceError call sites, 15,051 for thirteen DatabaseError
          -- ones.
          --
          -- The pattern is rendered from FRAME_LINE_PATTERN in fingerprint.ts,
          -- as is every redaction below. They used to be hand-copied here, which
          -- let the reference implementation the tests exercise drift away from
          -- the SQL that actually runs, silently.
          arraySlice(
            arrayFilter(
              line -> match(line, '^[ \\t]*at |^[ \\t]*File "|^[ \\t]+from [^ ]+:[0-9]+|^[^ \\t@]+@[^ \\t]*:[0-9]+|^[ \\t]+[^ \\t]+\\.(go|rs):[0-9]+|^[0-9]+ +\\S.* +0x[0-9a-fA-F]+')
                AND position(line, '(definition)') = 0
                AND NOT match(line, '^[ \\t]*at \\[')
                AND NOT match(line, '^[ \\t]*at ([^(:]|:[^0-9])*:?$'),
              splitByChar('\n', _exStack)
            ),
            -- A namespaced type plus its message names one failure: keep only the
            -- throw site. With no message the frames are all that is left; keep 3.
            1, if(position(_exType, '/') > 0 AND _msgSig != '', 1, 3)
          ) AS _rawFrames,
          -- Redact every volatile token a frame line can carry: the URL origin
          -- (so preview hosts share one fingerprint), Vite's 8-char bundle
          -- content hash (so a deploy does not re-split every triaged browser and
          -- Worker issue), then line numbers, hex pointers and long id runs. See
          -- FRAME_REDACTIONS for the order and the reasoning.
          arrayMap(
            line -> replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(line, '@[0-9]+\\.[0-9]+\\.[0-9]+[^/ )]*', '@#'), '\\(index [0-9]+\\)', '(index #)'), 'file:///bundle/', ''), 'https?://[^/ )]+', ''), '(-[A-Za-z0-9_-]{8})+\\.js', '.js'), '(-[A-Za-z0-9_-]{8})+\\.css', '.css'), '(/_astro/[^/ ).]+)\\.[A-Za-z0-9_-]{8}\\.js', '\\1.js'), ':[0-9]+|line [0-9]+|0x[0-9a-fA-F]+|[0-9a-fA-F]{8,}|[0-9]{6,}', ''),
            _rawFrames
          ) AS _topFrames,
          if(length(_topFrames) > 0, _topFrames[1], '') AS _topFrame,
          arrayStringConcat(_topFrames, '\n') AS _fpFrames,
          -- Display-only, best-effort human label (decoupled from the fingerprint:
          -- many labels may map to one hash). The broad key list here is a DISPLAY
          -- heuristic only; the fingerprint above makes no key-name assumption.
          multiIf(
            JSONExtractString(_msgText, 'title')   != '', JSONExtractString(_msgText, 'title'),
            JSONExtractString(_msgText, 'message') != '', JSONExtractString(_msgText, 'message'),
            JSONExtractString(_msgText, 'error')   != '', JSONExtractString(_msgText, 'error'),
            JSONExtractString(_msgText, '_tag')    != '', JSONExtractString(_msgText, '_tag'),
            JSONExtractString(_msgText, 'reason')  != '', JSONExtractString(_msgText, 'reason'),
            JSONExtractString(_msgText, 'name')    != '', JSONExtractString(_msgText, 'name'),
            JSONExtractString(_msgText, 'type')    != '', extract(JSONExtractString(_msgText, 'type'), '([^/]+)$'),
            'JSON error'
          ) AS _jsonLabel,
          multiIf(
            _msgText = '', 'Unknown Error',
            position(_msgText, '{ readonly') = 1 OR position(_msgText, '└─') > 0,
              if(
                extract(_msgText, 'readonly (\\w+)') != '',
                concat('Schema parse error: ', extract(_msgText, 'readonly (\\w+)')),
                'Schema parse error'
              ),
            _isJsonObj OR position(_msgText, '[') = 1, _jsonLabel,
            left(_msgText, multiIf(
              position(_msgText, ': ')  > 3, toInt64(position(_msgText, ': '))  - 1,
              position(_msgText, ' (')  > 3, toInt64(position(_msgText, ' (')) - 1,
              position(_msgText, '\n') > 3, toInt64(position(_msgText, '\n')) - 1,
              least(toInt64(length(_msgText)), 150)
            ))
          ) AS _statusLabel,
          if(_exType != '', _exType, _statusLabel) AS _errorLabel,
          -- Both semconv spellings; the current key wins when both are present.
          toUInt16OrZero(
            if(
              SpanAttributes['http.response.status_code'] != '',
              SpanAttributes['http.response.status_code'],
              SpanAttributes['http.status_code']
            )
          ) AS _httpStatus
        SELECT
          OrgId,
          toDateTime(Timestamp) AS Timestamp,
          TraceId,
          SpanId,
          ParentSpanId,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          _exType AS ExceptionType,
          _exMsg AS ExceptionMessage,
          _exStack AS ExceptionStacktrace,
          _topFrame AS TopFrame,
          cityHash64(OrgId, ServiceName, _exType, _fpFrames, _msgSig, 'fp3') AS FingerprintHash,
          StatusMessage,
          Duration,
          _errorLabel AS ErrorLabel,
          ResourceAttributes['service.version'] AS ServiceVersion
        FROM traces
        WHERE StatusCode = 'Error'
          -- Client-side runtimes (notably the native Cloudflare Workers
          -- observability) mark ANY non-2xx fetch span as Error, so 404s from bot
          -- traffic arrived here as unlabelled "Unknown Error" issues. Drop a
          -- span only when all hold: 4xx, no exception event, no exception.type
          -- attribute, and no error.type beyond the status code itself (HTTP
          -- semconv sets error.type to the bare status on a non-2xx response,
          -- which carries no exception). 5xx and anything carrying a real
          -- exception still count, and SpanKind is deliberately not consulted —
          -- these are Client spans.
          AND NOT (
            _httpStatus >= 400 AND _httpStatus < 500
            AND _ei = 0
            AND SpanAttributes['exception.type'] = ''
            AND (SpanAttributes['error.type'] = '' OR SpanAttributes['error.type'] = toString(_httpStatus))
          );

CREATE MATERIALIZED VIEW IF NOT EXISTS error_fingerprints_minutely_mv TO error_fingerprints_minutely AS
SELECT
          OrgId,
          toStartOfMinute(Timestamp) AS Minute,
          FingerprintHash,
          anyLast(ServiceName) AS ServiceName,
          anyLast(ExceptionType) AS ExceptionType,
          anyLast(ExceptionMessage) AS ExceptionMessage,
          anyLast(ErrorLabel) AS ErrorLabel,
          anyLast(TopFrame) AS TopFrame,
          count() AS OccurrenceCount,
          min(Timestamp) AS FirstSeen,
          max(Timestamp) AS LastSeen,
          -- Distinct builds, not a sample: see ServiceVersions on the datasource.
          groupUniqArray(ServiceVersion) AS ServiceVersions
        FROM error_events
        GROUP BY OrgId, Minute, FingerprintHash;

CREATE MATERIALIZED VIEW IF NOT EXISTS identity_links_mv TO identity_links AS
SELECT
          OrgId,
          VisitorId,
          UserId,
          StartTime AS FirstSeen
        FROM session_replays
        WHERE VisitorId != '' AND UserId != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS log_attribute_keys_mv TO attribute_keys_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          arrayJoin(mapKeys(LogAttributes)) AS AttributeKey,
          'log' AS AttributeScope,
          count() AS UsageCount
        FROM logs
        WHERE LogAttributes != map()
        GROUP BY OrgId, Hour, AttributeKey, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS log_attribute_values_mv TO attribute_values_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          AttributeKey,
          AttributeValue,
          'log' AS AttributeScope,
          count() AS UsageCount
        FROM logs
        ARRAY JOIN
          mapKeys(LogAttributes) AS AttributeKey,
          mapValues(LogAttributes) AS AttributeValue
        WHERE AttributeValue != ''
          AND length(AttributeValue) <= 128
          AND NOT (length(AttributeValue) > 4 AND match(AttributeValue, '^[0-9]+([.][0-9]+)?$'))
          AND NOT match(AttributeKey, '(_id|[.]id|Id|_ns)$')
          AND AttributeKey NOT LIKE 'http.request.header.%'
          AND AttributeKey NOT LIKE 'http.response.header.%'
        GROUP BY OrgId, Hour, AttributeKey, AttributeValue, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS logs_aggregates_hourly_mv TO logs_aggregates_hourly AS
SELECT
          OrgId,
          toStartOfHour(TimestampTime) AS Hour,
          ServiceName,
          SeverityText,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          count() AS Count,
          sum(length(Body) + 200) AS SizeBytes,
          ResourceAttributes['service.namespace'] AS ServiceNamespace
        FROM logs
        GROUP BY OrgId, Hour, ServiceName, SeverityText, DeploymentEnv, ServiceNamespace;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_attribute_keys_mv TO attribute_keys_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          arrayJoin(mapKeys(Attributes)) AS AttributeKey,
          'metric' AS AttributeScope,
          count() AS UsageCount
        FROM metrics_sum
        WHERE Attributes != map()
        GROUP BY OrgId, Hour, AttributeKey, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_attribute_values_mv TO attribute_values_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          AttributeKey,
          AttributeValue,
          'metric' AS AttributeScope,
          count() AS UsageCount
        FROM metrics_sum
        ARRAY JOIN
          mapKeys(Attributes) AS AttributeKey,
          mapValues(Attributes) AS AttributeValue
        WHERE AttributeValue != ''
          AND length(AttributeValue) <= 128
          AND NOT (length(AttributeValue) > 4 AND match(AttributeValue, '^[0-9]+([.][0-9]+)?$'))
          AND NOT match(AttributeKey, '(_id|[.]id|Id|_ns)$')
          AND AttributeKey NOT LIKE 'http.request.header.%'
          AND AttributeKey NOT LIKE 'http.response.header.%'
        GROUP BY OrgId, Hour, AttributeKey, AttributeValue, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_catalog_exp_histogram_mv TO metric_catalog AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          'exponential_histogram' AS MetricType,
          ServiceName,
          MetricName,
          anyLast(MetricDescription) AS MetricDescription,
          anyLast(MetricUnit) AS MetricUnit,
          toUInt8(0) AS IsMonotonic,
          count() AS DataPointCount,
          min(toDateTime(TimeUnix)) AS FirstSeen,
          max(toDateTime(TimeUnix)) AS LastSeen
        FROM metrics_exponential_histogram
        GROUP BY OrgId, Hour, MetricType, ServiceName, MetricName;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_catalog_gauge_mv TO metric_catalog AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          'gauge' AS MetricType,
          ServiceName,
          MetricName,
          anyLast(MetricDescription) AS MetricDescription,
          anyLast(MetricUnit) AS MetricUnit,
          toUInt8(0) AS IsMonotonic,
          count() AS DataPointCount,
          min(toDateTime(TimeUnix)) AS FirstSeen,
          max(toDateTime(TimeUnix)) AS LastSeen
        FROM metrics_gauge
        GROUP BY OrgId, Hour, MetricType, ServiceName, MetricName;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_catalog_histogram_mv TO metric_catalog AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          'histogram' AS MetricType,
          ServiceName,
          MetricName,
          anyLast(MetricDescription) AS MetricDescription,
          anyLast(MetricUnit) AS MetricUnit,
          toUInt8(0) AS IsMonotonic,
          count() AS DataPointCount,
          min(toDateTime(TimeUnix)) AS FirstSeen,
          max(toDateTime(TimeUnix)) AS LastSeen
        FROM metrics_histogram
        GROUP BY OrgId, Hour, MetricType, ServiceName, MetricName;

CREATE MATERIALIZED VIEW IF NOT EXISTS metric_catalog_sum_mv TO metric_catalog AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          'sum' AS MetricType,
          ServiceName,
          MetricName,
          anyLast(MetricDescription) AS MetricDescription,
          anyLast(MetricUnit) AS MetricUnit,
          anyLast(toUInt8(IsMonotonic)) AS IsMonotonic,
          count() AS DataPointCount,
          min(toDateTime(TimeUnix)) AS FirstSeen,
          max(toDateTime(TimeUnix)) AS LastSeen
        FROM metrics_sum
        GROUP BY OrgId, Hour, MetricType, ServiceName, MetricName;

CREATE MATERIALIZED VIEW IF NOT EXISTS product_events_mv TO product_events AS
SELECT
          OrgId,
          Timestamp,
          'browser' AS Source,
          SessionId,
          Seq,
          VisitorId,
          UserId,
          GroupId,
          Type AS Kind,
          if(Type = 'navigation', '$pageview', Message) AS EventName,
          domain(Url) AS Host,
          path(Url) AS PagePath,
          Url,
          '' AS ServiceName,
          Attributes,
          '' AS TraceId,
          '' AS SpanId
        FROM session_events
        WHERE Type IN ('navigation', 'custom');

CREATE MATERIALIZED VIEW IF NOT EXISTS product_events_traces_mv TO product_events AS
SELECT
          OrgId,
  Timestamp,
  'trace' AS Source,
  SpanAttributes['session.id'] AS SessionId,
  0 AS Seq,
  SpanAttributes['maple.product_event.visitor_id'] AS VisitorId,
  SpanAttributes['maple.product_event.user_id'] AS UserId,
  SpanAttributes['maple.product_event.group_id'] AS GroupId,
  'custom' AS Kind,
  SpanAttributes['maple.product_event.name'] AS EventName,
  domain(SpanAttributes['maple.product_event.url']) AS Host,
  path(SpanAttributes['maple.product_event.url']) AS PagePath,
  SpanAttributes['maple.product_event.url'] AS Url,
  ServiceName,
  mapUpdate(
    CAST(
      mapFilter(
        (k, v) -> NOT startsWith(k, 'maple.product_event.')
          AND (
            NOT has(mapKeys(SpanAttributes), 'maple.product_event.include')
            OR has(
              arrayMap(
                key -> trimBoth(key),
                splitByChar(',', SpanAttributes['maple.product_event.include'])
              ),
              k
            )
          ),
        SpanAttributes
      ),
      'Map(String, String)'
    ),
    mapApply(
      (k, v) -> (substring(k, 26), v),
      mapFilter((k, v) -> startsWith(k, 'maple.product_event.prop.'), SpanAttributes)
    )
  ) AS Attributes,
  TraceId,
  SpanId
        FROM traces
        WHERE SpanAttributes['maple.product_event.name'] != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS service_external_edges_hourly_mv TO service_external_edges_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          multiIf(
            coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']) != '' OR SpanAttributes['messaging.system'] != '', 'messaging',
            SpanAttributes['rpc.service'] != '' OR SpanAttributes['rpc.system'] != '', 'rpc',
            'http'
          ) AS TargetType,
          multiIf(
            coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']) != '' OR SpanAttributes['messaging.system'] != '', SpanAttributes['messaging.system'],
            SpanAttributes['rpc.service'] != '' OR SpanAttributes['rpc.system'] != '', SpanAttributes['rpc.system'],
            ''
          ) AS TargetSystem,
          multiIf(
            coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']) != '' OR SpanAttributes['messaging.system'] != '',
              if(coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']) != '', coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']), SpanAttributes['messaging.system']),
            SpanAttributes['rpc.service'] != '' OR SpanAttributes['rpc.system'] != '',
              if(SpanAttributes['rpc.service'] != '', SpanAttributes['rpc.service'], SpanAttributes['rpc.system']),
            if(SpanAttributes['server.address'] != '',
              SpanAttributes['server.address'],
              if(SpanAttributes['http.host'] != '',
                SpanAttributes['http.host'],
                SpanAttributes['url.authority']))
          ) AS TargetName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          count() AS CallCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sum(Duration / 1000000) AS DurationSumMs,
          max(Duration / 1000000) AS MaxDurationMs,
          sum(SampleRate) AS SampleRateSum,
          quantilesTDigestWeightedState(0.5, 0.95)(Duration, toUInt32(greatest(SampleRate, 1.0))) AS DurationQuantiles
        FROM traces
        WHERE SpanKind IN ('Client', 'Producer')
          AND coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) = ''
          AND ServiceName != ''
          AND (
               SpanAttributes['server.address'] != ''
            OR SpanAttributes['http.host'] != ''
            OR SpanAttributes['url.authority'] != ''
            OR coalesce(nullIf(SpanAttributes['messaging.destination.name'], ''), SpanAttributes['messaging.destination']) != ''
            OR SpanAttributes['messaging.system'] != ''
            OR SpanAttributes['rpc.service'] != ''
            OR SpanAttributes['rpc.system'] != ''
          )
        GROUP BY OrgId, Hour, ServiceName, TargetType, TargetSystem, TargetName, DeploymentEnv
        HAVING TargetName != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_children_mv TO service_map_children AS
SELECT
          OrgId,
          toDateTime(Timestamp) AS Timestamp,
          TraceId,
          ParentSpanId,
          ServiceName,
          SpanKind,
          Duration,
          StatusCode,
          TraceState,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv
        FROM traces
        WHERE SpanKind IN ('Server', 'Consumer')
          AND ParentSpanId != '';

CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_db_edges_hourly_mv TO service_map_db_edges_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) AS DbSystem,
          if(match(coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name']), '^([0-9a-fA-F]{32}|.*[.]hyperdrive[.]local)$'), 'hyperdrive', coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name'])) AS DbNamespace,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          count() AS CallCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sum(Duration / 1000000) AS DurationSumMs,
          max(Duration / 1000000) AS MaxDurationMs,
          countIf(TraceState LIKE '%th:%') AS SampledSpanCount,
          countIf(TraceState = '' OR TraceState NOT LIKE '%th:%') AS UnsampledSpanCount,
          sum(SampleRate) AS SampleRateSum,
          quantilesTDigestWeightedState(0.5, 0.95)(Duration, toUInt32(greatest(SampleRate, 1.0))) AS DurationQuantiles
        FROM traces
        WHERE SpanKind IN ('Client', 'Producer')
          AND coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) != ''
          AND ServiceName != ''
        GROUP BY OrgId, Hour, ServiceName, DbSystem, DbNamespace, DeploymentEnv;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_db_query_shapes_hourly_mv TO service_map_db_query_shapes_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) AS DbSystem,
          if(match(coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name']), '^([0-9a-fA-F]{32}|.*[.]hyperdrive[.]local)$'), 'hyperdrive', coalesce(nullIf(SpanAttributes['db.namespace'], ''), nullIf(SpanAttributes['db.name'], ''), nullIf(SpanAttributes['server.address'], ''), SpanAttributes['net.peer.name'])) AS DbNamespace,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          coalesce(
  nullIf(SpanAttributes['db.query.fingerprint'], ''),
  nullIf(SpanAttributes['db.statement.fingerprint'], ''),
  nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', toString(cityHash64(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(lower(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement'])), '\'[^\']*\'', '?'), '\\bin\\s*\\([^)]*\\)', 'in (?)'), '[0-9]+(\\.[0-9]+)?', '?'), '\\s+', ' '), '^\\s+|\\s+$', ''))), ''), ''),
  toString(cityHash64(coalesce(
  nullIf(SpanAttributes['db.query.summary'], ''),
  nullIf(if(SpanAttributes['db.operation.name'] != '', trimBoth(concat(SpanAttributes['db.operation.name'], if(coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace']) != '', concat(' ', coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace'])), ''))), ''), ''),
  nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', trimBoth(concat(upper(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '^\\s*(\\w+)')), if(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\s+\\W?([\\w.]+)') != '', concat(' ', extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\s+\\W?([\\w.]+)')), ''))), ''), ''),
  nullIf(SpanAttributes['query.context'], ''),
  nullIf(SpanAttributes['db.operation.name'], ''),
  nullIf(SpanAttributes['db.operation'], ''),
  SpanName
)))
) AS QueryKey,
          any(substring(coalesce(
  nullIf(SpanAttributes['db.query.summary'], ''),
  nullIf(if(SpanAttributes['db.operation.name'] != '', trimBoth(concat(SpanAttributes['db.operation.name'], if(coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace']) != '', concat(' ', coalesce(nullIf(SpanAttributes['db.collection.name'], ''), SpanAttributes['db.namespace'])), ''))), ''), ''),
  nullIf(if(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']) != '', trimBoth(concat(upper(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '^\\s*(\\w+)')), if(extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\s+\\W?([\\w.]+)') != '', concat(' ', extract(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), '(?i)(?:from|into|update|join|table)\\s+\\W?([\\w.]+)')), ''))), ''), ''),
  nullIf(SpanAttributes['query.context'], ''),
  nullIf(SpanAttributes['db.operation.name'], ''),
  nullIf(SpanAttributes['db.operation'], ''),
  SpanName
), 1, 220)) AS QueryLabel,
          any(substring(coalesce(nullIf(SpanAttributes['db.query.text'], ''), SpanAttributes['db.statement']), 1, 1000)) AS SampleStatement,
          count() AS CallCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sum(SampleRate) AS EstimatedCount,
          sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,
          sum(toFloat64(Duration) * SampleRate / 1000000) AS WeightedDurationSumMs,
          quantilesTDigestWeightedState(0.5, 0.95)(Duration, toUInt32(greatest(SampleRate, 1.0))) AS DurationQuantiles
        FROM traces
        WHERE SpanKind IN ('Client', 'Producer')
          AND coalesce(nullIf(SpanAttributes['db.system.name'], ''), SpanAttributes['db.system']) != ''
          AND ServiceName != ''
        GROUP BY OrgId, Hour, ServiceName, DbSystem, DbNamespace, DeploymentEnv, QueryKey;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_edges_hourly_ingest_mv TO service_map_edges_hourly AS
SELECT
          OrgId,
          Hour,
          SourceService,
          TargetService,
          DeploymentEnv,
          CallCount,
          ErrorCount,
          DurationSumMs,
          MaxDurationMs,
          SampledSpanCount,
          UnsampledSpanCount,
          SampleRateSum
        FROM service_map_edges_hourly_ingest;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_map_spans_mv TO service_map_spans AS
SELECT
          OrgId,
          toDateTime(Timestamp) AS Timestamp,
          TraceId,
          SpanId,
          ParentSpanId,
          ServiceName,
          SpanKind,
          Duration,
          StatusCode,
          TraceState,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv
        FROM traces
        WHERE SpanKind IN ('Client', 'Producer', 'Server', 'Consumer');

CREATE MATERIALIZED VIEW IF NOT EXISTS service_operations_hourly_mv TO service_operations_hourly AS
SELECT
          OrgId,
          toStartOfHour(Minute) AS Hour,
          ServiceName,
          DeploymentEnv,
          SpanName,
          sum(SpanCount) AS SpanCount,
          sum(EstimatedSpanCount) AS EstimatedSpanCount,
          sum(ErrorCount) AS ErrorCount,
          sum(EstimatedErrorCount) AS EstimatedErrorCount,
          sum(DurationSum) AS DurationSum,
          quantilesTDigestMergeState(0.5, 0.95)(DurationQuantiles) AS DurationQuantiles,
          sum(ClassifiedSpanCount) AS ClassifiedSpanCount,
          sum(ServerSpanCount) AS ServerSpanCount,
          sum(RoutedSpanCount) AS RoutedSpanCount
        FROM service_operations_minutely
        GROUP BY OrgId, Hour, ServiceName, DeploymentEnv, SpanName;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_operations_minutely_mv TO service_operations_minutely AS
SELECT
          OrgId,
          toStartOfMinute(toDateTime(Timestamp)) AS Minute,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          if(((SpanName LIKE 'http.server %' OR SpanName IN ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS')) AND (SpanAttributes['http.route'] != '' OR SpanAttributes['url.path'] != '')), concat(if(SpanName LIKE 'http.server %', replaceOne(SpanName, 'http.server ', ''), SpanName), ' ', if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], SpanAttributes['url.path'])), SpanName) AS SpanName,
          count() AS SpanCount,
          sum(SampleRate) AS EstimatedSpanCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,
          sum(toFloat64(Duration)) AS DurationSum,
          quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles,
          count() AS ClassifiedSpanCount,
          countIf(SpanKind IN ('Server', 'Consumer')) AS ServerSpanCount,
          countIf(SpanAttributes['http.route'] != '') AS RoutedSpanCount
        FROM traces
        GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, SpanName;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_overview_hourly_mv TO service_overview_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          ResourceAttributes['service.namespace'] AS ServiceNamespace,
          ResourceAttributes['vcs.ref.head.revision'] AS CommitSha,
          count() AS SpanCount,
          sum(SampleRate) AS EstimatedSpanCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,
          sum(toFloat64(Duration)) AS DurationSum,
          quantilesTDigestState(0.5, 0.95, 0.99)(Duration) AS DurationQuantiles,
          min(toDateTime(Timestamp)) AS FirstSeen,
          countIf(StatusCode != 'Error' AND Duration < 500000000) AS ApdexSatisfiedCount,
          countIf(StatusCode != 'Error' AND Duration >= 500000000 AND Duration < 2000000000) AS ApdexToleratingCount
        FROM traces
        WHERE SpanKind IN ('Server', 'Consumer') OR ParentSpanId = ''
        GROUP BY OrgId, Hour, ServiceName, DeploymentEnv, ServiceNamespace, CommitSha;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_overview_minutely_mv TO service_overview_minutely AS
SELECT
          OrgId,
          toStartOfMinute(toDateTime(Timestamp)) AS Minute,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          ResourceAttributes['service.namespace'] AS ServiceNamespace,
          ResourceAttributes['vcs.ref.head.revision'] AS CommitSha,
          count() AS SpanCount,
          sum(SampleRate) AS EstimatedSpanCount,
          countIf(StatusCode = 'Error') AS ErrorCount,
          sumIf(SampleRate, StatusCode = 'Error') AS EstimatedErrorCount,
          sum(toFloat64(Duration)) AS DurationSum,
          quantilesTDigestState(0.5, 0.95, 0.99)(Duration) AS DurationQuantiles,
          min(toDateTime(Timestamp)) AS FirstSeen,
          countIf(StatusCode != 'Error' AND Duration < 500000000) AS ApdexSatisfiedCount,
          countIf(StatusCode != 'Error' AND Duration >= 500000000 AND Duration < 2000000000) AS ApdexToleratingCount
        FROM traces
        WHERE SpanKind IN ('Server', 'Consumer') OR ParentSpanId = ''
        GROUP BY OrgId, Minute, ServiceName, DeploymentEnv, ServiceNamespace, CommitSha;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_overview_spans_mv TO service_overview_spans AS
SELECT
          OrgId,
          toDateTime(Timestamp) AS Timestamp,
          ServiceName,
          Duration,
          StatusCode,
          TraceState,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          ResourceAttributes['vcs.ref.head.revision'] AS CommitSha,
          SampleRate,
          ResourceAttributes['service.namespace'] AS ServiceNamespace,
          SpanKind,
          toUInt8(ParentSpanId = '') AS IsRoot
        FROM traces
        WHERE SpanKind IN ('Server', 'Consumer') OR ParentSpanId = '';

CREATE MATERIALIZED VIEW IF NOT EXISTS service_platforms_hourly_mv TO service_platforms_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          max(ResourceAttributes['k8s.cluster.name']) AS K8sCluster,
          max(ResourceAttributes['k8s.pod.name']) AS K8sPodName,
          max(ResourceAttributes['k8s.deployment.name']) AS K8sDeploymentName,
          max(ResourceAttributes['k8s.statefulset.name']) AS K8sStatefulSetName,
          max(ResourceAttributes['k8s.daemonset.name']) AS K8sDaemonSetName,
          max(ResourceAttributes['k8s.namespace.name']) AS K8sNamespaceName,
          max(ResourceAttributes['cloud.platform']) AS CloudPlatform,
          max(ResourceAttributes['cloud.provider']) AS CloudProvider,
          max(ResourceAttributes['faas.name']) AS FaasName,
          max(ResourceAttributes['maple.sdk.type']) AS MapleSdkType,
          max(ResourceAttributes['process.runtime.name']) AS ProcessRuntimeName,
          count() AS SpanCount
        FROM traces
        WHERE ServiceName != ''
        GROUP BY OrgId, Hour, ServiceName, DeploymentEnv;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_logs_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(TimestampTime) AS Hour,
          count() AS LogCount,
          sum(length(Body) + 200) AS LogSizeBytes,
          0 AS TraceCount,
          0 AS TraceSizeBytes,
          0 AS SumMetricCount,
          0 AS SumMetricSizeBytes,
          0 AS GaugeMetricCount,
          0 AS GaugeMetricSizeBytes,
          0 AS HistogramMetricCount,
          0 AS HistogramMetricSizeBytes,
          0 AS ExpHistogramMetricCount,
          0 AS ExpHistogramMetricSizeBytes
        FROM logs
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_metrics_exp_histogram_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          0 AS LogCount,
          0 AS LogSizeBytes,
          0 AS TraceCount,
          0 AS TraceSizeBytes,
          0 AS SumMetricCount,
          0 AS SumMetricSizeBytes,
          0 AS GaugeMetricCount,
          0 AS GaugeMetricSizeBytes,
          0 AS HistogramMetricCount,
          0 AS HistogramMetricSizeBytes,
          count() AS ExpHistogramMetricCount,
          count() * 300 AS ExpHistogramMetricSizeBytes
        FROM metrics_exponential_histogram
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_metrics_gauge_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          0 AS LogCount,
          0 AS LogSizeBytes,
          0 AS TraceCount,
          0 AS TraceSizeBytes,
          0 AS SumMetricCount,
          0 AS SumMetricSizeBytes,
          count() AS GaugeMetricCount,
          count() * 150 AS GaugeMetricSizeBytes,
          0 AS HistogramMetricCount,
          0 AS HistogramMetricSizeBytes,
          0 AS ExpHistogramMetricCount,
          0 AS ExpHistogramMetricSizeBytes
        FROM metrics_gauge
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_metrics_histogram_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          0 AS LogCount,
          0 AS LogSizeBytes,
          0 AS TraceCount,
          0 AS TraceSizeBytes,
          0 AS SumMetricCount,
          0 AS SumMetricSizeBytes,
          0 AS GaugeMetricCount,
          0 AS GaugeMetricSizeBytes,
          count() AS HistogramMetricCount,
          count() * 250 AS HistogramMetricSizeBytes,
          0 AS ExpHistogramMetricCount,
          0 AS ExpHistogramMetricSizeBytes
        FROM metrics_histogram
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_metrics_sum_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          0 AS LogCount,
          0 AS LogSizeBytes,
          0 AS TraceCount,
          0 AS TraceSizeBytes,
          count() AS SumMetricCount,
          count() * 150 AS SumMetricSizeBytes,
          0 AS GaugeMetricCount,
          0 AS GaugeMetricSizeBytes,
          0 AS HistogramMetricCount,
          0 AS HistogramMetricSizeBytes,
          0 AS ExpHistogramMetricCount,
          0 AS ExpHistogramMetricSizeBytes
        FROM metrics_sum
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS service_usage_traces_mv TO service_usage AS
SELECT
          OrgId,
          ServiceName,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          0 AS LogCount,
          0 AS LogSizeBytes,
          count() AS TraceCount,
          sum(length(SpanName) + 300) AS TraceSizeBytes,
          0 AS SumMetricCount,
          0 AS SumMetricSizeBytes,
          0 AS GaugeMetricCount,
          0 AS GaugeMetricSizeBytes,
          0 AS HistogramMetricCount,
          0 AS HistogramMetricSizeBytes,
          0 AS ExpHistogramMetricCount,
          0 AS ExpHistogramMetricSizeBytes
        FROM traces
        GROUP BY OrgId, ServiceName, Hour;

CREATE MATERIALIZED VIEW IF NOT EXISTS span_metrics_calls_hourly_mv TO span_metrics_calls_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(TimeUnix)) AS Hour,
          ServiceName,
          MetricName,
          Attributes['span.kind'] AS SpanKind,
          cityHash64(mapKeys(Attributes), mapValues(Attributes)) AS AttrFingerprint,
          cityHash64(mapKeys(ResourceAttributes), mapValues(ResourceAttributes)) AS ResourceFingerprint,
          StartTimeUnix,
          argMaxState(Value, TimeUnix) AS LastValue
        FROM metrics_sum
        -- 'traces.span.metrics.calls' is the name the collector actually emits:
        -- spanmetricsconnector output is namespaced by the pipeline it is attached
        -- to. Without it this MV matched nothing and the target sat at 0 rows since
        -- it was created, while ~880k rows / 2 days of the real counter flowed past
        -- into metrics_sum and every read fell back to the raw window-function scan
        -- (~7s p95 -- see queries/metrics.ts). Keep this list in sync with
        -- SPAN_METRICS_CALLS_NAMES on the read side.
        WHERE MetricName IN ('span.metrics.calls', 'calls', 'traces.span.metrics.calls') AND IsMonotonic
        GROUP BY OrgId, Hour, ServiceName, MetricName, SpanKind, AttrFingerprint, ResourceFingerprint, StartTimeUnix;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_facets_hourly_mv TO trace_facets_hourly AS
SELECT
          OrgId,
          toStartOfHour(Timestamp) AS Hour,
          ServiceName,
          SpanName,
          HttpMethod,
          HttpStatusCode,
          DeploymentEnv,
          ServiceNamespace,
          HasError,
          count() AS TraceCount,
          min(Duration) AS DurationMin,
          max(Duration) AS DurationMax,
          quantilesTDigestState(0.5, 0.95)(Duration) AS DurationQuantiles
        FROM trace_list_mv
        GROUP BY OrgId, Hour, ServiceName, SpanName, HttpMethod, HttpStatusCode, DeploymentEnv, ServiceNamespace, HasError;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_index_mv TO trace_index AS
SELECT
          OrgId,
          TraceId,
          ServiceName,
          toStartOfFiveMinutes(Timestamp) AS Bucket,
          min(Timestamp) AS MinTimestamp,
          max(Timestamp) AS MaxTimestamp
        FROM traces
        GROUP BY OrgId, TraceId, ServiceName, Bucket;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_list_mv_mv TO trace_list_mv AS
SELECT
          OrgId,
          TraceId,
          toDateTime(Timestamp) AS Timestamp,
          ServiceName,
          if(
            (SpanName LIKE 'http.server %' OR SpanName IN ('GET','POST','PUT','PATCH','DELETE','HEAD','OPTIONS'))
            AND (SpanAttributes['http.route'] != '' OR SpanAttributes['url.path'] != ''),
            concat(
              if(SpanName LIKE 'http.server %', replaceOne(SpanName, 'http.server ', ''), SpanName),
              ' ',
              if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], SpanAttributes['url.path'])
            ),
            SpanName
          ) AS SpanName,
          SpanKind,
          Duration,
          StatusCode,
          if(SpanAttributes['http.method'] != '', SpanAttributes['http.method'], SpanAttributes['http.request.method']) AS HttpMethod,
          if(SpanAttributes['http.route'] != '', SpanAttributes['http.route'], if(SpanAttributes['url.path'] != '', SpanAttributes['url.path'], SpanAttributes['http.target'])) AS HttpRoute,
          if(SpanAttributes['http.status_code'] != '', SpanAttributes['http.status_code'], SpanAttributes['http.response.status_code']) AS HttpStatusCode,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          toUInt8(
            StatusCode = 'Error'
            OR (SpanAttributes['http.status_code'] != '' AND toUInt16OrZero(SpanAttributes['http.status_code']) >= 500)
            OR (SpanAttributes['http.response.status_code'] != '' AND toUInt16OrZero(SpanAttributes['http.response.status_code']) >= 500)
          ) AS HasError,
          TraceState,
          ResourceAttributes['service.namespace'] AS ServiceNamespace
        FROM traces
        WHERE ParentSpanId = '';

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_resource_attribute_keys_mv TO attribute_keys_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          arrayJoin(mapKeys(ResourceAttributes)) AS AttributeKey,
          'resource' AS AttributeScope,
          count() AS UsageCount
        FROM traces
        WHERE ResourceAttributes != map()
        GROUP BY OrgId, Hour, AttributeKey, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_resource_attribute_values_mv TO attribute_values_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          AttributeKey,
          AttributeValue,
          'resource' AS AttributeScope,
          count() AS UsageCount
        FROM traces
        ARRAY JOIN
          mapKeys(ResourceAttributes) AS AttributeKey,
          mapValues(ResourceAttributes) AS AttributeValue
        WHERE AttributeValue != ''
          AND length(AttributeValue) <= 128
          AND NOT (length(AttributeValue) > 4 AND match(AttributeValue, '^[0-9]+([.][0-9]+)?$'))
          AND NOT match(AttributeKey, '(_id|[.]id|Id|_ns)$')
          AND AttributeKey NOT LIKE 'http.request.header.%'
          AND AttributeKey NOT LIKE 'http.response.header.%'
        GROUP BY OrgId, Hour, AttributeKey, AttributeValue, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_span_attribute_keys_mv TO attribute_keys_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          arrayJoin(mapKeys(SpanAttributes)) AS AttributeKey,
          'span' AS AttributeScope,
          count() AS UsageCount
        FROM traces
        WHERE SpanAttributes != map()
        GROUP BY OrgId, Hour, AttributeKey, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS trace_span_attribute_values_mv TO attribute_values_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          AttributeKey,
          AttributeValue,
          'span' AS AttributeScope,
          count() AS UsageCount
        FROM traces
        ARRAY JOIN
          mapKeys(SpanAttributes) AS AttributeKey,
          mapValues(SpanAttributes) AS AttributeValue
        WHERE AttributeValue != ''
          AND length(AttributeValue) <= 128
          AND NOT (length(AttributeValue) > 4 AND match(AttributeValue, '^[0-9]+([.][0-9]+)?$'))
          AND NOT match(AttributeKey, '(_id|[.]id|Id|_ns)$')
          AND AttributeKey NOT LIKE 'http.request.header.%'
          AND AttributeKey NOT LIKE 'http.response.header.%'
        GROUP BY OrgId, Hour, AttributeKey, AttributeValue, AttributeScope;

CREATE MATERIALIZED VIEW IF NOT EXISTS traces_aggregates_hourly_mv TO traces_aggregates_hourly AS
SELECT
          OrgId,
          toStartOfHour(toDateTime(Timestamp)) AS Hour,
          ServiceName,
          SpanName,
          SpanKind,
          StatusCode,
          IsEntryPoint,
          coalesce(nullIf(ResourceAttributes['deployment.environment.name'], ''), ResourceAttributes['deployment.environment']) AS DeploymentEnv,
          sum(SampleRate) AS WeightedCount,
          sum(toFloat64(Duration) * SampleRate) AS WeightedDurationSum,
          sumIf(SampleRate, StatusCode = 'Error') AS WeightedErrorCount,
          quantilesTDigestWeightedState(0.5, 0.95, 0.99)(Duration, toUInt32(SampleRate)) AS DurationQuantiles,
          min(Duration) AS DurationMin,
          max(Duration) AS DurationMax
        FROM traces
        GROUP BY OrgId, Hour, ServiceName, SpanName, SpanKind, StatusCode, IsEntryPoint, DeploymentEnv;