Why so long time query on small stream I have?

Question

I'm new in StreamInsight. My method for get stream looks like this:

public object GetStream(object args)
        {
            var streams = (List<object>)args;
            var streamBig = (CepStream<Payload>)streams[0];
            var cepApp = (Application)streams[1];
            
            // in tariffs about 1000 records  
            CepStream<int> tariffList = tariffs.ToIntervalStream(cepApp, e => IntervalEvent.CreateInsert(DateTimeOffset.MinValue, DateTimeOffset.MaxValue, e), AdvanceTimeSettings.IncreasingStartTime);

            // in bnumbers about 50 records
            CepStream<string> bnumberList = bnumbers.ToIntervalStream(cepApp, e => IntervalEvent.CreateInsert(DateTimeOffset.MinValue, DateTimeOffset.MaxValue, e), AdvanceTimeSettings.IncreasingStartTime);

            // in operatorsRange about 20 records 
            CepStream<KeyValuePair<int, int>> operatorsRangeList = operatorsRange.ToIntervalStream(cepApp, e => IntervalEvent.CreateInsert(DateTimeOffset.MinValue, DateTimeOffset.MaxValue, e),AdvanceTimeSettings.IncreasingStartTime);

            //(1) streamBig contains about 300000 records 
            var stream1 = from item in streamBig
                          let prefix = item.FieldInt2
                          from bnum in bnumberList
                          from tariff in tariffList
                          from opRange in operatorsRangeList
                          where item.FieldEntityId1 == bnum
                              && (item.FieldInt1 == 0 || item.FieldInt1 == tariff)
                              && (prefix >= opRange.Key && prefix <= opRange.Value)
                          select item;
              //(2)
              var stream = from item1 in stream1
                         group item1 by new { item1.FieldEntityId1, item1.FieldRegion, item1.FieldInt1, item1.FieldString1, item1.FieldString2, item1.FieldString3, item1.FieldString4 } into grp
                         from window in grp.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
                         select new Payload
                         {
                             FieldEntityId1 = grp.Key.FieldEntityId1,
                             FieldEntityId2 = string.Empty,
                             FieldEntityId3 = string.Empty,
                             FieldLong = window.Sum(x => x.FieldLong),
                             FieldRegion = grp.Key.FieldRegion,

                             FieldDateTime = window.Min(x => x.FieldDateTime),
                             FieldDateTime1 = window.Max(x => x.FieldDateTime1),
                             FieldDetail1 = grp.Key.FieldEntityId1,

                             FieldDetail2 = grp.Key.FieldString1,
                             FieldDetail3 = grp.Key.FieldString2,
                             FieldDetail4 = grp.Key.FieldString3,
                             FieldDetail5 = grp.Key.FieldInt1.ToString(),
                             FieldDetail6 = grp.Key.FieldString4,
                             FieldDetail7 = window.Min(x => x.FieldDateTime).ToString("dd.MM.yyyy HH:mm:ss"),
                             FieldDetail8 = window.Max(x => x.FieldDateTime1).ToString("dd.MM.yyyy HH:mm:ss"),
                             FieldDetail9 = window.Count().ToString(),
                             FieldDetail10 = window.Sum(x => x.FieldLong).ToString(),

                             FieldInt = 0,
                             FieldInt1 = 0,
                             FieldInt2 = 0,
                             FieldInt3 = 0,
                             FieldInt4 = 0,
                             FieldInt5 = 0,
                             FieldLong1 = 0,
                             FieldLong2 = 0,
                             FieldLong3 = 0,
                             FieldLong4 = 0,
                             FieldLong5 = 0,
                             FieldString = string.Empty,
                             FieldString1 = string.Empty,
                             FieldString2 = string.Empty,
                             FieldString3 = string.Empty,
                             FieldString4 = string.Empty,
                             FieldString5 = string.Empty
                         };

            return stream;
        }

Stream from database contains only 300000 records and I need filter records using 3 books tariffs (about 1000 items), bnumbers (about 50 items) and operatorsRange (about 20 items).

I don't get result many hours on high-power server. Please, help me. Where is my mistake? I see in debugger, query running on linq marked (1) and before grouping.

Answer 1

What is the specification for "streamBig?

It's likely synchronization with CTIs. Because you are specifying IncreasingStartTime, the CTIs for your 3 reference streams won't ever (really) get enqueued ... their start time never increases. Because their are no CTIs at all in the reference streams, the events aren't going to sync temporally. You'll want to actually import the CTIs from your data stream (I'm assuming that this is "streamBig") into your reference streams with an AdvanceTimePolicy of adjust (rather than drop).

To get a better idea, do a recording with the EventFlowDebugger. It'll be a big file but you should be able to see the when the CTIs move the streams forward.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.