mirror of
https://github.com/SigNoz/signoz.git
synced 2026-07-13 01:50:32 +01:00
Compare commits
1 Commits
worktree-t
...
worktree-h
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6cba50133f |
@@ -729,7 +729,11 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
|
||||
return ErrorConditionLiteral
|
||||
}
|
||||
|
||||
value := params[1:]
|
||||
value, err := normalizeFunctionValue(operator, functionName, params[1:])
|
||||
if err != nil {
|
||||
v.errors = append(v.errors, err.Error())
|
||||
return ErrorConditionLiteral
|
||||
}
|
||||
|
||||
conds, ok := v.buildConditions(key, matchingFieldKeys(key, v.fieldKeys), operator, value)
|
||||
if !ok {
|
||||
@@ -745,6 +749,44 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
|
||||
return v.builder.Or(conds...)
|
||||
}
|
||||
|
||||
// normalizeFunctionValue validates and normalizes the value argument(s) of a has-family
|
||||
// function call, returning them in the wrapper slice the condition builder unwraps.
|
||||
//
|
||||
// - has/hasToken take exactly one scalar value. More than one argument, or an array
|
||||
// argument, is rejected rather than silently dropping the extras.
|
||||
// - hasAny/hasAll take a set of values, supplied either as a single array literal
|
||||
// (hasAny(k, ['a','b'])) or as several scalar arguments (hasAny(k, 'a', 'b')); the
|
||||
// latter are folded into one list so no argument is silently ignored.
|
||||
func normalizeFunctionValue(operator qbtypes.FilterOperator, functionName string, valueParams []any) (any, error) {
|
||||
switch operator {
|
||||
case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasToken:
|
||||
if len(valueParams) != 1 {
|
||||
return nil, fmt.Errorf("function `%s` expects exactly one value argument", functionName)
|
||||
}
|
||||
if _, isArray := valueParams[0].([]any); isArray {
|
||||
return nil, fmt.Errorf("function `%s` expects a single scalar value, not an array", functionName)
|
||||
}
|
||||
return valueParams, nil
|
||||
case qbtypes.FilterOperatorHasAny, qbtypes.FilterOperatorHasAll:
|
||||
// A single array literal is already the value set.
|
||||
if len(valueParams) == 1 {
|
||||
if _, isArray := valueParams[0].([]any); isArray {
|
||||
return valueParams, nil
|
||||
}
|
||||
}
|
||||
// Otherwise fold the positional scalar arguments into one list.
|
||||
values := make([]any, 0, len(valueParams))
|
||||
for _, p := range valueParams {
|
||||
if _, isArray := p.([]any); isArray {
|
||||
return nil, fmt.Errorf("function `%s` expects either a single array literal or scalar values, not a mix of the two", functionName)
|
||||
}
|
||||
values = append(values, p)
|
||||
}
|
||||
return []any{values}, nil
|
||||
}
|
||||
return valueParams, nil
|
||||
}
|
||||
|
||||
// VisitFunctionParamList handles the parameter list for function calls.
|
||||
func (v *filterExpressionVisitor) VisitFunctionParamList(ctx *grammar.FunctionParamListContext) any {
|
||||
params := ctx.AllFunctionParam()
|
||||
|
||||
@@ -40,9 +40,8 @@ func isBodyJSONSearch(key *telemetrytypes.TelemetryFieldKey, columns []*schema.C
|
||||
return false
|
||||
}
|
||||
|
||||
// conditionForArrayFunction builds `has/hasAny/hasAll(<arrayFieldExpr>, value)` over a
|
||||
// body JSON array field. The field expression uses the JSON accessor (flag on) or
|
||||
// legacy string extraction (flag off); value[0] is the needle.
|
||||
// conditionForArrayFunction builds has/hasAny/hasAll over a body JSON path — via the JSON
|
||||
// access plan (flag on) or legacy typed extraction (flag off).
|
||||
func (c *conditionBuilder) conditionForArrayFunction(
|
||||
ctx context.Context,
|
||||
startNs, endNs uint64,
|
||||
@@ -62,24 +61,48 @@ func (c *conditionBuilder) conditionForArrayFunction(
|
||||
needle = args[0]
|
||||
}
|
||||
|
||||
var fieldExpr string
|
||||
if c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
|
||||
fe, err := c.fm.FieldFor(ctx, startNs, endNs, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
fieldExpr = fe
|
||||
} else {
|
||||
// legacy string-body path; value drives array-type inference (e.g. `[*]` paths)
|
||||
fieldExpr, _ = GetBodyJSONKey(ctx, key, qbtypes.FilterOperatorUnknown, value)
|
||||
// JSON access plan: data-type collision handling, nested array paths.
|
||||
valueType, needle := InferDataType(needle, operator, key)
|
||||
return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, needle, sb)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), fieldExpr, sb.Var(needle)), nil
|
||||
// legacy string-body path: type-matched array extraction, OR-ed with a scalar comparison
|
||||
// for a scalar body value.
|
||||
elemType := legacyElemType(needle)
|
||||
arrayExpr := getBodyJSONArrayKey(key, elemType)
|
||||
scalarExpr, scalarGuard, hasScalar := getBodyJSONScalarKey(key, elemType)
|
||||
if list, ok := needle.([]any); ok {
|
||||
vals := make([]any, len(list))
|
||||
for i, v := range list {
|
||||
vals[i] = legacyCoerceNeedle(v, elemType)
|
||||
}
|
||||
arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(vals))
|
||||
if !hasScalar {
|
||||
return arrayCond, nil
|
||||
}
|
||||
var membership string
|
||||
if operator == qbtypes.FilterOperatorHasAll {
|
||||
eqs := make([]string, len(vals))
|
||||
for i, v := range vals {
|
||||
eqs[i] = sb.E(scalarExpr, v)
|
||||
}
|
||||
membership = sb.And(eqs...)
|
||||
} else {
|
||||
membership = sb.In(scalarExpr, vals...)
|
||||
}
|
||||
return fmt.Sprintf("(%s OR %s)", arrayCond, sb.And(membership, scalarGuard)), nil
|
||||
}
|
||||
typedNeedle := legacyCoerceNeedle(needle, elemType)
|
||||
arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(typedNeedle))
|
||||
if !hasScalar {
|
||||
return arrayCond, nil
|
||||
}
|
||||
return fmt.Sprintf("(%s OR %s)", arrayCond, sb.And(sb.E(scalarExpr, typedNeedle), scalarGuard)), nil
|
||||
}
|
||||
|
||||
// conditionForHasToken builds `hasToken(LOWER(<bodyColumn>), LOWER(<needle>))`, a
|
||||
// full-text token search over the body column. It resolves the column from the key
|
||||
// name + use_json_body flag, validates the field/value, and tags errors with the doc URL.
|
||||
// conditionForHasToken builds a hasToken full-text search over the body column, resolving the
|
||||
// column from the key name + use_json_body flag.
|
||||
func (c *conditionBuilder) conditionForHasToken(
|
||||
ctx context.Context,
|
||||
key *telemetrytypes.TelemetryFieldKey,
|
||||
@@ -92,28 +115,34 @@ func (c *conditionBuilder) conditionForHasToken(
|
||||
needle = args[0]
|
||||
}
|
||||
|
||||
// TODO(Tushar): thread orgID here to evaluate correctly
|
||||
bodyJSONEnabled := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
|
||||
|
||||
columnName := LogsV2BodyColumn
|
||||
if bodyJSONEnabled {
|
||||
if key.Name != LogsV2BodyColumn && key.Name != bodyMessageField {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `hasToken` only supports body/body.message field as first parameter").WithUrl(hasTokenFunctionDocURL)
|
||||
}
|
||||
columnName = bodyMessageField
|
||||
} else if key.Name != LogsV2BodyColumn {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
|
||||
}
|
||||
|
||||
// hasToken matches string tokens only.
|
||||
if _, ok := needle.(string); !ok {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `hasToken` expects value parameter to be a string").WithUrl(hasTokenFunctionDocURL)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", columnName, sb.Var(needle)), nil
|
||||
// TODO(Tushar): thread orgID here to evaluate correctly
|
||||
bodyJSONEnabled := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
|
||||
|
||||
if !bodyJSONEnabled {
|
||||
// legacy: token search over the plain body string column only.
|
||||
if key.Name != LogsV2BodyColumn {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
|
||||
}
|
||||
return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", LogsV2BodyColumn, sb.Var(needle)), nil
|
||||
}
|
||||
|
||||
// JSON mode: a bare body/body.message key searches the body.message column; any other body
|
||||
// field is a token search over its JSON string field, incl. strings nested in arrays.
|
||||
if key.Name == LogsV2BodyColumn || key.Name == bodyMessageField {
|
||||
return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", bodyMessageField, sb.Var(needle)), nil
|
||||
}
|
||||
if key.FieldContext == telemetrytypes.FieldContextBody {
|
||||
return NewJSONConditionBuilder(key, telemetrytypes.FieldDataTypeString).buildTokenFunctionCondition(needle, sb)
|
||||
}
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `hasToken` only supports the body field or a body JSON string field as first parameter").WithUrl(hasTokenFunctionDocURL)
|
||||
}
|
||||
|
||||
func (c *conditionBuilder) conditionFor(
|
||||
@@ -124,8 +153,7 @@ func (c *conditionBuilder) conditionFor(
|
||||
value any,
|
||||
sb *sqlbuilder.SelectBuilder,
|
||||
) (string, error) {
|
||||
// hasToken is a token search over the body column resolved purely from the key
|
||||
// name + flag, independent of column resolution, so handle it before anything else.
|
||||
// hasToken resolves from the key name + flag alone (no column resolution), so handle it first.
|
||||
if operator == qbtypes.FilterOperatorHasToken {
|
||||
return c.conditionForHasToken(ctx, key, value, sb)
|
||||
}
|
||||
@@ -135,8 +163,7 @@ func (c *conditionBuilder) conditionFor(
|
||||
return "", err
|
||||
}
|
||||
|
||||
// has/hasAny/hasAll build `has(<arrayFieldExpr>, value)` over body JSON arrays
|
||||
// rather than going through the normal operator paths, so handle them up front.
|
||||
// has/hasAny/hasAll take the body-JSON path, not the normal operator paths.
|
||||
if operator.IsArrayFunctionOperator() {
|
||||
return c.conditionForArrayFunction(ctx, startNs, endNs, key, operator, value, columns, sb)
|
||||
}
|
||||
@@ -402,23 +429,6 @@ func (c *conditionBuilder) ConditionFor(
|
||||
}
|
||||
}
|
||||
|
||||
// has/hasAny/hasAll need an array field: in JSON-body mode drop non-array matches so a
|
||||
// scalar errors clearly instead of failing at ClickHouse runtime (legacy mode skips this).
|
||||
if operator.IsArrayFunctionOperator() &&
|
||||
c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
|
||||
arrayKeys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(keys))
|
||||
for _, k := range keys {
|
||||
if k.FieldDataType.IsArray() {
|
||||
arrayKeys = append(arrayKeys, k)
|
||||
}
|
||||
}
|
||||
if len(arrayKeys) == 0 {
|
||||
return nil, warnings, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"function `%s` expects key parameter to be an array field; no array fields found", operator.FunctionName())
|
||||
}
|
||||
keys = arrayKeys
|
||||
}
|
||||
|
||||
conds := make([]string, 0, len(keys))
|
||||
for _, k := range keys {
|
||||
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
|
||||
|
||||
@@ -44,34 +44,66 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
|
||||
category: "json",
|
||||
query: "has(body.requestor_list[*], 'index_service')",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE has(JSONExtract(JSON_QUERY(body, '$."requestor_list"[*]'), 'Array(String)'), ?)`,
|
||||
expectedArgs: []any{"index_service"},
|
||||
expectedQuery: `WHERE (has(JSONExtract(JSON_QUERY(body, '$."requestor_list"[*]'), 'Array(Nullable(String))'), ?) OR (JSON_VALUE(body, '$."requestor_list"') = ? AND JSONType(body, 'requestor_list') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{"index_service", "index_service"},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "has(body.int_numbers[*], 2)",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE has(JSONExtract(JSON_QUERY(body, '$."int_numbers"[*]'), 'Array(Float64)'), ?)`,
|
||||
expectedArgs: []any{float64(2)},
|
||||
expectedQuery: `WHERE (has(JSONExtract(JSON_QUERY(body, '$."int_numbers"[*]'), 'Array(Nullable(Float64))'), ?) OR (JSONExtract(JSON_VALUE(body, '$."int_numbers"'), 'Nullable(Float64)') = ? AND JSONType(body, 'int_numbers') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{float64(2), float64(2)},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "has(body.bool[*], true)",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE has(JSONExtract(JSON_QUERY(body, '$."bool"[*]'), 'Array(Bool)'), ?)`,
|
||||
expectedArgs: []any{true},
|
||||
expectedQuery: `WHERE (has(JSONExtract(JSON_QUERY(body, '$."bool"[*]'), 'Array(Nullable(String))'), ?) OR (JSON_VALUE(body, '$."bool"') = ? AND JSONType(body, 'bool') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{"true", "true"},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "NOT has(body.nested_num[*].float_nums[*], 2.2)",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE NOT (has(JSONExtract(JSON_QUERY(body, '$."nested_num"[*]."float_nums"[*]'), 'Array(Float64)'), ?))`,
|
||||
expectedQuery: `WHERE NOT (has(JSONExtract(JSON_QUERY(body, '$."nested_num"[*]."float_nums"[*]'), 'Array(Nullable(Float64))'), ?))`,
|
||||
expectedArgs: []any{float64(2.2)},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "has(body.tags, 'production')",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE (has(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR (JSON_VALUE(body, '$."tags"') = ? AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{"production", "production"},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "hasAny(body.tags, ['critical', 'test'])",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE (hasAny(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR (JSON_VALUE(body, '$."tags"') IN (?, ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{[]any{"critical", "test"}, "critical", "test"},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "hasAll(body.tags, ['production', 'web'])",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE (hasAll(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ((JSON_VALUE(body, '$."tags"') = ? AND JSON_VALUE(body, '$."tags"') = ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{[]any{"production", "web"}, "production", "web"},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "has(body.ids, \"200\")",
|
||||
shouldPass: true,
|
||||
expectedQuery: `WHERE (has(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), ?) OR (JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ? AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')))`,
|
||||
expectedArgs: []any{int64(200), int64(200)},
|
||||
expectedErrorContains: "",
|
||||
},
|
||||
{
|
||||
category: "json",
|
||||
query: "body.message = hello",
|
||||
|
||||
@@ -1561,6 +1561,25 @@ func TestFilterExprLogs(t *testing.T) {
|
||||
expectedArgs: []any{"download"},
|
||||
expectedErrorContains: "function `hasToken` expects value parameter to be a string",
|
||||
},
|
||||
// extra / mis-shaped value arguments are rejected, not silently dropped.
|
||||
{
|
||||
category: "hasExtraArgs",
|
||||
query: "has(body.tags[*], \"a\", \"b\")",
|
||||
shouldPass: false,
|
||||
expectedErrorContains: "function `has` expects exactly one value argument",
|
||||
},
|
||||
{
|
||||
category: "hasArrayArg",
|
||||
query: "has(body.tags[*], [\"a\", \"b\"])",
|
||||
shouldPass: false,
|
||||
expectedErrorContains: "function `has` expects a single scalar value, not an array",
|
||||
},
|
||||
{
|
||||
category: "hasTokenExtraArgs",
|
||||
query: "hasToken(body, \"a\", \"b\")",
|
||||
shouldPass: false,
|
||||
expectedErrorContains: "function `hasToken` expects exactly one value argument",
|
||||
},
|
||||
|
||||
// Basic materialized key
|
||||
{
|
||||
|
||||
@@ -96,6 +96,18 @@ func applyNotCondition(operator qbtypes.FilterOperator) (bool, qbtypes.FilterOpe
|
||||
return false, operator
|
||||
}
|
||||
|
||||
// branchArrayExpr returns the ClickHouse array expression for a given array-type branch
|
||||
// at this hop. The JSON branch reads Array(JSON(...)) directly; the Dynamic branch filters
|
||||
// the Array(Dynamic) down to its JSON elements and maps them to JSON.
|
||||
func (c *jsonConditionBuilder) branchArrayExpr(node *telemetrytypes.JSONAccessNode, branch telemetrytypes.JSONAccessBranchType) string {
|
||||
fieldPath := node.FieldPath()
|
||||
if branch == telemetrytypes.BranchDynamic {
|
||||
dynBaseExpr := fmt.Sprintf("dynamicElement(%s, 'Array(Dynamic)')", fieldPath)
|
||||
return fmt.Sprintf("arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), %s))", dynBaseExpr)
|
||||
}
|
||||
return fmt.Sprintf("dynamicElement(%s, 'Array(JSON(max_dynamic_types=%d, max_dynamic_paths=%d))')", fieldPath, node.MaxDynamicTypes, node.MaxDynamicPaths)
|
||||
}
|
||||
|
||||
// buildAccessNodeBranches builds conditions for each branch of the access node.
|
||||
func (c *jsonConditionBuilder) buildAccessNodeBranches(current *telemetrytypes.JSONAccessNode, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
if current == nil {
|
||||
@@ -103,31 +115,15 @@ func (c *jsonConditionBuilder) buildAccessNodeBranches(current *telemetrytypes.J
|
||||
}
|
||||
|
||||
currAlias := current.Alias()
|
||||
fieldPath := current.FieldPath()
|
||||
// Determine availability of Array(JSON) and Array(Dynamic) at this hop
|
||||
hasArrayJSON := current.Branches[telemetrytypes.BranchJSON] != nil
|
||||
hasArrayDynamic := current.Branches[telemetrytypes.BranchDynamic] != nil
|
||||
|
||||
// Then, at this hop, compute child per branch and wrap
|
||||
// At this hop, compute the child condition per array branch (JSON before Dynamic) and
|
||||
// wrap each in arrayExists over the corresponding array expression.
|
||||
branches := make([]string, 0, 2)
|
||||
if hasArrayJSON {
|
||||
jsonArrayExpr := fmt.Sprintf("dynamicElement(%s, 'Array(JSON(max_dynamic_types=%d, max_dynamic_paths=%d))')", fieldPath, current.MaxDynamicTypes, current.MaxDynamicPaths)
|
||||
childGroupJSON, err := c.recurseArrayHops(current.Branches[telemetrytypes.BranchJSON], operator, value, sb)
|
||||
for _, branch := range current.BranchesInOrder() {
|
||||
childGroup, err := c.recurseArrayHops(current.Branches[branch], operator, value, sb)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
branches = append(branches, fmt.Sprintf("arrayExists(%s-> %s, %s)", currAlias, childGroupJSON, jsonArrayExpr))
|
||||
}
|
||||
if hasArrayDynamic {
|
||||
dynBaseExpr := fmt.Sprintf("dynamicElement(%s, 'Array(Dynamic)')", fieldPath)
|
||||
dynFilteredExpr := fmt.Sprintf("arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), %s))", dynBaseExpr)
|
||||
|
||||
// Create the Query for Dynamic array
|
||||
childGroupDyn, err := c.recurseArrayHops(current.Branches[telemetrytypes.BranchDynamic], operator, value, sb)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
branches = append(branches, fmt.Sprintf("arrayExists(%s-> %s, %s)", currAlias, childGroupDyn, dynFilteredExpr))
|
||||
branches = append(branches, fmt.Sprintf("arrayExists(%s-> %s, %s)", currAlias, childGroup, c.branchArrayExpr(current, branch)))
|
||||
}
|
||||
|
||||
if len(branches) == 1 {
|
||||
@@ -309,6 +305,169 @@ func (c *jsonConditionBuilder) buildArrayMembershipCondition(node *telemetrytype
|
||||
return fmt.Sprintf("arrayExists(%s -> %s, %s)", key, op, arrayExpr), nil
|
||||
}
|
||||
|
||||
// buildArrayFunctionCondition builds a has/hasAny/hasAll condition over a body JSON path,
|
||||
// with contains-all semantics uniform across every leaf shape:
|
||||
// - has(v) = the path HAS v
|
||||
// - hasAny([v...]) = the path has ANY listed value (OR of has)
|
||||
// - hasAll([v...]) = the path has ALL listed values (AND of has)
|
||||
//
|
||||
// "the path has v" is an existential match resolved per leaf shape: for an array-typed leaf
|
||||
// (top-level `body.tags`, or nested `body.education[].scores`) it is native membership; for a
|
||||
// scalar leaf — whether reached through an array hop (`body.items[].sku`) or a plain scalar
|
||||
// path (`body.level`) — it is `<elem> = v`, wrapped in arrayExists over any array hops. So
|
||||
// `hasAll(body.education[].name, ['a','b'])` = "some element is a AND some element is b", and
|
||||
// for a plain scalar hasAll collapses to has (a one-element set can hold at most one value).
|
||||
//
|
||||
// Element comparisons reuse DataTypeCollisionHandledFieldName so a numeric literal against an
|
||||
// Int64 array (or a numeric literal against a String array) no longer silently misses.
|
||||
func (c *jsonConditionBuilder) buildArrayFunctionCondition(operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
if len(c.key.JSONPlan) == 0 {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "function `%s` could not resolve a JSON access plan for field `%s`", operator.FunctionName(), c.key.Name)
|
||||
}
|
||||
|
||||
switch operator {
|
||||
case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasAny:
|
||||
return c.buildOredRootChains(func(node *telemetrytypes.JSONAccessNode) (string, error) {
|
||||
return c.arrayFunctionLeaf(node, operator, value, sb)
|
||||
}, sb)
|
||||
case qbtypes.FilterOperatorHasAll:
|
||||
// contains-all: AND of a per-value "has" so the AND sits outside the array hops.
|
||||
values := toAnyList(value)
|
||||
conditions := make([]string, 0, len(values))
|
||||
for _, v := range values {
|
||||
v := v
|
||||
cond, err := c.buildOredRootChains(func(node *telemetrytypes.JSONAccessNode) (string, error) {
|
||||
return c.arrayFunctionLeaf(node, qbtypes.FilterOperatorHas, v, sb)
|
||||
}, sb)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
conditions = append(conditions, cond)
|
||||
}
|
||||
if len(conditions) == 1 {
|
||||
return conditions[0], nil
|
||||
}
|
||||
return sb.And(conditions...), nil
|
||||
}
|
||||
return "", qbtypes.ErrUnsupportedOperator
|
||||
}
|
||||
|
||||
// buildOredRootChains applies leafFn down every JSONPlan root (base + promoted), wrapping each
|
||||
// in its arrayExists chain, and ORs the per-root results.
|
||||
func (c *jsonConditionBuilder) buildOredRootChains(leafFn func(*telemetrytypes.JSONAccessNode) (string, error), sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
conditions := make([]string, 0, len(c.key.JSONPlan))
|
||||
for _, root := range c.key.JSONPlan {
|
||||
cond, err := c.buildArrayExistsChain(root, leafFn, sb)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
conditions = append(conditions, cond)
|
||||
}
|
||||
if len(conditions) == 1 {
|
||||
return conditions[0], nil
|
||||
}
|
||||
return sb.Or(conditions...), nil
|
||||
}
|
||||
|
||||
// buildArrayExistsChain wraps the terminal condition (produced by leafFn) in an arrayExists
|
||||
// over every array hop between the root and the terminal. For a terminal root (a top-level
|
||||
// array leaf) it simply returns leafFn(root).
|
||||
func (c *jsonConditionBuilder) buildArrayExistsChain(node *telemetrytypes.JSONAccessNode, leafFn func(*telemetrytypes.JSONAccessNode) (string, error), sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
if node == nil {
|
||||
return "", errors.NewInternalf(CodeArrayNavigationFailed, "navigation failed, current node is nil")
|
||||
}
|
||||
if node.IsTerminal {
|
||||
return leafFn(node)
|
||||
}
|
||||
|
||||
branches := make([]string, 0, 2)
|
||||
for _, branch := range node.BranchesInOrder() {
|
||||
childCond, err := c.buildArrayExistsChain(node.Branches[branch], leafFn, sb)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
branches = append(branches, fmt.Sprintf("arrayExists(%s-> %s, %s)", node.Alias(), childCond, c.branchArrayExpr(node, branch)))
|
||||
}
|
||||
if len(branches) == 1 {
|
||||
return branches[0], nil
|
||||
}
|
||||
return sb.Or(branches...), nil
|
||||
}
|
||||
|
||||
// arrayFunctionLeaf builds the existential comparison for has/hasAny at a terminal node (hasAll
|
||||
// composes from has in buildArrayFunctionCondition). For an array leaf it delegates to native
|
||||
// membership; for a scalar leaf it compares the element directly.
|
||||
func (c *jsonConditionBuilder) arrayFunctionLeaf(node *telemetrytypes.JSONAccessNode, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
if node.TerminalConfig.ElemType.IsArray {
|
||||
return c.arrayLeafMembership(node, operator, value, sb)
|
||||
}
|
||||
|
||||
switch operator {
|
||||
case qbtypes.FilterOperatorHas:
|
||||
return c.arrayFuncScalarLeaf(node, qbtypes.FilterOperatorEqual, value, sb)
|
||||
case qbtypes.FilterOperatorHasAny:
|
||||
return c.arrayFuncScalarLeaf(node, qbtypes.FilterOperatorIn, toAnyList(value), sb)
|
||||
}
|
||||
return "", qbtypes.ErrUnsupportedOperator
|
||||
}
|
||||
|
||||
// arrayLeafMembership builds native membership for an array-typed leaf, reusing
|
||||
// buildArrayMembershipCondition (which handles data-type collisions on each element).
|
||||
func (c *jsonConditionBuilder) arrayLeafMembership(node *telemetrytypes.JSONAccessNode, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
switch operator {
|
||||
case qbtypes.FilterOperatorHas:
|
||||
return c.buildArrayMembershipCondition(node, qbtypes.FilterOperatorEqual, value, sb)
|
||||
case qbtypes.FilterOperatorHasAny:
|
||||
return c.buildArrayMembershipCondition(node, qbtypes.FilterOperatorIn, toAnyList(value), sb)
|
||||
}
|
||||
return "", qbtypes.ErrUnsupportedOperator
|
||||
}
|
||||
|
||||
// arrayFuncScalarLeaf builds `<elemExpr> <op> value` for a scalar leaf reached through an
|
||||
// array hop, applying data-type collision handling like the standard primitive path.
|
||||
func (c *jsonConditionBuilder) arrayFuncScalarLeaf(node *telemetrytypes.JSONAccessNode, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
fieldExpr := fmt.Sprintf("dynamicElement(%s, '%s')", node.FieldPath(), node.TerminalConfig.ElemType.StringValue())
|
||||
fieldExpr, value = querybuilder.DataTypeCollisionHandledFieldName(node.TerminalConfig.Key, value, fieldExpr, operator)
|
||||
return c.applyOperator(sb, fieldExpr, operator, value)
|
||||
}
|
||||
|
||||
// buildTokenFunctionCondition builds a hasToken search over a body JSON string field:
|
||||
// hasToken(LOWER(<elem>), LOWER(?)) wrapped in arrayExists over any array hops between the
|
||||
// root and the terminal. The field must resolve to a String leaf or a String array.
|
||||
func (c *jsonConditionBuilder) buildTokenFunctionCondition(needle any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
if len(c.key.JSONPlan) == 0 {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` could not resolve a JSON access plan for field `%s`", c.key.Name)
|
||||
}
|
||||
|
||||
return c.buildOredRootChains(func(node *telemetrytypes.JSONAccessNode) (string, error) {
|
||||
return c.tokenLeaf(node, needle, sb)
|
||||
}, sb)
|
||||
}
|
||||
|
||||
// tokenLeaf builds the hasToken match at a terminal node: a direct match for a String leaf,
|
||||
// or an arrayExists over the elements for a String array leaf. hasToken is string-only, so
|
||||
// any other element type is rejected.
|
||||
func (c *jsonConditionBuilder) tokenLeaf(node *telemetrytypes.JSONAccessNode, needle any, sb *sqlbuilder.SelectBuilder) (string, error) {
|
||||
switch node.TerminalConfig.ElemType {
|
||||
case telemetrytypes.String:
|
||||
fieldExpr := fmt.Sprintf("dynamicElement(%s, 'String')", node.FieldPath())
|
||||
return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", fieldExpr, sb.Var(needle)), nil
|
||||
case telemetrytypes.ArrayString:
|
||||
arrayExpr := fmt.Sprintf("dynamicElement(%s, '%s')", node.FieldPath(), node.TerminalConfig.ElemType.StringValue())
|
||||
return fmt.Sprintf("arrayExists(x -> hasToken(LOWER(x), LOWER(%s)), %s)", sb.Var(needle), arrayExpr), nil
|
||||
default:
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` only supports string fields; field `%s` is `%s`", c.key.Name, node.TerminalConfig.Key.FieldDataType.StringValue())
|
||||
}
|
||||
}
|
||||
|
||||
// toAnyList normalizes a has-family value into a slice; a scalar becomes a one-element list.
|
||||
func toAnyList(value any) []any {
|
||||
if list, ok := value.([]any); ok {
|
||||
return list
|
||||
}
|
||||
return []any{value}
|
||||
}
|
||||
|
||||
func (c *jsonConditionBuilder) applyOperator(sb *sqlbuilder.SelectBuilder, fieldExpr string, operator qbtypes.FilterOperator, value any) (string, error) {
|
||||
switch operator {
|
||||
case qbtypes.FilterOperatorEqual:
|
||||
|
||||
@@ -602,7 +602,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
|
||||
name: "Simple has filter",
|
||||
filter: "has(body.education[].parameters, 1.65)",
|
||||
expected: TestExpected{
|
||||
WhereClause: "(has(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))'), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?) OR has(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?))",
|
||||
WhereClause: "(arrayExists(`body_v2.education`-> arrayExists(x -> toFloat64(x) = ?, dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education`-> arrayExists(x -> accurateCastOrNull(x, 'Float64') = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))",
|
||||
Args: []any{1.65, 1.65, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
Warnings: []string{
|
||||
"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic].",
|
||||
@@ -613,8 +613,8 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
|
||||
name: "Flat path hasAll filter",
|
||||
filter: "hasAll(body.user.permissions, ['read', 'write'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "hasAll(dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))'), ?)",
|
||||
Args: []any{[]any{"read", "write"}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
WhereClause: "(arrayExists(x -> x = ?, dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))')) AND arrayExists(x -> x = ?, dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))')))",
|
||||
Args: []any{"read", "write", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -739,8 +739,8 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
|
||||
name: "Nested path hasAny filter",
|
||||
filter: "hasAny(education[].awards[].participated[].members, ['Piyush', 'Tushar'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "hasAny(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->arrayConcat(arrayMap(`body_v2.education[].awards`->arrayConcat(arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')), arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), arrayMap(`body_v2.education[].awards`->arrayConcat(arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=256))')), arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?)",
|
||||
Args: []any{[]any{"Piyush", "Tushar"}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
WhereClause: "arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> (arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> x IN (?, ?), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> x IN (?, ?), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> (arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> x IN (?, ?), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=256))')) OR arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> x IN (?, ?), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"Piyush", "Tushar", "Piyush", "Tushar", "Piyush", "Tushar", "Piyush", "Tushar", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
@@ -755,8 +755,8 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
|
||||
name: "dynamic_array_element_compare_HAS_STRING",
|
||||
filter: "has(interests[].entities[].product_codes, '2002')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "has(arrayFlatten(arrayConcat(arrayMap(`body_v2.interests`->arrayMap(`body_v2.interests[].entities`->dynamicElement(`body_v2.interests[].entities`.`product_codes`, 'Array(Dynamic)'), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?)",
|
||||
Args: []any{"2002", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
WhereClause: "arrayExists(`body_v2.interests`-> arrayExists(`body_v2.interests[].entities`-> arrayExists(x -> accurateCastOrNull(x, 'Float64') = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.interests[].entities`.`product_codes`, 'Array(Dynamic)'))), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{int64(2002), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
@@ -771,10 +771,126 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
|
||||
name: "dynamic_array_element_compare_HAS_INT",
|
||||
filter: "has(interests[].entities[].product_codes, 1001)",
|
||||
expected: TestExpected{
|
||||
WhereClause: "has(arrayFlatten(arrayConcat(arrayMap(`body_v2.interests`->arrayMap(`body_v2.interests[].entities`->dynamicElement(`body_v2.interests[].entities`.`product_codes`, 'Array(Dynamic)'), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?)",
|
||||
WhereClause: "arrayExists(`body_v2.interests`-> arrayExists(`body_v2.interests[].entities`-> arrayExists(x -> accurateCastOrNull(x, 'Float64') = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.interests[].entities`.`product_codes`, 'Array(Dynamic)'))), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{float64(1001), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
|
||||
// ── scalar leaf reached through an array ───
|
||||
{
|
||||
name: "Nested primitive leaf has",
|
||||
filter: "has(body.education[].name, 'IIT')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`name`, 'String') = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"IIT", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Nested primitive leaf hasAny",
|
||||
filter: "hasAny(body.education[].name, ['IIT', 'MIT'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`name`, 'String') IN (?, ?), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"IIT", "MIT", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
// contains-all: single-value hasAll over a nested leaf collapses to has.
|
||||
name: "Nested primitive leaf hasAll single collapses to has",
|
||||
filter: "hasAll(body.education[].name, 'IIT')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`name`, 'String') = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"IIT", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
// contains-all: "some element is IIT AND some element is MIT" (AND of per-value has).
|
||||
name: "Nested primitive leaf hasAll multi (contains-all)",
|
||||
filter: "hasAll(body.education[].name, ['IIT', 'MIT'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "(arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`name`, 'String') = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')) AND arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`name`, 'String') = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))",
|
||||
Args: []any{"IIT", "MIT", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
// ── numeric literal against an Int64 array is collision-handled ───
|
||||
{
|
||||
name: "Nested Int64 array has collision",
|
||||
filter: "has(body.education[].scores, 90)",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> arrayExists(x -> toFloat64(x) = ?, dynamicElement(`body_v2.education`.`scores`, 'Array(Nullable(Int64))')), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{float64(90), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
// ── hasAny folds multiple scalar arguments into one value set ─────
|
||||
{
|
||||
name: "hasAny folds multiple scalar args",
|
||||
filter: "hasAny(body.user.permissions, 'read', 'write')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(x -> x IN (?, ?), dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))'))",
|
||||
Args: []any{"read", "write", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
// ── hasToken over JSON string fields (nested leaf, top-level array, nested array) ──
|
||||
{
|
||||
name: "hasToken nested string leaf",
|
||||
filter: "hasToken(body.education[].name, 'harvard')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> hasToken(LOWER(dynamicElement(`body_v2.education`.`name`, 'String')), LOWER(?)), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"harvard", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "hasToken top-level string array",
|
||||
filter: "hasToken(body.user.permissions, 'admin')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(x -> hasToken(LOWER(x), LOWER(?)), dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))'))",
|
||||
Args: []any{"admin", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "hasToken nested string array",
|
||||
filter: "hasToken(body.education[].awards[].participated[].members, 'piyush')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> (arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> hasToken(LOWER(x), LOWER(?)), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> hasToken(LOWER(x), LOWER(?)), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> (arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> hasToken(LOWER(x), LOWER(?)), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=256))')) OR arrayExists(`body_v2.education[].awards[].participated`-> arrayExists(x -> hasToken(LOWER(x), LOWER(?)), dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))')), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))",
|
||||
Args: []any{"piyush", "piyush", "piyush", "piyush", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
|
||||
// ── scalar (non-array) leaf: treated as a single-element set ─────────────
|
||||
{
|
||||
name: "Scalar leaf has",
|
||||
filter: "has(body.user.name, 'alice')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "dynamicElement(body_v2.`user.name`, 'String') = ?",
|
||||
Args: []any{"alice", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Scalar leaf hasAny",
|
||||
filter: "hasAny(body.user.name, ['alice', 'bob'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "dynamicElement(body_v2.`user.name`, 'String') IN (?, ?)",
|
||||
Args: []any{"alice", "bob", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
// contains-all: single-value hasAll over a plain scalar collapses to has.
|
||||
name: "Scalar leaf hasAll single collapses to has",
|
||||
filter: "hasAll(body.user.name, 'alice')",
|
||||
expected: TestExpected{
|
||||
WhereClause: "dynamicElement(body_v2.`user.name`, 'String') = ?",
|
||||
Args: []any{"alice", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
// contains-all over a one-element set: the scalar must equal every value, so
|
||||
// distinct values never match.
|
||||
name: "Scalar leaf hasAll multi (contains-all)",
|
||||
filter: "hasAll(body.user.name, ['alice', 'bob'])",
|
||||
expected: TestExpected{
|
||||
WhereClause: "(dynamicElement(body_v2.`user.name`, 'String') = ? AND dynamicElement(body_v2.`user.name`, 'String') = ?)",
|
||||
Args: []any{"alice", "bob", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
|
||||
@@ -130,3 +130,121 @@ func GetBodyJSONKey(_ context.Context, key *telemetrytypes.TelemetryFieldKey, op
|
||||
func GetBodyJSONKeyForExists(_ context.Context, key *telemetrytypes.TelemetryFieldKey, _ qbtypes.FilterOperator, _ any) string {
|
||||
return fmt.Sprintf("JSON_EXISTS(body, '$.%s')", getBodyJSONPath(key))
|
||||
}
|
||||
|
||||
// legacyElemType infers the has-family element type from the needle (legacy has no schema). It
|
||||
// scans EVERY value so the chosen array type and all coerced needles agree — else ClickHouse
|
||||
// raises "no supertype ... String" (code 386). Int64 stays distinct from Float64 so a quoted
|
||||
// integer is exact past 2^53 (unquoted literals already arrive as float64, parsed upstream).
|
||||
func legacyElemType(needle any) telemetrytypes.FieldDataType {
|
||||
list, ok := needle.([]any)
|
||||
if !ok {
|
||||
list = []any{needle}
|
||||
}
|
||||
if len(list) == 0 {
|
||||
return telemetrytypes.FieldDataTypeString
|
||||
}
|
||||
allInt, allNumeric := true, true
|
||||
for _, v := range list {
|
||||
switch t := v.(type) {
|
||||
case float32, float64:
|
||||
allInt = false
|
||||
case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
|
||||
// integer Go types stay int-exact
|
||||
case string:
|
||||
if _, err := strconv.ParseInt(t, 10, 64); err != nil {
|
||||
allInt = false
|
||||
}
|
||||
if _, err := strconv.ParseFloat(t, 64); err != nil {
|
||||
allNumeric = false
|
||||
}
|
||||
default:
|
||||
// booleans (and anything else) -> String; a bool renders to 'true'/'false', so a
|
||||
// bool needle only matches genuine JSON booleans, not truthy numbers/strings.
|
||||
allInt, allNumeric = false, false
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case allInt:
|
||||
return telemetrytypes.FieldDataTypeInt64
|
||||
case allNumeric:
|
||||
return telemetrytypes.FieldDataTypeFloat64
|
||||
default:
|
||||
return telemetrytypes.FieldDataTypeString
|
||||
}
|
||||
}
|
||||
|
||||
// legacyCoerceNeedle coerces a needle to elem type dt so its bound-arg type matches the
|
||||
// extracted column (legacyElemType guarantees it's coercible).
|
||||
func legacyCoerceNeedle(v any, dt telemetrytypes.FieldDataType) any {
|
||||
switch dt {
|
||||
case telemetrytypes.FieldDataTypeInt64:
|
||||
if s, ok := v.(string); ok {
|
||||
if i, err := strconv.ParseInt(s, 10, 64); err == nil {
|
||||
return i
|
||||
}
|
||||
}
|
||||
return v
|
||||
case telemetrytypes.FieldDataTypeFloat64:
|
||||
if s, ok := v.(string); ok {
|
||||
f, _ := strconv.ParseFloat(s, 64)
|
||||
return f
|
||||
}
|
||||
return v
|
||||
default:
|
||||
return bodyArrayNeedleString(v)
|
||||
}
|
||||
}
|
||||
|
||||
// getBodyJSONArrayKey extracts the leaf as Array(Nullable(<dt>)) — Nullable so a value of a
|
||||
// different JSON type maps to NULL instead of corrupting (e.g. a non-numeric string → 0).
|
||||
func getBodyJSONArrayKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) string {
|
||||
arrKey := *key
|
||||
if !strings.HasSuffix(arrKey.Name, "[*]") && !strings.HasSuffix(arrKey.Name, "[]") {
|
||||
arrKey.Name += "[*]"
|
||||
}
|
||||
return fmt.Sprintf("JSONExtract(JSON_QUERY(body, '$.%s'), 'Array(Nullable(%s))')", getBodyJSONPath(&arrKey), dt.CHDataType())
|
||||
}
|
||||
|
||||
// getBodyJSONScalarKey builds the single-element-set fallback for a scalar body value: the leaf
|
||||
// extracted as a scalar of type dt, plus a guard restricting it to a genuinely scalar body. The
|
||||
// guard is required because JSON_VALUE returns '' for an array/object/missing value, which would
|
||||
// otherwise zero-value match (has(x,0) / has(x,false) / has(x,'') on any array). ok=false when
|
||||
// the path still traverses an array ([*]/[]).
|
||||
func getBodyJSONScalarKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) (expr string, guard string, ok bool) {
|
||||
name := strings.TrimSuffix(strings.TrimSuffix(key.Name, "[*]"), "[]")
|
||||
if strings.Contains(name, "[") {
|
||||
return "", "", false
|
||||
}
|
||||
scalarKey := *key
|
||||
scalarKey.Name = name
|
||||
path := getBodyJSONPath(&scalarKey)
|
||||
if dt == telemetrytypes.FieldDataTypeString {
|
||||
expr = fmt.Sprintf("JSON_VALUE(body, '$.%s')", path)
|
||||
} else {
|
||||
// Nullable so a scalar of a different type (e.g. a bool/string where a number is
|
||||
// searched) extracts to NULL rather than the type's default (0/false), which would
|
||||
// otherwise zero-value match has(x, 0).
|
||||
expr = fmt.Sprintf("JSONExtract(JSON_VALUE(body, '$.%s'), 'Nullable(%s)')", path, dt.CHDataType())
|
||||
}
|
||||
keys := strings.Split(name, ".")
|
||||
for i, k := range keys {
|
||||
keys[i] = "'" + k + "'"
|
||||
}
|
||||
guard = fmt.Sprintf("JSONType(body, %s) NOT IN ('Array', 'Object', 'Null')", strings.Join(keys, ", "))
|
||||
return expr, guard, true
|
||||
}
|
||||
|
||||
func bodyArrayNeedleString(v any) string {
|
||||
switch t := v.(type) {
|
||||
case string:
|
||||
return t
|
||||
case bool:
|
||||
return strconv.FormatBool(t)
|
||||
case float64:
|
||||
return strconv.FormatFloat(t, 'f', -1, 64)
|
||||
case float32:
|
||||
return strconv.FormatFloat(float64(t), 'f', -1, 64)
|
||||
default:
|
||||
return fmt.Sprintf("%v", t)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -495,8 +495,8 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
|
||||
Limit: 10,
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(String)'), ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Args: []any{"john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(Nullable(String))'), ?) OR (JSON_VALUE(body, '$.\"user_names\"') = ? AND JSONType(body, 'user_names') NOT IN ('Array', 'Object', 'Null'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Args: []any{"john_doe", "john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
},
|
||||
|
||||
@@ -3,17 +3,18 @@ from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
from fixtures.querier import get_rows, make_query_request
|
||||
|
||||
# Positive coverage for the body-JSON array functions hasAny / hasAll in JSON-body
|
||||
# mode (BODY_JSON_QUERY_ENABLED=true). Here body_v2 arrays resolve to real ClickHouse
|
||||
# Arrays via dynamicElement, so these succeed — unlike legacy body mode, where
|
||||
# hasAny/hasAll xfail (see querierlogs/09_json_body_functions.py).
|
||||
# export_json_types registers the body paths + array element types (tags -> []string,
|
||||
# ids -> []int64) so the builder resolves body.tags/body.ids as arrays.
|
||||
from fixtures.querier import (
|
||||
RequestType,
|
||||
build_order_by,
|
||||
build_raw_query,
|
||||
get_rows,
|
||||
make_query_request,
|
||||
)
|
||||
|
||||
|
||||
def test_logs_json_body_has_any_string(
|
||||
@@ -23,7 +24,7 @@ def test_logs_json_body_has_any_string(
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""hasAny over a []string body array: matches logs sharing ANY listed value."""
|
||||
"""hasAny over a []string body array matches logs sharing any listed value."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body_v2=json.dumps({"tags": ["production", "api", "critical"]}), body_promoted="", severity_text="INFO"),
|
||||
@@ -34,16 +35,15 @@ def test_logs_json_body_has_any_string(
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# log1 has "critical", log2 has "test", log3 has neither -> 2 matches
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="raw",
|
||||
queries=[{"type": "builder_query", "spec": {"name": "A", "signal": "logs", "disabled": False, "limit": 100, "offset": 0, "filter": {"expression": "hasAny(body.tags, ['critical', 'test'])"}, "order": [{"key": {"name": "timestamp"}, "direction": "desc"}], "aggregations": [{"expression": "count()"}]}}],
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAny(body.tags, ['critical', 'test'])")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all(("critical" in row["data"]["body"]["tags"]) or ("test" in row["data"]["body"]["tags"]) for row in rows)
|
||||
@@ -56,7 +56,7 @@ def test_logs_json_body_has_all_string(
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""hasAll over a []string body array: matches only logs having ALL listed values."""
|
||||
"""hasAll over a []string body array matches only logs having all listed values."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"tags": ["production", "api", "critical"]}), body_promoted="", severity_text="INFO"),
|
||||
@@ -66,16 +66,15 @@ def test_logs_json_body_has_all_string(
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# only the second log has both "production" AND "web"
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="raw",
|
||||
queries=[{"type": "builder_query", "spec": {"name": "A", "signal": "logs", "disabled": False, "limit": 100, "offset": 0, "filter": {"expression": "hasAll(body.tags, ['production', 'web'])"}, "order": [{"key": {"name": "timestamp"}, "direction": "desc"}], "aggregations": [{"expression": "count()"}]}}],
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAll(body.tags, ['production', 'web'])")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1
|
||||
tags = rows[0]["data"]["body"]["tags"]
|
||||
@@ -99,16 +98,382 @@ def test_logs_json_body_has_any_number(
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# only the first log has 300 in ids; 999 matches nothing
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="raw",
|
||||
queries=[{"type": "builder_query", "spec": {"name": "A", "signal": "logs", "disabled": False, "limit": 100, "offset": 0, "filter": {"expression": "hasAny(body.ids, [300, 999])"}, "order": [{"key": {"name": "timestamp"}, "direction": "desc"}], "aggregations": [{"expression": "count()"}]}}],
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAny(body.ids, [300, 999])")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1
|
||||
assert 300 in rows[0]["data"]["body"]["ids"]
|
||||
|
||||
|
||||
def test_logs_json_body_array_membership(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has(body.<array>, value) matches logs whose body array contains the value, across
|
||||
string, numeric and boolean element types."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body_v2=json.dumps({"tags": ["production", "api", "critical"], "ids": [100, 200, 300], "flags": [True, False, True]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"tags": ["staging", "api", "test"], "ids": [200, 400, 500], "flags": [False, False, True]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"tags": ["production", "web", "important"], "ids": [100, 600, 700], "flags": [True, True, False]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='has(body.tags, "production")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all("production" in row["data"]["body"]["tags"] for row in rows)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.ids, 200)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all(200 in row["data"]["body"]["ids"] for row in rows)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.flags, true)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 3
|
||||
assert all(True in row["data"]["body"]["flags"] for row in rows)
|
||||
|
||||
|
||||
def test_logs_json_body_has_token(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""hasToken(body, token) matches logs whose body message contains the token."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"message": "request served from production node", "tags": ["api"]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"message": "request served from staging node", "tags": ["api"]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='hasToken(body, "production")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1
|
||||
assert "production" in rows[0]["data"]["body"]["message"]
|
||||
|
||||
|
||||
def test_logs_json_body_function_scalar_path(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has/hasAny/hasAll on a scalar (non-array) body path treat the scalar as a single-element
|
||||
set: has = equals, hasAny/hasAll = membership. So has and hasAll coincide for a scalar."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"level": "info", "code": 200}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"level": "debug", "code": 500}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
# Each expression matches only the first log (level=info, code=200).
|
||||
for expression in ["has(body.level, 'info')", "hasAny(body.code, [200, 999])", "hasAll(body.level, ['info'])"]:
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: expected 200, got {response.status_code}: {response.text}"
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1, f"{expression}: expected 1 match, got {len(rows)}"
|
||||
assert rows[0]["data"]["body"]["level"] == "info", f"{expression}: matched wrong row: {rows[0]['data']['body']}"
|
||||
|
||||
# hasAll is contains-all: a scalar (single-element set) can't hold two distinct values, so nothing matches.
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAll(body.level, ['info', 'debug'])")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert get_rows(response) == [], "hasAll(scalar, [two distinct values]) should match nothing (contains-all)"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"expression",
|
||||
[
|
||||
pytest.param('has(code.function, "main")', id="has_non_body_key"),
|
||||
pytest.param('hasToken(code.function, "main")', id="hastoken_non_body_key"),
|
||||
pytest.param("hasToken(body, 123)", id="hastoken_non_string_value"),
|
||||
],
|
||||
)
|
||||
def test_logs_json_body_function_errors(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
expression: str,
|
||||
) -> None:
|
||||
"""has/hasToken misuse (non-body key, non-string token) is rejected (400)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"expression, expected_error",
|
||||
[
|
||||
pytest.param('has(body.tags, "a", "b")', "expects exactly one value argument", id="has_multiple_values"),
|
||||
pytest.param('hasToken(body, "a", "b")', "expects exactly one value argument", id="hastoken_multiple_values"),
|
||||
pytest.param('has(body.tags, ["a", "b"])', "expects a single scalar value, not an array", id="has_array_value"),
|
||||
pytest.param('hasAny(body.tags, ["a"], "b")', "not a mix of the two", id="hasany_mixed_array_and_scalar"),
|
||||
],
|
||||
)
|
||||
def test_logs_json_body_function_argument_shape_errors(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
expression: str,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
"""has-family value arguments are shape-validated; mis-shaped arguments are rejected (400)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST, f"{expression}: expected 400, got {response.status_code}: {response.text}"
|
||||
assert expected_error in response.text, f"{expression}: expected error {expected_error!r} in {response.text}"
|
||||
|
||||
|
||||
|
||||
def test_logs_json_body_large_integer_exact(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""Large ids (> 2^53) stay exact only when QUOTED (compared as Int64). An unquoted numeric
|
||||
literal is parsed as float64 upstream, so it false-matches neighbouring ids (documented)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"id": [1234567890123456789]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"id": [1234567890123456788]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
# Quoted -> Int64 -> exact: matches only the exact id.
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='has(body.id, "1234567890123456789")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
# exactly one match (the query compares as Int64); the returned body value itself is
|
||||
# float64-serialized in the response, so assert on the match count, not the echoed id.
|
||||
assert len(get_rows(response)) == 1, "quoted big id should match exactly one log"
|
||||
|
||||
# Unquoted -> float64 upstream -> imprecise: false-matches the off-by-one neighbour too.
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.id, 1234567890123456789)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert len(get_rows(response)) == 2, "unquoted big id collides with neighbour (float64 precision) - quote large ids"
|
||||
|
||||
|
||||
def test_logs_json_body_no_supertype_safety(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""A needle whose type doesn't match the body array returns 200 with no match (never a
|
||||
ClickHouse type error)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"names": ["alice", "bob"]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"nums": [100, 200]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression in ["has(body.names, 200)", "has(body.nums, 'alice')"]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: expected 200, got {response.status_code}: {response.text}"
|
||||
assert get_rows(response) == [], f"{expression}: type mismatch should match nothing"
|
||||
|
||||
|
||||
def test_logs_json_body_type_isolation(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""JSON-body mode is inherently free of the legacy coercion quirks (it compares typed
|
||||
dynamicElement values off body_v2): a bool needle matches only genuine booleans, and
|
||||
empty-string / zero needles never false-match array bodies. Genuine scalars still match."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=4), resources={"service.name": "app-service"}, body_v2=json.dumps({"nums": [100, 200, 300]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body_v2=json.dumps({"flags": [True, False]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"names": ["a", "b"]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"code": 0}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [
|
||||
("has(body.nums, true)", 0), # bool needle does NOT truthy-match a numeric array
|
||||
("has(body.flags, true)", 1), # genuine bool array matches
|
||||
("has(body.names, '')", 0), # empty-string does NOT false-match an array
|
||||
("has(body.nums, 0)", 0), # zero does NOT false-match an array without a 0
|
||||
("has(body.code, 0)", 1), # genuine scalar zero matches
|
||||
]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
|
||||
def test_logs_json_body_dynamic_array(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has() over a mixed-type body array (stored as Array(Dynamic)) matches by the needle's
|
||||
type: a numeric needle matches numeric elements, a string needle matches string elements."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"mix": [100, "abc", 300]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [
|
||||
("has(body.mix, 100)", 1),
|
||||
("has(body.mix, 'abc')", 1),
|
||||
("has(body.mix, 999)", 0),
|
||||
("has(body.mix, 'xyz')", 0),
|
||||
]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
|
||||
def test_logs_json_body_nested_array_path(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
export_json_types: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has over a scalar leaf reached through an array (body.education[].name) checks each element."""
|
||||
now = datetime.now(tz=UTC)
|
||||
logs = [
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body_v2=json.dumps({"education": [{"name": "IIT"}, {"name": "MIT"}]}), body_promoted="", severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body_v2=json.dumps({"education": [{"name": "Stanford"}]}), body_promoted="", severity_text="INFO"),
|
||||
]
|
||||
export_json_types(logs)
|
||||
insert_logs(logs)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [("has(body.education[].name, 'MIT')", 1), ("has(body.education[].name, 'Yale')", 0)]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
@@ -563,214 +563,6 @@ def test_logs_json_body_nested_keys(
|
||||
assert all(code == 200 for code in status_codes)
|
||||
|
||||
|
||||
def test_logs_json_body_array_membership(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""
|
||||
Setup:
|
||||
Insert logs with JSON bodies containing arrays
|
||||
|
||||
Tests:
|
||||
1. Search by has(body.tags[*], "value") - string array
|
||||
2. Search by has(body.ids[*], 123) - numeric array
|
||||
3. Search by has(body.flags[*], true) - boolean array
|
||||
"""
|
||||
now = datetime.now(tz=UTC)
|
||||
|
||||
log1_body = json.dumps(
|
||||
{
|
||||
"tags": ["production", "api", "critical"],
|
||||
"ids": [100, 200, 300],
|
||||
"flags": [True, False, True],
|
||||
"users": [
|
||||
{"name": "alice", "role": "admin"},
|
||||
{"name": "bob", "role": "user"},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
log2_body = json.dumps(
|
||||
{
|
||||
"tags": ["staging", "api", "test"],
|
||||
"ids": [200, 400, 500],
|
||||
"flags": [False, False, True],
|
||||
"users": [
|
||||
{"name": "charlie", "role": "user"},
|
||||
{"name": "david", "role": "admin"},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
log3_body = json.dumps(
|
||||
{
|
||||
"tags": ["production", "web", "important"],
|
||||
"ids": [100, 600, 700],
|
||||
"flags": [True, True, False],
|
||||
"users": [
|
||||
{"name": "alice", "role": "admin"},
|
||||
{"name": "eve", "role": "user"},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=3),
|
||||
resources={"service.name": "app-service"},
|
||||
attributes={},
|
||||
body=log1_body,
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=2),
|
||||
resources={"service.name": "app-service"},
|
||||
attributes={},
|
||||
body=log2_body,
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
resources={"service.name": "app-service"},
|
||||
attributes={},
|
||||
body=log3_body,
|
||||
severity_text="INFO",
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# Test 1: Search by has(body.tags[*], "production")
|
||||
response = requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v5/query_range"),
|
||||
timeout=2,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json={
|
||||
"schemaVersion": "v1",
|
||||
"start": int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
"end": int(now.timestamp() * 1000),
|
||||
"requestType": "raw",
|
||||
"compositeQuery": {
|
||||
"queries": [
|
||||
{
|
||||
"type": "builder_query",
|
||||
"spec": {
|
||||
"name": "A",
|
||||
"signal": "logs",
|
||||
"disabled": False,
|
||||
"limit": 100,
|
||||
"offset": 0,
|
||||
"filter": {"expression": 'has(body.tags[*], "production")'},
|
||||
"order": [
|
||||
{"key": {"name": "timestamp"}, "direction": "desc"},
|
||||
],
|
||||
"aggregations": [{"expression": "count()"}],
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
"formatOptions": {"formatTableResultForUI": False, "fillGaps": False},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["status"] == "success"
|
||||
results = response.json()["data"]["data"]["results"]
|
||||
assert len(results) == 1
|
||||
rows = results[0]["rows"]
|
||||
assert len(rows) == 2 # log1 and log3 have "production" in tags
|
||||
tags_list = [json.loads(row["data"]["body"])["tags"] for row in rows]
|
||||
assert all("production" in tags for tags in tags_list)
|
||||
|
||||
# Test 2: Search by has(body.ids[*], 200)
|
||||
response = requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v5/query_range"),
|
||||
timeout=2,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json={
|
||||
"schemaVersion": "v1",
|
||||
"start": int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
"end": int(now.timestamp() * 1000),
|
||||
"requestType": "raw",
|
||||
"compositeQuery": {
|
||||
"queries": [
|
||||
{
|
||||
"type": "builder_query",
|
||||
"spec": {
|
||||
"name": "A",
|
||||
"signal": "logs",
|
||||
"disabled": False,
|
||||
"limit": 100,
|
||||
"offset": 0,
|
||||
"filter": {"expression": "has(body.ids[*], 200)"},
|
||||
"order": [
|
||||
{"key": {"name": "timestamp"}, "direction": "desc"},
|
||||
],
|
||||
"aggregations": [{"expression": "count()"}],
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
"formatOptions": {"formatTableResultForUI": False, "fillGaps": False},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["status"] == "success"
|
||||
results = response.json()["data"]["data"]["results"]
|
||||
assert len(results) == 1
|
||||
rows = results[0]["rows"]
|
||||
assert len(rows) == 2 # log1 and log2 have 200 in ids
|
||||
ids_list = [json.loads(row["data"]["body"])["ids"] for row in rows]
|
||||
assert all(200 in ids for ids in ids_list)
|
||||
|
||||
# Test 3: Search by has(body.flags[*], true)
|
||||
response = requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v5/query_range"),
|
||||
timeout=2,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json={
|
||||
"schemaVersion": "v1",
|
||||
"start": int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
"end": int(now.timestamp() * 1000),
|
||||
"requestType": "raw",
|
||||
"compositeQuery": {
|
||||
"queries": [
|
||||
{
|
||||
"type": "builder_query",
|
||||
"spec": {
|
||||
"name": "A",
|
||||
"signal": "logs",
|
||||
"disabled": False,
|
||||
"limit": 100,
|
||||
"offset": 0,
|
||||
"filter": {"expression": "has(body.flags[*], true)"},
|
||||
"order": [
|
||||
{"key": {"name": "timestamp"}, "direction": "desc"},
|
||||
],
|
||||
"aggregations": [{"expression": "count()"}],
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
"formatOptions": {"formatTableResultForUI": False, "fillGaps": False},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["status"] == "success"
|
||||
results = response.json()["data"]["data"]["results"]
|
||||
assert len(results) == 1
|
||||
rows = results[0]["rows"]
|
||||
assert len(rows) == 3 # All logs have true in flags
|
||||
flags_list = [json.loads(row["data"]["body"])["flags"] for row in rows]
|
||||
assert all(True in flags for flags in flags_list)
|
||||
|
||||
|
||||
def test_logs_json_body_listing(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
|
||||
@@ -16,10 +16,161 @@ from fixtures.querier import (
|
||||
make_query_request,
|
||||
)
|
||||
|
||||
# Body-JSON array/token functions on the logs body. has() success paths are already
|
||||
# covered by 06_json_body.py::test_logs_json_body_array_membership; this file adds the
|
||||
# sibling functions hasAny / hasAll / hasToken (success) and the function-operator error
|
||||
# paths (has/hasToken on a non-body key, non-string token) which must be rejected.
|
||||
|
||||
def test_logs_json_body_array_membership(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has(body.<array>[*], value) matches logs whose body array contains the value, across
|
||||
string, numeric and boolean element types."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=3),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"tags": ["production", "api", "critical"], "ids": [100, 200, 300], "flags": [True, False, True]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=2),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"tags": ["staging", "api", "test"], "ids": [200, 400, 500], "flags": [False, False, True]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"tags": ["production", "web", "important"], "ids": [100, 600, 700], "flags": [True, True, False]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='has(body.tags[*], "production")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all("production" in json.loads(row["data"]["body"])["tags"] for row in rows)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.ids[*], 200)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all(200 in json.loads(row["data"]["body"])["ids"] for row in rows)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.flags[*], true)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 3
|
||||
assert all(True in json.loads(row["data"]["body"])["flags"] for row in rows)
|
||||
|
||||
|
||||
def test_logs_json_body_has_type_collision(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""A numeric-looking string needle matches a numeric body array (the needle is coerced to a number)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=2),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"ids": [100, 200, 300]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"ids": [400, 500]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='has(body.ids, "200")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1
|
||||
assert 200 in json.loads(rows[0]["data"]["body"])["ids"]
|
||||
|
||||
|
||||
def test_logs_json_body_has_mixed_type_array(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has() over a mixed-type body array must not error (no ClickHouse "no supertype"). The
|
||||
needle's type selects the view: a numeric needle matches numeric elements (non-numeric →
|
||||
NULL, skipped), a string needle matches every element rendered as text."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"ids": [100, "abc", 300]}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [
|
||||
("has(body.ids, 100)", 1), # numeric needle → matches the numeric element
|
||||
("has(body.ids, 999)", 0), # numeric needle → no match, non-numeric "abc" is NULL not 0
|
||||
("has(body.ids, 'abc')", 1), # string needle → matches the string element
|
||||
]:
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: expected 200, got {response.status_code}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} match(es)"
|
||||
|
||||
|
||||
def test_logs_json_body_has_any(
|
||||
@@ -28,7 +179,7 @@ def test_logs_json_body_has_any(
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""hasAny(body.tags, [...]) matches a log whose array shares ANY value."""
|
||||
"""hasAny(body.tags, [...]) matches a log whose array shares any listed value."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
@@ -54,26 +205,18 @@ def test_logs_json_body_has_any(
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# log1 has "critical", log2 has "test", log3 has neither -> 2 matches
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[
|
||||
build_raw_query(
|
||||
"A",
|
||||
"logs",
|
||||
order=[build_order_by("timestamp")],
|
||||
limit=100,
|
||||
filter_expression="hasAny(body.tags, ['critical', 'test'])",
|
||||
)
|
||||
],
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAny(body.tags, ['critical', 'test'])")],
|
||||
)
|
||||
# BUG: hasAny on a flat body array path extracts a scalar String; ClickHouse rejects
|
||||
# it ("Argument 0 ... must be an array") -> HTTP 500.
|
||||
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 2
|
||||
assert all(("critical" in json.loads(row["data"]["body"])["tags"]) or ("test" in json.loads(row["data"]["body"])["tags"]) for row in rows)
|
||||
|
||||
|
||||
def test_logs_json_body_has_all(
|
||||
@@ -82,7 +225,7 @@ def test_logs_json_body_has_all(
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""hasAll(body.tags, [...]) matches only a log whose array has ALL values."""
|
||||
"""hasAll(body.tags, [...]) matches only a log whose array has all listed values."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
@@ -102,26 +245,19 @@ def test_logs_json_body_has_all(
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# only the second log has both "production" AND "web"
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[
|
||||
build_raw_query(
|
||||
"A",
|
||||
"logs",
|
||||
order=[build_order_by("timestamp")],
|
||||
limit=100,
|
||||
filter_expression="hasAll(body.tags, ['production', 'web'])",
|
||||
)
|
||||
],
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAll(body.tags, ['production', 'web'])")],
|
||||
)
|
||||
# BUG: hasAll on a flat body array path extracts a scalar String; ClickHouse rejects
|
||||
# it ("Argument 0 ... must be an array") -> HTTP 500.
|
||||
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1
|
||||
tags = json.loads(rows[0]["data"]["body"])["tags"]
|
||||
assert "production" in tags and "web" in tags
|
||||
|
||||
|
||||
def test_logs_json_body_has_token(
|
||||
@@ -156,22 +292,13 @@ def test_logs_json_body_has_token(
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# "production" appears in the first and third log bodies
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[
|
||||
build_raw_query(
|
||||
"A",
|
||||
"logs",
|
||||
order=[build_order_by("timestamp")],
|
||||
limit=100,
|
||||
filter_expression='hasToken(body, "production")',
|
||||
)
|
||||
],
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='hasToken(body, "production")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
rows = get_rows(response)
|
||||
@@ -179,28 +306,35 @@ def test_logs_json_body_has_token(
|
||||
assert all("production" in row["data"]["body"] for row in rows)
|
||||
|
||||
|
||||
def test_logs_json_body_function_scalar_path_errors(
|
||||
def test_logs_json_body_function_scalar_path(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""BUG: has/hasAny/hasAll on a scalar body path extract a scalar String; ClickHouse
|
||||
rejects the array function -> HTTP 500."""
|
||||
"""has/hasAny/hasAll on a scalar (non-array) body path treat the scalar as a single-element
|
||||
set: has = equals, hasAny/hasAll = membership. So has and hasAll coincide for a scalar."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
timestamp=now - timedelta(seconds=2),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"level": "info", "code": 200}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
resources={"service.name": "app-service"},
|
||||
body=json.dumps({"level": "debug", "code": 500}),
|
||||
severity_text="INFO",
|
||||
),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
for expression in ["has(body.level, 'info')", "hasAny(body.code, [200])", "hasAll(body.level, ['info'])"]:
|
||||
# Each expression matches only the first log (level=info, code=200).
|
||||
for expression in ["has(body.level, 'info')", "hasAny(body.code, [200, 999])", "hasAll(body.level, ['info'])"]:
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
@@ -209,7 +343,22 @@ def test_logs_json_body_function_scalar_path_errors(
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR, f"{expression}: expected 500, got {response.status_code}: {response.text}"
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: expected 200, got {response.status_code}: {response.text}"
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1, f"{expression}: expected 1 match, got {len(rows)}"
|
||||
assert json.loads(rows[0]["data"]["body"])["level"] == "info", f"{expression}: matched wrong row"
|
||||
|
||||
# hasAll is contains-all: a scalar (single-element set) can't hold two distinct values, so nothing matches.
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="hasAll(body.level, ['info', 'debug'])")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert get_rows(response) == [], "hasAll(scalar, [two distinct values]) should match nothing (contains-all)"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -238,3 +387,286 @@ def test_logs_json_body_function_errors(
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"expression, expected_error",
|
||||
[
|
||||
pytest.param('has(body.tags, "a", "b")', "expects exactly one value argument", id="has_multiple_values"),
|
||||
pytest.param('hasToken(body, "a", "b")', "expects exactly one value argument", id="hastoken_multiple_values"),
|
||||
pytest.param('has(body.tags, ["a", "b"])', "expects a single scalar value, not an array", id="has_array_value"),
|
||||
pytest.param('hasAny(body.tags, ["a"], "b")', "not a mix of the two", id="hasany_mixed_array_and_scalar"),
|
||||
],
|
||||
)
|
||||
def test_logs_json_body_function_argument_shape_errors(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
expression: str,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
"""has-family value arguments are shape-validated; mis-shaped arguments are rejected (400)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST, f"{expression}: expected 400, got {response.status_code}: {response.text}"
|
||||
assert expected_error in response.text, f"{expression}: expected error {expected_error!r} in {response.text}"
|
||||
|
||||
|
||||
def test_logs_json_body_type_match_by_needle(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""The element type is inferred from the needle (legacy has no schema): a string needle
|
||||
searches as String, a quoted-integer needle as Int64, a decimal needle as Float64. A
|
||||
numeric-string body array (["100","200"]) is parsed, so numeric needles still match it."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=4), resources={"service.name": "app-service"}, body=json.dumps({"vals": ["prod", "api"]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body=json.dumps({"vals": [100, 200, 300]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"vals": [1.5, 2.5]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"vals": ["100", "200"]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [
|
||||
("has(body.vals, 'prod')", 1), # String -> only the string array
|
||||
('has(body.vals, "200")', 2), # quoted Int64 -> [100,200,300] and ["100","200"]
|
||||
("has(body.vals, 1.5)", 1), # Float64 -> only the float array
|
||||
]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
|
||||
def test_logs_json_body_large_integer_exact(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""Large ids (> 2^53) stay exact only when QUOTED (extracted as Int64). An unquoted numeric
|
||||
literal is parsed as float64 upstream, so it false-matches neighbouring ids (documented)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"id": [1234567890123456789]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"id": [1234567890123456788]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
# Quoted -> Int64 -> exact: matches only the exact id, not the neighbour.
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='has(body.id, "1234567890123456789")')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1, "quoted big id should match exactly one log"
|
||||
assert json.loads(rows[0]["data"]["body"])["id"] == [1234567890123456789]
|
||||
|
||||
# Unquoted -> float64 upstream -> imprecise: false-matches the off-by-one neighbour too.
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.id, 1234567890123456789)")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert len(get_rows(response)) == 2, "unquoted big id collides with neighbour (float64 precision) - quote large ids"
|
||||
|
||||
|
||||
def test_logs_json_body_no_supertype_safety(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""A needle whose type doesn't match the body array must never raise a ClickHouse "no
|
||||
supertype" error (code 386): it returns 200 with no match (Nullable extraction -> NULL)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"names": ["alice", "bob"]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"nums": [100, 200]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression in ["has(body.names, 200)", "has(body.nums, 'alice')"]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: expected 200 (no supertype error), got {response.status_code}: {response.text}"
|
||||
assert get_rows(response) == [], f"{expression}: type mismatch should match nothing"
|
||||
|
||||
|
||||
def test_logs_json_body_mixed_needle_args(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""A value list mixing types (e.g. [200, "abc"]) is NOT rejected; it falls back to a string
|
||||
comparison (safe, no error). Numbers render to their text form, so both logs match."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"vals": [100, 200, 300]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"vals": ["abc", "xyz"]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz, token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression='hasAny(body.vals, [200, "abc"])')],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert len(get_rows(response)) == 2, "mixed-type args fall back to string comparison; both logs match"
|
||||
|
||||
|
||||
def test_logs_json_body_nested_array_path(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has over a path that traverses arrays (body.edu[*].names) searches the flattened leaf."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"edu": [{"names": ["IIT", "MIT"]}, {"names": ["CMU"]}]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"edu": [{"names": ["Stanford"]}]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [("has(body.edu[*].names, 'MIT')", 1), ("has(body.edu[*].names, 'Yale')", 0)]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
|
||||
def test_logs_json_body_missing_empty_null(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""has matches nothing for a missing key, an empty array, or a null value (only the log
|
||||
that actually contains the value matches)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=4), resources={"service.name": "app-service"}, body=json.dumps({"other": [1, 2]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body=json.dumps({"vals": []}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"vals": None}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"vals": ["x"]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz, token,
|
||||
start_ms=int((now - timedelta(seconds=10)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression="has(body.vals, 'x')")],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
rows = get_rows(response)
|
||||
assert len(rows) == 1, "only the log whose vals contains 'x' matches"
|
||||
assert json.loads(rows[0]["data"]["body"])["vals"] == ["x"]
|
||||
|
||||
|
||||
def test_logs_json_body_negative_and_special_chars(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""Negative integers and strings with spaces/special chars match correctly."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"nums": [-5, -10]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"tags": ["a b", "c/d"]}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [('has(body.nums, "-5")', 1), ("has(body.tags, 'a b')", 1)]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
|
||||
def test_logs_json_body_type_isolation(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
"""Type isolation (regression guard). A bool needle matches only genuine booleans, not
|
||||
"truthy" numbers/strings; an empty-string or zero needle does not false-match array bodies;
|
||||
genuine scalar bool / zero / empty-string bodies still match (the single-element-set path)."""
|
||||
now = datetime.now(tz=UTC)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(timestamp=now - timedelta(seconds=5), resources={"service.name": "app-service"}, body=json.dumps({"nums": [100, 200, 300]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=4), resources={"service.name": "app-service"}, body=json.dumps({"flags": [True, False]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=3), resources={"service.name": "app-service"}, body=json.dumps({"tags": ["a", "b"]}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=2), resources={"service.name": "app-service"}, body=json.dumps({"flag": True}), severity_text="INFO"),
|
||||
Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "app-service"}, body=json.dumps({"code": 0}), severity_text="INFO"),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
|
||||
end_ms = int(now.timestamp() * 1000)
|
||||
|
||||
for expression, expected in [
|
||||
("has(body.nums, true)", 0), # bool needle does NOT truthy-match a numeric array
|
||||
("has(body.flags, true)", 1), # genuine bool array matches
|
||||
("has(body.flag, true)", 1), # genuine scalar bool matches
|
||||
("has(body.tags, '')", 0), # empty-string does NOT false-match an array body
|
||||
("has(body.nums, 0)", 0), # zero does NOT false-match an array body without a 0
|
||||
("has(body.code, 0)", 1), # genuine scalar zero matches
|
||||
]:
|
||||
response = make_query_request(
|
||||
signoz, token, start_ms=start_ms, end_ms=end_ms, request_type=RequestType.RAW,
|
||||
queries=[build_raw_query("A", "logs", order=[build_order_by("timestamp")], limit=100, filter_expression=expression)],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
|
||||
assert len(get_rows(response)) == expected, f"{expression}: expected {expected} rows"
|
||||
|
||||
Reference in New Issue
Block a user