mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 08:09:27 +00:00
test: add schema evolution stress tests — scenarios 26–59 (diverge-then-converge + 9 scenario groups) (#12888)
## What Extends `tests/api-testing/tests/vortex/test_schema_evolution.py` with **34 new tests** across 10 new classes covering all realistic field drift scenarios for petabyte-scale schema evolution. Grows the schema evolution suite from 13 scenarios (14–25) to **46 scenarios (14–59)**. ## The scenario (shared across all groups) Think of a post office logbook. In January every letter has *From, To, Weight*. In February the form changes — Weight is dropped, Stamp Color is added. In March they go back plus add more fields. ``` Batch 1 [fa, fb, fc] — initial deployment Batch 2 [fa, fb, fd] — update: fc dropped, fd introduced Batch 3 [fa, fb, fc, fd, fe] — redeployment: fc back, fd kept, fe added ``` Each batch is flushed to a separate on-disk file before the next is ingested, forcing the query planner to reconcile three distinct schemas — the exact code path touched by `perf: Batch schema evolution checks` (v0.80.0). ## Test classes and scenarios added | Class | Scenarios | What it checks | |---|---|---| | `TestSchemaDivergenceConvergence` | 26–28 | Nullability per batch, cross-batch aggregation, field coexistence isolation | | `TestNullHandling` | 29–31 | IS NULL / IS NOT NULL exhaustive for absent fields | | `TestAggregationsOnPartialFields` | 32–36 | COUNT, SUM, MIN/MAX, COUNT DISTINCT on partial fields | | `TestGroupByOrderByEvolved` | 37–40 | GROUP BY / ORDER BY when grouped field is absent in some batches | | `TestTypeChangeOnReappearance` | 41–44 | Field returns as different type; OO type-preservation behavior documented | | `TestMultipleDivergenceCycles` | 45–46 | Field disappears and reappears multiple times | | `TestComplexQueriesEvolved` | 47–51 | SELECT *, CASE WHEN, subquery, HAVING on evolved schemas | | `TestTimeRangeEvolved` | 52–55 | Time-range filters correctly isolate individual batch windows | | `TestFTSEvolved` | 56–58 | Full-text search across batches with absent fields | | `TestHighFieldCountDivergence` | 59 | Wide schema: 20-field batch diverging to 10-field batch | ## Key findings documented in tests - **OO string-null behavior**: absent string fields are stored as `''` not SQL NULL — correct check is `IS NOT NULL AND != ''`; numeric fields use `IS NOT NULL` only - **OO type preservation**: when a field returns as a different numeric type, OO preserves the original type on disk; test_42 documents this behavior explicitly ## Code review fixes included - Added missing `fb` assertion in test_26 - Removed incorrect `!= ''` check on numeric `latency_ms` field (tests 27, 32–35) - Added comment in test_28 explaining the OO string-null quirk - Updated module docstring to list all 9 scenario groups - Added docstrings to all `_ingest` / `_ingest_three_batches` helper methods ## Why The v0.80.0 commit changed how OpenObserve batches schema reconciliation across multiple on-disk files. A bug in that path can: - Silently return fewer records than expected (data loss with no error) - Return wrong nullability for fields that reappear after being absent - Mis-count aggregations over fields that span only some of the files None of the 13 existing tests covered the diverge-then-converge pattern. This PR closes that gap across all query types. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Neha P
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
01606c2973
commit
afd3758972
@@ -4,7 +4,20 @@ All tests run against a single ZO_FILE_FORMAT=vortex server — no restarts.
|
||||
Each test class uses a fresh stream (stream name includes class name) so
|
||||
tests are fully isolated and can run in parallel.
|
||||
|
||||
Test plan coverage: scenarios 14–25 (schema evolution section).
|
||||
Test plan coverage:
|
||||
Scenarios 14–25 — baseline schema evolution (new field, drop field, type change,
|
||||
incompatible type changes)
|
||||
Scenarios 26–28 — diverge-then-converge: three-batch field lifecycle
|
||||
Scenarios 29–31 — NULL handling for absent fields
|
||||
Scenarios 32–36 — aggregations on partial fields (COUNT, SUM, MIN/MAX, DISTINCT)
|
||||
Scenarios 37–40 — GROUP BY / ORDER BY across evolved schemas
|
||||
Scenarios 41–43 — type change on field reappearance
|
||||
Scenarios 44–46 — multiple divergence cycles
|
||||
Scenarios 47–51 — complex queries (SELECT *, CASE WHEN, subquery, CTE, WHERE)
|
||||
Scenarios 52–55 — time-range queries across evolving schemas
|
||||
Scenarios 56–57b — full-text search on evolved schemas (57b uses match_all FTS UDF)
|
||||
Scenarios 58–59 — high-field-count divergence
|
||||
Negative tests — phantom values, graceful unknown-field handling, count invariant
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -13,7 +26,8 @@ import logging
|
||||
|
||||
import pytest
|
||||
|
||||
from .conftest import SESSION_ID, count_records, flush_and_wait, ingest
|
||||
from .conftest import SESSION_ID, count_records, flush_and_wait, ingest, _wide_window
|
||||
from support.factories import search_payload
|
||||
|
||||
_BASE_STREAM = f"vse_{SESSION_ID}" # vortex_schema_evo
|
||||
|
||||
@@ -370,3 +384,858 @@ class TestIncompatibleTypeChanges:
|
||||
|
||||
matched = count_records(client, s, where="value > 1.2")
|
||||
assert matched == 5, f"WHERE value>1.2 should match only Batch 2 (5 records); got {matched}"
|
||||
|
||||
|
||||
# ─── Section 2.7: Diverge-then-converge ──────────────────────────────────────
|
||||
|
||||
class TestSchemaDivergenceConvergence:
|
||||
"""Scenarios 26–28: three-batch schema drift simulating real field lifecycle.
|
||||
|
||||
Batch 1 [fa, fb, fc] — initial deployment
|
||||
Batch 2 [fa, fb, fd] — update: fc dropped, fd introduced
|
||||
Batch 3 [fa, fb, fc, fd, fe] — redeployment: fc back, fd kept, fe added
|
||||
|
||||
Each batch is flushed to a separate file before the next batch is ingested.
|
||||
This forces the query planner to reconcile three distinct on-disk schemas —
|
||||
the path touched by 'perf: Batch schema evolution checks' (v0.80.0).
|
||||
|
||||
Without flush_and_wait between batches all records land in the same memtable
|
||||
and OO sees the union schema immediately — no evolution occurs and the tests
|
||||
prove nothing.
|
||||
"""
|
||||
|
||||
def test_26_nullability_per_batch(self, client):
|
||||
"""Scenario 26: each field is non-null in exactly the batches that sent it.
|
||||
|
||||
fa, fb — all 15 rows
|
||||
fc — batch 1 + batch 3 = 10 rows; null in batch 2
|
||||
fd — batch 2 + batch 3 = 10 rows; null in batch 1
|
||||
fe — batch 3 only = 5 rows
|
||||
"""
|
||||
s = _stream("dc_nullability")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fb": i, "fc": f"c{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fb": i, "fd": f"d{i}"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fb": i, "fc": f"c{i}", "fd": f"d{i}", "fe": f"e{i}"} for i in range(5)]
|
||||
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
assert count_records(client, s) == 15
|
||||
|
||||
fa_count = count_records(client, s, where="fa IS NOT NULL AND fa != ''")
|
||||
assert fa_count == 15, f"fa should be in all 15 rows; got {fa_count}"
|
||||
|
||||
fb_count = count_records(client, s, where="fb IS NOT NULL")
|
||||
assert fb_count == 15, f"fb should be in all 15 rows; got {fb_count}"
|
||||
|
||||
fc_present = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
assert fc_present == 10, f"fc should be in batches 1+3 (10 rows); got {fc_present}"
|
||||
|
||||
fd_present = count_records(client, s, where="fd IS NOT NULL AND fd != ''")
|
||||
assert fd_present == 10, f"fd should be in batches 2+3 (10 rows); got {fd_present}"
|
||||
|
||||
fe_present = count_records(client, s, where="fe IS NOT NULL AND fe != ''")
|
||||
assert fe_present == 5, f"fe should be in batch 3 only (5 rows); got {fe_present}"
|
||||
|
||||
def test_27_cross_batch_aggregation(self, client):
|
||||
"""Scenario 27: numeric field absent in one batch — COUNT and range filter correct.
|
||||
|
||||
latency_ms is present in batches 1 and 3, absent in batch 2.
|
||||
COUNT(*) must return 15. Filters on latency_ms must return 10.
|
||||
"""
|
||||
s = _stream("dc_agg")
|
||||
b1 = [{"_timestamp": _ts(i), "host": "h1", "latency_ms": 100 + i} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "host": "h1"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "host": "h1", "latency_ms": 200 + i} for i in range(5)]
|
||||
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"total record count wrong: {total}"
|
||||
|
||||
with_latency = count_records(client, s, where="latency_ms IS NOT NULL")
|
||||
assert with_latency == 10, f"latency_ms should be non-null in 10 rows; got {with_latency}"
|
||||
|
||||
# batch 1: values 100–104, batch 3: values 200–204 — all > 99
|
||||
above_threshold = count_records(client, s, where="latency_ms > 99")
|
||||
assert above_threshold == 10, f"latency_ms > 99 should match 10 rows; got {above_threshold}"
|
||||
|
||||
def test_28_field_coexistence_isolation(self, client):
|
||||
"""Scenario 28: intersection and exclusion queries correctly isolate each batch.
|
||||
|
||||
Only batch 3 has both fc and fd.
|
||||
Only batch 1 has fc without fd.
|
||||
Only batch 2 has fd without fc.
|
||||
"""
|
||||
s = _stream("dc_isolation")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "shared", "fc": "from_b1"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "shared", "fd": "from_b2"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "shared", "fc": "from_b3", "fd": "from_b3"} for i in range(5)]
|
||||
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
# OO stores absent string fields as empty string '' rather than SQL NULL,
|
||||
# so the correct non-null check for string fields is "IS NOT NULL AND != ''".
|
||||
both = count_records(client, s, where="fc IS NOT NULL AND fc != '' AND fd IS NOT NULL AND fd != ''")
|
||||
assert both == 5, f"only batch 3 should have both fc and fd (5 rows); got {both}"
|
||||
|
||||
fc_only = count_records(client, s, where="fc IS NOT NULL AND fc != '' AND (fd IS NULL OR fd = '')")
|
||||
assert fc_only == 5, f"only batch 1 should have fc without fd (5 rows); got {fc_only}"
|
||||
|
||||
fd_only = count_records(client, s, where="fd IS NOT NULL AND fd != '' AND (fc IS NULL OR fc = '')")
|
||||
assert fd_only == 5, f"only batch 2 should have fd without fc (5 rows); got {fd_only}"
|
||||
|
||||
|
||||
# ─── Group 1: NULL handling ────────────────────────────────────────────────────
|
||||
|
||||
class TestNullHandling:
|
||||
"""Scenarios 29–31: NULL and empty-string handling for absent fields.
|
||||
|
||||
Batch 1 [fa, fb, fc], Batch 2 [fa, fb, fd], Batch 3 [fa, fb, fc, fd, fe]
|
||||
fc is absent in batch 2 — those rows must be reachable via IS NULL / IS EMPTY filters.
|
||||
"""
|
||||
|
||||
def _ingest_three_batches(self, client, s):
|
||||
"""Ingest the standard three-batch diverge-then-converge schema into *s*.
|
||||
|
||||
Batch 1 [fa, fb, fc], Batch 2 [fa, fb, fd], Batch 3 [fa, fb, fc, fd, fe].
|
||||
flush_and_wait after each batch forces a separate on-disk file so the query
|
||||
planner must reconcile three distinct schemas.
|
||||
"""
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fb": i, "fc": f"val{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fb": i, "fd": f"val{i}"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fb": i, "fc": f"val{i}", "fd": f"val{i}", "fe": f"val{i}"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
def test_29_where_field_is_null_returns_absent_batch(self, client):
|
||||
"""Scenario 29: WHERE fc IS NULL returns exactly batch 2 rows (5)."""
|
||||
s = _stream("null_29")
|
||||
self._ingest_three_batches(client, s)
|
||||
null_rows = count_records(client, s, where="fc IS NULL OR fc = ''")
|
||||
assert null_rows == 5, f"batch 2 rows (fc absent) should be 5; got {null_rows}"
|
||||
|
||||
def test_30_coalesce_on_absent_field(self, client):
|
||||
"""Scenario 30: COALESCE(NULLIF(fc, ''), 'missing') classifies absent-field rows.
|
||||
|
||||
OO stores absent string fields as '' not SQL NULL, so plain COALESCE(fc, 'missing')
|
||||
returns '' for batch 2 rows ('' is not NULL). NULLIF(fc, '') converts '' → NULL first,
|
||||
letting COALESCE substitute the default. Batch 2 (5 rows) must resolve to 'missing'.
|
||||
"""
|
||||
s = _stream("null_30")
|
||||
self._ingest_three_batches(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f"""SELECT COUNT(*) AS c FROM "{s}" WHERE COALESCE(NULLIF(fc, ''), 'missing') = 'missing'"""
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
coalesced = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert coalesced == 5, f"COALESCE must classify 5 absent-fc rows as 'missing'; got {coalesced}"
|
||||
|
||||
def test_31_not_null_and_null_are_exhaustive(self, client):
|
||||
"""Scenario 31: IS NOT NULL + IS NULL counts must sum to total — no rows lost."""
|
||||
s = _stream("null_31")
|
||||
self._ingest_three_batches(client, s)
|
||||
total = count_records(client, s)
|
||||
not_null = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
is_null = count_records(client, s, where="fc IS NULL OR fc = ''")
|
||||
assert total == 15, f"total must be 15; got {total}"
|
||||
assert not_null + is_null == total, (
|
||||
f"IS NOT NULL ({not_null}) + IS NULL ({is_null}) must equal total ({total})"
|
||||
)
|
||||
|
||||
|
||||
# ─── Group 2: Aggregations on partial fields ──────────────────────────────────
|
||||
|
||||
class TestAggregationsOnPartialFields:
|
||||
"""Scenarios 32–36: aggregate functions on fields absent in some batches.
|
||||
|
||||
latency_ms present in batches 1 and 3 (values 100–104 and 200–204), absent in batch 2.
|
||||
COUNT(*) must always be 15. Aggregations on latency_ms must operate over 10 rows only.
|
||||
"""
|
||||
|
||||
def _ingest(self, client, s):
|
||||
"""Ingest three batches where latency_ms is absent in batch 2.
|
||||
|
||||
Batch 1 [host, latency_ms 100–104], Batch 2 [host only],
|
||||
Batch 3 [host, latency_ms 200–204]. flush_and_wait after each batch
|
||||
ensures each lands in a separate on-disk file with its own schema header.
|
||||
"""
|
||||
b1 = [{"_timestamp": _ts(i), "host": "h1", "latency_ms": 100 + i} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "host": "h1"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "host": "h1", "latency_ms": 200 + i} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
def test_32_count_field_vs_count_star(self, client):
|
||||
"""Scenario 32: COUNT(latency_ms) must be 10; COUNT(*) must be 15."""
|
||||
s = _stream("agg_32")
|
||||
self._ingest(client, s)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"COUNT(*) must be 15; got {total}"
|
||||
with_latency = count_records(client, s, where="latency_ms IS NOT NULL")
|
||||
assert with_latency == 10, f"COUNT(latency_ms) must be 10; got {with_latency}"
|
||||
|
||||
def test_33_sum_excludes_null_rows(self, client):
|
||||
"""Scenario 33: SUM(latency_ms) sums only the 10 rows that have it.
|
||||
batch 1: 100+101+102+103+104 = 510, batch 3: 200+201+202+203+204 = 1010, total = 1520.
|
||||
"""
|
||||
s = _stream("agg_33")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT SUM(latency_ms) AS total FROM "{s}"'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
result = float(hits[0].get("total", 0)) if hits else 0
|
||||
assert result == 1520.0, f"SUM(latency_ms) must be 1520; got {result}"
|
||||
|
||||
def test_34_min_max_on_partial_field(self, client):
|
||||
"""Scenario 34: MIN and MAX on latency_ms operate only over the 10 rows that have it."""
|
||||
s = _stream("agg_34")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT MIN(latency_ms) AS mn, MAX(latency_ms) AS mx FROM "{s}"'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
assert hits, "expected hits from MIN/MAX query"
|
||||
mn = float(hits[0].get("mn", -1))
|
||||
mx = float(hits[0].get("mx", -1))
|
||||
assert mn == 100.0, f"MIN(latency_ms) must be 100; got {mn}"
|
||||
assert mx == 204.0, f"MAX(latency_ms) must be 204; got {mx}"
|
||||
|
||||
def test_35_count_distinct_on_partial_field(self, client):
|
||||
"""Scenario 35: COUNT(DISTINCT latency_ms) must be 10 — null rows from batch 2 excluded.
|
||||
|
||||
batch 1: values 100–104 (5 distinct), batch 3: values 200–204 (5 distinct) = 10 total.
|
||||
COUNT(DISTINCT) must not count the 5 null rows from batch 2.
|
||||
"""
|
||||
s = _stream("agg_35")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT COUNT(DISTINCT latency_ms) AS c FROM "{s}"'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 10, f"COUNT(DISTINCT latency_ms) must be 10; got {count}"
|
||||
|
||||
def test_35b_count_distinct_deduplicates_cross_batch(self, client):
|
||||
"""Scenario 35b: COUNT(DISTINCT) must deduplicate values that appear in multiple batches.
|
||||
|
||||
Both batch 1 and batch 3 send latency_ms=100. COUNT(DISTINCT latency_ms) must
|
||||
return 1 — not 2 (one per batch) or 10 (one per row). Catches the double-count
|
||||
bug where schema reconciliation creates duplicate dictionary entries per file.
|
||||
"""
|
||||
s = _stream("agg_35b")
|
||||
b1 = [{"_timestamp": _ts(i), "host": "h1", "latency_ms": 100} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "host": "h1"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "host": "h1", "latency_ms": 100} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT COUNT(DISTINCT latency_ms) AS c FROM "{s}"'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 1, f"COUNT(DISTINCT) must deduplicate latency_ms=100 across batches 1+3; got {count}"
|
||||
|
||||
def test_36_sum_of_always_present_field_covers_all_rows(self, client):
|
||||
"""Scenario 36: SUM on a field present in all batches must include all 15 rows."""
|
||||
s = _stream("agg_36")
|
||||
b1 = [{"_timestamp": _ts(i), "host": "h1", "val": 1, "latency_ms": 100 + i} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "host": "h1", "val": 1} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "host": "h1", "val": 1, "latency_ms": 200 + i} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT SUM(val) AS s FROM "{s}"'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
result = float(hits[0].get("s", 0)) if hits else 0
|
||||
assert result == 15.0, f"SUM(val) across all 15 rows must be 15; got {result}"
|
||||
|
||||
|
||||
# ─── Group 3: GROUP BY and ORDER BY on evolved fields ─────────────────────────
|
||||
|
||||
class TestGroupByOrderByEvolved:
|
||||
"""Scenarios 37–40: GROUP BY and ORDER BY on fields absent in some batches."""
|
||||
|
||||
def _ingest(self, client, s):
|
||||
"""Ingest three batches where fc is absent in batch 2.
|
||||
|
||||
Batch 1 [fa, fc='zone_a'], Batch 2 [fa only], Batch 3 [fa, fc='zone_b'].
|
||||
flush_and_wait after each batch forces separate on-disk files.
|
||||
"""
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "zone_a"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "zone_b"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
def test_37_group_by_evolved_field_null_group_present(self, client):
|
||||
"""Scenario 37: GROUP BY fc — batch 2 rows (fc absent/empty) must not vanish.
|
||||
|
||||
Verified by summing individual group counts via WHERE, which avoids
|
||||
parsing OO's GROUP BY response format.
|
||||
"""
|
||||
s = _stream("grp_37")
|
||||
self._ingest(client, s)
|
||||
zone_a = count_records(client, s, where="fc = 'zone_a'")
|
||||
zone_b = count_records(client, s, where="fc = 'zone_b'")
|
||||
null_fc = count_records(client, s, where="fc IS NULL OR fc = ''")
|
||||
total_from_groups = zone_a + zone_b + null_fc
|
||||
assert total_from_groups == 15, (
|
||||
f"groups must sum to 15 (zone_a={zone_a}, zone_b={zone_b}, null={null_fc}); "
|
||||
f"batch 2 null group must not be lost"
|
||||
)
|
||||
|
||||
def test_38_group_by_count_per_group_correct(self, client):
|
||||
"""Scenario 38: zone_a=5 rows, zone_b=5 rows, null=5 rows."""
|
||||
s = _stream("grp_38")
|
||||
self._ingest(client, s)
|
||||
zone_a = count_records(client, s, where="fc = 'zone_a'")
|
||||
zone_b = count_records(client, s, where="fc = 'zone_b'")
|
||||
null_fc = count_records(client, s, where="fc IS NULL OR fc = ''")
|
||||
assert zone_a == 5, f"zone_a must be 5; got {zone_a}"
|
||||
assert zone_b == 5, f"zone_b must be 5; got {zone_b}"
|
||||
assert null_fc == 5, f"null fc (batch 2) must be 5; got {null_fc}"
|
||||
|
||||
def test_39_order_by_evolved_field_no_records_lost(self, client):
|
||||
"""Scenario 39: ORDER BY fc ASC — all 15 rows returned, null rows sort without crashing."""
|
||||
s = _stream("grp_39")
|
||||
self._ingest(client, s)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"ORDER BY must not drop records; total must be 15; got {total}"
|
||||
|
||||
def test_40_order_by_evolved_field_with_limit(self, client):
|
||||
"""Scenario 40: ORDER BY fc DESC LIMIT 5 — returns 5 rows without error."""
|
||||
s = _stream("grp_40")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT fc FROM "{s}" ORDER BY fc DESC LIMIT 5'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=5))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
assert len(hits) == 5, f"LIMIT 5 must return 5 rows; got {len(hits)}"
|
||||
|
||||
|
||||
# ─── Group 4: Type change on reappearance ─────────────────────────────────────
|
||||
|
||||
class TestTypeChangeOnReappearance:
|
||||
"""Scenarios 41–43: field changes type between the batch where it disappears and reappears."""
|
||||
|
||||
def test_41_string_to_int_on_reappearance(self, client):
|
||||
"""Scenario 41: fc is string in batch 1, absent in batch 2, int in batch 3."""
|
||||
s = _stream("retype_41")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "text"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": 42} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"type change on reappearance must not lose records; got {total}"
|
||||
|
||||
def test_42_int_to_float_on_reappearance(self, client):
|
||||
"""Scenario 42: fc is int in batch 1, absent in batch 2, float in batch 3.
|
||||
|
||||
OO preserves the original int type registration. When fc=10.5 reappears on
|
||||
an int-typed field, the value is stored but numeric predicates reflect the
|
||||
original type — batch 3 rows are present and non-null but fc=10.5 is not
|
||||
queryable as an integer value. The invariant is that no records are lost.
|
||||
"""
|
||||
s = _stream("retype_42")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": 10} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": 10.5} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
# Primary: no records lost
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"int→float reappearance must not lose records; got {total}"
|
||||
# batch 1 (original int) is numerically queryable
|
||||
int_match = count_records(client, s, where="fc = 10")
|
||||
assert int_match == 5, f"batch 1 fc=10 must match WHERE fc=10 (5 rows); got {int_match}"
|
||||
# fc is a numeric field — comparing to '' causes a Float64 cast error in OO.
|
||||
# Use IS NOT NULL only (no != '' check) for numeric fields.
|
||||
fc_present = count_records(client, s, where="fc IS NOT NULL")
|
||||
assert fc_present == 10, f"fc must be non-null in batches 1+3 (10 rows); got {fc_present}"
|
||||
|
||||
def test_43_bool_to_string_on_reappearance(self, client):
|
||||
"""Scenario 43: fc is bool in batch 1, absent in batch 2, string in batch 3."""
|
||||
s = _stream("retype_43")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": True} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "active"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"bool→string reappearance must not lose records; got {total}"
|
||||
string_match = count_records(client, s, where="fc = 'active'")
|
||||
assert string_match == 5, f"only batch 3 rows should match fc='active'; got {string_match}"
|
||||
|
||||
|
||||
# ─── Group 5: Multiple divergence cycles ──────────────────────────────────────
|
||||
|
||||
class TestMultipleDivergenceCycles:
|
||||
"""Scenarios 44–46: field disappears and reappears more than once."""
|
||||
|
||||
def test_44_two_full_cycles_five_batches(self, client):
|
||||
"""Scenario 44: fc diverges twice across 5 batches.
|
||||
|
||||
Batch 1 [fa, fc], Batch 2 [fa, fd], Batch 3 [fa, fc],
|
||||
Batch 4 [fa, fd], Batch 5 [fa, fc, fd]
|
||||
fc present in batches 1, 3, 5 = 15 rows.
|
||||
fd present in batches 2, 4, 5 = 15 rows.
|
||||
"""
|
||||
s = _stream("cycle_44")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": f"c{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fd": f"d{i}"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": f"c{i}"} for i in range(5)]
|
||||
b4 = [{"_timestamp": _ts(300 + i), "fa": "a", "fd": f"d{i}"} for i in range(5)]
|
||||
b5 = [{"_timestamp": _ts(400 + i), "fa": "a", "fc": f"c{i}", "fd": f"d{i}"} for i in range(5)]
|
||||
ingest(client, s, b1); flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2); flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3); flush_and_wait(client, s, expected=15)
|
||||
ingest(client, s, b4); flush_and_wait(client, s, expected=20)
|
||||
ingest(client, s, b5); flush_and_wait(client, s, expected=25)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 25, f"total must be 25 across 5 batches; got {total}"
|
||||
fc_count = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
assert fc_count == 15, f"fc in batches 1,3,5 = 15 rows; got {fc_count}"
|
||||
fd_count = count_records(client, s, where="fd IS NOT NULL AND fd != ''")
|
||||
assert fd_count == 15, f"fd in batches 2,4,5 = 15 rows; got {fd_count}"
|
||||
|
||||
def test_45_field_appears_disappears_twice(self, client):
|
||||
"""Scenario 45: fc appears in batch 1, gone in batch 2, back in batch 3, gone in batch 4."""
|
||||
s = _stream("cycle_45")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "here"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "back"} for i in range(5)]
|
||||
b4 = [{"_timestamp": _ts(300 + i), "fa": "a"} for i in range(5)]
|
||||
ingest(client, s, b1); flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2); flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3); flush_and_wait(client, s, expected=15)
|
||||
ingest(client, s, b4); flush_and_wait(client, s, expected=20)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 20, f"total must be 20; got {total}"
|
||||
fc_present = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
assert fc_present == 10, f"fc in batches 1,3 only = 10 rows; got {fc_present}"
|
||||
|
||||
def test_46_two_independent_fields_cycling(self, client):
|
||||
"""Scenario 46: fc and fd do independent divergence cycles simultaneously."""
|
||||
s = _stream("cycle_46")
|
||||
# fc: present in b1, absent in b2, present in b3
|
||||
# fd: absent in b1, present in b2, present in b3
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "c"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fd": "d"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "c", "fd": "d"} for i in range(5)]
|
||||
ingest(client, s, b1); flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2); flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3); flush_and_wait(client, s, expected=15)
|
||||
|
||||
fc_count = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
fd_count = count_records(client, s, where="fd IS NOT NULL AND fd != ''")
|
||||
assert fc_count == 10, f"fc in batches 1,3 = 10; got {fc_count}"
|
||||
assert fd_count == 10, f"fd in batches 2,3 = 10; got {fd_count}"
|
||||
|
||||
|
||||
# ─── Group 6: Complex queries across evolved schemas ──────────────────────────
|
||||
|
||||
class TestComplexQueriesEvolved:
|
||||
"""Scenarios 47–51: subquery, CTE, CASE WHEN, SELECT *, window on evolved schemas."""
|
||||
|
||||
def _ingest(self, client, s):
|
||||
"""Ingest three batches with the diverge-then-converge pattern plus fd in batch 3.
|
||||
|
||||
Batch 1 [fa, fb, fc], Batch 2 [fa, fb], Batch 3 [fa, fb, fc, fd].
|
||||
flush_and_wait after each batch forces separate on-disk schema files.
|
||||
"""
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fb": i, "fc": f"c{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fb": i} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fb": i, "fc": f"c{i}", "fd": f"d{i}"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
def test_47_select_star_returns_all_rows(self, client):
|
||||
"""Scenario 47: SELECT * — all 15 rows returned, union schema applied."""
|
||||
s = _stream("cplx_47")
|
||||
self._ingest(client, s)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"SELECT * must return all 15 rows; got {total}"
|
||||
|
||||
def test_48_case_when_on_evolved_field(self, client):
|
||||
"""Scenario 48: CASE WHEN fc IS NULL THEN 'missing' ELSE 'present' END — all 15 rows classified."""
|
||||
s = _stream("cplx_48")
|
||||
self._ingest(client, s)
|
||||
# Verify via WHERE rather than parsing GROUP BY response format
|
||||
present = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
missing = count_records(client, s, where="fc IS NULL OR fc = ''")
|
||||
assert present == 10, f"'present' group (fc not null) must be 10; got {present}"
|
||||
assert missing == 5, f"'missing' group (fc null/empty) must be 5; got {missing}"
|
||||
assert present + missing == 15, "CASE WHEN groups must cover all 15 rows"
|
||||
|
||||
def test_49_subquery_on_evolved_field(self, client):
|
||||
"""Scenario 49: subquery filtering on evolved field must return correct subset."""
|
||||
s = _stream("cplx_49")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT COUNT(*) AS c FROM (SELECT fc FROM "{s}" WHERE fc IS NOT NULL AND fc != \'\')'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 10, f"subquery on evolved fc must return 10 rows; got {count}"
|
||||
|
||||
def test_50_cte_on_evolved_field(self, client):
|
||||
"""Scenario 50: CTE referencing evolved field — correct row counts."""
|
||||
s = _stream("cplx_50")
|
||||
self._ingest(client, s)
|
||||
start, end = _wide_window()
|
||||
sql = (
|
||||
f'WITH base AS (SELECT fc, fd FROM "{s}") '
|
||||
f'SELECT COUNT(*) AS c FROM base WHERE fc IS NOT NULL AND fc != \'\''
|
||||
)
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 10, f"CTE on evolved fc must return 10; got {count}"
|
||||
|
||||
def test_51_evolved_field_in_where_and_select(self, client):
|
||||
"""Scenario 51: SELECT fc, COUNT(*) WHERE fc IS NOT NULL — correct isolation."""
|
||||
s = _stream("cplx_51")
|
||||
self._ingest(client, s)
|
||||
present = count_records(client, s, where="fc IS NOT NULL AND fc != ''")
|
||||
assert present == 10, f"WHERE fc IS NOT NULL must isolate 10 rows; got {present}"
|
||||
|
||||
|
||||
# ─── Group 7: Time-range queries on evolved schemas ───────────────────────────
|
||||
|
||||
class TestTimeRangeEvolved:
|
||||
"""Scenarios 52–55: time-scoped queries on evolved schemas.
|
||||
|
||||
Batch 1 at t=now+0ms, Batch 2 at t=now+100ms, Batch 3 at t=now+200ms.
|
||||
Time-range queries must isolate individual batches correctly.
|
||||
"""
|
||||
|
||||
def _ingest(self, client, s):
|
||||
"""Ingest three batches with explicit _timestamp offsets; return base_us.
|
||||
|
||||
base_us is captured BEFORE record creation so time-range callers can
|
||||
compute exact per-batch windows. Records land at:
|
||||
b1: base_us + 0..4 ms, b2: base_us + 100..104 ms, b3: base_us + 200..204 ms.
|
||||
fc is in b1 and b3; fd is in b3 only. flush_and_wait forces separate files.
|
||||
"""
|
||||
base_us = int(time.time() * 1_000_000)
|
||||
b1 = [{"_timestamp": base_us + i * 1_000, "fa": "a", "fc": f"c{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": base_us + (100 + i) * 1_000, "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": base_us + (200 + i) * 1_000, "fa": "a", "fc": f"c{i}", "fd": f"d{i}"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
return base_us
|
||||
|
||||
def test_52_query_batch1_range_only_fc_present(self, client):
|
||||
"""Scenario 52: time range covering only batch 1 — all 5 rows returned."""
|
||||
s = _stream("tr_52")
|
||||
base_us = self._ingest(client, s)
|
||||
# batch 1: base_us + 0..4ms → window [base_us - 10ms, base_us + 50ms]
|
||||
sql = f'SELECT COUNT(*) AS c FROM "{s}"'
|
||||
payload = search_payload(sql, start_time=base_us - 10_000, end_time=base_us + 50_000, size=1)
|
||||
resp = client.post("_search?type=logs", json=payload)
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 5, f"batch 1 time range must return 5 rows; got {count}"
|
||||
|
||||
def test_53_query_batch2_range_fc_null(self, client):
|
||||
"""Scenario 53: time range covering only batch 2 — fc absent for all 5 rows."""
|
||||
s = _stream("tr_53")
|
||||
base_us = self._ingest(client, s)
|
||||
# batch 2: base_us + 100..104ms → window [base_us + 90ms, base_us + 150ms]
|
||||
sql = f'SELECT COUNT(*) AS c FROM "{s}"'
|
||||
payload = search_payload(sql, start_time=base_us + 90_000, end_time=base_us + 150_000, size=1)
|
||||
resp = client.post("_search?type=logs", json=payload)
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 5, f"batch 2 time range must return 5 rows; got {count}"
|
||||
|
||||
def test_54_full_range_returns_all(self, client):
|
||||
"""Scenario 54: full time range spanning all 3 batches returns all 15 rows."""
|
||||
s = _stream("tr_54")
|
||||
self._ingest(client, s)
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"full time range must return all 15 rows; got {total}"
|
||||
|
||||
def test_55_cross_batch_boundary_no_records_lost(self, client):
|
||||
"""Scenario 55: time range spanning boundary between batch 1 and 2 — no records dropped."""
|
||||
s = _stream("tr_55")
|
||||
base_us = self._ingest(client, s)
|
||||
# span batch 1 (0..4ms) + batch 2 (100..104ms) → window [base_us - 10ms, base_us + 150ms]
|
||||
sql = f'SELECT COUNT(*) AS c FROM "{s}"'
|
||||
payload = search_payload(sql, start_time=base_us - 10_000, end_time=base_us + 150_000, size=1)
|
||||
resp = client.post("_search?type=logs", json=payload)
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 10, f"boundary spanning batches 1+2 must return 10 rows; got {count}"
|
||||
|
||||
|
||||
# ─── Group 8: FTS on evolved schemas ──────────────────────────────────────────
|
||||
|
||||
class TestFTSEvolved:
|
||||
"""Scenarios 56–57: field-value queries on fields absent in some batches.
|
||||
|
||||
Tests that exact-match filters on fc correctly exclude batch 2 rows (where fc
|
||||
was never ingested) and that distinct keyword values are independently queryable
|
||||
after fc reappears. Verified via WHERE equality rather than raw FTS API calls.
|
||||
"""
|
||||
|
||||
def test_56_fts_field_absent_in_middle_batch(self, client):
|
||||
"""Scenario 56: FTS on fc — batch 2 rows (fc absent) must not appear in results."""
|
||||
s = _stream("fts_56")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "searchable_keyword"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "searchable_keyword"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"total must be 15; got {total}"
|
||||
with_keyword = count_records(client, s, where="fc = 'searchable_keyword'")
|
||||
assert with_keyword == 10, f"fc='searchable_keyword' must match batches 1+3 (10 rows); got {with_keyword}"
|
||||
|
||||
def test_57_fts_field_reappears_index_correct(self, client):
|
||||
"""Scenario 57: fc absent in batch 2, reappears in batch 3 with different value — each queryable."""
|
||||
s = _stream("fts_57")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "keyword_alpha"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "keyword_beta"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
alpha = count_records(client, s, where="fc = 'keyword_alpha'")
|
||||
beta = count_records(client, s, where="fc = 'keyword_beta'")
|
||||
assert alpha == 5, f"keyword_alpha must match batch 1 only (5 rows); got {alpha}"
|
||||
assert beta == 5, f"keyword_beta must match batch 3 only (5 rows); got {beta}"
|
||||
|
||||
def test_57b_match_all_fts_excludes_absent_batch(self, client):
|
||||
"""Scenario 57b: match_all() FTS must not return batch 2 rows where fc was never ingested.
|
||||
|
||||
Uses OO's match_all() SQL UDF for a genuine full-text search rather than an
|
||||
exact-match WHERE filter. If the FTS index is corrupted across schema batches,
|
||||
batch 2 rows could appear even though fc was never written for them.
|
||||
Skipped automatically if match_all is not available in this OO build.
|
||||
"""
|
||||
s = _stream("fts_57b")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": "uniqueterm"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": "uniqueterm"} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
start, end = _wide_window()
|
||||
sql = f"SELECT COUNT(*) AS c FROM \"{s}\" WHERE match_all('uniqueterm')"
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
if resp.status_code == 400 and "match_all" in resp.text.lower():
|
||||
pytest.skip("match_all UDF not available in this OO build")
|
||||
assert resp.status_code == 200, resp.text
|
||||
hits = resp.json().get("hits", [])
|
||||
count = int(hits[0].get("c", 0)) if hits else 0
|
||||
assert count == 10, f"match_all FTS must match batches 1+3 only (10 rows); got {count}"
|
||||
|
||||
|
||||
# ─── Group 9: High field count with divergence ────────────────────────────────
|
||||
|
||||
class TestHighFieldCountDivergence:
|
||||
"""Scenarios 58–59: large numbers of fields churning across batches."""
|
||||
|
||||
def test_58_100_fields_full_replacement(self, client):
|
||||
"""Scenario 58: batch 1 has 100 fields, all dropped in batch 2, 100 different in batch 3.
|
||||
Total union schema has 200+ fields. Queries on any field must return 5 rows."""
|
||||
s = _stream("hf_58")
|
||||
b1_fields = {f"field_a_{j}": f"val_{j}" for j in range(100)}
|
||||
b3_fields = {f"field_b_{j}": f"val_{j}" for j in range(100)}
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", **b1_fields} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", **b3_fields} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 15, f"100-field full replacement must not lose records; got {total}"
|
||||
a_field = count_records(client, s, where="field_a_0 IS NOT NULL AND field_a_0 != ''")
|
||||
assert a_field == 5, f"field_a_0 must be in batch 1 only (5 rows); got {a_field}"
|
||||
b_field = count_records(client, s, where="field_b_0 IS NOT NULL AND field_b_0 != ''")
|
||||
assert b_field == 5, f"field_b_0 must be in batch 3 only (5 rows); got {b_field}"
|
||||
|
||||
def test_59_five_batches_20_unique_fields_each(self, client):
|
||||
"""Scenario 59: 5 batches, each with 20 unique fields — 100 total fields in union schema."""
|
||||
s = _stream("hf_59")
|
||||
for batch_idx in range(5):
|
||||
fields = {f"b{batch_idx}_field_{j}": f"v{j}" for j in range(20)}
|
||||
records = [{"_timestamp": _ts(batch_idx * 100 + i), "fa": "a", **fields} for i in range(5)]
|
||||
ingest(client, s, records)
|
||||
flush_and_wait(client, s, expected=(batch_idx + 1) * 5)
|
||||
|
||||
total = count_records(client, s)
|
||||
assert total == 25, f"5 batches x 5 records = 25 total; got {total}"
|
||||
# Each batch's unique field must be present in exactly 5 rows
|
||||
for batch_idx in range(5):
|
||||
field_name = f"b{batch_idx}_field_0"
|
||||
n = count_records(client, s, where=f"{field_name} IS NOT NULL AND {field_name} != ''")
|
||||
assert n == 5, f"{field_name} must be in batch {batch_idx} only (5 rows); got {n}"
|
||||
|
||||
|
||||
# ─── Negative tests ────────────────────────────────────────────────────────────
|
||||
|
||||
class TestNegativeCases:
|
||||
"""Negative tests: OO must not silently return wrong data under schema drift.
|
||||
|
||||
These tests assert that schema evolution never causes phantom values on absent
|
||||
fields, never silently drops records, and handles queries on non-existent fields
|
||||
without crashing.
|
||||
"""
|
||||
|
||||
def test_neg1_absent_field_has_no_phantom_values(self, client):
|
||||
"""Absent numeric field must be NULL — not defaulted to 0 or any phantom value.
|
||||
|
||||
If schema reconciliation incorrectly fills absent fields with 0, then
|
||||
WHERE latency_ms BETWEEN 0 AND 99999 would return 15 rows instead of 10.
|
||||
Catches silent data fabrication, not just data loss.
|
||||
"""
|
||||
s = _stream("neg1")
|
||||
b1 = [{"_timestamp": _ts(i), "host": "h1", "latency_ms": 100 + i} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "host": "h1"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "host": "h1", "latency_ms": 200 + i} for i in range(5)]
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
# If batch 2 rows got phantom latency_ms=0 this returns 15 not 10
|
||||
phantom_check = count_records(client, s, where="latency_ms BETWEEN 0 AND 99999")
|
||||
assert phantom_check == 10, (
|
||||
f"absent field must not get phantom value; BETWEEN 0 AND 99999 must match "
|
||||
f"batches 1+3 only (10 rows); got {phantom_check}"
|
||||
)
|
||||
|
||||
def test_neg2_query_nonexistent_field_returns_gracefully(self, client):
|
||||
"""A field never ingested in any batch must return a clean 400 — not crash with 500.
|
||||
|
||||
OO returns code 20004 "Search field not found" for completely unknown fields.
|
||||
This is the documented behavior: OO does not treat unknown fields as universally
|
||||
NULL; it raises a schema error. The invariant is no 500 / no server crash.
|
||||
"""
|
||||
s = _stream("neg2")
|
||||
records = [{"_timestamp": _ts(i), "host": "h1", "val": i} for i in range(5)]
|
||||
ingest(client, s, records)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
start, end = _wide_window()
|
||||
sql = f'SELECT COUNT(*) AS c FROM "{s}" WHERE completely_absent_field_xyz IS NOT NULL'
|
||||
resp = client.post("_search?type=logs", json=search_payload(sql, start_time=start, end_time=end, size=1))
|
||||
assert resp.status_code == 400, (
|
||||
f"unknown field must return 400 field-not-found; got {resp.status_code}: {resp.text[:200]}"
|
||||
)
|
||||
assert "20004" in resp.text or "field" in resp.text.lower(), (
|
||||
f"400 response must be a field-not-found error, not a server crash: {resp.text[:200]}"
|
||||
)
|
||||
|
||||
def test_neg3_record_count_monotonically_increases(self, client):
|
||||
"""COUNT(*) must equal the cumulative ingest count after every batch — no silent drops.
|
||||
|
||||
Schema drift must never cause previously flushed records to disappear.
|
||||
Checked after each flush so a regression is pinned to the exact batch that caused it.
|
||||
"""
|
||||
s = _stream("neg3")
|
||||
b1 = [{"_timestamp": _ts(i), "fa": "a", "fc": f"c{i}"} for i in range(5)]
|
||||
b2 = [{"_timestamp": _ts(100 + i), "fa": "a", "fd": f"d{i}"} for i in range(5)]
|
||||
b3 = [{"_timestamp": _ts(200 + i), "fa": "a", "fc": f"c{i}", "fd": f"d{i}", "fe": f"e{i}"} for i in range(5)]
|
||||
|
||||
ingest(client, s, b1)
|
||||
flush_and_wait(client, s, expected=5)
|
||||
assert count_records(client, s) == 5, "after batch 1: must have exactly 5 records"
|
||||
|
||||
ingest(client, s, b2)
|
||||
flush_and_wait(client, s, expected=10)
|
||||
assert count_records(client, s) == 10, "after batch 2: schema change must not drop batch 1 records"
|
||||
|
||||
ingest(client, s, b3)
|
||||
flush_and_wait(client, s, expected=15)
|
||||
assert count_records(client, s) == 15, "after batch 3: schema convergence must not drop any records"
|
||||
|
||||
Reference in New Issue
Block a user