Skip to content

feat(qwp): stop resending the full symbol dictionary on every message #66

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
glasstiger wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(qwp): stop resending the full symbol dictionary on every message #66

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 131 additions & 13 deletions core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import io.questdb.client.cutlass.qwp.client.sf.cursor.DefaultSenderConnectionListener;
import io.questdb.client.cutlass.qwp.client.sf.cursor.DefaultSenderErrorHandler;
import io.questdb.client.cutlass.qwp.client.sf.cursor.DefaultSenderProgressHandler;
import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderProgressDispatcher;
Expand Down Expand Up @@ -221,6 +222,14 @@ public class QwpWebSocketSender implements Sender {
private CursorSendEngine cursorEngine;
private CursorWebSocketSendLoop cursorSendLoop;
private boolean deferCommit;
// True when the sender emits incremental (delta) symbol dictionaries: each
// message carries only symbol ids not yet sent on the wire, rather than the
// full dictionary from id 0. Enabled only in memory-mode, where a reconnect
// replays from the in-process ring and the I/O thread re-registers the whole
// dictionary via a catch-up frame before replaying. File-mode store-and-forward
// keeps full self-sufficient frames (recovery/orphan-drain can replay mid-stream
// to a fresh server, where a partial delta would leave gaps). Set in setCursorEngine.
private boolean deltaDictEnabled;
// User-supplied observer for background orphan-slot drainer events.
// Volatile: written by setDrainerListener (any thread, before or after
// startOrphanDrainers) and read at pool-creation time. Null -> drainers
Expand Down Expand Up @@ -313,6 +322,12 @@ public class QwpWebSocketSender implements Sender {
// beginRound(true) call. roundSeq=1 is the first round; CONNECTED in the
// first round indicates the initial connect.
private long roundSeq;
// Highest global symbol id the producer has baked into a frame so far, or -1.
// Lifetime-monotonic in delta mode -- it is NOT reset on reconnect, because
// the I/O thread re-registers the full dictionary via a catch-up frame before
// replaying, so the producer's delta baseline stays valid across the wire
// boundary. Used only when deltaDictEnabled; ignored in full-dict mode.
private int sentMaxSymbolId = -1;
// When true, auto-flush sends messages with FLAG_DEFER_COMMIT and only
// explicit flush() triggers the server-side commit. Enables accumulating
// arbitrarily large datasets that exceed the server's recv buffer.
Expand Down Expand Up @@ -2215,6 +2230,18 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
}
this.cursorEngine = engine;
this.ownsCursorEngine = takeOwnership && engine != null;
// Delta encoding is available in memory-mode (in-process catch-up) and in
// file-mode when the persisted dictionary opened (recovery / orphan-drain
// rebuild the dictionary from it). Otherwise fall back to full self-
// sufficient frames. See CursorSendEngine.isDeltaDictEnabled.
this.deltaDictEnabled = engine != null && engine.isDeltaDictEnabled();
// Recovery: repopulate the producer's global dictionary from the slot's
// persisted dictionary so newly ingested symbols continue from the
// recovered ids (rather than colliding with them at 0), and the delta
// baseline resumes where the crashed session left off.
if (deltaDictEnabled && engine.wasRecoveredFromDisk()) {
seedGlobalDictionaryFromPersisted(engine.getPersistedSymbolDict());
}
}

