microsoft
/
Trill
зеркало из https://github.com/microsoft/Trill.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

Merge pull request #73 from Microsoft/Provider 

Add XML documentation, and fix an issue in EquiJoin code generation
This commit is contained in:
James Terwilliger 2019-05-01 10:00:44 -07:00 коммит произвёл GitHub
Родитель 4bd6a5c146 a31556f533
Коммит c3f4d48dfa
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
17 изменённых файлов: 122 добавлений и 115 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

10
Sources/Core/Microsoft.StreamProcessing.Provider/IGroupedWindow.cs
Просмотреть файл

@ -6,19 +6,19 @@
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// A streaming analogue to the framework interface IGrouping
/// </summary> /// </summary>
/// <typeparam name="TKey"></typeparam> /// <typeparam name="TKey">The type of the grouping key</typeparam>
/// <typeparam name="TPayload"></typeparam> /// <typeparam name="TPayload">The type of the payload being grouped</typeparam>
public interface IGroupedWindow<out TKey, out TPayload> public interface IGroupedWindow<out TKey, out TPayload>
{ {
/// <summary> /// <summary>
/// /// A concrete reference to the key of the group
/// </summary> /// </summary>
TKey Key { get; } TKey Key { get; }
/// <summary> /// <summary>
/// /// A reference to the payload data contained in the group
/// </summary> /// </summary>
IWindow<TPayload> Window { get; } IWindow<TPayload> Window { get; }
} }

8
Sources/Core/Microsoft.StreamProcessing.Provider/IQStreamable.cs
Просмотреть файл

@ -2,23 +2,23 @@
// Copyright (c) Microsoft Corporation. All rights reserved. // Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License // Licensed under the MIT License
// ********************************************************************* // *********************************************************************
using System;
using System.Linq.Expressions; using System.Linq.Expressions;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// The core interface of the streaming provider API, representing a temporal or time-series stream of data
/// </summary> /// </summary>
/// <typeparam name="TPayload">The type of the payload of the underlying stream query.</typeparam>
public interface IQStreamable<out TPayload> public interface IQStreamable<out TPayload>
{ {
/// <summary> /// <summary>
/// /// The expression representing the full query so far in the pipeline.
/// </summary> /// </summary>
Expression Expression { get; } Expression Expression { get; }
/// <summary> /// <summary>
/// /// The assigned provider whose job it is to evaluate the query once it is constructed.
/// </summary> /// </summary>
IQStreamableProvider Provider { get; } IQStreamableProvider Provider { get; }
} }

18
Sources/Core/Microsoft.StreamProcessing.Provider/IQStreamableProvider.cs
Просмотреть файл

@ -7,24 +7,24 @@ using System.Linq.Expressions;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// Provides the base interface for the implementation of a stream provider.
/// </summary> /// </summary>
public interface IQStreamableProvider public interface IQStreamableProvider
{ {
/// <summary> /// <summary>
/// /// This method represents a link in the chain for building new IQStreamable queries.
/// </summary> /// </summary>
/// <typeparam name="TElement"></typeparam> /// <typeparam name="TElement">The type of the underlying payload.</typeparam>
/// <param name="expression"></param> /// <param name="expression">The expression representing the streaming query that has been created.</param>
/// <returns></returns> /// <returns>A new IQStreamable object from which the query can continue to be built or evaluated.</returns>
IQStreamable<TElement> CreateQuery<TElement>(Expression expression); IQStreamable<TElement> CreateQuery<TElement>(Expression expression);
/// <summary> /// <summary>
/// /// The method that is called when it is time to evaluate the constructed query.
/// </summary> /// </summary>
/// <typeparam name="TResult"></typeparam> /// <typeparam name="TResult">The type of the result of the streaming query.</typeparam>
/// <param name="expression"></param> /// <param name="expression">The expression to be evaluated and run.</param>
/// <returns></returns> /// <returns>The result of evaluating the query expression.</returns>
TResult Execute<TResult>(Expression expression); TResult Execute<TResult>(Expression expression);
} }
} }

6
Sources/Core/Microsoft.StreamProcessing.Provider/IWindow.cs
Просмотреть файл

