[v3] Core 3.10.1: query results for an overwritten point depend on query plan shape — non-tag-filtered plans serve the old field values indefinitely (survives restart; can tear a single atomic write)

## Summary

After an in-place overwrite (same measurement/tags/timestamp, new field values), two
SQL queries against the same table return **different generations of the same row**,
and which one you get depends on the *shape* of the query:

- A query with an **equality predicate on a tag** (`component_id = 'X'`) always
returns the new values (fresh within ~1 s; never observed stale in 85/85 checks
across six probe campaigns).
- Queries **without** that predicate — a bare table/time-range scan, an `IN` list,
or a `count(*)` filtered only on a *field* — can permanently return the **old**
values for some or all of the overwritten series. "Permanently" is literal: no
affected plan was ever observed to recover (watched ≥ 20 min), and the stale
state **survives a full server restart**.

The mis-arbitration happens per *(query plan × per-series chunk of the write)*, so a
single atomic write (one `write_lp` call, 12 rows across 2 series) can be **torn**:
the wide scan serves series A's new rows and series B's old rows indefinitely, while
per-series queries serve both correctly. Incidence is roughly 40–50 % per overwritten
series in our environment; 27 of 28 two-series overwrite trials left at least one
query plan permanently stale.

## Environment

- InfluxDB 3 Core **3.10.1** (revision `ff6872a3d8`), official Docker image `influxdb:3-core`
- Default server config: `influxdb3 serve --node-id=node0 --object-store=file --data-dir=/var/lib/influxdb3`
- Linux x86_64 host; single node; one database
- Reproduced identically: via the `influxdb3-python` (Flight) client **and** via the
raw HTTP API (`/api/v3/write_lp` + `/api/v3/query_sql`) with a stdlib-only Python
script (attached) — so no client library is involved
- Reproduced identically with writes using `no_sync=true` and default sync writes
- Also reproduced under `--disable-parquet-mem-cache` and under `--wal-flush-interval=5s` (see "Additional evidence", below)

## Reproduction