/**
Expand Down Expand Up @@ -3423,14 +3450,15 @@ private void flushPendingRows(boolean deferCommit) {
}

ensureActiveBufferReady();
// Cursor SF requires every on-disk frame to be self-sufficient:
// recorded frames replay to fresh server connections (orphan-slot
// drainers and post-reconnect replay), so always emit the full
// symbol-dict delta from id=0 and the full column schema inline,
// never a back-reference the target server may not have seen.
// In full-dict mode every frame is self-sufficient: it carries the whole
// symbol dictionary from id 0 so orphan-drain / recovery replay to a fresh
// server never dangles a symbol id. In delta mode (memory-mode only) each
// frame carries only ids above sentMaxSymbolId; a reconnect re-registers
// the dictionary via an I/O-thread catch-up frame before replay, so the
// producer's monotonic baseline stays valid across the wire boundary.
encoder.setDeferCommit(deferCommit);
encoder.beginMessage(tableCount, globalSymbolDictionary,
/*confirmedMaxId=*/ -1, currentBatchMaxSymbolId);
symbolDeltaBaseline(), currentBatchMaxSymbolId);
for (int i = 0, n = keys.size(); i < n; i++) {
CharSequence tableName = keys.getQuick(i);
if (tableName == null) {
Expand All @@ -3456,10 +3484,18 @@ private void flushPendingRows(boolean deferCommit) {
return;
}

// Write-ahead: durably persist this frame's new symbols BEFORE it is
// published, so a recovered/orphan-drained slot can always rebuild the
// dictionary the (non-self-sufficient) delta frame references. No-op in
// memory mode and when the frame introduces no new symbols.
persistNewSymbolsBeforePublish();
activeBuffer.ensureCapacity(messageSize);
activeBuffer.write(buffer.getBufferPtr(), messageSize);
activeBuffer.incrementRowCount();
sealAndSwapBuffer();
// The frame carrying ids up to currentBatchMaxSymbolId is now on the ring;
// advance the delta baseline so the next frame ships only newer ids.
advanceSentMaxSymbolId();

hasDeferredMessages = deferCommit;
if (!deferCommit) {
Expand Down Expand Up @@ -3514,8 +3550,12 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
boolean deferThis = deferCommit || !isLast;

encoder.setDeferCommit(deferThis);
// Each split frame emits the delta above sentMaxSymbolId; the first
// frame ships the whole batch's new ids and advances the baseline, so
// the remaining frames carry an empty delta and just reference ids the
// first frame already registered.
encoder.beginMessage(1, globalSymbolDictionary,
/*confirmedMaxId=*/ -1, currentBatchMaxSymbolId);
symbolDeltaBaseline(), currentBatchMaxSymbolId);
encoder.addTable(tableBuffer);
int messageSize = encoder.finishMessage();
QwpBufferWriter buffer = encoder.getBuffer();
Expand All @@ -3528,11 +3568,18 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
.put(", serverMaxBatchSize=").put(serverMaxBatchSize).put(']');
}

// Write-ahead persist before publish (see flushPendingRows). The
// first split frame carries the batch's new symbols; the rest are
// no-ops once the baseline has advanced past them.
persistNewSymbolsBeforePublish();
ensureActiveBufferReady();
activeBuffer.ensureCapacity(messageSize);
activeBuffer.write(buffer.getBufferPtr(), messageSize);
activeBuffer.incrementRowCount();
sealAndSwapBuffer();
// Frame queued: advance so the next split frame's delta starts above
// the ids this one just registered.
advanceSentMaxSymbolId();
}

encoder.setDeferCommit(false);
Expand Down Expand Up @@ -3573,8 +3620,10 @@ private void sendCommitMessage() {
LOG.debug("Sending commit message for deferred batch");
}
encoder.setDeferCommit(false);
// A commit carries no rows and no new symbols; in delta mode its empty
// delta simply starts at the server's current dictionary size.
encoder.beginMessage(0, globalSymbolDictionary,
/*confirmedMaxId=*/ -1, currentBatchMaxSymbolId);
symbolDeltaBaseline(), currentBatchMaxSymbolId);
int messageSize = encoder.finishMessage();
QwpBufferWriter buffer = encoder.getBuffer();
ensureActiveBufferReady();
Expand All @@ -3586,15 +3635,84 @@ private void sendCommitMessage() {
lastCommitBoundaryFsn = cursorEngine.publishedFsn();
}

/**
* Advances the delta baseline once a frame carrying the current batch's
* symbols has been queued onto the ring. No-op in full-dict mode. Only ever
* moves the baseline forward, so a batch that used no new symbols leaves it
* unchanged.
*/
private void advanceSentMaxSymbolId() {
if (deltaDictEnabled && currentBatchMaxSymbolId > sentMaxSymbolId) {
sentMaxSymbolId = currentBatchMaxSymbolId;
}
}