@ -6,13 +6,13 @@
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// A window of data in a grouped window, similar to the group in an IGrouping.
/// </summary> /// </summary>
/// <typeparam name="TPayload"></typeparam> /// <typeparam name="TPayload">The type of the underlying payload in the window of an IGroupedWindow.</typeparam>
public interface IWindow<out TPayload> public interface IWindow<out TPayload>
{ {
/// <summary> /// <summary>
/// /// Take the count of the number of elements in the current window.
/// </summary> /// </summary>
int Count(); int Count();
} }

8
Sources/Core/Microsoft.StreamProcessing.Provider/QStreamable.cs
Просмотреть файл

@ -7,9 +7,9 @@ using System.Linq.Expressions;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// A concrete implementation of the IQStreamable interface for building streaming query expressions.
/// </summary> /// </summary>
/// <typeparam name="TPayload"></typeparam> /// <typeparam name="TPayload">The type of the payload of the underlying stream query.</typeparam>
public sealed class QStreamable<TPayload> : IQStreamable<TPayload> public sealed class QStreamable<TPayload> : IQStreamable<TPayload>
{ {
internal QStreamable(Expression expression, IQStreamableProvider provider) internal QStreamable(Expression expression, IQStreamableProvider provider)
@ -19,12 +19,12 @@ namespace Microsoft.StreamProcessing.Provider
} }
/// <summary> /// <summary>
/// /// The expression representing the full query so far in the pipeline.
/// </summary> /// </summary>
public Expression Expression { get; } public Expression Expression { get; }
/// <summary> /// <summary>
/// /// The assigned provider whose job it is to evaluate the query once it is constructed.
/// </summary> /// </summary>
public IQStreamableProvider Provider { get; } public IQStreamableProvider Provider { get; }
} }

18
Sources/Core/Microsoft.StreamProcessing.Provider/QStreamableProviderBase.cs
Просмотреть файл

@ -8,25 +8,25 @@ using System.Linq.Expressions;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// Provides the base implementation for a Stream provider.
/// </summary> /// </summary>
public abstract class QStreamableProviderBase : IQStreamableProvider public abstract class QStreamableProviderBase : IQStreamableProvider
{ {
/// <summary> /// <summary>
/// /// This method represents a link in the chain for building new IQStreamable queries.
/// </summary> /// </summary>
/// <typeparam name="TElement"></typeparam> /// <typeparam name="TElement">The type of the underlying payload.</typeparam>
/// <param name="expression"></param> /// <param name="expression">The expression representing the streaming query that has been created.</param>
/// <returns></returns> /// <returns>A new IQStreamable object from which the query can continue to be built or evaluated.</returns>
public IQStreamable<TElement> CreateQuery<TElement>(Expression expression) public IQStreamable<TElement> CreateQuery<TElement>(Expression expression)
=> new QStreamable<TElement>(expression, this); => new QStreamable<TElement>(expression, this);
/// <summary> /// <summary>
/// /// The method that is called when it is time to evaluate the constructed query.
/// </summary> /// </summary>
/// <typeparam name="TResult"></typeparam> /// <typeparam name="TResult">The type of the result of the streaming query.</typeparam>
/// <param name="expression"></param> /// <param name="expression">The expression to be evaluated and run.</param>
/// <returns></returns> /// <returns>The result of evaluating the query expression.</returns>
public TResult Execute<TResult>(Expression expression) => throw new NotImplementedException(); public TResult Execute<TResult>(Expression expression) => throw new NotImplementedException();
} }
} }

2
Sources/Core/Microsoft.StreamProcessing.Provider/QStreamableStatic.cs
Просмотреть файл