[`repro_standalone.py`](https://gist.github.com/PascalRaux-EP/9d2490c1aaed286e57081b19717880bc) (Python 3, stdlib only, ~200 lines; also inlined at the bottom of this issue). Steps it performs:

1. **Seed**: one table (`repro_pac`), tags `plant_id` (1 value) and `component_id`
(20 values), fields `value` (float) and `run_id` (string); 144 points per series
at 10-min cadence over one day = 2 880 rows. Then 3 value-identical full rewrites
(so keys carry a few duplicate generations, as any idempotent ingest produces).
2. **Overwrite**: for two series, rewrite one hour (6 points each, `value` + 1000,
fresh `run_id`) — 12 rows in **one** `write_lp` call.
3. **Poll** three query shapes until each reflects the new values (90 s cap):
- A. `... WHERE plant_id='P' AND component_id='X' AND time >= .. AND time < ..`
- B. `... WHERE plant_id='P' AND time >= .. AND time < ..` (no series predicate)
- C. `SELECT count(*) WHERE plant_id='P' AND run_id=''` (field predicate)

Typical output (this exact run captured 2026-07-17; 3 of 4 trials stuck, all torn):

```
server: InfluxDB 3 Core 3.10.1 (ff6872a3d8)
trial 2: overwriting RINV02+RINV12 @ 03:00 (12 rows, ONE write_lp call, run a187dcc1)
t=0.0s single-component(RINV02) reflects
t=0.0s single-component(RINV12) reflects
t=0.0s plant-wide reflects for RINV02
RESULT: {"single_component_s": {"RINV02": 0.0, "RINV12": 0.0},
"plant_wide_s": {"RINV02": 0.0, "RINV12": null},
"count_run_id_s": null, "count_final": 6,
"stuck_plant_wide": ["RINV12"], "torn": true}
```

The stale rows are then directly demonstrable — same server, same instant, the only
difference is the tag predicate:

```sql
-- fresh (new generation, the correction's run_id):
SELECT time, component_id, value, run_id FROM repro_pac
WHERE plant_id='REPRO01' AND component_id='RINV14'
AND time >= '2023-03-01T07:00:00Z' AND time < '2023-03-01T08:00:00Z';
--> value=2001.914..2001.919, run_id=9d4da7ec-... (6 rows, correct)

-- stale (old generation, a run superseded 20+ minutes earlier):
SELECT time, component_id, value, run_id FROM repro_pac
WHERE plant_id='REPRO01'
AND time >= '2023-03-01T07:00:00Z' AND time < '2023-03-01T08:00:00Z';
--> RINV14 rows: value=1001.914..1001.919, run_id=2bc39942-... (old values)
```

## Where the divergence happens: inside the queryable buffer, at dedup

At capture time `SELECT count(*) FROM system.parquet_files WHERE table_name='repro_pac'`
returned **0** — every row involved was still in the in-memory queryable buffer. Both
plans scan the **same** source (`RecordBatchesExec: chunks=144`, no parquet, no object
store, no cache); `EXPLAIN` (full plans in [`evidence_explain_and_values.txt`](https://gist.github.com/PascalRaux-EP/9d2490c1aaed286e57081b19717880bc))
shows they differ only in the merge/dedup topology above the scan:

- plant-wide (stale): `SortExec [component_id, time, __chunk_order]` →
`SortPreservingMergeExec` → `DeduplicateExec`
- single-component (fresh): `SortExec [time, __chunk_order]` →
`ReorderPartitionsExec` → `ProgressiveEvalExec` → `DeduplicateExec`

Same chunks in, different duplicate-resolution out: the two topologies disagree on
which generation of the same key wins deduplication. (Whether the same arbitration
error also exists post-persist we can't isolate as cleanly, but behaviorally the
stale state carries over hours of uptime and a server restart — see below.)

## Additional evidence (49 instrumented overwrite trials across 6 campaigns, 2 stacks)

1. **Never recovers.** No affected plan ever flipped to the new values: watched
continuously ≥ 20 min, re-checked ~3.5 h later, and after restart. The only
action that clears it is *rewriting the affected keys again* — which re-rolls
the dice (~40 % chance of a new stale plan).
2. **Survives full server restart** — 0 of 12 stuck (plan × series) states cleared
after a container restart ~3.5 h post-write; per-series reads of the same keys
remained correct throughout. So the mis-arbitration is reproducible against
persisted/WAL state, not a transient in-memory artifact.
3. **Tears atomic writes.** In 24 of 28 two-series trials, the wide scan
served exactly one series' new rows while the other series' rows *from the same
write call and same run_id* stayed old; `count(*) WHERE run_id=''` sat
frozen at 6 of 12 indefinitely.
4. **Not WAL/ingest chunking.** Under `--wal-flush-interval=5s`, both series' rows
provably landed in the same WAL flush (the two per-series polls flipped 12–19 ms
apart at ~28 ms poll resolution, after the expected ~4 s flush delay) — and the
wide plan still served only one of them.
5. **Not the parquet memory cache.** `--disable-parquet-mem-cache` changed neither
incidence (14/16 trials with ≥ 1 stuck plan) nor read latency (median wide-scan
~15–50 ms in both configs) — consistent with the `EXPLAIN` finding that the
affected data is in `RecordBatchesExec`, not parquet.
6. **Not monotonic in predicate selectivity.** An `IN` list of 10 of the 20 series
was observed stuck *alone* while the bare plant scan and the count were fresh —
and the inverse. `IN ()` behaves like no predicate. A `component_id IS
NOT NULL` predicate doesn't help. Only single-value equality was never stale.
7. **Field predicates don't help.** `count(*) ... AND run_id=''` (maximally
selective, but on a field) goes stale along with the wide scans.
8. **Correlated flips.** When plans *do* flip to fresh, disjoint plans flip together
within ≤ 70 ms — so whatever state arbitrates the winner is shared per chunk.
9. **First writes are fine.** New keys become visible to all shapes in ~1 s,
atomically and monotonically. Only in-place overwrites exhibit this; overwrites
with *identical* values are unaffected (or rather: a stale winner is
indistinguishable). Value-*changing* overwrites (data corrections, recomputation
pipelines) are the exposed case.

## Expected behavior

The result set for a given key should not depend on which otherwise-equivalent
plan produced it; a committed overwrite should be visible to all query shapes
(eventual consistency would be acceptable — the issue is that affected plans
**never** converge).

## Impact

Any workflow that overwrites points and reads them back with a multi-series query —
correction ingestion, idempotent re-processing, downsampling pipelines — can compute
on a mix of old and new data, silently and permanently. We hit it as a KPI pipeline
recomputing from corrected raw data: the recompute's wide input scan served the
superseded generation.

Workarounds we've validated: fan out reads per series (tag-equality reads were fresh
in 85/85 checks) — or rewrite the affected keys until no plan is stuck, which is
detectable only by comparing shaped reads.

## Related issues (searched 2026-07-17; no existing report of this defect found)

- #27337 (`max_dedup_split`) and #26993 document the `DedupSortOptimizer` /
`group_potential_duplicates` machinery from the *performance* side — plans split
chunks into dedup groups and skip dedup for groups deemed non-overlapping. This
issue looks like the correctness face of the same plan-shape-dependent dedup
planning: which generation of an overwritten key survives depends on the plan
topology built over the identical chunk set.

## Supporting data

All in one gist: https://gist.github.com/PascalRaux-EP/9d2490c1aaed286e57081b19717880bc

- `repro_standalone.py` — stdlib-only HTTP repro (inlined below)
- `evidence_explain_and_values.txt` — captured side-by-side values + full `EXPLAIN`
/ `EXPLAIN ANALYZE` plans + `system.parquet_files` count
- `probe_results_v3.jsonl`, `probe_results_c2.jsonl`, `probe_results_c5a.jsonl` —
full timing records of the instrumented campaigns (torn writes, flip skews,
restart check)

repro_standalone.py

```python
"""Standalone repro: value-changing overwrite invisible to non-tag-filtered
queries in InfluxDB 3 Core (plan-dependent stale reads, torn multi-series
writes).

Zero dependencies beyond the Python standard library — talks to the server
over the plain HTTP API (/api/v3/write_lp + /api/v3/query_sql), so no client
library or Flight caching is involved.

Usage:
INFLUX_TOKEN=... python3 repro_standalone.py
# optional: INFLUX_HOST (default http://127.0.0.1:8181), INFLUX_DB (default inverter)
# from this repo's root (token lives in .env; beware a stale INFLUX_TOKEN
# already exported in your shell — it wins over nothing, and an old stack's
# token 401s):
# INFLUX_TOKEN=$(grep -m1 '^INFLUX_TOKEN' .env | cut -d= -f2-) \
# python3 .superpowers/probes/2026-07-16-dedup-shape-matrix/repro_standalone.py

What it does:
1. Seeds 20 series (tag component_id=RINV01..RINV20) x 144 points at 10-min
cadence over one day (2880 rows), plus 3 value-identical rewrites so keys
carry a few duplicate generations (as any idempotent ingest produces).
2. Per trial: overwrites ONE HOUR (6 points) for TWO components in a single
write_lp call, changing `value` (+1000) and the `run_id` string field.
3. Polls three query shapes until each reflects the new values (90 s cap):
A. single-component: ... WHERE plant_id=.. AND component_id='X' (full day)
B. plant-wide: ... WHERE plant_id=.. (full day)
C. count(*): ... WHERE plant_id=.. AND run_id='' (field predicate)
4. Reports per-shape reflect latencies + classification (fresh / stuck / torn).

Expected (observed on 3-core 3.10.1): shape A reflects <1 s in every trial;
shapes B and C frequently serve the OLD values for one or both components
indefinitely (never recovering, surviving server restart) — including TORN
results where B/C show exactly half of a single atomic write_lp call.
"""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import UTC, datetime, timedelta
from uuid import uuid4

HOST = os.environ.get("INFLUX_HOST", "http://127.0.0.1:8181").rstrip("/")
TOKEN = os.environ.get("INFLUX_TOKEN") or exit("INFLUX_TOKEN is not set")
DB = os.environ.get("INFLUX_DB", "inverter")

TABLE = "repro_pac"
PLANT = "REPRO01"
COMPONENTS = [f"RINV{i:02d}" for i in range(1, 21)]
WINDOW_START = datetime(2023, 3, 1, tzinfo=UTC)
POINTS_PER_COMPONENT = 144 # 24 h at 10-min cadence
TOTAL_ROWS = POINTS_PER_COMPONENT * len(COMPONENTS)
TRIALS = 4
CAP_S = 90.0

def _http(url: str, data: bytes, headers: dict[str, str]) -> bytes:
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.read()

def write_lp(lines: list[str]) -> None:
url = f"{HOST}/api/v3/write_lp?" + urllib.parse.urlencode(
{"db": DB, "precision": "second"}
)
_http(
url,
"\n".join(lines).encode(),
{"Authorization": f"Bearer {TOKEN}", "Content-Type": "text/plain"},
)

def query(sql: str) -> list[dict]:
body = json.dumps({"db": DB, "q": sql, "format": "json"}).encode()
out = _http(
f"{HOST}/api/v3/query_sql",
body,
{"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
)
return json.loads(out)

def lp_line(comp: str, ts: datetime, value: float, run_id: str) -> str:
return (
f"{TABLE},plant_id={PLANT},component_id={comp} "
f'value={value},run_id="{run_id}" {int(ts.timestamp())}'
)

def baseline_lines(run_id: str) -> list[str]:
lines = []
for ci, comp in enumerate(COMPONENTS):
for pi in range(POINTS_PER_COMPONENT):
ts = WINDOW_START + timedelta(minutes=10 * pi)
lines.append(lp_line(comp, ts, 1000.0 + (ci * POINTS_PER_COMPONENT + pi) * 0.001, run_id))
return lines

def reflects(rows: list[dict], comp: str, targets: dict[str, float]) -> bool:
got = {
r["time"]: r["value"]
for r in rows
if r.get("component_id") == comp and r["time"] in targets
}
return len(got) == len(targets) and all(
abs(got[t] - v) < 1e-6 for t, v in targets.items()
)

def seed() -> None:
run_id = str(uuid4())
write_lp(baseline_lines(run_id))
t0 = time.monotonic()
while time.monotonic() - t0 < 60:
n = query(f"SELECT count(*) AS n FROM {TABLE} WHERE plant_id = '{PLANT}'")[0]["n"]
if n == TOTAL_ROWS:
print(f"seed: {TOTAL_ROWS} rows visible after {time.monotonic() - t0:.1f}s")
break
time.sleep(1)
else:
raise SystemExit("seed never became visible")
for i in range(3): # value-identical generations
write_lp(baseline_lines(str(uuid4())))
print(f"regen rewrite {i + 1}/3 done")
time.sleep(5)

def run_trial(idx: int) -> dict:
comp_a, comp_b = COMPONENTS[idx], COMPONENTS[10 + idx]
hour = WINDOW_START + timedelta(hours=1 + 2 * idx)
new_run = str(uuid4())

targets: dict[str, dict[str, float]] = {}
lines: list[str] = []
for comp in (comp_a, comp_b):
rows = query(
f"SELECT time, value FROM {TABLE} WHERE plant_id = '{PLANT}' "
f"AND component_id = '{comp}' AND time >= '{hour.isoformat()}' "
f"AND time < '{(hour + timedelta(hours=1)).isoformat()}'"
)
assert len(rows) == 6, f"expected 6 rows for {comp}, got {len(rows)}"
tmap = {}
for r in rows:
ts = datetime.fromisoformat(r["time"]).replace(tzinfo=UTC)
new_val = r["value"] + 1000.0
tmap[r["time"]] = new_val
lines.append(lp_line(comp, ts, new_val, new_run))
targets[comp] = tmap

print(f"\ntrial {idx + 1}: overwriting {comp_a}+{comp_b} @ {hour:%H:%M} "
f"(12 rows, ONE write_lp call, run {new_run[:8]})")
write_lp(lines)
t0 = time.monotonic()

day_end = WINDOW_START + timedelta(days=1)
single: dict[str, float | None] = {comp_a: None, comp_b: None}
plant_wide: dict[str, float | None] = {comp_a: None, comp_b: None}
count_s: float | None = None
count_last = 0

while time.monotonic() - t0 < CAP_S:
now = round(time.monotonic() - t0, 2)
for comp in (comp_a, comp_b):
if single[comp] is None and reflects(
query(
f"SELECT time, component_id, value FROM {TABLE} "
f"WHERE plant_id = '{PLANT}' AND component_id = '{comp}' "
f"AND time >= '{WINDOW_START.isoformat()}' AND time < '{day_end.isoformat()}'"
),
comp,
targets[comp],
):
single[comp] = now
print(f" t={now}s single-component({comp}) reflects")
if any(v is None for v in plant_wide.values()):
rows = query(
f"SELECT time, component_id, value FROM {TABLE} "
f"WHERE plant_id = '{PLANT}' "
f"AND time >= '{WINDOW_START.isoformat()}' AND time < '{day_end.isoformat()}'"
)
for comp in (comp_a, comp_b):
if plant_wide[comp] is None and reflects(rows, comp, targets[comp]):
plant_wide[comp] = now
print(f" t={now}s plant-wide reflects for {comp}")
if count_s is None:
count_last = query(
f"SELECT count(*) AS n FROM {TABLE} WHERE plant_id = '{PLANT}' "
f"AND run_id = '{new_run}'"
)[0]["n"]
if count_last == 12:
count_s = now
print(f" t={now}s count(run_id) reflects (12 rows)")
if all(single.values()) and all(plant_wide.values()) and count_s is not None:
break
time.sleep(0.2)

stuck = [c for c, v in plant_wide.items() if v is None]
result = {
"trial": idx + 1,
"components": [comp_a, comp_b],
"single_component_s": single,
"plant_wide_s": plant_wide,
"count_run_id_s": count_s,
"count_final": count_last if count_s is None else 12,
"stuck_plant_wide": stuck,
"torn": len(stuck) == 1 or (count_s is None and 0 < count_last < 12),
}
print(f" RESULT: {json.dumps(result)}")
return result

def main() -> None:
try:
ping = json.loads(
urllib.request.urlopen(
urllib.request.Request(
f"{HOST}/ping", headers={"Authorization": f"Bearer {TOKEN}"}
),
timeout=10,
).read()
)
except urllib.error.HTTPError as e:
if e.code == 401:
raise SystemExit(f"{HOST} rejected the token (401).") from None
raise
print(f"server: {ping.get('product_name')} {ping.get('version')} ({ping.get('revision')})")
seed()
results = [run_trial(i) for i in range(TRIALS)]
n_stuck = sum(1 for r in results if r["stuck_plant_wide"] or r["count_run_id_s"] is None)
n_torn = sum(1 for r in results if r["torn"])
print(f"\nSUMMARY: {n_stuck}/{len(results)} trials with a stale plan at {CAP_S:.0f}s cap; "
f"{n_torn} torn (wide plan serves exactly half of one atomic write). "
f"single-component reads fresh in all trials: "
f"{all(v is not None for r in results for v in r['single_component_s'].values())}")

if __name__ == "__main__":
main()
```

Start by running repro_standalone.py against InfluxDB 3 Core 3.10.1 and compare the three query shapes with EXPLAIN and EXPLAIN ANALYZE. Read the DedupSortOptimizer and group_potential_duplicates machinery referenced in issues #27337 and #26993, then follow the RecordBatchesExec and DeduplicateExec plan paths. Done means every query shape returns the latest values consistently, including both series from one atomic write.

Written by the indexing model from the issue text.