/**
* Appends the symbols this frame introduces ({@code [sentMaxSymbolId+1 ..
* currentBatchMaxSymbolId]}) to the slot's persisted dictionary BEFORE the
* frame is published to the ring. This write-ahead ordering keeps the
* persisted dictionary a superset of every process-crash-recoverable frame's
* references, so recovery and orphan-drain can re-register it on a fresh
* server. Not fsync'd (see PersistedSymbolDict) -- a host crash that tears it
* is caught by the send loop's replay guard. No-op in memory mode (no
* persisted dictionary) and when the frame introduces no new symbols.
*/
private void persistNewSymbolsBeforePublish() {
if (!deltaDictEnabled || cursorEngine == null) {
return;
}
PersistedSymbolDict pd = cursorEngine.getPersistedSymbolDict();
if (pd == null) {
return;
}
int from = sentMaxSymbolId + 1;
int to = currentBatchMaxSymbolId;
if (to < from) {
return;
}
for (int id = from; id <= to; id++) {
pd.appendSymbol(globalSymbolDictionary.getSymbol(id));
}
}

private void resetSymbolDictStateForNewConnection() {
// The new server has an empty symbol dictionary, so the next batch
// must ship a delta starting at id 0. beginMessage() always passes
// confirmedMaxId = -1; resetting the batch watermark here keeps a
// stale value from suppressing re-emission of symbol ids the new
// server has never seen.
// Runs on the foreground (initial) connect only -- NOT on the I/O thread's
// reconnect/failover path. The per-batch watermark is drained state, so
// clearing it here is harmless. sentMaxSymbolId is deliberately left
// untouched: in delta mode the I/O thread re-registers the whole
// dictionary with a catch-up frame on reconnect, so the producer's
// monotonic baseline must survive the wire boundary; resetting it would
// desync the producer from the I/O thread's sent-dictionary count.
currentBatchMaxSymbolId = -1;
}

/**
* On recovery, repopulates the producer's {@link GlobalSymbolDictionary} from
* the slot's persisted dictionary (ids assigned in the same ascending order,
* so they match the recovered frames) and resumes the delta baseline at the
* recovered tip, so newly ingested symbols continue above the recovered ids.
*/
private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
if (pd == null || pd.size() == 0) {
return;
}
ObjList<String> symbols = pd.readLoadedSymbols();
for (int i = 0, n = symbols.size(); i < n; i++) {
globalSymbolDictionary.getOrAddSymbol(symbols.getQuick(i));
}
sentMaxSymbolId = globalSymbolDictionary.size() - 1;
}

/**
* The symbol id below which the server already holds every dictionary entry,
* used as {@code confirmedMaxId} when encoding a frame. In delta mode this is
* the producer's monotonic sent watermark; in full-dict mode it is -1 so every
* frame re-ships the dictionary from id 0.
*/
private int symbolDeltaBaseline() {
return deltaDictEnabled ? sentMaxSymbolId : -1;
}

