feat(linq): optimize Flux Query for querying one time-series (#197)

bednar · web-flow · commit 5760e0f89b22 · 2021-05-07T10:13:48.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ### Features
 1. [#194](https://github.com/influxdata/influxdb-client-csharp/pull/194): Add possibility to handle HTTP response from InfluxDB server [write]
+1. [#197](https://github.com/influxdata/influxdb-client-csharp/pull/197): Optimize Flux Query for querying one time-series [LINQ]
 
 ### Bug Fixes
 1. [#193](https://github.com/influxdata/influxdb-client-csharp/pull/193): Create services without API implementation
diff --git a/Client.Linq.Test/InfluxDBQueryVisitorTest.cs b/Client.Linq.Test/InfluxDBQueryVisitorTest.cs
diff --git a/Client.Linq.Test/ItInfluxDBQueryableTest.cs b/Client.Linq.Test/ItInfluxDBQueryableTest.cs
@@ -92,6 +92,18 @@ public void QueryTake()
 
             var sensors = query.ToList();
 
+            Assert.AreEqual(2*2, sensors.Count);
+        }
+
+        [Test]
+        public void QueryTakeMultipleTimeSeries()
+        {
+            var query = (from s in InfluxDBQueryable<Sensor>.Queryable("my-bucket", "my-org", _client.GetQueryApiSync(),
+                    new QueryableOptimizerSettings {QueryMultipleTimeSeries = true})
+                select s).Take(2);
+
+            var sensors = query.ToList();
+
             Assert.AreEqual(2, sensors.Count);
         }
 
diff --git a/Client.Linq.Test/QueryAggregatorTest.cs b/Client.Linq.Test/QueryAggregatorTest.cs
@@ -1,4 +1,5 @@
 using InfluxDB.Client.Core.Test;
+using InfluxDB.Client.Linq;
 using InfluxDB.Client.Linq.Internal;
 using NUnit.Framework;
 
@@ -76,10 +77,10 @@ public void Range()
                                "from(bucket: p1) " +
                                $"|> {range} " +
                                "|> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\") " +
-                               "|> drop(columns: [\"_start\", \"_stop\", \"_measurement\"]) " +
-                               "|> group()";
+                               "|> drop(columns: [\"_start\", \"_stop\", \"_measurement\"])";
 
-                Assert.AreEqual(expected, _aggregator.BuildFluxQuery(), $"Expected Range: {range}, Shift: {shift}");
+                var settings = new QueryableOptimizerSettings();
+                Assert.AreEqual(expected, _aggregator.BuildFluxQuery(settings), $"Expected Range: {range}, Shift: {shift}");
             }
         }
     }
