perf: batch time-series HTTP ingest into one append transaction per measurement#5419
perf: batch time-series HTTP ingest into one append transaction per measurement#5419robfrank wants to merge 2 commits into
Conversation
…easurement PostTimeSeriesWriteHandler and PostPrometheusWriteHandler called TimeSeriesEngine.appendSamples once per parsed sample. Each of those opens its own begin/commit inside TimeSeriesShard.appendSamples, and on a Raft HA leader that commit is a full replicated quorum round trip - so a 100-sample request cost 100 sequential round trips, all serialized behind the per-shard append lock. Measured on a 3-node localhost cluster with ~19 ms inter-node RTT, sustained line protocol ingest ran at ~18 samples/sec. TimeSeriesThreeNodeWalGapReproIT (12,000 samples) needed roughly 11 minutes and timed out against its 3-minute writer budget; it now completes in ~19 s. TimeSeriesEngine.appendBatch already exists for this and was only reachable from tests. The line protocol handler now groups samples by measurement (so the schema lookup and instanceof narrowing happen once per measurement rather than once per sample) and issues one appendBatch per group; the Prometheus handler issues one per remote-write TimeSeries, whose samples already share a type and labels. Drop-set ordering, the written/dropped counts and the partial-write 400 response are unchanged - LinkedHashMap and LinkedHashSet preserve first-occurrence order. PostTimeSeriesWriteHandlerBatchAppendIT guards the batching contract using the writeTx database statistic, which counts commits that produced WAL and so counts shard append transactions directly, giving a deterministic assertion with no timing dependency. It measures 200 transactions for 200 samples before this change and at most 2 after.
|
Tick the box to add this pull request to the merge queue (same as
|
Up to standards ✅🟢 Issues
|
| Metric | Results |
|---|---|
| Coverage variation | ✅ -6.32% coverage variation |
| Diff coverage | ✅ 96.83% diff coverage |
Coverage variation details
Coverable lines Covered lines Coverage Common ancestor commit (c885235) 143368 105728 73.75% Head commit (95f212b) 175651 (+32283) 118436 (+12708) 67.43% (-6.32%) Coverage variation is the difference between the coverage for the head and common ancestor commits of the pull request branch:
<coverage of head commit> - <coverage of common ancestor commit>
Diff coverage details
Coverable lines Covered lines Diff coverage Pull request (#5419) 63 61 96.83% Diff coverage is the percentage of lines that are covered by tests out of the coverable lines that the pull request added or modified:
<covered lines added or modified>/<coverable lines added or modified> * 100%
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
|
Review: batch time-series HTTP ingest Solid, well-motivated change. It reuses the existing TimeSeriesEngine.appendBatch (previously test-only) instead of inventing new machinery, and the per-measurement grouping with LinkedHashMap/LinkedHashSet correctly preserves the first-occurrence drop-set ordering the client-facing 400 relies on. The PR description, thread-dump evidence, and before/after table are excellent. Both handlers call appendBatch inside an active database.begin(), so appendBatch takes its in-thread sequential per-shard path - the HA-safe one that avoids publishing shard pages out of band with the enclosing commit. Good. A few points, none blocking: 1. Prometheus handler: hoist tag/label values out of the per-sample loop (perf). In PostPrometheusWriteHandler.java (~L120-133), the labels of a remote-write TimeSeries are per-series, not per-sample, yet findLabelValue(ts.getLabels(), col.getName()) - a linear scan of the label list - is now recomputed for every sample x every TAG column, always returning the same value. Before batching this ran once (one sample); now it runs count times per tag with identical results. Since the value grid is column-major, you can resolve each TAG column value once before the sample loop and just fill the row. Given the project performance mantra this is worth doing. It does NOT apply to the line-protocol handler, where sample.getTags().get(...) genuinely varies per line - keeping that per-sample is correct. 2. Prometheus batching has no regression test asserting the contract (test coverage). The new PostTimeSeriesWriteHandlerBatchAppendIT is a nice deterministic, timing-free assertion via the writeTx statistic - but it only covers the line-protocol handler. The PostPrometheusWriteHandler change is exercised only for correctness by the existing PrometheusRemoteWriteReadIT, which does not assert the batching (transaction-count) contract. A regression there could silently revert Prometheus to per-sample commits without any test failing. Consider an analogous writeTx-delta assertion for a multi-sample remote-write series. 3. Empty remote-write series still auto-creates a type (minor, pre-existing). PostPrometheusWriteHandler calls getOrCreateType(...) before the (count == 0) continue guard, so a TimeSeries carrying labels but zero samples still creates a schema type with no data. This matches pre-change behavior (the old sample loop simply did not execute), so it is not a regression - but if you are touching this path, moving the empty check ahead of getOrCreateType would avoid the surprise. 4. Partial-write window on multi-measurement failure (pre-existing, worth documenting). TimeSeriesShard.appendSamples runs its own begin/commit on getWrappedDatabaseInstance(), so each appendBatch commits at the shard level immediately rather than deferring to the handler outer commit(). If a later measurement appendBatch throws, the handler database.rollback() cannot undo earlier measurements already-committed shard writes, and the client gets a 500 with partial data persisted. This predates the PR (per-sample appendSamples committed independently too), so it is not introduced here - but batching reshapes the granularity (per-measurement rather than per-sample); a line in the code/docs would keep the outer begin/commit from being mistaken for atomicity across measurements. Minor. Schema lookups (existsType/getType) in PostTimeSeriesWriteHandler now run before database.begin() rather than inside it. Fine for reads, just noting the move. Overall the correctness reasoning holds and the fix targets a real, well-evidenced HA hot spot. Addressing (1) and (2) would round it out. |
…us batching Review follow-ups on the time-series ingest batching change. PostPrometheusWriteHandler filled the value grid sample-major, calling findLabelValue for every TAG column of every sample. Those labels are per-series, not per-sample, so every call after the first returned the same value from a linear scan that re-sanitizes each label name it visits. The grid is column-major, so filling it column by column lets a TAG column resolve once and broadcast down its column. TimeSeriesIngestBatchAppendIT (renamed from PostTimeSeriesWriteHandlerBatchAppendIT) now covers the Prometheus remote_write handler too, so a regression there cannot silently restore per-sample commits. The bound is read back from the type rather than hardcoded: an auto-created remote-write type gets ASYNC_WORKER_THREADS shards, and appendBatch opens at most one transaction per shard that received samples, so a fixed bound would vary with the CPU count of the machine running the test. Verified the guard discriminates: restoring the per-sample shape measures 200 transactions against a bound of 12. Both handlers now record that the outer begin/commit is not atomicity across batches - TimeSeriesShard.appendSamples commits its own shard writes on getWrappedDatabaseInstance(), so a later failure cannot roll back batches already written. That predates this change, but batching moves the granularity from per-sample to per-measurement and the outer transaction reads as if it covered the whole request.
|
Addressed in 95f212b. Notes per item. 1. Prometheus tag hoisting — done, but the premise needs a correction. The optimization is right and is now implemented. The stated reason is not: this is not extra work introduced by batching. The pre-change code also looped What is true, and what makes it worth fixing, is that the value is invariant across samples and the column-major grid makes it trivial to exploit — plus Filling the grid column by column instead of sample by sample lets each TAG column resolve once and broadcast down its column via 2. Prometheus batching regression test — done, and it needed the shard count.
Writing it surfaced something worth flagging: my first attempt reused the line-protocol tests' hardcoded Verified the guard actually discriminates rather than merely passing: with the per-sample shape temporarily restored, both new cases fail at 200 transactions against a bound of 12. 3. Empty remote-write series auto-creating a type — not changing this. Correctly identified as pre-existing and not a regression; 4. Partial-write window — documented in both handlers. The claim checks out, and there is direct evidence for it in this PR: the Minor (schema lookups before Verification after these changes: server TS/Prometheus/Grafana/PromQL ITs 58/58; |
Review: batch time-series HTTP ingestReviewed the diff against both handlers, Strengths
Prometheus handler micro-optimisationResolving each TAG once via Minor / non-blocking notes
Nothing blocking from my side. Solid performance fix with a well-targeted regression test. Automated review by Claude Opus 4.8. Reviewed statically; I did not run the build/tests myself. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #5419 +/- ##
============================================
+ Coverage 64.82% 65.28% +0.45%
- Complexity 1057 1058 +1
============================================
Files 1726 1726
Lines 143368 143652 +284
Branches 30697 30771 +74
============================================
+ Hits 92934 93777 +843
+ Misses 37683 37110 -573
- Partials 12751 12765 +14 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
1 participant
What does this PR do?
Batches time-series HTTP ingest so each measurement's samples are appended in one shard transaction instead of one per sample.
PostTimeSeriesWriteHandler(InfluxDB line protocol) groups parsed samples by measurement, then issues oneTimeSeriesEngine.appendBatchper group. The schema lookup andinstanceofnarrowing now happen once per measurement rather than once per sample.PostPrometheusWriteHandlerissues oneappendBatchper remote-writeTimeSeries— those samples already share a type and label set.TimeSeriesEngine.appendBatchalready existed for exactly this and was only reachable from tests.Behaviour is otherwise unchanged: drop-set ordering, the
written/droppedcounts and the partial-write400response are preserved (LinkedHashMap/LinkedHashSetkeep first-occurrence order).Motivation
Both handlers called
TimeSeriesEngine.appendSamplesonce per parsed sample. Each of those opens its ownbegin/commitinsideTimeSeriesShard.appendSamples, and on a Raft HA leader that commit is a full replicated quorum round trip — so a 100-sample request cost 100 sequential quorum round trips, all serialized behind the per-shardappendLock.TimeSeriesThreeNodeWalGapReproITwas failing on this. Evidence gathered while reproducing:appendLock(TimeSeriesShard.java:196); the fourth insideRaftGroupCommitter.submitAndWait, reached fromappendSamples→RaftReplicatedDatabase.commit.compactAll()loop compounded it — every append commit waits out the active compaction recording session viawaitForActiveRecordingSession— but the per-sample round trip alone already exceeded the budget.TimeSeriesThreeNodeWalGapReproITThe
NullPointerException: ...getTransactionBroker() is nullseen in that test's log is downstream, not the cause: the assertion fails,@AfterEachstops the servers, and the still-in-flight HTTP writes then hit a torn-downRaftHAServer.Related issues
None — found while investigating the
TimeSeriesThreeNodeWalGapReproITfailure.Additional Notes
TimeSeriesIngestBatchAppendITguards the batching contract for both ingest handlers using thewriteTxdatabase statistic. It counts commits that produced WAL, so it counts shard append transactions directly — a deterministic assertion with no timing dependency. Verified to discriminate rather than merely pass: with the per-sample shape temporarily restored it measures 200 transactions in every case.The line-protocol cases pin
SHARDS 1in DDL and bind to≤ 2. The remote_write cases cannot: an auto-created type getsASYNC_WORKER_THREADSshards andappendBatchopens at most one transaction per shard that received samples, so a hardcoded bound would vary with the CPU count of the machine running the test. Those assertions read the shard count back off the type instead.Left alone deliberately, flagging for a follow-up:
RaftHAServer.getTransactionBroker()returningnullafterstop()surfaces as a raw NPE to in-flight writers rather than a clean "server is shutting down" error. Harmless here now, but it will keep producing confusing stack traces on any shutdown-under-load.Verification:
TimeSeriesIngestBatchAppendIT(new)enginecom.arcadedb.engine.timeseries.*TestIssue4458TsWalVersionGapIT,RaftTimeSeriesReplication3NodesIT)TimeSeriesThreeNodeWalGapReproITChecklist
mvn clean packagecommand — not run; verified with targetedmvn -pl server/-pl engine/-pl ha-raftbuilds and the test suites listed above