diff --git a/rsql/join_test.go b/rsql/join_test.go index ade1a8e..cba8a21 100644 --- a/rsql/join_test.go +++ b/rsql/join_test.go @@ -63,7 +63,7 @@ func TestParseLeftJoinAndCompositeKey(t *testing.T) { if jc.JoinType != "LEFT" { t.Errorf("joinType = %q, want LEFT", jc.JoinType) } - want := []types.JoinOnPair{{"k1", "k1"}, {"k2", "k2"}} + want := []types.JoinOnPair{{StreamField: "k1", TableField: "k1"}, {StreamField: "k2", TableField: "k2"}} if len(jc.OnPairs) != 2 { t.Fatalf("onpairs len = %d, want 2", len(jc.OnPairs)) } diff --git a/stream/processor_data.go b/stream/processor_data.go index 01c90d8..d2c468f 100644 --- a/stream/processor_data.go +++ b/stream/processor_data.go @@ -362,11 +362,29 @@ func (dp *DataProcessor) processWindowBatch(batch []types.Row) { // Get and send aggregation results if results, err := dp.stream.aggregator.GetResults(); err == nil { + stampWindowID(results, batch) dp.processAggregationResults(results) dp.stream.aggregator.Reset() } } +// stampWindowID stamps a stable window_id (window time bounds) onto each +// result. It is identical across the initial emit and accumulating late +// re-emits (AllowedLateness>0), so sinks can dedup/replace by group + window_id. +func stampWindowID(results []map[string]interface{}, batch []types.Row) { + if len(batch) == 0 { + return + } + slot := batch[0].Slot + if slot == nil || slot.Start == nil || slot.End == nil { + return + } + id := fmt.Sprintf("%d_%d", slot.Start.UnixNano(), slot.End.UnixNano()) + for _, r := range results { + r["window_id"] = id + } +} + // processAggregationResults processes aggregation results func (dp *DataProcessor) processAggregationResults(results []map[string]interface{}) { var finalResults []map[string]interface{} diff --git a/window/counting_window.go b/window/counting_window.go index 5d9e54d..e69fc8c 100644 --- a/window/counting_window.go +++ b/window/counting_window.go @@ -33,16 +33,15 @@ import ( var _ Window = (*CountingWindow)(nil) type CountingWindow struct { - config types.WindowConfig - threshold int - count int - mu sync.Mutex - callback func([]types.Row) - dataBuffer []types.Row - outputChan chan []types.Row - ctx context.Context - cancelFunc context.CancelFunc - triggerChan chan types.Row + config types.WindowConfig + threshold int + mu sync.Mutex + callback func([]types.Row) + dataBuffer []types.Row + outputChan chan []types.Row + ctx context.Context + cancelFunc context.CancelFunc + triggerChan chan types.Row // keyedBuffer/keyedCount accumulate rows per group key until threshold is hit. // No eviction: with high-cardinality GroupByKeys where each key receives fewer // than `threshold` rows, these grow unbounded over long runs. Count windows @@ -53,8 +52,8 @@ type CountingWindow struct { lastActive map[string]time.Time countStateTTL time.Duration sentCount int64 - droppedCount int64 - stopped bool + droppedCount int64 + stopped bool } func NewCountingWindow(config types.WindowConfig) (*CountingWindow, error) { @@ -86,13 +85,13 @@ func NewCountingWindow(config types.WindowConfig) (*CountingWindow, error) { } cw := &CountingWindow{ - config: config, - threshold: threshold, - dataBuffer: make([]types.Row, 0, threshold), - outputChan: make(chan []types.Row, bufferSize), - ctx: ctx, - cancelFunc: cancel, - triggerChan: make(chan types.Row, bufferSize), + config: config, + threshold: threshold, + dataBuffer: make([]types.Row, 0, threshold), + outputChan: make(chan []types.Row, bufferSize), + ctx: ctx, + cancelFunc: cancel, + triggerChan: make(chan types.Row, bufferSize), keyedBuffer: make(map[string][]types.Row), keyedCount: make(map[string]int), lastActive: make(map[string]time.Time), @@ -285,7 +284,6 @@ func (cw *CountingWindow) Reset() { cw.mu.Lock() defer cw.mu.Unlock() - cw.count = 0 cw.dataBuffer = nil cw.keyedBuffer = make(map[string][]types.Row) cw.keyedCount = make(map[string]int) diff --git a/window/factory.go b/window/factory.go index 7972bdd..65411c1 100644 --- a/window/factory.go +++ b/window/factory.go @@ -81,7 +81,9 @@ func GetTimestamp(data interface{}, tsProp string, timeUnit time.Duration) time. // extractTimestamp returns the event timestamp and true when one can be derived // from the data (a GetTimestamp() time.Time method, or a tsProp field of -// time.Time or int64). Returns (zero, false) otherwise. +// time.Time or int64 epoch). An int64 epoch requires a non-zero TimeUnit; +// otherwise the unit is ambiguous and it is treated as unplaceable. +// Returns (zero, false) otherwise. func extractTimestamp(data interface{}, tsProp string, timeUnit time.Duration) (time.Time, bool) { if ts, ok := data.(interface{ GetTimestamp() time.Time }); ok { return ts.GetTimestamp(), true @@ -103,6 +105,10 @@ func extractTimestamp(data interface{}, tsProp string, timeUnit time.Duration) ( if t, ok := value.Interface().(time.Time); ok { return t, true } else if timestampInt, isInt := value.Interface().(int64); isInt { + // int epoch without TimeUnit is ambiguous (s vs ms); don't guess. + if timeUnit == 0 { + return time.Time{}, false + } return cast.ConvertIntToTime(timestampInt, timeUnit), true } } diff --git a/window/session_window.go b/window/session_window.go index b9d7477..39efe72 100644 --- a/window/session_window.go +++ b/window/session_window.go @@ -360,18 +360,20 @@ func (sw *SessionWindow) checkExpiredSessions() { // otherwise, with allowedLateness > 0, they accumulate forever (the event-time // path does this in checkAndTriggerSessions). sw.closeExpiredSessions(now) + callback := sw.callback sw.mu.Unlock() - sw.sendResults(resultsToSend) + sw.sendResults(resultsToSend, callback) } func (sw *SessionWindow) checkAndTriggerSessions(watermarkTime time.Time) { sw.mu.Lock() resultsToSend := sw.collectExpiredSessions(watermarkTime) sw.closeExpiredSessions(watermarkTime) + callback := sw.callback sw.mu.Unlock() - sw.sendResults(resultsToSend) + sw.sendResults(resultsToSend, callback) } func (sw *SessionWindow) collectExpiredSessions(currentTime time.Time) [][]types.Row { @@ -411,15 +413,17 @@ func (sw *SessionWindow) collectExpiredSessions(currentTime time.Time) [][]types return resultsToSend } -func (sw *SessionWindow) sendResults(resultsToSend [][]types.Row) { +// sendResults dispatches collected results to the callback and output channel. +// callback must be captured under sw.mu by the caller to avoid racing SetCallback. +func (sw *SessionWindow) sendResults(resultsToSend [][]types.Row, callback func([]types.Row)) { for _, result := range resultsToSend { // Skip empty results to avoid filling up channels if len(result) == 0 { continue } - if sw.callback != nil { - sw.callback(result) + if callback != nil { + callback(result) } sw.sendResult(result) @@ -500,23 +504,11 @@ func (sw *SessionWindow) Trigger() { // Clear all sessions sw.sessionMap = make(map[string]*session) - // Release lock before sending to channel and calling callback to avoid blocking + // Capture callback under the lock; release before sending to avoid blocking. + callback := sw.callback sw.mu.Unlock() - // Send results and call callbacks outside of lock to avoid blocking - for _, result := range resultsToSend { - // Skip empty results to avoid filling up channels - if len(result) == 0 { - continue - } - - // If callback function is set, execute it - if sw.callback != nil { - sw.callback(result) - } - - sw.sendResult(result) - } + sw.sendResults(resultsToSend, callback) } // Reset resets session window data