diff --git a/Client.Linq/InfluxDBQueryable.cs b/Client.Linq/InfluxDBQueryable.cs
@@ -8,6 +8,27 @@
 
 namespace InfluxDB.Client.Linq
 {
+    /// <summary>
+    /// The settings for a Query optimization.
+    /// </summary>
+    public class QueryableOptimizerSettings
+    {
+        public QueryableOptimizerSettings()
+        {
+            QueryMultipleTimeSeries = false;
+        }
+        
+        /// <summary>
+        /// Gets or sets whether the drive is used to query multiple time series.
+        /// Setting this variable to true will change how the produced Flux Query looks like:
+        /// <list type="bullet">
+        /// <item>appends <a href="https://docs.influxdata.com/influxdb/v2.0/reference/flux/stdlib/built-in/transformations/group/">group operator</a></item>
+        /// <item>enable use default sorting: <i>sort(columns: ["_time"], desc: false)</i></item>
+        /// </list>
+        /// </summary>
+        public bool QueryMultipleTimeSeries { get; set; }
+    }
+
     /// <summary>
     /// Main entry point to query InfluxDB by LINQ
     /// </summary>
@@ -19,34 +40,40 @@ public class InfluxDBQueryable<T> : QueryableBase<T>
         /// <param name="bucket">Specifies the source bucket.</param>
         /// <param name="org">Specifies the source organization.</param>
         /// <param name="queryApi">The underlying API to execute Flux Query.</param>
+        /// <param name="queryableOptimizerSettings">Settings for a Query optimization</param>
         /// <returns>new instance for of Queryable</returns>
-        public static InfluxDBQueryable<T> Queryable(string bucket, string org, QueryApiSync queryApi)
+        public static InfluxDBQueryable<T> Queryable(string bucket, string org, QueryApiSync queryApi,
+            QueryableOptimizerSettings queryableOptimizerSettings = default)
         {
-            return Queryable(bucket, org, queryApi, new DefaultMemberNameResolver());
+            return Queryable(bucket, org, queryApi, new DefaultMemberNameResolver(), queryableOptimizerSettings);
         }
-        
+
         /// <summary>
         /// Create a new instance of IQueryable.
         /// </summary>
         /// <param name="bucket">Specifies the source bucket.</param>
         /// <param name="org">Specifies the source organization.</param>
         /// <param name="queryApi">The underlying API to execute Flux Query.</param>
         /// <param name="memberResolver">Resolver for customized names.</param>
+        /// <param name="queryableOptimizerSettings">Settings for a Query optimization</param>
         /// <returns>new instance for of Queryable</returns>
-        public static InfluxDBQueryable<T> Queryable(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver)
+        public static InfluxDBQueryable<T> Queryable(string bucket, string org, QueryApiSync queryApi,
+            IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default)
         {
-            return new InfluxDBQueryable<T>(bucket, org, queryApi, memberResolver);
+            return new InfluxDBQueryable<T>(bucket, org, queryApi, memberResolver, queryableOptimizerSettings);
         }
-        
+
         /// <summary>
         /// Create a new instance of IQueryable.
         /// </summary>
         /// <param name="bucket">Specifies the source bucket.</param>
         /// <param name="org">Specifies the source organization.</param>
         /// <param name="queryApi">The underlying API to execute Flux Query.</param>
         /// <param name="memberResolver">Resolver for customized names.</param>
-        public InfluxDBQueryable(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver) : base(CreateQueryParser(),
-            CreateExecutor(bucket, org, queryApi, memberResolver))
+        /// <param name="queryableOptimizerSettings">Settings for a Query optimization</param>
+        public InfluxDBQueryable(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver,
+            QueryableOptimizerSettings queryableOptimizerSettings = default) : base(CreateQueryParser(),
+            CreateExecutor(bucket, org, queryApi, memberResolver, queryableOptimizerSettings))
         {
         }
 
@@ -77,18 +104,20 @@ public Api.Domain.Query ToDebugQuery()
             return generateQuery;
         }
 
-        private static IQueryExecutor CreateExecutor(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver)
+        private static IQueryExecutor CreateExecutor(string bucket, string org, QueryApiSync queryApi,
+            IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default)
         {
             Arguments.CheckNonEmptyString(bucket, nameof(bucket));
             Arguments.CheckNonEmptyString(org, nameof(org));
             Arguments.CheckNotNull(queryApi, nameof(queryApi));
-                
-            return new InfluxDBQueryExecutor(bucket, org, queryApi, memberResolver);
+
+            return new InfluxDBQueryExecutor(bucket, org, queryApi, memberResolver,
+                queryableOptimizerSettings ?? new QueryableOptimizerSettings());
         }
 
         private static QueryParser CreateQueryParser()
         {
             return QueryParser.CreateDefault();
         }
     }
-}
+}
diff --git a/Client.Linq/Internal/QueryAggregator.cs b/Client.Linq/Internal/QueryAggregator.cs
@@ -64,15 +64,15 @@ internal class QueryAggregator
         private ResultFunction _resultFunction;
         private readonly List<string> _filterByTags;
         private readonly List<string> _filterByFields;
-        private readonly List<(string, string)> _orders;
+        private readonly List<(string, string, bool, string)> _orders;
 
         internal QueryAggregator()
         {
             _resultFunction = ResultFunction.None;
             _limitNOffsetAssignments = new List<LimitOffsetAssignment>();
             _filterByTags = new List<string>();
             _filterByFields = new List<string>();
-            _orders = new List<(string, string)>();
+            _orders = new List<(string, string, bool, string)>();
         }
 
         internal void AddBucket(string bucket)
@@ -133,9 +133,9 @@ internal void AddSubQueries(QueryAggregator aggregator)
             _orders.AddRange(aggregator._orders);
         }
 
-        internal void AddOrder(string orderPart, string desc)
+        internal void AddOrder(string column, string columnVariable, bool descending, string descendingVariable)
         {
-            _orders.Add((orderPart, desc));
+            _orders.Add((column, columnVariable, descending, descendingVariable));
         }
 
         internal void AddResultFunction(ResultFunction resultFunction)
@@ -145,8 +145,10 @@ internal void AddResultFunction(ResultFunction resultFunction)
             _resultFunction = resultFunction;
         }
 