private void rollbackRow() {
if (currentTableBuffer != null) {
currentTableBuffer.cancelCurrentRow();
Expand Down
70 changes: 68 additions & 2 deletions core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,13 @@ public final class CursorSendEngine implements QuietCloseable {
// in the constructor, closed by {@link #close()}. The segment manager
// writes through this on every tick where ackedFsn has advanced.
private final AckWatermark watermark;
// Engine-owned per-slot symbol dictionary file (disk mode only; {@code null}
// in memory mode and if open() failed). Enables delta-encoded SF frames:
// recovery / orphan-drain load it to re-register the dictionary on the fresh
// server before replaying non-self-sufficient frames. Opened in the
// constructor, closed by {@link #close()}. When null in disk mode the engine
// reports delta encoding as unavailable and the sender keeps full-dict frames.
private final PersistedSymbolDict persistedSymbolDict;
// close() is publicly callable from any thread (Sender.close from a user
// thread, JVM shutdown hooks, test cleanup). volatile + synchronized
// close() makes the check-and-set atomic and gives readers a fence.
Expand Down Expand Up @@ -199,6 +206,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
// reference instead of orphaning the mmap'd segments + fds.
SegmentRing ringInProgress = null;
AckWatermark watermarkInProgress = null;
PersistedSymbolDict persistedDictInProgress = null;
try {
// Disk mode: try to recover any *.sfa files left behind by a prior
// session before deciding to start fresh. Without this the engine
Expand Down Expand Up @@ -257,6 +265,10 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
// mmap doesn't take down the engine -- we just fall
// back to the bare lowestBase - 1 seed.
watermarkInProgress = AckWatermark.open(sfDir);
// Load the persisted symbol dictionary so delta-encoded frames
// in this recovered slot can be re-registered on the fresh
// server before replay. Null on open failure -> delta disabled.
persistedDictInProgress = PersistedSymbolDict.open(sfDir);
long baseSeed = lowestBase - 1;
long watermarkFsn = watermarkInProgress != null
? watermarkInProgress.read()
Expand Down Expand Up @@ -309,6 +321,10 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
if (!memoryMode) {
AckWatermark.removeOrphan(sfDir);
watermarkInProgress = AckWatermark.open(sfDir);
// Same stale-side-file hygiene for the symbol dictionary: a
// fresh slot starts with an empty dictionary.
PersistedSymbolDict.removeOrphan(sfDir);
persistedDictInProgress = PersistedSymbolDict.open(sfDir);
}
MmapSegment initial;
String initialPath = null;
Expand All @@ -333,10 +349,11 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
manager.start();
}
manager.register(ringInProgress, sfDir, watermarkInProgress);
// All construction succeeded — commit the ring and
// watermark references.
// All construction succeeded — commit the ring, watermark and
// symbol-dictionary references.
this.ring = ringInProgress;
this.watermark = watermarkInProgress;
this.persistedSymbolDict = persistedDictInProgress;
} catch (Throwable t) {
// Stop an owned manager before freeing the ring and watermark it may
// touch, then release the slot lock. Each cleanup is in its own
Expand All @@ -362,6 +379,12 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
} catch (Throwable ignored) {
}
}
if (persistedDictInProgress != null) {
try {
persistedDictInProgress.close();
} catch (Throwable ignored) {
}
}
if (acquiredLock != null) {
try {
acquiredLock.close();
Expand Down Expand Up @@ -541,6 +564,12 @@ public synchronized void close() {
} catch (Throwable ignored) {
}
}
if (persistedSymbolDict != null) {
try {
persistedSymbolDict.close();
} catch (Throwable ignored) {
}
}
if (fullyDrained) {
try {
unlinkAllSegmentFiles(sfDir);
Expand All @@ -550,6 +579,11 @@ public synchronized void close() {
AckWatermark.removeOrphan(sfDir);
} catch (Throwable ignored) {
}
try {
// Slot fully drained: the dictionary has no frames behind it.
PersistedSymbolDict.removeOrphan(sfDir);
} catch (Throwable ignored) {
}
}
} finally {
if (slotLock != null) {
Expand Down Expand Up @@ -586,6 +620,38 @@ public long getTotalBackpressureStalls() {
return backpressureStallCount.get();
}

/**
* True when this engine has no store-and-forward directory: the ring lives
* only in malloc'd memory, so it cannot be recovered after a crash and no
* orphan drainer ever replays it. Only in-process reconnect/failover replays
* its frames, which is what makes send-time symbol-dict catch-up (rather than
* fully self-sufficient frames) safe.
*/
public boolean isMemoryMode() {
return sfDir == null;
}

/**
* Whether the sender may delta-encode symbol dictionaries on this engine.
* Always true in memory mode (the send loop keeps an in-process catch-up
* mirror). In disk mode it requires the persisted dictionary to have opened,
* since delta frames are not self-sufficient and recovery / orphan-drain must
* be able to rebuild the dictionary from disk. When false in disk mode the
* sender falls back to full self-sufficient frames.
*/
public boolean isDeltaDictEnabled() {
return sfDir == null || persistedSymbolDict != null;
}

/**
* The engine's persisted symbol dictionary, or {@code null} in memory mode
* (and in disk mode if it failed to open). The producer appends new symbols
* to it; recovery / orphan-drain read its loaded entries to seed catch-up.
*/
public PersistedSymbolDict getPersistedSymbolDict() {
return persistedSymbolDict;
}

/**
* Pass-through to {@link SegmentRing#nextSealedAfter(MmapSegment)}.
*/
Expand Down
Loading
Loading