java-topology/defects/elasticsearch/unit/test_elasticsearch_scale.py

"""
Elasticsearch CWE-407 scaling benchmark — projected wall-clock savings.

Simulates bulk ingest with append dedup (elasticsearch-0001) at production scale.
Each document has an array field with k existing values and receives N new values
with ~50% overlap (realistic for tag/label append pipelines).

The cost per document:
  Before: O(N×k) for the list.contains() calls (k grows as values append)
  After:  O(N) for the set.add() calls (O(1) each)

Total cost for D documents:
  Before: D × O(N×k) ≈ D × N × k/2 comparisons (average)
  After:  D × O(N) ≈ D × N hash lookups
"""

import time
import sys


def append_values_before(existing, new_values):
    """Original: list.contains O(k) per value."""
    result = list(existing)
    for val in new_values:
        if val not in result:
            result.append(val)
    return result


def append_values_after(existing, new_values):
    """Fixed: HashSet shadow O(1) per value."""
    result = list(existing)
    seen = set(result)
    for val in new_values:
        if val not in seen:
            result.append(val)
            seen.add(val)
    return result


def benchmark_bulk_ingest(num_docs, field_size, new_per_doc, overlap_pct=50):
    """
    Simulate bulk ingest of num_docs documents.
    Each document has an array field with field_size existing values.
    Each append adds new_per_doc values with overlap_pct% already present.
    """
    overlap = int(new_per_doc * overlap_pct / 100)
    unique_new = new_per_doc - overlap

    # Build representative document field and new values
    existing = list(range(field_size))
    new_values = list(range(field_size - overlap, field_size - overlap + new_per_doc))

    # Time BEFORE (list.contains)
    t0 = time.perf_counter()
    for _ in range(num_docs):
        append_values_before(existing, new_values)
    t_before = time.perf_counter() - t0

    # Time AFTER (set shadow)
    t0 = time.perf_counter()
    for _ in range(num_docs):
        append_values_after(existing, new_values)
    t_after = time.perf_counter() - t0

    speedup = t_before / t_after if t_after > 1e-9 else float('inf')

    return t_before, t_after, speedup


def format_time(seconds):
    if seconds < 0.001:
        return f"{seconds*1_000_000:.0f}us"
    elif seconds < 1:
        return f"{seconds*1000:.1f}ms"
    elif seconds < 60:
        return f"{seconds:.2f}s"
    elif seconds < 3600:
        return f"{seconds/60:.1f}min"
    else:
        return f"{seconds/3600:.1f}hr"


def main():
    print("=" * 90)
    print("Elasticsearch CWE-407 Scaling Benchmark — IngestDocument.appendValues()")
    print("=" * 90)
    print()

    # Scenario 1: Small array fields (k=20, N=10) — tags, labels, categories
    print("Scenario 1: Tag/label append (k=20 existing, N=10 new, 50% overlap)")
    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
    print("-" * 65)
    for num_docs in [10_000, 100_000, 1_000_000]:
        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 20, 10)
        saved = t_b - t_a
        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")

    print()

    # Scenario 2: Medium array fields (k=100, N=50) — enrichment pipelines
    print("Scenario 2: Enrichment pipeline (k=100 existing, N=50 new, 50% overlap)")
    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
    print("-" * 65)
    for num_docs in [10_000, 100_000, 1_000_000]:
        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 100, 50)
        saved = t_b - t_a
        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")

    print()

    # Scenario 3: Large array fields (k=500, N=200) — IoT sensor tags, log enrichment
    print("Scenario 3: Large field append (k=500 existing, N=200 new, 50% overlap)")
    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
    print("-" * 65)
    for num_docs in [10_000, 100_000, 1_000_000]:
        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 500, 200)
        saved = t_b - t_a
        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")

    print()

    # Scenario 4: Projected at 10M/100M scale (extrapolate from 10K measurement)
    print("Scenario 4: Projected savings at production scale (k=100, N=50)")
    print("(Extrapolated linearly from measured 10K document benchmark)")
    print("-" * 65)
    t_b_10k, t_a_10k, spd = benchmark_bulk_ingest(10_000, 100, 50)
    for label, multiplier in [("10K", 1), ("100K", 10), ("1M", 100),
                               ("10M", 1_000), ("100M", 10_000)]:
        proj_before = t_b_10k * multiplier
        proj_after = t_a_10k * multiplier
        proj_saved = proj_before - proj_after
        print(f"  {label:>6} docs: before={format_time(proj_before):>10}  "
              f"after={format_time(proj_after):>10}  "
              f"saved={format_time(proj_saved):>10}  ({spd:.0f}x)")

    print()
    print("=" * 90)
    print(f"Conclusion: at k=100, N=50 (typical enrichment pipeline),")
    print(f"  our patch saves {format_time(t_b_10k - t_a_10k)} per 10K documents ({spd:.0f}x faster).")
    print(f"  At 100M documents: projected savings of {format_time((t_b_10k - t_a_10k) * 10_000)}.")
    print("=" * 90)


if __name__ == "__main__":
    main()