Qfyilyi

I have to process large CSV files (up to tens of GB), that looks like this:

Key,CompletedA,CompletedB

1,true,NULL

2,true,NULL

3,false,NULL

1,NULL,true

2,NULL,true  

I have a parser that yields parsed lines as IEnumerable<Record>, so that I reads only one line at a time into memory.

Now I have to group records by Key and check whether columns CompletedA and CompletedB have value within the group. On the output I need records, that does not have both CompletedA,CompletedB within the group.

In this case it is record with key 3.

However, there is many similar processings going on the same dataset and I don't wont to iterate over it multiple times.

I think I can convert IEnumerable into IObservable and use Reactive Extentions to find the records.

Is it possible to do it in memory efficient way with simple Linq expression over the IObservable collection?

Providing that Key is an integer we can try using a Dictionary and one scan:

 // value: 0b00 - neither A nor B

 //        0b01 - A only

 //        0b10 - B only

 //        0b11 - Both A and B    

 Dictionary<int, byte> Status = new Dictionary<int, byte>();



 var query = File

   .ReadLines(@"c:MyFile.csv")

   .Where(line => !string.IsNullOrWhiteSpace(line))

   .Skip(1) // skip header 

   .Select(line => YourParserHere(line));



 foreach (var record in query) {

   int mask = (record.CompletedA != null ? 1 : 0) |

              (record.CompletedB != null ? 2 : 0); 



   if (Status.TryGetValue(record.Key, out var value))

     Status[record.Key] = (byte) (value | mask);

   else

     Status.Add(record.Key, (byte) mask);

 }



 // All keys that don't have 3 == 0b11 value (both A and B)  

 var bothAandB = Status

   .Where(pair => pair.Value != 3)

   .Select(pair => pair.Key); 

I think this will do what you need:

var result =

    source

        .GroupBy(x => x.Key)

        .SelectMany(xs =>

            (xs.Select(x => x.CompletedA).Any(x => x != null && x == true) && xs.Select(x => x.CompletedA).Any(x => x != null && x == true))

            ? new List<Record>()

            : xs.ToList());

Using Rx doesn't help here.