Fix Coconut disk streaming with bounded LSM updates

CONTINUITY.md

@@ -211,3 +211,20 @@ Update `CONTINUITY.md` when:
 - Python binding structure changes materially
 - Tests, demos, or benchmarks gain a new required pattern
 - A major known rough edge is fixed or a new one appears
+
+
+## Coconut materialized disk streaming
+
+Coconut uses a bounded raw memtable (default 1M records), external-sort bulk build,
+immutable sorted disk runs and background binary compaction. Resident SAX remains
+O(N); raw vectors persist to disk. Include waitForMerges() when timing completed
+insertion. Temporary indexes have no reopen/WAL/recovery or concurrent caller
+write/search support. Gaussian SAX only; inputs are never normalized internally.
+
+Top-k uses per-run sortable-key seeds, safe SAX pruning and independent exact
+heaps, sharing bounded candidate reads across configurable groups of up to 64
+queries (default 32). Recent-window top-k filters stable insertion IDs. Python
+bindings expose layout/storage/merge/query counters; BUILD_COCONUT remains optional.
+Tests cover strict physical neighbors through flushes/merges, batch/scalar parity,
+lookup-bound parity, ranges, corruption and API validation. See
+`docs/coconut-disk-streaming.md` and `docs/coconut-paper-implementation.md`.

README.md

