Package io.deephaven.extensions.barrage.chunk

Class SharedDictionaryWriterState

java.lang.Object
io.deephaven.extensions.barrage.chunk.SharedDictionaryWriterState
All Implemented Interfaces:
DictionaryWriterState
public final class SharedDictionaryWriterState extends Object implements DictionaryWriterState
Per-subscriber DictionaryWriterState for full subscriptions (and growing subscriptions targeting a full subscription). Delegates value-to-index lookups to a SharedWriterDictionary shared across all full subscribers on the same table, and tracks an independent flushedOffset so each subscriber only receives values it has not yet seen.

On first use (needsFullBatch() == true), buildDeltaChunk() returns a chunk covering the complete current value list so the subscriber receives an isDelta=false reset batch covering all values accumulated before it joined. After resetDelta() is called, only values added after that point are included in future delta batches.

Other subscribers add values to the shared dictionary, and this subscriber is sent them only in a batch whose own rows reference the dictionary, that is, a batch for which fillIndexChunk(io.deephaven.chunk.Chunk<io.deephaven.chunk.attributes.Values>, io.deephaven.engine.rowset.RowSet, io.deephaven.extensions.barrage.BarrageOptions, io.deephaven.chunk.WritableIntChunk<io.deephaven.chunk.attributes.Values>) was called. A subscriber that never writes a column using this dictionary, such as a subscriber not subscribed to that column, is sent only the first batch, carrying no values, which defines the dictionary for strict Arrow readers. It then holds an empty dictionary, so a later batch that references the dictionary is sent every value from the start, as a delta.

resetDelta() advances past exactly the values that the last buildDeltaChunk() covered, not past whatever the shared dictionary holds by then. Full subscribers write in parallel, so other subscribers may add values between this subscriber's building a DictionaryBatch and its sending it; those values were not in the batch, and they stay in this subscriber's next delta.

Thread-safety: each instance belongs to one subscription and is used by one thread at a time. The SharedWriterDictionary it delegates to is shared, and is safe to use from several threads at once.

  • Constructor Details

    • SharedDictionaryWriterState

      public SharedDictionaryWriterState(@NotNull @NotNull SharedWriterDictionary shared)

  • Method Details

    • shared

      @InternalUseOnly @NotNull public @NotNull SharedWriterDictionary shared()
      The shared dictionary this state delegates to.

    • flushedOffset

      @InternalUseOnly public int flushedOffset()
      Returns the number of values of the shared dictionary's current generation that have been sent, after detecting any reset since this state last looked.

    • advanceTo

      @InternalUseOnly public void advanceTo(int flushedOffset, boolean needsFullBatch)
      Advances this state to a position: from now on it behaves as though it had sent exactly the values below flushedOffset of the shared dictionary's current generation, and as though its last batch had been accepted. A producer that writes one message for a group of subscribers uses this to start the group's state where its members are, and then to advance each member to where the group ends once the group's messages have been written to it.

      A state never moves back within a generation: one credited with fewer values than its subscriber holds would send them again, and every later index would refer to the wrong value. Only a reset of the shared dictionary, which starts a new generation, returns a state to the start, and this first detects any such reset.

      Parameters:
      flushedOffset - the number of values sent; no fewer than this state has sent, and no more than the shared dictionary holds
      needsFullBatch - whether the next DictionaryBatch must be isDelta=false; only while this state still needs one, and so has sent no value
      Throws:
      IllegalArgumentException - if the position is behind this state's own, or past the end of the dictionary

    • getDictId

      public long getDictId()
      Specified by:
      getDictId in interface DictionaryWriterState

    • fillIndexChunk

      public void fillIndexChunk(@NotNull @NotNull Chunk<Values> source, @Nullable @Nullable RowSet subset, @NotNull @NotNull BarrageOptions options, @NotNull @NotNull WritableIntChunk<Values> out)
      Description copied from interface: DictionaryWriterState
      Fills out with one dictionary index per logical row in source/subset. Null rows (in non-deephaven-nulls mode) produce QueryConstants.NULL_INT; non-null rows produce a non-negative dictionary index, registering new values as needed.
      Specified by:
      fillIndexChunk in interface DictionaryWriterState
      Parameters:
      source - the source chunk containing column values
      subset - row positions within source to include; null means all rows
      options - barrage serialization options (e.g. useDeephavenNulls)
      out - pre-sized output chunk to fill with dictionary indices

    • hasDelta

      public boolean hasDelta()
      Description copied from interface: DictionaryWriterState
      Returns true if a DictionaryBatch message needs to be emitted before the current RecordBatch — either because this is the first batch for this subscriber (isDelta=false) or because new values have been added since the last reset. A subscriber is owed values that others added to a shared dictionary only once a batch of its own references the dictionary; see SharedDictionaryWriterState.
      Specified by:
      hasDelta in interface DictionaryWriterState

    • needsFullBatch

      public boolean needsFullBatch()
      Specified by:
      needsFullBatch in interface DictionaryWriterState

    • buildDeltaChunk

      @NotNull public @NotNull WritableChunk<Values> buildDeltaChunk()
      Description copied from interface: DictionaryWriterState
      Builds and returns a typed chunk containing the current delta values (values added since the last DictionaryWriterState.resetDelta() call, or all values if this is the first batch for this subscriber). The returned chunk is owned by the caller and must be closed when no longer needed.
      Specified by:
      buildDeltaChunk in interface DictionaryWriterState

    • resetDelta

      public void resetDelta()
      Accepts the batch just written: advances past the values that the last buildDeltaChunk() covered, and forgets that the batch referenced the dictionary. With no delta built since the last advance, nothing was sent and the offset does not change.
      Specified by:
      resetDelta in interface DictionaryWriterState

    • totalSize

      public int totalSize()
      Description copied from interface: DictionaryWriterState
      Current number of distinct values in the dictionary (resets to zero after DictionaryWriterState.reset()).
      Specified by:
      totalSize in interface DictionaryWriterState

    • reset

      public void reset()
      Not supported: the shared dictionary is compacted by calling SharedWriterDictionary.reset() directly; this per-subscriber wrapper detects that reset lazily via syncGeneration(). Calling reset() here is always a caller error.
      Specified by:
      reset in interface DictionaryWriterState