Merge pull request #77 from jessemcdowell-AI/BatchLargeAppends

Issue-73 Batch large appends

Author: Doug Schmidt

TimeSeries/PublicApis/SdkExamples/PointZilla/Context.cs

@@ -34,6 +34,7 @@ public class Context
         public Instant StartTime { get; set; } = Instant.FromDateTimeUtc(DateTime.UtcNow);
         public TimeSpan PointInterval { get; set; } = TimeSpan.FromMinutes(1);
         public int NumberOfPoints { get; set; } // 0 means "derive the point count from number of periods"
+        public int BatchSize { get; set; } = 500_000;
         public double NumberOfPeriods { get; set; } = 1;
         public WaveformType WaveformType { get; set; } = WaveformType.SineWave;
         public double WaveformOffset { get; set; } = 0;

TimeSeries/PublicApis/SdkExamples/PointZilla/PointsAppender.cs

@@ -60,60 +60,95 @@ public void AppendPoints()
                         ? $"Appending {Points.Count} {pointExtents} within TimeRange={GetTimeRange()} to {timeSeries.Identifier} ({timeSeries.TimeSeriesType}) ..."
                         : $"Appending {Points.Count} {pointExtents} to {timeSeries.Identifier} ({timeSeries.TimeSeriesType}) ...");
 
+                var numberOfPointsAppended = 0;
+                var numberOfPointsDeleted = 0;
+                var batchCount = 0;
                 var stopwatch = Stopwatch.StartNew();
 