-        internal string BuildFluxQuery()
+        internal string BuildFluxQuery(QueryableOptimizerSettings settings)
         {
+            Arguments.CheckNotNull(settings, nameof(settings));
+            
             var transforms = new List<string>();
             var parts = new List<string>
             {
@@ -155,14 +157,20 @@ internal string BuildFluxQuery()
                 BuildFilter(_filterByTags),
                 "pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")",
                 "drop(columns: [\"_start\", \"_stop\", \"_measurement\"])",
-                "group()",
+                settings.QueryMultipleTimeSeries ? "group()" : "",
                 BuildFilter(_filterByFields)
             };
 
             // https://docs.influxdata.com/influxdb/cloud/reference/flux/stdlib/built-in/transformations/sort/
-            foreach (var (column, desc) in _orders)
+            foreach (var ((column, columnVariable, descending, descendingVariable), index) in _orders.Select((value, i) => (value, i)))
             {
-                parts.Add(BuildOperator("sort", "columns", new List<string> {column}, "desc", desc));
+                // skip default sorting if don't query to multiple time series
+                if (!settings.QueryMultipleTimeSeries && index == 0 && column == "_time" && !descending)
+                {
+                    continue;
+                }
+                
+                parts.Add(BuildOperator("sort", "columns", new List<string> {columnVariable}, "desc", descendingVariable));
             }
 
             // https://docs.influxdata.com/influxdb/cloud/reference/flux/stdlib/built-in/transformations/limit/
diff --git a/Client.Linq/Internal/QueryExecutor.cs b/Client.Linq/Internal/QueryExecutor.cs
@@ -21,19 +21,24 @@ internal class InfluxDBQueryExecutor : IQueryExecutor
         private readonly string _org;
         private readonly QueryApiSync _queryApi;
         private readonly IMemberNameResolver _memberResolver;
+        private readonly QueryableOptimizerSettings _queryableOptimizerSettings;
 
         /// <summary>
         /// 
         /// </summary>
         /// <param name="bucket">Specifies the source bucket.</param>
         /// <param name="org">Specifies the source organization.</param>
         /// <param name="queryApi">The underlying API to execute Flux Query.</param>
-        public InfluxDBQueryExecutor(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver)
+        /// <param name="memberResolver">Resolver for customized names.</param>
+        /// <param name="queryableOptimizerSettings">Settings for a Query optimization</param>
+        public InfluxDBQueryExecutor(string bucket, string org, QueryApiSync queryApi,
+            IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings)
         {
             _bucket = bucket;
             _org = org;
             _queryApi = queryApi;
             _memberResolver = memberResolver;
+            _queryableOptimizerSettings = queryableOptimizerSettings;
         }
 
         /// <summary>
@@ -85,11 +90,22 @@ public IEnumerable<T> ExecuteCollection<T>(QueryModel queryModel)
         /// <returns>Query to Invoke</returns>
         internal Query GenerateQuery(QueryModel queryModel, out QueryResultsSettings settings)
         {
-            var visitor = new InfluxDBQueryVisitor(_bucket, _memberResolver);
-            visitor.VisitQueryModel(queryModel);
+            var visitor = QueryVisitor(queryModel);
 
             settings = new QueryResultsSettings(queryModel);
             return visitor.GenerateQuery();
         }
+
+        /// <summary>
+        /// Create QueryVisitor for specified model.
+        /// </summary>
+        /// <param name="queryModel">Query Model</param>
+        /// <returns>Query Visitor</returns>
+        internal InfluxDBQueryVisitor QueryVisitor(QueryModel queryModel)
+        {
+            var visitor = new InfluxDBQueryVisitor(_bucket, _memberResolver, _queryableOptimizerSettings);
+            visitor.VisitQueryModel(queryModel);
+            return visitor;
+        }
     }
 }
diff --git a/Client.Linq/Internal/QueryGenerationContext.cs b/Client.Linq/Internal/QueryGenerationContext.cs
@@ -5,18 +5,23 @@ internal class QueryGenerationContext
         internal readonly QueryAggregator QueryAggregator;
         internal readonly IMemberNameResolver MemberResolver;
         internal readonly VariableAggregator Variables;
+        internal readonly QueryableOptimizerSettings QueryableOptimizerSettings;
 
