Splitting a list of numbers into multiple lists

Question

I need split the list I get from the select into multiple list so I can use Parallel.For() to generate the next query in different threads.

• I create the main list
• Create the sub list
• Keep filling sub list until reach max capacity.
• Add sublist to main list
  • Create new sub list
• When finish the loop save the last sublist

This works but I bet there is an easy / clean way to do this, I can change the db if needed.

List<List<long>> result = new List<List<long>>();
using (var conn = new NpgsqlConnection(strConnection))
{
    conn.Open();
    using (var cmd = new NpgsqlCommand())
    {
        cmd.Connection = conn;
        cmd.CommandText = "SELECT route_source_id " +                                      
                          "FROM route_sources " +
                          "WHERE has_route IS NULL " +          
                          "LIMIT 1000; ";

        using (var reader = cmd.ExecuteReader())
        {
            const long list_size = 100;                

            long row_id = 0;
            List<long> route_list = new List<long>();

            while (reader.Read())
            {                                
               if (row_id++ > list_size) // Add the sublist to the result
               {
                   result.Add(route_list); 
                   row_id = 0;
               }

               if (row_id == 0)  // Create new sublist
               {
                  route_list = new List<long>();
               }

               if (!reader.IsDBNull(0)) // fill the sublist
               {
                    route_list.Add(reader.GetInt64(0));                                
               }
            }

            if (route_list.Count > 0 ) // add the last sublist
            {
               result.Add(route_list);
            }
        }
   }
 }
 return result;

EDIT: aditional part of the process.

I have ticker to check the db for pending id to be process

// read 1000 rows split into 10 list, using above function
List<List<long>> route_pending = getRouteRange(); 
long total_rows = 0;

while (route_pending.Count > 0) 
{
     foreach (List<long> range in route_pending)
     {
         total_rows += range.Count;                    
     }
     routeParallelGeneration(route_pending);
     route_pending = getRouteRange(); // keep reading next 1000 rows
}

And the parallel function, here I call my db storeproc with an array parameter. So max 10 connections with array of size 100. Here was where got issues with the excesive number of connections and decide split the data into 10 lists

public void routeParallelGeneration(List<List<long>> route_pending)
{
    Action<List<long>> forEachLoop = route_source_id => //Begin definition of forLoop
    {
        using (var conn = new NpgsqlConnection(strConnection))
        {
            conn.Open();
            using (var cmd = new NpgsqlCommand())
            {
                cmd.Connection = conn;
                cmd.CommandText = "SELECT avl_db.process_route(@route_source)";
                cmd.CommandType = CommandType.Text;

                NpgsqlParameter p = new NpgsqlParameter("@route_source", NpgsqlDbType.Array | NpgsqlDbType.Bigint);
                p.Value = route_source_id;
                cmd.Parameters.Add(p);
                cmd.ExecuteNonQuery();
            }
        }
    };
    Parallel.ForEach(route_pending, forEachLoop);
}

Answer 1

Calculating ranges

I need split the list I get from the select into multiple list so I can use Parallel.For() to generate the next query in different threads.

Nope. You don't need it. Use Parallel.ForEach that automatically can distribute the load on multiple threads if necessary or use the Partitioner to calculate the ranges e.g.:

var partitions = Partitioner
    .Create(fromInclusive: 0, toExclusive: 100, rangeSize: 30)
    .GetDynamicPartitions();

the result

Item1  Item2

    0     30 
   30     60 
   60     90 
   90    100

---

Example

Here's an example of how it could be implemented:

• The first change is that you don't split the list but pass all ids to this method. It'll partition it and will use these partitions for reading from the id-list.

• The second change is that Parallel.ForEach doesn't run over the ids but over the partitions. Each partition (index range) is passed to the body.

• The third change is that you initialize the connection and the command once per thread via the localInit. It'll be disposed/closed by localFinally.

• The fourth change is that you don't have to recreate the command each time. All you have to do is to set the new parameter value and call ExecuteNonQuery.

Code:

public void routeParallelGeneration(List<long> route_pending)
{
    var parallelOptions = new ParallelOptions
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount
    };

    var partitions =
        Partitioner
            .Create(
                fromInclusive: 0,
                toExclusive: route_pending.Count,
                rangeSize: 100
            )
            .GetDynamicPartitions();

    Parallel.ForEach(
        source: partitions,
        parallelOptions: parallelOptions,
        localInit: () => 
        {
            var conn = new NpgsqlConnection(strConnection);
            conn.Open();

            var cmd = new NpgsqlCommand();
            cmd.Connection = conn;
            cmd.CommandText = "SELECT avl_db.process_route(@route_source)";
            cmd.CommandType = CommandType.Text;

            NpgsqlParameter p = new NpgsqlParameter("@route_source", NpgsqlDbType.Array | NpgsqlDbType.Bigint);
            cmd.Parameters.Add(p);

            return new { Connection = conn, Command = cmd };
        },
        body: (source, state, local) =>
        {
            local.Command.Parameters["@route_source"].Value = route_pending.GetRange(
                index: source.Item1, 
                count: source.Item2 - source.Item1 - 1
            );
            local.Command.ExecuteNonQuery();
            return local;
        },
        localFinally: local =>
        {
            local.Connection?.Dispose();
            local.Command?.Dispose();
        }
    );
}

Improvable

Don't do everything in a single method. Write one method to get the data from the server and another one to process it and/or calculate the partitions. You need to pay more attentions to the separation of concerns. Mixing all features in a single method makes it super hard to maintain and to test.

Answer 2

Pretty minor but you don't need the if (row_id == 0)

if (row_id++ > list_size) // Add the sublist to the result
{
    result.Add(route_list); 
    row_id = 0;
    route_list = new List<long>();
}

Also it seem like you should also test for

if (route_list.Count > 0 ) 

Why not just put where route_source_id is not null in the query?

Connection, Command, and Reader behave the same with many DB so this is pretty portable.

I think you can use

conn.CreateCommand()

Long is not required for row_id or list_size

Why bump row_id++ if there is no value?

In summary

    using (var reader = cmd.ExecuteReader())
    {
        const int list_size = 100;                
        List<long> route_list = new List<long>(list_size);

        while (reader.Read())
        {                                
           if (route_list.Count >= list_size) // Add the sublist to the result
           {
               result.Add(route_list); 
               route_list = new List<long>(list_size);
           }
           route_list.Add(reader.GetInt64(0));
        }

        if (route_list.Count > 0 ) // add the last sublist
        {
           result.Add(route_list);
        }
    }

Answer 3

I found this version. I create one query to return an array. here just hard code the size 100.

using (var cmd = new NpgsqlCommand())
{
    cmd.Connection = conn;
    cmd.CommandText = "with cte as ( " +
                      "  SELECT route_source_id::bigint, " +
                      "          (row_number() OVER() - 1) / 100 AS grp " +
                      "  FROM avl_db.route_sources " +
                      "  WHERE  has_route IS NULL " +
                      "  limit 1000 " +
                      ") " +
                      "SELECT array_agg(route_source_id)" +
                      "FROM cte " +
                      "GROUP BY grp " +
                      "ORDER BY grp; ";
    using (var reader = cmd.ExecuteReader())
    {
       while (reader.Read())
       {
           long[] sublist = (long[])reader.GetValue(0);
           result.Add(sublist.ToList());
       }
    }
}