I'm using azure table storage, and I'm trying to iterate over the table quickish.
I must be doing it wrong, but I can't for the life of me see why;
When I specify just a single partition I end up getting back results for MULTIPLE partitions. I.e. If I constrain the query by just using "pkey1", I get 1000 results back for "pkey1", and then 325 for "pkey2"
Completely confused as to how this can happen..
This is the code I'm using:
private CloudTableClient _client;
private string _tableName;
private class QueryState
{
public CloudTableQuery<T> Ctq;
public Action<IEnumerable<T>> Populator;
public ManualResetEvent Mre;
public string Pkey;
public QueryState(CloudTableQuery<T> ctq, Action<IEnumerable<T>> populator, ManualResetEvent mre, string pkey)
{
Populator = populator;
Ctq = ctq;
Mre = mre;
Pkey = pkey;
}
}
public void ParallelQueryWithClause(Action<IEnumerable<T>> populator, string[] partitionKeys)
{
List<ManualResetEvent> mre = new List<ManualResetEvent>();
foreach (string pKey in partitionKeys)
{
//_retry.Go(tsc =>
// {
TableServiceContext tsc = _client.GetDataServiceContext();
ManualResetEvent m = new ManualResetEvent(false);
mre.Add(m);
CloudTableQuery<T> query = tsc.CreateQuery<T>(_tableName).Where(e => e.PartitionKey == pKey).AsTableServiceQuery<T>();
Action<IAsyncResult> act = null;
act = result =>
{
int retries = 0;
while (retries++ < 5)
{
try
{
QueryState qsInternal = result.AsyncState as QueryState;
CloudTableQuery<T> ctq = qsInternal.Ctq;
ResultSegment<T> seg = ctq.EndExecuteSegmented(result);
if (seg.Results.Count() > 0)
populator(seg.Results);
if (seg.ContinuationToken != null)
{
ctq.BeginExecuteSegmented(seg.ContinuationToken, iasync => act(iasync), qsInternal);
}
else
{
m.Set();
}
break;
}
catch(Exception ex)
{
Logger.LogError(ex);
}
}
};
query.BeginExecuteSegmented(iasync => act(iasync), new QueryState(query, populator, m, pKey));
//});
}
ManualResetEvent.WaitAll(mre.ToArray());
}
And sample calling code:
AzureTableStorage<ProductEntity> _ats = new AzureTableStorage<ProductEntity>("Products");
string[] partitions = new string[] { "pkey1" };
Dictionary<string, int> cntr = new Dictionary<string, int>();
_ats.ParallelQueryWithClause(p =>
{
lock (cntr)
{
foreach (ProductEntity pe in p)
{
if (cntr.ContainsKey(pe.PartitionKey))
cntr[pe.PartitionKey]++;
else
cntr.Add(pe.PartitionKey, 1);
}
}
}, partitions);
Hopefully this makes sense, and someone can help!