@ -9,7 +9,7 @@ using System.Reflection;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// The extension methods over interface IQStreamable
/// </summary> /// </summary>
public static class QStreamableStatic public static class QStreamableStatic
{ {

44
Sources/Core/Microsoft.StreamProcessing.Provider/QStreamableVisitor.cs
Просмотреть файл

@ -8,7 +8,7 @@ using System.Reflection;
namespace Microsoft.StreamProcessing.Provider namespace Microsoft.StreamProcessing.Provider
{ {
/// <summary> /// <summary>
/// /// An augmented expression visitor that requires implementors to provide transformations for the method API for IQStreamable
/// </summary> /// </summary>
public abstract class QStreamableVisitor : ExpressionVisitor public abstract class QStreamableVisitor : ExpressionVisitor
{ {
@ -19,10 +19,10 @@ namespace Microsoft.StreamProcessing.Provider
private static readonly MethodInfo WhereMethod = typeof(QStreamableStatic).GetMethod("Where"); private static readonly MethodInfo WhereMethod = typeof(QStreamableStatic).GetMethod("Where");
/// <summary> /// <summary>
/// /// Visits the children of the System.Linq.Expressions.MethodCallExpression.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected sealed override Expression VisitMethodCall(MethodCallExpression node) protected sealed override Expression VisitMethodCall(MethodCallExpression node)
{ {
var method = node.Method; var method = node.Method;
@ -39,45 +39,45 @@ namespace Microsoft.StreamProcessing.Provider
} }
/// <summary> /// <summary>
/// /// An overrideable method for handling any method that is not part of the IQStreamable API.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitNonStreamingMethodCall(MethodCallExpression node) => node; protected virtual Expression VisitNonStreamingMethodCall(MethodCallExpression node) => base.VisitMethodCall(node);
/// <summary> /// <summary>
/// /// Visits an IQStreamable Alter Duration call.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitAlterDurationCall(MethodCallExpression node) => node; protected virtual Expression VisitAlterDurationCall(MethodCallExpression node) => node;
/// <summary> /// <summary>
/// /// Visits an IQStreamable Quantize Lifetime call.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitQuantizeLifetimeCall(MethodCallExpression node) => node; protected virtual Expression VisitQuantizeLifetimeCall(MethodCallExpression node) => node;
/// <summary> /// <summary>
/// /// Visits an IQStreamable Select call.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitSelectCall(MethodCallExpression node) => node; protected virtual Expression VisitSelectCall(MethodCallExpression node) => node;
/// <summary> /// <summary>
/// /// Visits an IQStreamable Where call.
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitWhereCall(MethodCallExpression node) => node; protected virtual Expression VisitWhereCall(MethodCallExpression node) => node;
/// <summary> /// <summary>
/// ///
/// </summary> /// </summary>
/// <param name="node"></param> /// <param name="node">The expression to visit.</param>
/// <returns></returns> /// <returns>The modified expression, if it or any subexpression was modified; otherwise, returns the original expression.</returns>
protected virtual Expression VisitGroupByCall(MethodCallExpression node) => node; protected virtual Expression VisitGroupByCall(MethodCallExpression node) => node;
} }
} }

10
Sources/Core/Microsoft.StreamProcessing/Operators/EquiJoin/Basic/EquijoinTransformer.cs
Просмотреть файл

@ -112,9 +112,10 @@ namespace Microsoft.StreamProcessing
#region LeftBatchSelector #region LeftBatchSelector
{ {
var leftBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("i");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = "i", parameterRepresentation = template.leftMessageRepresentation, }), Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = leftBatchIndexVariable, parameterRepresentation = template.leftMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -127,7 +128,7 @@ namespace Microsoft.StreamProcessing
var d = new Dictionary<ParameterExpression, string> var d = new Dictionary<ParameterExpression, string>
{ {
{ Expression.Variable(leftMessageType, "leftBatch"), leftBatch }, { Expression.Variable(leftMessageType, "leftBatch"), leftBatch },
{ Expression.Variable(typeof(int), "i"), leftIndex }, { Expression.Variable(typeof(int), leftBatchIndexVariable), leftIndex },
{ selector.Parameters[1], rightEvent } { selector.Parameters[1], rightEvent }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
@ -158,9 +159,10 @@ namespace Microsoft.StreamProcessing
#endregion #endregion
#region RightBatchSelector #region RightBatchSelector
{ {
var rightBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("j");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = "j", parameterRepresentation = template.rightMessageRepresentation, }), Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = rightBatchIndexVariable, parameterRepresentation = template.rightMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -174,7 +176,7 @@ namespace Microsoft.StreamProcessing
{ {
{ selector.Parameters[0], leftEvent }, { selector.Parameters[0], leftEvent },
{ Expression.Variable(rightMessageType, "rightBatch"), rightBatch }, { Expression.Variable(rightMessageType, "rightBatch"), rightBatch },
{ Expression.Variable(typeof(int), "j"), rightIndex } { Expression.Variable(typeof(int), rightBatchIndexVariable), rightIndex }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
sb.AppendLine("{"); sb.AppendLine("{");

20
Sources/Core/Microsoft.StreamProcessing/Operators/EquiJoin/EquiJoinStreamable.cs
Просмотреть файл

@ -32,12 +32,18 @@ namespace Microsoft.StreamProcessing
// This operator uses the equality method on payloads // This operator uses the equality method on payloads
if (left.Properties.IsColumnar && !left.Properties.IsStartEdgeOnly && !left.Properties.PayloadEqualityComparer.CanUsePayloadEquality()) if (left.Properties.IsColumnar && !left.Properties.IsStartEdgeOnly && !left.Properties.PayloadEqualityComparer.CanUsePayloadEquality())
{ {
throw new InvalidOperationException($"Type of left side of join, '{typeof(TLeft).FullName}', does not have a valid equality operator for columnar mode."); this.errorMessages = $"The left input payload type, '{typeof(TLeft).FullName}', to Equijoin does not implement the interface {nameof(IEqualityComparerExpression<TLeft>)}. This interface is needed for code generation of this operator for columnar mode. Furthermore, the equality expression in the interface can only refer to input variables if used in field or property references.";
if (Config.CodegenOptions.DontFallBackToRowBasedExecution)
throw new StreamProcessingException(this.errorMessages);
else this.Left = this.Left.ColumnToRow();
} }
// This operator uses the equality method on payloads // This operator uses the equality method on payloads
if (right.Properties.IsColumnar && !right.Properties.IsStartEdgeOnly && !right.Properties.PayloadEqualityComparer.CanUsePayloadEquality()) if (right.Properties.IsColumnar && !right.Properties.IsStartEdgeOnly && !right.Properties.PayloadEqualityComparer.CanUsePayloadEquality())
{ {
throw new InvalidOperationException($"Type of right side of join, '{typeof(TRight).FullName}', does not have a valid equality operator for columnar mode."); this.errorMessages = $"The right input payload type, '{typeof(TRight).FullName}', to Equijoin does not implement the interface {nameof(IEqualityComparerExpression<TRight>)}. This interface is needed for code generation of this operator for columnar mode. Furthermore, the equality expression in the interface can only refer to input variables if used in field or property references.";
if (Config.CodegenOptions.DontFallBackToRowBasedExecution)
throw new StreamProcessingException(this.errorMessages);
else this.Right = this.Right.ColumnToRow();
} }
if (left.Properties.IsStartEdgeOnly && right.Properties.IsStartEdgeOnly) if (left.Properties.IsStartEdgeOnly && right.Properties.IsStartEdgeOnly)
@ -138,15 +144,11 @@ namespace Microsoft.StreamProcessing
} }
protected override IBinaryObserver<TKey, TLeft, TRight, TResult> CreatePipe(IStreamObserver<TKey, TResult> observer) protected override IBinaryObserver<TKey, TLeft, TRight, TResult> CreatePipe(IStreamObserver<TKey, TResult> observer)
{ => typeof(TKey).GetPartitionType() != null
if (typeof(TKey).GetPartitionType() == null) ? this.partitionedGenerator(this, this.Selector, observer)
{ : this.properties.IsColumnar
return this.properties.IsColumnar
? GetPipe(observer) ? GetPipe(observer)
: this.fallbackGenerator(this, this.Selector, observer); : this.fallbackGenerator(this, this.Selector, observer);
}
return this.partitionedGenerator(this, this.Selector, observer);
}
protected override bool CanGenerateColumnar() protected override bool CanGenerateColumnar()
{ {

10
Sources/Core/Microsoft.StreamProcessing/Operators/EquiJoin/FixedInterval/FixedIntervalEquijoinTransformer.cs
Просмотреть файл

@ -117,9 +117,10 @@ namespace Microsoft.StreamProcessing
#region LeftBatchSelector #region LeftBatchSelector
{ {
var leftBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("i");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = "i", parameterRepresentation = template.leftMessageRepresentation, }), Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = leftBatchIndexVariable, parameterRepresentation = template.leftMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -132,7 +133,7 @@ namespace Microsoft.StreamProcessing
var d = new Dictionary<ParameterExpression, string> var d = new Dictionary<ParameterExpression, string>
{ {
{ Expression.Variable(leftMessageType, "leftBatch"), leftBatch }, { Expression.Variable(leftMessageType, "leftBatch"), leftBatch },
{ Expression.Variable(typeof(int), "i"), leftIndex }, { Expression.Variable(typeof(int), leftBatchIndexVariable), leftIndex },
{ selector.Parameters[1], rightEvent } { selector.Parameters[1], rightEvent }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
@ -163,9 +164,10 @@ namespace Microsoft.StreamProcessing
#endregion #endregion
#region RightBatchSelector #region RightBatchSelector
{ {
var rightBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("j");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = "j", parameterRepresentation = template.rightMessageRepresentation, }), Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = rightBatchIndexVariable, parameterRepresentation = template.rightMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultMessageRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -179,7 +181,7 @@ namespace Microsoft.StreamProcessing
{ {
{ selector.Parameters[0], leftEvent }, { selector.Parameters[0], leftEvent },
{ Expression.Variable(rightMessageType, "rightBatch"), rightBatch }, { Expression.Variable(rightMessageType, "rightBatch"), rightBatch },
{ Expression.Variable(typeof(int), "j"), rightIndex } { Expression.Variable(typeof(int), rightBatchIndexVariable), rightIndex }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
sb.AppendLine("{"); sb.AppendLine("{");

10
Sources/Core/Microsoft.StreamProcessing/Operators/EquiJoin/IncreasingOrder/IncreasingOrderEquiJoinTransformer.cs
Просмотреть файл

@ -93,9 +93,10 @@ namespace Microsoft.StreamProcessing
return null; return null;
#region LeftBatchSelector #region LeftBatchSelector
{ {
var leftBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("i");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = "i", parameterRepresentation = template.leftMessageRepresentation, }), Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = leftBatchIndexVariable, parameterRepresentation = template.leftMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -108,7 +109,7 @@ namespace Microsoft.StreamProcessing
var d = new Dictionary<ParameterExpression, string> var d = new Dictionary<ParameterExpression, string>
{ {
{ Expression.Variable(leftMessageType, "leftBatch"), leftBatch }, { Expression.Variable(leftMessageType, "leftBatch"), leftBatch },
{ Expression.Variable(typeof(int), "i"), leftIndex }, { Expression.Variable(typeof(int), leftBatchIndexVariable), leftIndex },
{ selector.Parameters[1], rightEvent } { selector.Parameters[1], rightEvent }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
@ -139,9 +140,10 @@ namespace Microsoft.StreamProcessing
#endregion #endregion
#region RightBatchSelector #region RightBatchSelector
{ {
var rightBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("j");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = "j", parameterRepresentation = template.rightMessageRepresentation, }), Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = rightBatchIndexVariable, parameterRepresentation = template.rightMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -155,7 +157,7 @@ namespace Microsoft.StreamProcessing
{ {
{ selector.Parameters[0], leftEvent }, { selector.Parameters[0], leftEvent },
{ Expression.Variable(rightMessageType, "rightBatch"), rightBatch }, { Expression.Variable(rightMessageType, "rightBatch"), rightBatch },
{ Expression.Variable(typeof(int), "j"), rightIndex } { Expression.Variable(typeof(int), rightBatchIndexVariable), rightIndex }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
sb.AppendLine("{"); sb.AppendLine("{");

10
Sources/Core/Microsoft.StreamProcessing/Operators/EquiJoin/StartEdge/StartEdgeEquiJoinTransformer.cs
Просмотреть файл

@ -75,9 +75,10 @@ namespace Microsoft.StreamProcessing
var rightMessageType = StreamMessageManager.GetStreamMessageType<TKey, TRight>(); var rightMessageType = StreamMessageManager.GetStreamMessageType<TKey, TRight>();
#region LeftBatchSelector #region LeftBatchSelector
{ {
var leftBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("i");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = "i", parameterRepresentation = template.leftMessageRepresentation, }), Tuple.Create(selector.Parameters[0], new SelectParameterInformation() { BatchName = "leftBatch", BatchType = leftMessageType, IndexVariableName = leftBatchIndexVariable, parameterRepresentation = template.leftMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -90,7 +91,7 @@ namespace Microsoft.StreamProcessing
var d = new Dictionary<ParameterExpression, string> var d = new Dictionary<ParameterExpression, string>
{ {
{ Expression.Variable(leftMessageType, "leftBatch"), leftBatch }, { Expression.Variable(leftMessageType, "leftBatch"), leftBatch },
{ Expression.Variable(typeof(int), "i"), leftIndex }, { Expression.Variable(typeof(int), leftBatchIndexVariable), leftIndex },
{ selector.Parameters[1], rightEvent } { selector.Parameters[1], rightEvent }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
@ -121,9 +122,10 @@ namespace Microsoft.StreamProcessing
#endregion #endregion
#region RightBatchSelector #region RightBatchSelector
{ {
var rightBatchIndexVariable = selector.Parameters.GenerateFreshVariableName("j");
var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>() var parameterSubsitutions = new List<Tuple<ParameterExpression, SelectParameterInformation>>()
{ {
Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = "j", parameterRepresentation = template.rightMessageRepresentation, }), Tuple.Create(selector.Parameters[1], new SelectParameterInformation() { BatchName = "rightBatch", BatchType = rightMessageType, IndexVariableName = rightBatchIndexVariable, parameterRepresentation = template.rightMessageRepresentation, }),
}; };
var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true); var projectionResult = SelectTransformer.Transform(selector, parameterSubsitutions, resultRepresentation, true);
if (projectionResult.Error) if (projectionResult.Error)
@ -137,7 +139,7 @@ namespace Microsoft.StreamProcessing
{ {
{ selector.Parameters[0], leftEvent }, { selector.Parameters[0], leftEvent },
{ Expression.Variable(rightMessageType, "rightBatch"), rightBatch }, { Expression.Variable(rightMessageType, "rightBatch"), rightBatch },
{ Expression.Variable(typeof(int), "j"), rightIndex } { Expression.Variable(typeof(int), rightBatchIndexVariable), rightIndex }
}; };
var sb = new System.Text.StringBuilder(); var sb = new System.Text.StringBuilder();
sb.AppendLine("{"); sb.AppendLine("{");

12
Sources/Core/Microsoft.StreamProcessing/Operators/LeftAntiSemiJoin/LeftAntiSemiJoinStreamable.cs
Просмотреть файл

@ -4,7 +4,6 @@
// ********************************************************************* // *********************************************************************
using System; using System;
using System.Diagnostics.Contracts; using System.Diagnostics.Contracts;
using System.Globalization;
using Microsoft.StreamProcessing.Internal.Collections; using Microsoft.StreamProcessing.Internal.Collections;
namespace Microsoft.StreamProcessing namespace Microsoft.StreamProcessing
@ -25,13 +24,16 @@ namespace Microsoft.StreamProcessing
this.LeftComparer = left.Properties.PayloadEqualityComparer; this.LeftComparer = left.Properties.PayloadEqualityComparer;
Initialize();
// This operator uses the equality method on payloads // This operator uses the equality method on payloads
if (this.Properties.IsColumnar && !this.Properties.PayloadEqualityComparer.CanUsePayloadEquality()) if (left.Properties.IsColumnar && !this.LeftComparer.CanUsePayloadEquality())
{ {
throw new InvalidOperationException($"Type of payload, '{typeof(TLeft).FullName}', to LASJ does not have a valid equality operator for columnar mode."); this.errorMessages = $"The payload type, '{typeof(TLeft).FullName}', to Left Antisemijoin does not implement the interface {nameof(IEqualityComparerExpression<TLeft>)}. This interface is needed for code generation of this operator for columnar mode. Furthermore, the equality expression in the interface can only refer to input variables if used in field or property references.";
if (Config.CodegenOptions.DontFallBackToRowBasedExecution)
throw new StreamProcessingException(this.errorMessages);
else this.Left = this.Left.ColumnToRow();
} }
Initialize();
} }
protected override IBinaryObserver<TKey, TLeft, TRight, TLeft> CreatePipe(IStreamObserver<TKey, TLeft> observer) protected override IBinaryObserver<TKey, TLeft, TRight, TLeft> CreatePipe(IStreamObserver<TKey, TLeft> observer)

37
Sources/Core/Microsoft.StreamProcessing/StreamProperties/StreamProperties.cs
Просмотреть файл

@ -182,7 +182,7 @@ namespace Microsoft.StreamProcessing
/// <summary> /// <summary>
/// Equality comparers for possible key selectors (selector -> IECE) /// Equality comparers for possible key selectors (selector -> IECE)
/// </summary> /// </summary>
private Dictionary<Expression, object> EqualityComparerSelectorMap; private readonly Dictionary<Expression, object> EqualityComparerSelectorMap;
/// <summary> /// <summary>
/// Selectors that identify what the data is sorted by (could be more than one) /// Selectors that identify what the data is sorted by (could be more than one)
@ -506,19 +506,15 @@ namespace Microsoft.StreamProcessing
/// Clone /// Clone
/// </summary> /// </summary>
internal StreamProperties<TKey, TPayload> Clone() internal StreamProperties<TKey, TPayload> Clone()
{ => new StreamProperties<TKey, TPayload>
return new StreamProperties<TKey, TPayload>
(this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer, this.PayloadEqualityComparer, this.KeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer); (this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer, this.PayloadEqualityComparer, this.KeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer);
}
/// <summary> /// <summary>
/// Clone /// Clone
/// </summary> /// </summary>
internal StreamProperties<TKey, TPayload> CloneDelayed() internal StreamProperties<TKey, TPayload> CloneDelayed()
{ => new StreamProperties<TKey, TPayload>
return new StreamProperties<TKey, TPayload>
(this.isColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer, this.PayloadEqualityComparer, this.KeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer); (this.isColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer, this.PayloadEqualityComparer, this.KeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer);
}
/// <summary> /// <summary>
/// Clone /// Clone
@ -526,12 +522,10 @@ namespace Microsoft.StreamProcessing
internal StreamProperties<TNewKey, TPayload> CloneToNewKeyType<TNewKey>( internal StreamProperties<TNewKey, TPayload> CloneToNewKeyType<TNewKey>(
IEqualityComparerExpression<TNewKey> newKeyEqualityComparer, IEqualityComparerExpression<TNewKey> newKeyEqualityComparer,
IComparerExpression<TNewKey> newKeyComparer) IComparerExpression<TNewKey> newKeyComparer)
{ => new StreamProperties<TNewKey, TPayload>
return new StreamProperties<TNewKey, TPayload>
(this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, (this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree,
newKeyEqualityComparer, this.PayloadEqualityComparer, newKeyEqualityComparer, this.PayloadEqualityComparer,
newKeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer); newKeyComparer, this.PayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer);
}
/// <summary> /// <summary>
/// Clone /// Clone
@ -539,20 +533,15 @@ namespace Microsoft.StreamProcessing
internal StreamProperties<TKey, TNewPayload> CloneToNewPayloadType<TNewPayload>( internal StreamProperties<TKey, TNewPayload> CloneToNewPayloadType<TNewPayload>(
IEqualityComparerExpression<TNewPayload> newPayloadEqualityComparer, IEqualityComparerExpression<TNewPayload> newPayloadEqualityComparer,
IComparerExpression<TNewPayload> newPayloadComparer) IComparerExpression<TNewPayload> newPayloadComparer)
{ => new StreamProperties<TKey, TNewPayload>
return new StreamProperties<TKey, TNewPayload>
(this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer, (this.IsColumnar, this.IsConstantDuration, this.ConstantDurationLength, this.IsConstantHop, this.ConstantHopLength, this.ConstantHopOffset, this.IsIntervalFree, this.IsSyncTimeSimultaneityFree, this.IsSnapshotSorted, this.IsEventOverlappingFree, this.KeyEqualityComparer,
newPayloadEqualityComparer, this.KeyComparer, newPayloadEqualityComparer, this.KeyComparer,
newPayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer); newPayloadComparer, this.EqualityComparerSelectorMap.Clone(), this.SortSelectorMap.Clone(), this.QueryContainer);
}
/// <summary> /// <summary>
/// Where /// Where
/// </summary> /// </summary>
internal StreamProperties<TKey, TPayload> Where(Expression<Func<TPayload, bool>> predicate) internal StreamProperties<TKey, TPayload> Where(Expression<Func<TPayload, bool>> predicate) => this;
{
return this;
}
/// <summary> /// <summary>
/// Select /// Select
@ -875,9 +864,7 @@ namespace Microsoft.StreamProcessing
public NullStreamable(StreamProperties<TKey, TPayload> properties) : base(properties) { } public NullStreamable(StreamProperties<TKey, TPayload> properties) : base(properties) { }
public override IDisposable Subscribe(IStreamObserver<TKey, TPayload> observer) public override IDisposable Subscribe(IStreamObserver<TKey, TPayload> observer)
{ => throw new InvalidOperationException("Only for use in property derivation");
throw new InvalidOperationException("Only for use in property derivation");
}
} }
internal sealed class VerifyPropertiesStreamable<TKey, TPayload> : UnaryStreamable<TKey, TPayload, TPayload> internal sealed class VerifyPropertiesStreamable<TKey, TPayload> : UnaryStreamable<TKey, TPayload, TPayload>
@ -885,9 +872,7 @@ namespace Microsoft.StreamProcessing
public VerifyPropertiesStreamable(IStreamable<TKey, TPayload> source) : base(source, source.Properties) { } public VerifyPropertiesStreamable(IStreamable<TKey, TPayload> source) : base(source, source.Properties) { }
internal override IStreamObserver<TKey, TPayload> CreatePipe(IStreamObserver<TKey, TPayload> observer) internal override IStreamObserver<TKey, TPayload> CreatePipe(IStreamObserver<TKey, TPayload> observer)
{ => new VerifyPropertiesPipe<TKey, TPayload>(this.properties, observer);
return new VerifyPropertiesPipe<TKey, TPayload>(this.properties, observer);
}
protected override bool CanGenerateColumnar() => true; protected override bool CanGenerateColumnar() => true;
} }
@ -967,6 +952,7 @@ namespace Microsoft.StreamProcessing
this.ConstantDurationValidator(s, o); this.ConstantDurationValidator(s, o);
this.ConstantHopValidator(s, o); this.ConstantHopValidator(s, o);
this.IntervalFreeValidator(s, o); this.IntervalFreeValidator(s, o);
this.SyncTimeValidator(s, k);
this.SyncTimeSimultaneityFreeValidator(s, o, k); this.SyncTimeSimultaneityFreeValidator(s, o, k);
} }
@ -1058,10 +1044,7 @@ namespace Microsoft.StreamProcessing
key.ToString())); key.ToString()));
} }
public override void ProduceQueryPlan(PlanNode previous) public override void ProduceQueryPlan(PlanNode previous) => throw new NotImplementedException();
{
throw new NotImplementedException();
}
public override int CurrentlyBufferedOutputCount => 0; public override int CurrentlyBufferedOutputCount => 0;

9
Sources/Core/Microsoft.StreamProcessing/Utilities/Utility.cs
Просмотреть файл

@ -44,6 +44,15 @@ namespace Microsoft.StreamProcessing
yield return element; yield return element;
} }
internal static string GenerateFreshVariableName(this IEnumerable<ParameterExpression> parameters, string baseName)
{
var names = new HashSet<string>(parameters.Select(p => p.Name));
var suffix = 0;
var testName = baseName;
while (names.Contains(testName)) testName = baseName + suffix++;
return testName;
}
/// <summary> /// <summary>
/// Tries to get the first element in the given sorted dictionary. /// Tries to get the first element in the given sorted dictionary.
/// </summary> /// </summary>

5
Sources/Test/SimpleTesting/AdHocTests.cs
Просмотреть файл

@ -1686,6 +1686,9 @@ namespace SimpleTesting
[TestClass] [TestClass]
public class LeftComparerPayload_WithCodegen : TestWithConfigSettingsAndMemoryLeakDetection public class LeftComparerPayload_WithCodegen : TestWithConfigSettingsAndMemoryLeakDetection
{ {
public LeftComparerPayload_WithCodegen()
: base(new ConfigModifier().DontFallBackToRowBasedExecution(true)) { }
public class ClassOverridingEquals public class ClassOverridingEquals
{ {
public int x; public int x;
@ -1729,7 +1732,7 @@ namespace SimpleTesting
.ToEnumerable() .ToEnumerable()
.ToArray(); .ToArray();
} }
catch (Exception) catch (StreamProcessingException)
{ {
exceptionHappened = true; exceptionHappened = true;
} }