-        internal QueryGenerationContext(QueryAggregator queryAggregator, VariableAggregator variableAggregator,
-            IMemberNameResolver memberResolver)
+        internal QueryGenerationContext(
+            QueryAggregator queryAggregator, 
+            VariableAggregator variableAggregator,
+            IMemberNameResolver memberResolver, 
+            QueryableOptimizerSettings queryableOptimizerSettings)
         {
             QueryAggregator = queryAggregator;
             Variables = variableAggregator;
             MemberResolver = memberResolver;
+            QueryableOptimizerSettings = queryableOptimizerSettings;
         }
 
         internal QueryGenerationContext Clone(QueryAggregator queryAggregator)
         {
-            return new QueryGenerationContext(queryAggregator, Variables, MemberResolver);
+            return new QueryGenerationContext(queryAggregator, Variables, MemberResolver, QueryableOptimizerSettings);
         }
     }
 }
diff --git a/Client.Linq/Internal/QueryVisitor.cs b/Client.Linq/Internal/QueryVisitor.cs
@@ -17,8 +17,11 @@ internal class InfluxDBQueryVisitor : QueryModelVisitorBase
     {
         private readonly QueryGenerationContext _context;
 
-        internal InfluxDBQueryVisitor(string bucket, IMemberNameResolver memberResolver) : 
-            this(new QueryGenerationContext(new QueryAggregator(), new VariableAggregator(), memberResolver))
+        internal InfluxDBQueryVisitor(
+            string bucket, 
+            IMemberNameResolver memberResolver,
+            QueryableOptimizerSettings queryableOptimizerSettings) :
+            this(new QueryGenerationContext(new QueryAggregator(), new VariableAggregator(), memberResolver, queryableOptimizerSettings))
         {
             var bucketVariable = _context.Variables.AddNamedVariable(bucket);
             _context.QueryAggregator.AddBucket(bucketVariable);
@@ -59,7 +62,7 @@ internal File BuildFluxAST()
 
         internal string BuildFluxQuery()
         {
-            return _context.QueryAggregator.BuildFluxQuery();
+            return _context.QueryAggregator.BuildFluxQuery(_context.QueryableOptimizerSettings);
         }
 
         public override void VisitWhereClause(WhereClause whereClause, QueryModel queryModel, int index)
@@ -155,11 +158,11 @@ public override void VisitOrderByClause(OrderByClause orderByClause, QueryModel
 
             foreach (var ordering in orderByClause.Orderings)
             {
-                var orderPart = _context.Variables
-                    .AddNamedVariable(GetFluxExpression(ordering.Expression, orderByClause));
-                var desc = _context.Variables
-                    .AddNamedVariable(ordering.OrderingDirection == OrderingDirection.Desc);
-                _context.QueryAggregator.AddOrder(orderPart, desc);
+                var column = GetFluxExpression(ordering.Expression, orderByClause);
+                var columnVariable = _context.Variables.AddNamedVariable(column);
+                var descending = ordering.OrderingDirection == OrderingDirection.Desc;
+                var descendingVariable = _context.Variables.AddNamedVariable(descending);
+                _context.QueryAggregator.AddOrder(column, columnVariable, descending, descendingVariable);
             }
         }
 
diff --git a/Client.Linq/README.md b/Client.Linq/README.md

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`using InfluxDB.Client.Core.Test;`
	`2`	`+using InfluxDB.Client.Linq;`
`2`	`3`	`using InfluxDB.Client.Linq.Internal;`
`3`	`4`	`using NUnit.Framework;`
`4`	`5`
`@@ -76,10 +77,10 @@ public void Range()`
`76`	`77`	`"from(bucket: p1) " +`
`77`	`78`	`$"\|> {range} " +`
`78`	`79`	`"\|> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\") " +`
`79`		`- "\|> drop(columns: [\"_start\", \"_stop\", \"_measurement\"]) " +`
`80`		`- "\|> group()";`
	`80`	`+ "\|> drop(columns: [\"_start\", \"_stop\", \"_measurement\"])";`
`81`	`81`
`82`		`- Assert.AreEqual(expected, _aggregator.BuildFluxQuery(), $"Expected Range: {range}, Shift: {shift}");`
	`82`	`+ var settings = new QueryableOptimizerSettings();`
	`83`	`+ Assert.AreEqual(expected, _aggregator.BuildFluxQuery(settings), $"Expected Range: {range}, Shift: {shift}");`
`83`	`84`	`}`
`84`	`85`	`}`
`85`	`86`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,18 +5,23 @@ internal class QueryGenerationContext`
`5`	`5`	`internal readonly QueryAggregator QueryAggregator;`
`6`	`6`	`internal readonly IMemberNameResolver MemberResolver;`
`7`	`7`	`internal readonly VariableAggregator Variables;`
	`8`	`+ internal readonly QueryableOptimizerSettings QueryableOptimizerSettings;`
`8`	`9`
`9`		`- internal QueryGenerationContext(QueryAggregator queryAggregator, VariableAggregator variableAggregator,`
`10`		`- IMemberNameResolver memberResolver)`
	`10`	`+ internal QueryGenerationContext(`
	`11`	`+ QueryAggregator queryAggregator,`
	`12`	`+ VariableAggregator variableAggregator,`
	`13`	`+ IMemberNameResolver memberResolver,`
	`14`	`+ QueryableOptimizerSettings queryableOptimizerSettings)`
`11`	`15`	`{`
`12`	`16`	`QueryAggregator = queryAggregator;`
`13`	`17`	`Variables = variableAggregator;`
`14`	`18`	`MemberResolver = memberResolver;`
	`19`	`+ QueryableOptimizerSettings = queryableOptimizerSettings;`
`15`	`20`	`}`
`16`	`21`
`17`	`22`	`internal QueryGenerationContext Clone(QueryAggregator queryAggregator)`
`18`	`23`	`{`
`19`		`- return new QueryGenerationContext(queryAggregator, Variables, MemberResolver);`
	`24`	`+ return new QueryGenerationContext(queryAggregator, Variables, MemberResolver, QueryableOptimizerSettings);`
`20`	`25`	`}`
`21`	`26`	`}`
`22`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,8 +17,11 @@ internal class InfluxDBQueryVisitor : QueryModelVisitorBase`
`17`	`17`	`{`
`18`	`18`	`private readonly QueryGenerationContext _context;`
`19`	`19`
`20`		`- internal InfluxDBQueryVisitor(string bucket, IMemberNameResolver memberResolver) :`
`21`		`- this(new QueryGenerationContext(new QueryAggregator(), new VariableAggregator(), memberResolver))`
	`20`	`+ internal InfluxDBQueryVisitor(`
	`21`	`+ string bucket,`
	`22`	`+ IMemberNameResolver memberResolver,`
	`23`	`+ QueryableOptimizerSettings queryableOptimizerSettings) :`
	`24`	`+ this(new QueryGenerationContext(new QueryAggregator(), new VariableAggregator(), memberResolver, queryableOptimizerSettings))`
`22`	`25`	`{`
`23`	`26`	`var bucketVariable = _context.Variables.AddNamedVariable(bucket);`
`24`	`27`	`_context.QueryAggregator.AddBucket(bucketVariable);`
`@@ -59,7 +62,7 @@ internal File BuildFluxAST()`
`59`	`62`
`60`	`63`	`internal string BuildFluxQuery()`
`61`	`64`	`{`
`62`		`- return _context.QueryAggregator.BuildFluxQuery();`
	`65`	`+ return _context.QueryAggregator.BuildFluxQuery(_context.QueryableOptimizerSettings);`
`63`	`66`	`}`
`64`	`67`
`65`	`68`	`public override void VisitWhereClause(WhereClause whereClause, QueryModel queryModel, int index)`
`@@ -155,11 +158,11 @@ public override void VisitOrderByClause(OrderByClause orderByClause, QueryModel`
`155`	`158`
`156`	`159`	`foreach (var ordering in orderByClause.Orderings)`
`157`	`160`	`{`
`158`		`- var orderPart = _context.Variables`
`159`		`- .AddNamedVariable(GetFluxExpression(ordering.Expression, orderByClause));`
`160`		`- var desc = _context.Variables`
`161`		`- .AddNamedVariable(ordering.OrderingDirection == OrderingDirection.Desc);`
`162`		`- _context.QueryAggregator.AddOrder(orderPart, desc);`
	`161`	`+ var column = GetFluxExpression(ordering.Expression, orderByClause);`
	`162`	`+ var columnVariable = _context.Variables.AddNamedVariable(column);`
	`163`	`+ var descending = ordering.OrderingDirection == OrderingDirection.Desc;`
	`164`	`+ var descendingVariable = _context.Variables.AddNamedVariable(descending);`
	`165`	`+ _context.QueryAggregator.AddOrder(column, columnVariable, descending, descendingVariable);`
`163`	`166`	`}`
`164`	`167`	`}`
`165`	`168`