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\")'.");
}
}
}