53 increase csv ingest speed#54
Open
wmehling wants to merge 5 commits into
Open
Conversation
- Implement KvinIngestionCsvDiagnosticBenchmark for performance testing of CSV ingestion. - Create KvinIngestionJsonDiagnosticBenchmark for benchmarking JSON ingestion. - Introduce KvinIngestionWorkload to generate test data for benchmarks. - Add unit tests for KvinIngestionCsvDiagnosticBenchmark and KvinIngestionWorkload. - Remove obsolete KvinServiceBenchmark Scala file. - Document benchmark setup and execution in markdown format.
Keep the existing row-major CsvFormatParser.parse() contract, and add a bounded 1000-row column-major iterator for CSV ingestion. The service feeds that iterator through one KVIN iterable write, preserving one request-level batch while improving locality. Listener encounter order within each group is intentionally column-major. JMH 30,000 tuples, JDK 21.0.11, 3 warmups, 5 measurements, 2 forks, 1 thread: postCsv A 76.581 -> 56.823 ms/op (392k -> 528k tuples/s, 25.8% faster); B 59.683 -> 53.465 ms/op (503k -> 561k, 10.4% faster). putCsvGrouped: A 45.397 ms/op (661k/s); B 48.841 ms/op (614k/s). CSV diagnostics A/B were also captured under .cache-main/benchmarks.
Cache the last item/property/context prefix and its striped lock inside KvinLevelDb.put(Iterable). The cache is invalidated naturally whenever tuple identity changes; key encoding, fetch behavior, batching, and listener APIs are unchanged. JMH 30,000 tuples, JDK 21.0.11, 3 warmups, 5 measurements, 2 forks, 1 thread, incremental over a9f5a0d: postCsv A 56.823 -> 49.763 ms/op (528k -> 603k tuples/s, 12.4% faster); B 53.465 -> 47.179 ms/op (561k -> 636k, 11.8% faster). LevelDB KvinLevelDbTest: 7 passed.
Parse 1,000-row column-major groups on the shared executor while one iterable LevelDB write drains a two-group queue. Keep benchmark controls for direct overlap and shared-store concurrency, and record paired A/B results and read-contention caveats.
kenwenzel
reviewed
Jul 19, 2026
kenwenzel
left a comment
kenwenzel
left a comment
Member
There was a problem hiding this comment.
The CsvIngestionPipeline could in general also be used by other formats and there be generalized. Furthermore, a lock-free communication can be faster especially for smaller batch sizes.
| } | ||
|
|
||
| IExtendedIterator<List<KvinTuple>> groups = parser.parseRowGroupBatches(rowsPerGroup); | ||
| BlockingQueue<Object> queue = new ArrayBlockingQueue<>(queueCapacity); |
Member
There was a problem hiding this comment.
This is slower than using a CircularBuffer and Thread.onSpinWait() that allows to lock-free send data between too threads.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Increased CSV Ingest from ~400k Tokens/s to ~ 800k Tokens/s
Depends on #51 / Builds on branch, so review #52 first
The optimization was split into three commits so each idea remains independently reviewable, benchmarkable, and revertible:
Column-major row groups: CSV rows are collected in bounded groups of 1,000 and emitted column by column. Creates consecutive tuples with the same item/property/context. Throughput increased from the baseline 391.7k/502.7k tuples/s to **528.0k/561.1k tuples/**s in two independent runs.
Prefix and lock reuse: The LevelDB batch writer reuses the resolved prefix and lock for consecutive tuples with the same identity. Throughput increased further to 602.9k/635.9k tuples/s.
Bounded parser/storage overlap: One task parses CSV row groups while the request thread stores them through a bounded two-group queue. Final endpoint throughput reached 851.9k tuples/s in Run A and 694.7k tuples/s in Run B.
The existing row-major parse() API and read path remain unchanged; only the optimized ingestion route uses column-major ordering within each bounded group.