using System; using System.Collections.Generic; using System.Linq; using System.Threading; using InfluxDB.Client.Core; using InfluxDB.Client.Linq.Internal; using InfluxDB.Client.Linq.Internal.NodeTypes; using Remotion.Linq; using Remotion.Linq.Parsing.Structure; using Remotion.Linq.Parsing.Structure.NodeTypeProviders; using Expression = System.Linq.Expressions.Expression; namespace InfluxDB.Client.Linq { /// /// The settings for a Query optimization. /// public class QueryableOptimizerSettings { public QueryableOptimizerSettings() { QueryMultipleTimeSeries = false; AlignFieldsWithPivot = true; AlignLimitFunctionAfterPivot = true; DropMeasurementColumn = true; DropStartColumn = true; DropStopColumn = true; RangeStartValue = null; RangeStopValue = null; } /// /// 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: /// /// Appends group operator /// Enable use default sorting: sort(columns: ["_time"], desc: false) /// /// public bool QueryMultipleTimeSeries { get; set; } /// /// Gets or set whether the drive should use a Flux pivot() function /// to align fields to tabular way. /// public bool AlignFieldsWithPivot { get; set; } /// /// Gets or set whether the drive should align limit() and tail() functions /// after pivot() function. /// public bool AlignLimitFunctionAfterPivot { get; set; } /// /// Gets or sets whether the _measurement column will be dropped from query results. /// Setting this variable to true will change how the produced Flux Query looks like: /// /// Appends drop operator /// Drops the _measurement column: drop(columns: ["_measurement"]) /// /// public bool DropMeasurementColumn { get; set; } /// /// Gets or sets whether the _start column will be dropped from query results. /// Setting this variable to true will change how the produced Flux Query looks like: /// /// Appends drop operator /// Drops the _start column: drop(columns: ["_start"]) /// /// public bool DropStartColumn { get; set; } /// /// Gets or sets whether the _stop column will be dropped from query results. /// Setting this variable to true will change how the produced Flux Query looks like: /// /// Appends drop operator /// Drops the _stop column: drop(columns: ["_stop"]) /// /// public bool DropStopColumn { get; set; } /// /// Gets or sets the default value for a start parameter of a range function. /// The `start` is earliest time to include in results. Results include points that match the specified start time. /// Defaults to `0`. /// public DateTime? RangeStartValue { get; set; } /// /// Gets or sets the default value for a stop parameter of a range function. /// The `stop` is latest time to include in results. Results exclude points that match the specified stop time. /// Defaults to `now()`. /// public DateTime? RangeStopValue { get; set; } } /// /// Main entry point to query InfluxDB by LINQ /// public class InfluxDBQueryable : QueryableBase { /// /// Create a new instance of IQueryable for synchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Settings for a Query optimization /// new instance for of Queryable public static InfluxDBQueryable Queryable(string bucket, string org, QueryApiSync queryApi, QueryableOptimizerSettings queryableOptimizerSettings = default) { return Queryable(bucket, org, queryApi, new DefaultMemberNameResolver(), queryableOptimizerSettings); } /// /// Create a new instance of IQueryable for asynchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Settings for a Query optimization /// new instance for of Queryable public static InfluxDBQueryable Queryable(string bucket, string org, QueryApi queryApi, QueryableOptimizerSettings queryableOptimizerSettings = default) { return Queryable(bucket, org, queryApi, new DefaultMemberNameResolver(), queryableOptimizerSettings); } /// /// Create a new instance of IQueryable for synchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Resolver for customized names. /// Settings for a Query optimization /// new instance for of Queryable public static InfluxDBQueryable Queryable(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default) { return new InfluxDBQueryable(bucket, org, queryApi, memberResolver, queryableOptimizerSettings); } /// /// Create a new instance of IQueryable for asynchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Resolver for customized names. /// Settings for a Query optimization /// new instance for of Queryable public static InfluxDBQueryable Queryable(string bucket, string org, QueryApi queryApi, IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default) { return new InfluxDBQueryable(bucket, org, queryApi, memberResolver, queryableOptimizerSettings); } /// /// Create a new instance of IQueryable for synchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Resolver for customized names. /// Settings for a Query optimization public InfluxDBQueryable(string bucket, string org, QueryApiSync queryApi, IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default) : base(CreateQueryParser(), CreateExecutor(bucket, org, queryApi, memberResolver, queryableOptimizerSettings)) { } /// /// Create a new instance of IQueryable for asynchronous Queries. /// /// Specifies the source bucket. /// Specifies the source organization. /// The underlying API to execute Flux Query. /// Resolver for customized names. /// Settings for a Query optimization public InfluxDBQueryable(string bucket, string org, QueryApi queryApi, IMemberNameResolver memberResolver, QueryableOptimizerSettings queryableOptimizerSettings = default) : base(CreateQueryParser(), CreateExecutor(bucket, org, queryApi, memberResolver, queryableOptimizerSettings)) { } /// /// Call by ReLinq. /// /// /// public InfluxDBQueryable(IQueryProvider provider, Expression expression) : base(provider, expression) { } /// /// Create a object that will be used for Querying. /// /// Query that will be used to Querying public Api.Domain.Query ToDebugQuery() { var provider = Provider as DefaultQueryProvider; var executor = provider?.Executor as InfluxDBQueryExecutor; if (executor == null) { throw new NotSupportedException("InfluxDBQueryable should use InfluxDBQueryExecutor"); } var parsedQuery = provider.QueryParser.GetParsedQuery(Expression); var generateQuery = executor.GenerateQuery(parsedQuery, out _); return generateQuery; } 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, queryableOptimizerSettings ?? new QueryableOptimizerSettings()); } private static IQueryExecutor CreateExecutor(string bucket, string org, QueryApi 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, queryableOptimizerSettings ?? new QueryableOptimizerSettings()); } internal static QueryParser CreateQueryParser() { var queryParser = QueryParser.CreateDefault(); var compoundNodeTypeProvider = queryParser.NodeTypeProvider as CompoundNodeTypeProvider; compoundNodeTypeProvider?.InnerProviders.Add(new InfluxDBNodeTypeProvider()); return queryParser; } public IAsyncEnumerable GetAsyncEnumerator(CancellationToken cancellationToken = default) { var provider = Provider as DefaultQueryProvider; if (!(provider?.Executor is InfluxDBQueryExecutor executor)) { throw new NotSupportedException("InfluxDBQueryable should use InfluxDBQueryExecutor"); } var parsedQuery = provider.QueryParser.GetParsedQuery(Expression); return executor.ExecuteCollectionAsync(parsedQuery, cancellationToken); } } public static class QueryableExtensions { public static InfluxDBQueryable ToInfluxQueryable(this IQueryable source) { if (source == null) { throw new InvalidCastException("Queryable source is null"); } if (!(source is InfluxDBQueryable queryable)) { throw new InvalidCastException("Queryable should be InfluxDBQueryable"); } return queryable; } /// /// The extension to use Flux window operator. For more info see https://docs.influxdata.com/flux/v0.x/stdlib/universe/aggregatewindow/ /// /// /// /// var query = from s in InfluxDBQueryable<Sensor>.Queryable("my-bucket", "my-org", _queryApi) /// where s.Timestamp.AggregateWindow(TimeSpan.FromSeconds(20), TimeSpan.FromSeconds(40), "mean") /// where s.Value == 5 /// select s; /// /// /// /// The entity value which is market as a Timestamp. /// Duration of time between windows. /// Duration of the window. /// Aggregate or selector function used to operate on each window of time. /// NotSupportedException if it's called outside LINQ expression. /// Caused by calling outside of LINQ expression. // ReSharper disable UnusedParameter.Global public static bool AggregateWindow(this DateTime timestamp, TimeSpan every, TimeSpan? period = null, string fn = "mean") { throw new NotSupportedException("This should be used only in LINQ expression. " + "Something like: 'where s.Timestamp.AggregateWindow(TimeSpan.FromSeconds(20), TimeSpan.FromSeconds(40), \"mean\")'."); } } }