-                AppendResponse appendResponse;
+                foreach (var batch in GetPointBatches())
+                {
+                    var result = AppendPointBatch(client, timeSeries, batch.Item1, batch.Item2, isReflected, hasTimeRange);
+                    numberOfPointsAppended += result.NumberOfPointsAppended;
+                    numberOfPointsDeleted += result.NumberOfPointsDeleted;
+                    batchCount++;
+
+                    if (result.AppendStatus != AppendStatusCode.Completed)
+                        throw new ExpectedException($"Unexpected append status={result.AppendStatus}");
+                }
+
+                var batchText = batchCount > 1 ? $" using {batchCount} appends" : "";
+                Log.Info($"Appended {numberOfPointsAppended} points (deleting {numberOfPointsDeleted} points) in {stopwatch.ElapsedMilliseconds / 1000.0:F1} seconds{batchText}.");
+            }
+        }
+
+        private TimeSeriesAppendStatus AppendPointBatch(IAquariusClient client, TimeSeries timeSeries, List<ReflectedTimeSeriesPoint> points, Interval timeRange, bool isReflected, bool hasTimeRange)
+        {
+            AppendResponse appendResponse;
+
+            if (isReflected)
+            {
+                appendResponse = client.Acquisition.Post(new PostReflectedTimeSeries
+                {
+                    UniqueId = timeSeries.UniqueId,
+                    TimeRange = timeRange,
+                    Points = points
+                });
+            }
+            else
+            {
+                var basicPoints = points
+                    .Select(p => new TimeSeriesPoint
+                    {
+                        Time = p.Time,
+                        Value = p.Value
+                    })
+                    .ToList();
 
-                if (isReflected)
+                if (hasTimeRange)
                 {
-                    appendResponse = client.Acquisition.Post(new PostReflectedTimeSeries
+                    appendResponse = client.Acquisition.Post(new PostTimeSeriesOverwriteAppend
                     {
                         UniqueId = timeSeries.UniqueId,
-                        TimeRange = GetTimeRange(),
-                        Points = Points
+                        TimeRange = timeRange,
+                        Points = basicPoints
                     });
                 }
                 else
                 {
-                    var basicPoints = Points
-                        .Select(p => new TimeSeriesPoint
-                        {
-                            Time = p.Time,
-                            Value = p.Value
-                        })
-                        .ToList();
-
-                    if (hasTimeRange)
-                    {
-                        appendResponse = client.Acquisition.Post(new PostTimeSeriesOverwriteAppend
-                        {
-                            UniqueId = timeSeries.UniqueId,
-                            TimeRange = GetTimeRange(),
-                            Points = basicPoints
-                        });
-                    }
-                    else
+                    appendResponse = client.Acquisition.Post(new PostTimeSeriesAppend
                     {
-                        appendResponse = client.Acquisition.Post(new PostTimeSeriesAppend
-                        {
-                            UniqueId = timeSeries.UniqueId,
-                            Points = basicPoints
-                        });
-                    }
+                        UniqueId = timeSeries.UniqueId,
+                        Points = basicPoints
+                    });
                 }
+            }
 
-                var result = client.Acquisition.RequestAndPollUntilComplete(
-                    acquisition => appendResponse,
-                    (acquisition, response) => acquisition.Get(new GetTimeSeriesAppendStatus { AppendRequestIdentifier = response.AppendRequestIdentifier }),
-                    polledStatus => polledStatus.AppendStatus != AppendStatusCode.Pending,
-                    null,
-                    Context.AppendTimeout);
+            return client.Acquisition.RequestAndPollUntilComplete(
+                acquisition => appendResponse,
+                (acquisition, response) => acquisition.Get(new GetTimeSeriesAppendStatus { AppendRequestIdentifier = response.AppendRequestIdentifier }),
+                polledStatus => polledStatus.AppendStatus != AppendStatusCode.Pending,
+                null,
+                Context.AppendTimeout);
+        }
+
+        private IEnumerable<Tuple<List<ReflectedTimeSeriesPoint>, Interval>> GetPointBatches()
+        {
+            var points = GetPoints();
+            var remainingTimeRange = GetTimeRange();
 
-                if (result.AppendStatus != AppendStatusCode.Completed)
-                    throw new ExpectedException($"Unexpected append status={result.AppendStatus}");
+            var index = 0;
+            while (points.Count - index > Context.BatchSize)
+            {
+                var batchPoints = points.Skip(index).Take(Context.BatchSize).ToList();
+                var batchTimeRange = new Interval(remainingTimeRange.Start, batchPoints.Last().Time.GetValueOrDefault().PlusTicks(1));
+                remainingTimeRange = new Interval(batchTimeRange.End, remainingTimeRange.End);
+
+                yield return new Tuple<List<ReflectedTimeSeriesPoint>, Interval>(batchPoints, batchTimeRange);
 
-                Log.Info($"Appended {result.NumberOfPointsAppended} points (deleting {result.NumberOfPointsDeleted} points) in {stopwatch.ElapsedMilliseconds / 1000.0:F1} seconds.");
+                index += Context.BatchSize;
             }
+
+            yield return new Tuple<List<ReflectedTimeSeriesPoint>, Interval>(points.Skip(index).ToList(), remainingTimeRange);
         }
 
         private List<ReflectedTimeSeriesPoint> GetPoints()

TimeSeries/PublicApis/SdkExamples/PointZilla/Program.cs

@@ -97,6 +97,7 @@ private static Context ParseArgs(string[] args)
 
                 new Option {Key = nameof(context.Wait), Setter = value => context.Wait = bool.Parse(value), Getter = () => context.Wait.ToString(), Description = "Wait for the append request to complete"},
                 new Option {Key = nameof(context.AppendTimeout), Setter = value => context.AppendTimeout = TimeSpan.Parse(value), Getter = () => context.AppendTimeout.ToString(), Description = "Timeout period for append completion, in .NET TimeSpan format."},
+                new Option {Key = nameof(context.BatchSize), Setter = value => context.BatchSize = int.Parse(value), Getter = () => context.BatchSize.ToString(), Description = "Maximum number of points to send in a single append request"},
 
                 new Option(), new Option {Description = "Time-series options:"},
                 new Option {Key = nameof(context.TimeSeries), Setter = value => context.TimeSeries = value, Getter = () => context.TimeSeries, Description = "Target time-series identifier or unique ID"},