Files
openobserve/tests/test-data/query-agent/data_gen.py
Neha PandClaude Opus 4.7 b0f4b0e46f test: expand query-agent suite from 672 to 1,000 queries (#12856)
## Summary

Expands the query-agent regression test suite from **672 → 1,000
queries** by adding 328 complex queries (Q672–Q999) across all 13
categories.

## What changed

- **`data_gen.py`**: `NUM_QUERIES` bumped from 675 → 1000 so
`build_dataset()` generates records for all query IDs
- **13 query JSON files**: 328 new queries appended, with
DuckDB-computed expected results via `compute_counts.py`
- **`gen_q672_q999.py`**: new generation script (idempotent — re-running
skips all existing queries)

## New query patterns added

Targets DataFusion-upgrade-fragile areas identified from past breakage:

| Pattern | Examples |
|---|---|
| 3-stream JOINs + correlated EXISTS | Q672–Q677 |
| FULL OUTER JOIN between streams | Q693–Q696 |
| Window + DISTINCT (FIRST_VALUE, LAST_VALUE, NTH_VALUE) | Q717–Q730 |
| Named WINDOW clauses | Q731–Q736 |
| LAG/LEAD with CASE trend detection | Q742–Q751 |
| match_all + GROUP BY + HAVING | Q792–Q796 |
| INTERSECT/EXCEPT with CTEs | Q822–Q849 |
| Deep CTE chains (3+ CTEs, self-joins) | Q757–Q791 |
| Correlated scalar subqueries in SELECT | Q769–Q776 |
| Histogram with JOIN between streams | Q901–Q905 |

## Issues fixed

- **4 duplicate IDs** (Q545–Q548) existed in both `cte_subquery.json`
and `full_text_search.json` — renumbered FTS copies to available gaps
(Q025, Q611, Q631, Q1000)
- **27 queries** using OO-specific functions (match_all, spath,
regexp_like, window-inside-aggregate) annotated with `skip_sqllogictest:
true`

## Final breakdown

| Category | Before | After |
|---|---|---|
| cte_subquery | 89 | 124 |
| aggregation | 71 | 99 |
| string_functions | 70 | 92 |
| combined | 73 | 88 |
| full_text_search | 58 | 88 |
| window | 48 | 88 |
| basic_select | 71 | 86 |
| histogram | 57 | 79 |
| union | 34 | 62 |
| cross_stream | 16 | 61 |
| date_time | 37 | 60 |
| pagination | 22 | 42 |
| math_functions | 26 | 31 |
| **Total** | **672** | **1,000** |

- **956** queries use sqllogictest mode (DuckDB-verified results)
- **44** use legacy assertion mode (OO-specific functions, histograms)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-24 14:24:11 +00:00

672 lines
25 KiB
Python

"""Shared deterministic data generator for query agent tests.
Used by both conftest.py (test fixture) and compute_counts.py (DuckDB oracle).
"""
import json
import random
from datetime import datetime, timedelta, timezone
from pathlib import Path
random.seed(42)
# Number of queries this generator produces data for.
# build_dataset() default must match this value.
# Must be >= the highest query ID (e.g. Q1000 -> NUM_QUERIES = 1000).
# Changing this requires re-running compute_counts.py to regenerate
# expected results for all query JSON files.
NUM_QUERIES = 1000
# Check for a BASE_TS override saved by compute_counts.py so that the
# compute oracle and the test harness share the same BASE_TS. Without
# this, a minute boundary crossing between compute and test shifts
# all histogram bucket timestamps.
_OVERRIDE_FILE = Path(__file__).parent / "base_ts_override.json"
if _OVERRIDE_FILE.exists():
BASE_TS = json.loads(_OVERRIDE_FILE.read_text())["BASE_TS"]
else:
_now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
# Anchor BASE_TS 2 hours in the past. Records span forward from there
# (BASE_TS to BASE_TS + NUM_QUERIES*60s), so most land in the future.
# 2h keeps BASE_TS within any server's ZO_INGEST_ALLOWED_UPTO limit
# (default 5h) while the fixture's wait_until uses end_us=max(now,
# max_ts)+1h to cover future-dated records on vortex and parquet.
_span_hours = 2
BASE_TS = int((_now - timedelta(hours=_span_hours)).timestamp() * 1_000_000)
FIELD_POOL = {
# --- Original 17 fields (warehouse/sorter themed) ---
"pallet_id": ['PL-001', 'PL-002', 'PL-003', 'PL-004', 'PL-005', 'PL-006', 'PL-007', 'PL-008'],
"load_factor": [12.5, 33.0, 55.2, 72.8, 88.1, 94.6, 45.3, 67.9, 21.4, 99.0],
"charge_remaining": [10.0, 25.0, 42.5, 58.0, 75.3, 88.7, 95.0, 5.0, 33.3, 66.6],
"throughput_rate": [150.0, 320.0, 480.0, 610.0, 725.0, 840.0, 290.0, 555.0, 190.0, 900.0],
"sorter_model": ['SORT-X1', 'SORT-A7', 'SORT-M3', 'SORT-Z9', 'SORT-X1', 'SORT-A7', 'SORT-M3', 'SORT-Q5'],
"conveyor_lane": ['LANE-A', 'LANE-B', 'LANE-C', 'LANE-D', 'LANE-A', 'LANE-B', 'LANE-C', 'LANE-D'],
"facility_zone": ['ZONE-1', 'ZONE-2', 'ZONE-3', 'ZONE-4', 'ZONE-1', 'ZONE-2', 'ZONE-3', 'ZONE-4'],
"control_center": ['CC-ALPHA', 'CC-BETA', 'CC-GAMMA', 'CC-DELTA', 'CC-ALPHA', 'CC-BETA', 'CC-GAMMA', 'CC-DELTA'],
"item_count": [50, 120, 200, 340, 88, 155, 410, 275, 60, 500],
"defect_limit": [0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 0.8, 1.2, 2.8, 3.5],
"runtime_hours": [24, 72, 168, 360, 720, 48, 120, 500, 90, 1000],
"operation_mode": ['auto', 'manual', 'hybrid', 'auto', 'manual', 'hybrid', 'auto', 'manual'],
"package_size": [100, 250, 500, 750, 1000, 200, 350, 600, 850, 1200],
"scan_attempts": [1, 2, 3, 1, 2, 4, 1, 3, 2, 5],
"conveyor_segment": [1, 2, 3, 4, 5, 1, 2, 3, 4, 5],
"cell_temp": [18.5, 22.0, 26.3, 30.1, 35.7, 19.2, 24.8, 28.9, 32.5, 40.0],
"build_signature": ["BS-2026-A1", "BS-2026-B2", "BS-2026-C3", "BS-2025-D4", "BS-2026-A1",
"BS-2026-B2", "BS-2026-C3", "BS-2025-D4", "BS-2026-A1", "BS-2026-B2"],
# --- New fields for 392-query regression suite ---
# Text patterns for regexp extraction testing
"event_detail": [
"batch ABC123 processed successfully",
"segment 42 scan completed with no defects",
"track TRK-abc-def diverted to LANE-B",
"batch XYZ789 timed out at checkpoint 3",
"segment 17 weight threshold exceeded",
"track TRK-ghi-jkl rerouted to ZONE-3",
"batch MNO456 merged into pallet PL-005",
"segment 88 speed anomaly detected",
"track TRK-mno-pqr alignment corrected",
"batch DEF234 registered at loading dock",
],
# ACK batch reference patterns
"ack_detail": [
"ACK batch batchIdList:BT-001-2026 received",
"ACK batch batchIdList:BT-002-2026 acknowledged",
"ACK batch batchIdList:BT-003-2026 confirmed",
"ACK batch batchIdList:BT-004-2026 validated",
"ACK batch batchIdList:BT-005-2026 processed",
"ACK batch batchIdList:BT-006-2026 queued",
"ACK batch batchIdList:BT-007-2026 dispatched",
"ACK batch batchIdList:BT-008-2026 archived",
"ACK batch batchIdList:BT-009-2026 retried",
"ACK batch batchIdList:BT-010-2026 completed",
],
# Log entries with URLs and patterns for LIKE/regexp
"log_message": [
"GET /api/v1/warehouse/status 200 OK",
"POST /api/v2/sorter/scan 201 Created",
"INFO: batch sync completed, url=http://warehouse.local/sync",
"WARN: lane jam detected at conveyor_segment=3",
"ERROR: pallet PL-003 mismatch, retry url=https://warehouse.local/retry",
"DEBUG: charge_remaining below threshold 15.0",
"INFO: throughput_rate=840.0 spike at facility_zone=ZONE-4",
"WARN: cell_temp=38.2 exceeding limit at sorter_model=SORT-Z9",
"ERROR: scan_attempts exhausted for pallet_id=PL-007",
"INFO: runtime_hours=1000 maintenance due http://warehouse.local/maint",
],
# Error codes for string matching
"error_code": [
"E001", "E002", "E003", "E004", "E005",
"E006", "E007", "E008", "E009", "E010",
],
# Service owner identifiers (warehouse/sorter themed)
"service_owner": [
"@warehouse-Android-sort-v1",
"@warehouse-iOS-sort-v2",
"@warehouse-Linux-sort-v3",
"@warehouse-Windows-sort-v1",
"@warehouse-Android-scan-v2",
"@warehouse-iOS-scan-v1",
"@warehouse-Linux-scan-v3",
"@warehouse-Windows-scan-v2",
"@warehouse-Android-track-v1",
"@warehouse-iOS-track-v3",
],
# Upstream error codes
"upstream_error_code": [
"UP-E001", "UP-E002", "UP-E003", "UP-E004", "UP-E005",
"UP-E006", "UP-E007", "UP-E008", "UP-E009", "UP-E010",
],
# Exception text snippets for regexp extraction
"exception_message": [
"NullPointerException at ConveyorController.handle()",
"TimeoutException: lane LANE-A did not respond within 5000ms",
"DataIntegrityException: pallet_id PL-002 checksum mismatch",
"ResourceExhaustedException: sorter_model SORT-M3 queue full",
"SecurityException: unauthorized access to control_center CC-ALPHA",
"ConfigurationException: missing facility_zone ZONE-3 override",
"NetworkException: connection to batch sync failed",
"ValidationException: item_count exceeds defect_limit bounds",
"StateTransitionException: operation_mode hybrid invalid state",
"OverflowException: throughput_rate 999.0 exceeds max capacity",
],
# JSON strings with nested message objects for json_get_str / json_get_json
"request_payload": [
'{"message":{"endpoint":"searchItems","status":"ok"}}',
'{"message":{"endpoint":"updateBatch","status":"pending"}}',
'{"message":{"endpoint":"deleteRecord","status":"failed"}}',
'{"message":{"endpoint":"scanPallet","status":"ok"}}',
'{"message":{"endpoint":"rerouteTrack","status":"queued"}}',
'{"message":{"endpoint":"mergeBatch","status":"ok"}}',
'{"message":{"endpoint":"checkThreshold","status":"warning"}}',
'{"message":{"endpoint":"archiveLog","status":"ok"}}',
'{"message":{"endpoint":"retryScan","status":"retrying"}}',
'{"message":{"endpoint":"syncZone","status":"syncing"}}',
],
# JSON array of JSON-encoded strings for spath/unnest/flatten/cast_to_arr testing
"phase_data": [
'["{\\"name\\":\\"init\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"process\\",\\"startTime\\":\\"150\\"}"]',
'["{\\"name\\":\\"init\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"validate\\",\\"startTime\\":\\"200\\"}", "{\\"name\\":\\"complete\\",\\"startTime\\":\\"400\\"}"]',
'["{\\"name\\":\\"boot\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"calibrate\\",\\"startTime\\":\\"100\\"}"]',
'["{\\"name\\":\\"scan\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"sort\\",\\"startTime\\":\\"180\\"}", "{\\"name\\":\\"pack\\",\\"startTime\\":\\"360\\"}"]',
'["{\\"name\\":\\"precheck\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"execute\\",\\"startTime\\":\\"50\\"}", "{\\"name\\":\\"finalize\\",\\"startTime\\":\\"300\\"}"]',
'["{\\"name\\":\\"queue\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"dequeue\\",\\"startTime\\":\\"70\\"}", "{\\"name\\":\\"dispatch\\",\\"startTime\\":\\"140\\"}"]',
'["{\\"name\\":\\"load\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"unload\\",\\"startTime\\":\\"250\\"}"]',
'["{\\"name\\":\\"acquire\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"process\\",\\"startTime\\":\\"90\\"}", "{\\"name\\":\\"release\\",\\"startTime\\":\\"270\\"}"]',
'["{\\"name\\":\\"merge\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"split\\",\\"startTime\\":\\"120\\"}", "{\\"name\\":\\"route\\",\\"startTime\\":\\"240\\"}"]',
'["{\\"name\\":\\"start\\",\\"startTime\\":\\"0\\"}", "{\\"name\\":\\"pause\\",\\"startTime\\":\\"160\\"}", "{\\"name\\":\\"resume\\",\\"startTime\\":\\"320\\"}", "{\\"name\\":\\"stop\\",\\"startTime\\":\\"480\\"}"]',
],
# JSON array strings for array_element testing
"db_info": [
'["cid-abc123","status-ok"]',
'["cid-def456","status-pending"]',
'["cid-ghi789","status-failed"]',
'["cid-jkl012","status-ok"]',
'["cid-mno345","status-retrying"]',
'["cid-pqr678","status-ok"]',
'["cid-stu901","status-warning"]',
'["cid-vwx234","status-ok"]',
'["cid-yza567","status-error"]',
'["cid-bcd890","status-ok"]',
],
# API names for str_match_ignore_case testing
"api_name": [
"SearchItems", "UpdateBatch", "DeleteRecord", "ScanPallet",
"RerouteTrack", "MergeBatch", "CheckThreshold", "ArchiveLog",
"RetryScan", "SyncZone",
],
# IP addresses
"node_ip": [
"10.12.0.101", "10.12.0.102", "10.12.0.103", "10.12.0.104", "10.12.0.105",
"10.12.1.201", "10.12.1.202", "10.12.1.203", "10.12.1.204", "10.12.1.205",
],
# Source IPs
"source_ip": [
"192.168.1.10", "192.168.1.11", "192.168.1.12", "192.168.1.13", "192.168.1.14",
"192.168.2.20", "192.168.2.21", "192.168.2.22", "192.168.2.23", "192.168.2.24",
],
# String tags
"tag_a": [
"prod", "staging", "dev", "prod", "staging",
"dev", "prod", "staging", "dev", "prod",
],
"tag_b": [
"critical", "high", "medium", "low", "critical",
"high", "medium", "low", "critical", "high",
],
# Source identifiers
"query_source": [
"dashboard", "cli", "api", "scheduler", "dashboard",
"cli", "api", "scheduler", "dashboard", "cli",
],
# Region codes
"region_code": [
"US-EAST", "US-WEST", "EU-WEST", "EU-NORTH", "AP-SOUTH",
"AP-EAST", "SA-EAST", "US-CENTRAL", "EU-CENTRAL", "AP-CENTRAL",
],
# User agent strings
"client_agent": [
"Mozilla/5.0 WarehouseScanner/1.0",
"curl/7.88.1 (x86_64-pc-linux-gnu)",
"Python-urllib/3.11 warehouse-client",
"Java/17.0.8 (OpenJDK; sorter-agent)",
"Go-http-client/2.0 conveyor-monitor",
"PostmanRuntime/7.35.0",
"axios/1.6.0 (Node.js; warehouse-ui)",
"okhttp/4.12.0 (Android; sort-app)",
"Alamofire/5.8.0 (iOS; scan-app)",
"node-fetch/3.3.2 (tracking-service)",
],
# Operation names
"operation_name": [
"SearchItems", "UpdateBatch", "DeleteRecord", "ScanPallet",
"RerouteTrack", "MergeBatch", "CheckThreshold", "ArchiveLog",
"RetryScan", "SyncZone",
],
# Site identifiers
"site_name": [
"site-nyc-01", "site-lon-02", "site-tok-03", "site-syd-04", "site-sfo-05",
"site-ber-06", "site-dub-07", "site-sin-08", "site-mum-09", "site-tor-10",
],
# URL paths
"endpoint_path": [
"/api/v1/warehouse/search",
"/api/v1/warehouse/update",
"/api/v1/warehouse/delete",
"/api/v2/sorter/scan",
"/api/v2/sorter/reroute",
"/api/v2/sorter/merge",
"/api/v1/health/check",
"/api/v1/archive/log",
"/api/v2/sorter/retry",
"/api/v1/zone/sync",
],
# HTTP methods
"http_method": [
"GET", "POST", "PUT", "DELETE", "GET",
"POST", "PUT", "DELETE", "GET", "POST",
],
# HTTP status codes
"response_code": [
200, 201, 204, 400, 404,
500, 503, 401, 403, 200,
],
# Scan type categories
"scan_category": [
"inbound", "outbound", "internal", "inbound", "outbound",
"internal", "inbound", "outbound", "internal", "inbound",
],
# Auth results
"auth_result": [
"pass", "fail", "denied", "pass", "pass",
"fail", "denied", "pass", "fail", "pass",
],
# Task identifiers
"task_id": [
"task-a1b2c3", "task-d4e5f6", "task-g7h8i9", "task-j0k1l2", "task-m3n4o5",
"task-p6q7r8", "task-s9t0u1", "task-v2w3x4", "task-y5z6a7", "task-b8c9d0",
],
# Entity identifiers
"entity_id": [
"ent-001-wh", "ent-002-wh", "ent-003-wh", "ent-004-wh", "ent-005-wh",
"ent-006-wh", "ent-007-wh", "ent-008-wh", "ent-009-wh", "ent-010-wh",
],
# Alert tag strings
"alert_tags": [
"latency", "error", "capacity", "latency", "security",
"error", "capacity", "latency", "error", "capacity",
],
# Issue type strings
"issue_type": [
"timeout", "overflow", "mismatch", "jam", "timeout",
"overflow", "mismatch", "jam", "timeout", "overflow",
],
# Filter result strings
"filter_result": [
"allow", "block", "allow", "allow", "block",
"allow", "block", "allow", "allow", "block",
],
# Guard mode values
"guard_mode": [
"active", "passive", "disabled", "active", "passive",
"active", "passive", "disabled", "active", "passive",
],
# Record type identifiers
"record_type": [
"scan", "sort", "merge", "track", "scan",
"sort", "merge", "track", "scan", "sort",
],
# File paths
"resource_path": [
"/logs/warehouse/app.log",
"/data/sorter/output.dat",
"/config/zone/ZONE-1.yaml",
"/logs/warehouse/error.log",
"/data/sorter/input.dat",
"/config/zone/ZONE-2.yaml",
"/logs/warehouse/access.log",
"/data/sorter/temp.dat",
"/config/zone/ZONE-3.yaml",
"/logs/warehouse/audit.log",
],
# Integer latency values
"latency_ms": [
12, 45, 230, 1450, 89,
567, 3200, 78, 1900, 15,
],
# Organization names
"org_name": [
"warehouse-east", "warehouse-west", "warehouse-north", "warehouse-south",
"warehouse-central", "warehouse-east", "warehouse-west", "warehouse-north",
"warehouse-south", "warehouse-central",
],
# Boolean-like values
"threat_flag": [
"false", "true", "false", "false", "true",
"false", "true", "false", "false", "true",
],
# Version strings
"build_version": [
"v1.0.0", "v1.1.2", "v2.0.0", "v2.1.3", "v2.2.0",
"v3.0.1", "v3.1.0", "v1.0.5", "v2.0.8", "v3.2.0",
],
# Bin codes
"bin_code": [
"BIN-A1", "BIN-B2", "BIN-C3", "BIN-D4", "BIN-E5",
"BIN-F6", "BIN-G7", "BIN-H8", "BIN-I9", "BIN-J0",
],
# Integer sizes
"payload_size": [
512, 1024, 2048, 4096, 8192,
16384, 256, 32768, 65536, 128,
],
# Component identifiers
"component_name": [
"sorter-engine", "conveyor-driver", "zone-controller", "scan-hub",
"track-router", "batch-sync", "pallet-mapper", "lane-monitor",
"temp-regulator", "charge-checker",
],
# Routing keys
"route_key": [
"rk-warehouse-001", "rk-warehouse-002", "rk-warehouse-003",
"rk-warehouse-004", "rk-warehouse-005", "rk-warehouse-006",
"rk-warehouse-007", "rk-warehouse-008", "rk-warehouse-009",
"rk-warehouse-010",
],
# Token strings
"auth_token": [
"tok-wh-abc123def456", "tok-wh-ghi789jkl012", "tok-wh-mno345pqr678",
"tok-wh-stu901vwx234", "tok-wh-yza567bcd890", "tok-wh-efg123hij456",
"tok-wh-klm789nop012", "tok-wh-qrs345tuv678", "tok-wh-wxy901zab234",
"tok-wh-cde567fgh890",
],
# Application labels
"app_label": [
"warehouse-sort", "warehouse-scan", "warehouse-track", "warehouse-merge",
"warehouse-sync", "warehouse-sort", "warehouse-scan", "warehouse-track",
"warehouse-merge", "warehouse-sync",
],
# --- Synthetic fields for Q393-Q417 (A/B testing + security/bot-detection) ---
"variant_tag": [
"control", "variant_a", "variant_b", "variant_c", "control",
"variant_a", "variant_b", "variant_c", "control", "variant_a",
],
"cookie_id": [
"sess_abc123", "sess_def456", "ctl_xyz789", "var1_pqr012", "var2_mno345",
"sess_abc123", "var3_stu678", "ctl_vwx901", "sess_def456", "var1_yza234",
],
"status_code": [
"200", "200", "200", "500", "404",
"200", "502", "200", "403", "200",
],
"lcp_micros": [
1200000, 2500000, 1800000, 3200000, 900000,
1500000, 2800000, 2100000, 3400000, 1100000,
],
"inp_micros": [
80000, 150000, 200000, 95000, 300000,
120000, 180000, 250000, 70000, 160000,
],
"cls_score": [
0.05, 0.12, 0.08, 0.25, 0.03,
0.15, 0.10, 0.20, 0.30, 0.06,
],
"device_type": [
"Mobile", "Computer", "Tablet", "Mobile", "Computer",
"Mobile", "Tablet", "Computer", "Mobile", "Computer",
],
"page_slug": [
"itemPage", "searchPage", "cartPage", "itemPage", "checkoutPage",
"itemPage", "homePage", "itemPage", "searchPage", "itemPage",
],
"action_category": [
"performanceMetric", "performanceMetric", "interaction", "navigation", "performanceMetric",
"interaction", "performanceMetric", "navigation", "performanceMetric", "performanceMetric",
],
"action_subcategory": [
"vitalsLcp", "vitalsInp", "vitalsCls", "clickEvent", "vitalsLcp",
"vitalsInp", "vitalsCls", "pageView", "vitalsLcp", "vitalsInp",
],
"page_url": [
"https://shop.example.com/item/1001",
"https://shop.example.com/item/1002",
"https://shop.example.com/search?q=test",
"https://shop.example.com/item/1003",
"https://shop.example.com/cart",
"https://shop.example.com/item/1004",
"https://shop.example.com/",
"https://shop.example.com/item/1005",
"https://shop.example.com/item/1006",
"https://shop.example.com/item/1007",
],
"render_scope": [
"SSR", "CSR", "SSR", "SSR", "CSR",
"SSR", "CSR", "SSR", "SSR", "CSR",
],
"os_platform": [
"Android 14", "iOS 17.4", "Android 13", "Mac OS X (iPhone)", "Windows 11",
"Android 14", "iOS 18.1", "Android 12", "Mac OS X (iPhone)", "Android 14",
],
"attack_name": [
"SQL Injection (Param)", "XSS Reflected (Header)", "Path Traversal", "CSRF Attempt", "SSRF Probe",
"Command Injection", "XSS Stored (Body)", "File Inclusion", "Desync Attack", "Evasion Technique",
],
# Named http_request instead of request_payload to avoid collision with the
# existing request_payload field (JSON message objects, lines 130-141 above).
"http_request": [
"GET /api/users?id=1",
"POST /api/order {\"item\":123}",
"GET /search?q=test",
"GET /favicon.ico",
"POST /admin/config",
"GET /api/products",
"PUT /api/cart/update",
"DELETE /api/session",
"GET /.env",
"POST /api/login {\"user\":\"admin\"}",
],
"user_agent_str": [
"Mozilla/5.0 Chrome/120",
"curl/7.88.1",
"python-requests/2.31",
"Mozilla/5.0 Firefox/121",
"PostmanRuntime/7.35",
"Mozilla/5.0 Safari/17.2",
"axios/1.6.0",
"Go-http-client/2.0",
"Java/11.0.20",
"Mozilla/5.0 Edge/120",
],
"uri_path": [
"/api/v1/data", "/swag/graphql", "/ip/192.168.1.1", "/admin/login", "/api/v2/search",
"/assets/js/app.js", "/swag/graphql", "/api/v1/users", "/health", "/ip/10.0.0.1",
],
"geo_org": [
"Acme Corp", "GlobalTech Inc", "DigitalOcean LLC", "FastHost Ltd", "CloudServe GmbH",
"Acme Corp", "NetGuard AS", "HostPro Inc", "CloudServe GmbH", "DataCenter EU",
],
"geo_country": [
"US", "DE", "UK", "US", "BR",
"US", "JP", "US", "IN", "US",
],
"geo_user_type": [
"business", "hosting", "business", "business", "hosting",
"business", "education", "business", "hosting", "business",
],
"risk_rules_list": [
'["rate_limit_exceeded","geo_anomaly"]',
'["torbot_ua_mismatch","cookie_reuse"]',
'["credential_stuffing","rapid_login"]',
'["session_hijack","ip_hop"]',
'["scraping_pattern","high_freq"]',
'["rate_limit_exceeded"]',
'["torbot_ua_mismatch"]',
'["cookie_reuse","session_hijack"]',
'["credential_stuffing"]',
'["scraping_pattern","ip_hop","rapid_login"]',
],
"visitor_session_id": [
"sess_001", "sess_001", "sess_002", "sess_003", "sess_002",
"sess_001", "sess_004", "sess_003", "sess_005", "sess_001",
],
"bot_flag": [
None, None, None, "suspicious", None,
None, "verified_bot", None, None, None,
],
"passthrough_flag": [
"false", "false", "false", "true", "false",
"false", "false", "true", "false", "false",
],
"order_ref": [
"ORD-001", "ORD-002", "ORD-001", "ORD-003", "ORD-004",
"ORD-005", "ORD-003", "ORD-006", "ORD-007", "ORD-008",
],
"info_tag": [
None, None, "Allowlisted", None, None,
None, "Blocklisted", None, None, None,
],
"graphql_operation": [
"updateItems", "placeOrder", "verifyPayment", "updateCart", "createAccount",
"updateItems", "getInventory", "placeOrder", "updateItems", "verifyPayment",
],
"datacenter": [
"dc-east", "dc-west", "dc-east", "dc-central", "dc-west",
"dc-east", "dc-central", "dc-west", "dc-east", "dc-west",
],
}
STREAM_VALUES = ["stdout", "stdout", "stdout", "stderr"]
# Fields that are NULL for some records to exercise COALESCE / IS NULL /
# IS NOT NULL paths. Two out of every five records per query get None
# for these fields.
_NULLABLE_FIELDS = {
"bot_flag", "info_tag", "upstream_error_code", "page_url",
"endpoint_path", "resource_path", "variant_tag", "threat_flag",
}
def make_record(ts, idx, qid, stream_offset=0):
"""Build a single deterministic data record.
Each field uses a different rotation offset so field-value
combinations vary independently — no two fields pick from the same
pool position for the same (idx, qi) pair.
stream_offset shifts the per-field rotation so the same (qi, idx)
produces different values for a secondary stream while keeping
partial overlap on join keys (pallet_id, org_name, etc.).
IMPORTANT: stream_offset only affects field VALUES, NOT timestamps.
Both streams generate records at the same _timestamp values (same
BASE_TS, same per-query time windows). This means cross-stream
JOIN queries can use a single time_offset — both streams' records
for a given query fall within the same 74-second window.
The log field always uses the original qid (unshifted) so that log
entries are tagged with the query that owns them, regardless of
stream_offset. Only qi (used for per-field rotation) is shifted.
"""
qi = int(qid[1:]) + stream_offset
# Vary the log field: some records get an ACK batch suffix for
# str_match_ignore_case testing on ack_detail-like patterns.
if idx % 3 == 0:
log = f"{qid} warehouse event record {idx} ACK batch batchIdList:{qid}-{idx}"
elif idx % 5 == 0:
log = f"{qid} warehouse event record {idx} error E{idx:03d} traceback"
else:
log = f"{qid} warehouse event record {idx}"
r = {
"_timestamp": ts,
"log": log,
"stream": STREAM_VALUES[(idx + qi) % len(STREAM_VALUES)],
}
# Per-field rotation: multipliers 13 and 7 are coprime with all pool
# sizes (8, 10) and with each other, so every (qi, idx) pair produces
# a unique value fingerprint. Two records with the same idx but
# different qi get different values for every field.
# NULL injection: (idx + qi) % 5 ∈ {1, 3} gives a deterministic 40 %
# rate across the 5-record window, varying which records are nulled
# per query so both nullable and non-nullable combinations appear.
inject_null = (idx + qi) % 5 in (1, 3)
for i, (field, pool) in enumerate(FIELD_POOL.items()):
if inject_null and field in _NULLABLE_FIELDS:
r[field] = None
continue
offset = (qi * 13 + i * 7) % len(pool)
r[field] = pool[(idx + offset) % len(pool)]
return r
def build_dataset(num_queries=NUM_QUERIES, stream_offset=0):
"""Generate deterministic records for queries Q001-Q{num_queries}.
Each query gets 5 records spaced 18 seconds apart within its own
non-overlapping 60-second time window.
stream_offset shifts field rotation for secondary streams while
preserving shared time windows and partial join-key overlap.
"""
records = []
for qi in range(1, num_queries + 1):
qid = f"Q{qi:03d}"
base = BASE_TS + (qi - 1) * 60_000_000
for i in range(5):
ts = base + i * 18_000_000
records.append(make_record(ts, i, qid, stream_offset))
return records