@@ -56,7 +56,7 @@ The following table summarizes the key features of each algorithm:
 | **[Hercules](https://helios2.mi.parisdescartes.fr/~themisp/publications/pvldb22-hercules.pdf)** | In-memory hierarchical similarity search using EAPCA and SAX-based pruning |
 | **[DumpyOS](https://helios2.mi.parisdescartes.fr/~themisp/publications/vldbj24-dumpyos.pdf)** | In-memory scalable data series similarity search using an adaptive multi-ary iSAX index |
 | **[FreSH](http://publications.ics.forth.gr/tech-reports/2023/2023.TR489_FreSh_A_LockFree_Data_Series_Index.pdf)** | In-memory lock-free parallel similarity search using an iSAX index (SRDS 2023) |
-| **[COCONUT](http://www.vldb.org/pvldb/vol11/p677-kondylakis.pdf)** | Sortable-SAX index built bottom-up; supports both static datasets and **streaming** (incremental) inserts (PVLDB 2018) |
+| **[COCONUT](https://arxiv.org/abs/2006.11474)** | Paper-based materialized Coconut-LSM-Full: bottom-up disk index, streaming inserts, resident-summary SIMS and recent-window top-k |
 | **[FAISS](https://github.com/facebookresearch/faiss) flat** | Exact flat baseline backed by `faiss::IndexFlatL2`; static build plus streaming inserts with no retraining. Optional: build with `-DBUILD_FAISS=ON` |
 
 ### Incremental streaming inserts
@@ -91,18 +91,45 @@ data and the index stays exact after every insert. It supports `L2_SQUARED` only
 Streaming updates are not concurrent with queries. Inserts can reallocate the owned database,
 so callers should not retain a pointer returned by `getDatabase()` across them.
 
-`Coconut` uses the upstream ordered approximate search before exact top-k refinement:
-it locates the query's sortable-SAX key in a resident sorted directory and seeds a
-heap with nearby distinct records. Exact search then checks the remaining SAX bounds.
-Top-k and range queries read candidate leaves as complete blocks into private worker
-buffers, avoiding per-vector file seeks. The ordered lookup uses the resident directory
-instead of importing Berkeley DB's disk B-tree engine; raw vectors remain in leaf files.
-Search supports squared L2 only. Inserts append to an unmerged in-memory buffer;
-Coconut-LSM flush/merge handling and an out-of-core key directory remain unimplemented.
+`Coconut` now uses bounded disk streaming: a RAM memtable of at most
+`getBufferCapacity()` records (default 1,000,000) flushes to sorted disk runs.
+Binary LSM compaction merges adjacent temporal runs in the background, including
+the raw vectors. Include `waitForMerges()` after `insertBatch()` when measuring
+completed insertion time (sorting, writes, `fsync` and compaction). Initial build
+uses external sorting; SIMS keeps packed SAX summaries resident and parallelizes
+bounds over resident summaries before ordered raw refinement. Up to 32 independent
+queries share bounded candidate reads; each retains its own exact heap and bound.
+Sparse reads coalesce short gaps, while dense reads share blocks of at most 8 MiB.
+Ordered approximate seeding precedes exact SAX-bound refinement across every run
+and any partially full memtable. Candidate raw vectors are read in leaf blocks.
+Top-k and inclusive range search support squared L2 with Gaussian SAX breakpoints;
+vectors are never normalized internally. Configure layout before building:
+
+```python
+index = daisy.Coconut(daisy.DistanceType.L2_SQUARED)
+index.setBufferCapacity(1_000_000)  # records, not the external insertion batch size
+index.setIndexDirectory("/path/to/storage")  # unique temporary child directory
+index.buildIndex(initial_vectors)
+index.insertBatch(new_vectors)  # full tables persist and schedule compaction
+index.waitForMerges()           # include this barrier in completed-insert timing
+index.setQueryBatchSize(32)     # share candidate reads across independent queries
+ids, distances = index.searchIndexRecent(queries, 1, 50_000_000)  # latest records
+print(index.getDiskRecords(), index.getBufferedRecords(), index.getMergeCount())
+```
+
+This implements the paper's materialized Coconut-LSM-Full algorithms, with explicit
+[implementation choices and tests](docs/coconut-paper-implementation.md); it is
+not the authors' binary or an upstream performance reproduction. POSIX disk storage is required. The OS may
+cache disk pages, so measure cgroup memory as well as the bounded explicit buffers.
+Indexes are temporary and removed on destruction; crash recovery, reopening,
+deletion and concurrent caller writes/reads are unsupported. Recent-window queries
+use insertion counts, not arbitrary user timestamps. Resident SAX uses O(N) RAM; bound/read scratch is bounded independently of N; the raw vectors and bounded insertion buffer have separate budgets.
+See [the implementation notes](docs/coconut-disk-streaming.md). Earlier results
+from the unmerged RAM insertion buffer do not evaluate disk-based streaming.
 
 ### Range (distance-r) queries
 
-Every algorithm except Coconut's streaming-only paths answers range queries through
+Algorithms that implement range search answer range queries through
 `SearchConfig`. Instead of a fixed `k`, each query returns however many series fall within the
 radius, so the results come back as one vector per query:
 

benchmark/CMakeLists.txt

@@ -147,6 +147,8 @@ endif()
 
 # ////// COCONUT //////
 if(BUILD_COCONUT)
+    add_executable(bm_Coconut_Streaming bm_Coconut_Streaming.cpp)
+    target_link_libraries(bm_Coconut_Streaming PRIVATE benchmark::benchmark dino_lib commons_lib)
     add_executable(bm_Coconut_L2Square
         bm_Coconut_L2Square.cpp
         bm_utils.cpp

benchmark/bm_Coconut_Streaming.cpp

@@ -0,0 +1,32 @@
+#include <benchmark/benchmark.h>
+#include "../commons/dataloaders.hpp"
+#include "../lib/algos/Coconut.hpp"
+#include <memory>
+#include <vector>
+
+// Match the existing streaming benchmark pattern. Unlike an enqueue-only
+// measurement, wall time includes every flush and background compaction.
+static void BM_Coconut_Streaming(benchmark::State &state)
+{
+    constexpr daisy::idx_t dim = 96, initial = 4096, queries = 4;
+    const auto batch = static_cast<daisy::idx_t>(state.range(0));
+    std::unique_ptr<float[]> data(loadRandomData(initial + batch, dim, 100, true));
+    std::unique_ptr<float[]> query(loadRandomData(queries, dim, 50, true));
+    daisy::Coconut index(daisy::DistanceType::L2_SQUARED);
+    index.setBufferCapacity(512); index.setNumThreads(2);
+    index.setQueryBatchSize(static_cast<daisy::idx_t>(state.range(1)));
+    index.buildIndex(data.get(), initial, dim);
+    std::vector<daisy::idx_t> ids(queries); std::vector<float> distances(queries);
+    for (auto _ : state)
+    {
+        index.insertBatch(data.get() + initial * dim, batch);
+        index.waitForMerges();
+        index.searchIndex(query.get(), queries, 1, ids.data(), distances.data());
+        benchmark::ClobberMemory();
+    }
+    state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) * batch);
+    state.counters["n_database_final"] = static_cast<double>(index.getNDatabase());
+    state.counters["resident_summary_bytes"] = static_cast<double>(index.getResidentSummaryBytes());
+}
+BENCHMARK(BM_Coconut_Streaming)->Args({1024, 1})->Args({1024, 32})->Args({8192, 32})->Iterations(5)->UseRealTime()->Unit(benchmark::kMillisecond);
+BENCHMARK_MAIN();

demos/demo_Coconut_Streaming.cpp

@@ -19,6 +19,9 @@ int main()
     float *query = loadRandomData(n_query, dim, 50, true);
 
     daisy::Coconut coconut(daisy::DistanceType::L2_SQUARED);
+    coconut.setBufferCapacity(10000);  // auto flush and background binary disk merges
+    coconut.setNumThreads(2);
+    coconut.setQueryBatchSize(32);       // shared reads; exact per-query SAX bounds/heaps
 
     // 1. Build on the initial batch.
     coconut.buildIndex(stream, initial, dim);
@@ -42,9 +45,14 @@ int main()
     for (int s = 0; s < steps; s++)
     {
         coconut.insertBatch(stream + (size_t)seen * dim, batch);
+        coconut.waitForMerges(); // include pending compaction in completed-insert timing
         seen += batch;
         printf("Streamed a batch; index now holds %llu series.\n", seen);
+        printf("Disk records: %llu; buffered: %llu; merges: %llu\n",
+               coconut.getDiskRecords(), coconut.getBufferedRecords(), coconut.getMergeCount());
         run_query("after insert");
+        coconut.searchIndexRecent(query, n_query, 1, batch, I, D);
+        printf("Most recent batch, query 0 top-1: %llu (%.1f)\n", I[0], D[0]);
     }
 
     delete[] stream;

demos/demo_Coconut_Streaming.py

@@ -23,7 +23,9 @@ def main():
     query = np.random.randn(n_query, dim).astype(np.float32)
 
     index = Coconut(DistanceType.L2_SQUARED)
-    index.setNumThreads(1)
+    index.setNumThreads(2)
+    index.setQueryBatchSize(32)  # shared bounded candidate reads; independent exact heaps
+    index.setBufferCapacity(10_000)  # bounded RAM table; full buffers persist and schedule merges
     index.buildIndex(stream[:initial])         # build on the initial batch
 
     def show(when):
@@ -35,9 +37,14 @@ def show(when):
     seen = initial
     for s in range(steps):
         index.insertBatch(stream[seen:seen + batch])   # stream the next batch
+        index.waitForMerges()  # include this barrier when timing completed insertions
         seen += batch
-        print(f"streamed a batch; index now holds {seen} series")
+        print(f"streamed {seen} series; disk={index.getDiskRecords()}, "
+              f"RAM buffer={index.getBufferedRecords()}, merges={index.getMergeCount()}")
         show("after insert")
+        recent_ids, _ = index.searchIndexRecent(query, 1, batch)
+        print("top-1 in the most recent batch:", recent_ids[:, 0])
+        print("SIMS phases and reads:", index.getLastSearchStats())
 
 if __name__ == "__main__":
     main()

docs/coconut-disk-streaming.md

@@ -0,0 +1,60 @@
+# Coconut disk streaming
+
+`Coconut` follows the materialized sorted-run design of
+[Coconut-LSM](https://arxiv.org/abs/2006.11474), adapted to DaiSy's exact top-k and
+inclusive range APIs. It is a new implementation of that storage/update policy,
+not an import of the 2018 Berkeley DB executable or a claim of performance parity.
+
+* A configurable memtable holds at most `setBufferCapacity(records)` raw vectors
+  plus SAX summaries and sortable keys. Sorting adds one bounded permutation.
+* Full tables are sorted and persisted as immutable runs with stable insertion
+  IDs. The initial build also processes bounded chunks, including file sources.
+* Each run has a fixed-width `[sortable key][SAX word][ID]` directory and a raw
+  float32 data file in the corresponding order. Directories are reclaimable
+  read-only file mappings; no full vector database or heap key directory is kept.
+* Equal-level runs are merged with sequential bounded input/output buffers,
+  using binary size tiers in a background worker. A new run is written, flushed and synced before its
+  input runs are retired. Bulk build consolidates its sorted chunks into a run.
+* Queries seed a shared bound from each run's sortable-key neighborhood, then
+  refine every surviving SAX summary and inspect the remaining bounded memtable.
+  Top-k uses resident SAX, exact query interval lookups and bounded shared
+  candidate reads across independent queries. Returned IDs refer to the original stream.
+* `insertBatch` includes automatic flushes and schedules asynchronous merges.
+  Include `waitForMerges()` in completed-insert timing; `flush()` can persist an
+  incomplete table. `getDiskRecords`, `getBufferedRecords`, `getDiskBytes`,
+  `getRunCount`, `getFlushCount`, `getMergeCount`, and `getPeakBufferedRecords`
+  provide evidence that inserted data are actually reaching disk.
+
+There is no WAL, recoverable manifest, restart/reopen support, deletion or arbitrary-timestamp
+window API. Writes and searches must not run concurrently. The temporary directory
+is removed when its instance is destroyed. A failed batch can have accepted a
+prefix before an I/O failure; inspect the count and treat the error as a failed
+operation, not an atomic transaction. Invalid shapes/non-finite vectors are rejected.
+Changing SAX/leaf/buffer/storage layout after building is rejected. Gaussian SAX
+is supported; selecting adaptive breakpoints is explicitly rejected in C++.
+No additional dependency is introduced; `BUILD_COCONUT=OFF` still excludes it.
+
+## Memory and measurements
+
+Raw vectors persist in run files; the memtable is bounded by record count.
+Resident SAX summaries cost one byte per segment per persisted record, so they
+still require O(N) RAM. Mapped directories, OS file cache and summaries retained
+by in-flight merges also count toward memory usage. Record cgroup/RSS alongside
+explicit buffer counters; this is not a constant-memory index.
+
+Freeze PAA segments, leaf size, query batch size and memtable capacity across
+streaming stages. Include `waitForMerges()` after `insertBatch()` inside the timed
+operation, then audit live disk/buffer counters. Do not charge outstanding merge
+work to the subsequent query. Build and batch loading are separate measurements.
+Query phase/read counters describe logical requests, not physical device I/O.
+
+See [the paper mapping and optimization details](coconut-paper-implementation.md).
+
+## Validation
+
+Validation includes strict float64 physical-neighbor checks after repeated flushes
+and merges, stable IDs, inclusive ranges, partial buffers, file-backed builds,
+strided Python inputs, corruption detection and optional-build exclusion. The
+legacy Astronomy range fixture has non-finite rows 10321 and 40380; those nine
+parameterized cases explicitly skip the invalid fixture rather than filtering it
+or claiming exact range accuracy on NaNs. Finite-input rejection has its own test.