-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
optimized streaming, handling error on sortmarking while doing parall…
…el querying to the same file, added Loaj (load from json)
- Loading branch information
Showing
6 changed files
with
165 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| using FKala.Core.DataLayer.Infrastructure; | ||
| using FKala.Core.Interfaces; | ||
| using FKala.Core.KalaQl.Windowing; | ||
| using FKala.Core.Model; | ||
| using Newtonsoft.Json; | ||
| using Newtonsoft.Json.Linq; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Text; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace FKala.Core.KalaQl | ||
| { | ||
| public class Op_JsonQuery : Op_Base, IKalaQlOperation | ||
| { | ||
| public string Name { get; } | ||
| public string Measurement { get; } | ||
| public DateTime StartTime { get; } | ||
| public DateTime EndTime { get; } | ||
| public CacheResolution CacheResolution { get; } | ||
| public bool NewestOnly { get; } | ||
| public bool DoSortRawFiles { get; } | ||
| public string FieldPath { get; } | ||
|
|
||
| public Op_JsonQuery(string? line, string name, string measurement, string fieldPath, DateTime startTime, DateTime endTime, CacheResolution cacheResolution, bool newestOnly = false, bool doSortRawFiles = false) : base(line) | ||
| { | ||
| this.Name = name; | ||
| this.Measurement = measurement; | ||
| this.StartTime = startTime; | ||
| this.EndTime = endTime; | ||
| this.CacheResolution = cacheResolution; | ||
| this.NewestOnly = newestOnly; | ||
| this.DoSortRawFiles = doSortRawFiles; | ||
| this.FieldPath = fieldPath; | ||
| } | ||
|
|
||
| public override bool CanExecute(KalaQlContext context) | ||
| { | ||
| return true; | ||
| } | ||
|
|
||
| public override void Execute(KalaQlContext context) | ||
| { | ||
| //var result = context.DataLayer.LoadData(this.Measurement, this.StartTime, this.EndTime, CacheResolution, NewestOnly, DoSortRawFiles); | ||
|
|
||
| //TODODODODODODDODODO REBUILD IST SOMIT DEFEKT | ||
| // bei ForceRebuild auch ohne Ausgabe etc. den Rebuild durchführen, ..was erst geschieht beim Materialisieren | ||
| //if (CacheResolution.ForceRebuild) result = result.ToList(); | ||
|
|
||
| var pathParts = FieldPath.Split("/"); | ||
|
|
||
| context.IntermediateDatasources.Add( | ||
| new ResultPromise() | ||
| { | ||
| Name = this.Name, | ||
| Query_StartTime = StartTime, | ||
| Query_EndTime = EndTime, | ||
| Creator = this, | ||
| ResultsetFactory = () => | ||
| { | ||
| var result = context.DataLayer.LoadData(this.Measurement, this.StartTime, this.EndTime, CacheResolution, NewestOnly, DoSortRawFiles, context); | ||
| return ReadJson(result, pathParts); | ||
| } | ||
| }); | ||
| this.hasExecuted = true; | ||
| } | ||
|
|
||
| public IEnumerable<DataPoint> ReadJson(IEnumerable<DataPoint> jsonEnum, string[]? pathParts) | ||
| { | ||
| int index = -1; | ||
| if (pathParts[pathParts.Length - 1].EndsWith("]")) | ||
| { | ||
| var ps = pathParts[pathParts.Length - 1].Split('['); | ||
| pathParts[pathParts.Length - 1] = ps[0]; | ||
| index = int.Parse(ps[1].TrimEnd(']')); | ||
| } | ||
| foreach (var item in jsonEnum) | ||
| { | ||
| var jsonDict = JsonConvert.DeserializeObject<Dictionary<string, object>>(item.ValueText); | ||
| foreach (var pathPart in pathParts) | ||
| { | ||
| if (jsonDict.ContainsKey(pathPart)) | ||
| { | ||
| object jsonDictEntry = jsonDict[pathPart]; | ||
| if (jsonDictEntry is JToken) | ||
| { | ||
| if ((jsonDictEntry as JToken).Type == JTokenType.Object) | ||
| { | ||
| jsonDict = (jsonDictEntry as JObject).ToObject<Dictionary<string, object>>(); | ||
| } | ||
| else if ((jsonDictEntry as JToken).Type == JTokenType.Array) | ||
| { | ||
| var jarray = (jsonDictEntry as JArray); | ||
| var dp = Pools.DataPoint.Get(); | ||
| dp.StartTime = item.StartTime; | ||
| dp.EndTime = item.EndTime; | ||
| dp.Source = item.Source; | ||
| dp.ValueText = jarray.ToList()[index].ToString(); | ||
| yield return dp; | ||
| } | ||
| } | ||
| else | ||
| { | ||
| if (decimal.TryParse(jsonDictEntry.ToString(), out decimal decimalValue)) | ||
| { | ||
| var dp = Pools.DataPoint.Get(); | ||
| dp.StartTime = item.StartTime; | ||
| dp.EndTime = item.EndTime; | ||
| dp.Source = item.Source; | ||
| dp.Value = decimalValue; | ||
| yield return dp; | ||
| } | ||
| else | ||
| { | ||
| var dp = Pools.DataPoint.Get(); | ||
| dp.StartTime = item.StartTime; | ||
| dp.EndTime = item.EndTime; | ||
| dp.Source = item.Source; | ||
| dp.ValueText = jsonDictEntry.ToString(); | ||
| yield return dp; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |