fix(window): 聚合结果加 window_id 字段；整数时间戳缺单位时不猜测；修 session 回调竞争；删 counting 无用字段

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))
 	}

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{}

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)

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
 				}
 			}

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