berkus
/
Dryad
mirror of https://github.com/MicrosoftResearch/Dryad.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
Dryad/LinqToDryad/DryadLinqQueryGen.cs

4757 lines
245 KiB
C#
Raw Permalink Blame History

/*
Copyright (c) Microsoft Corporation
All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
compliance with the License. You may obtain a copy of the License
at http://www.apache.org/licenses/LICENSE-2.0
THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER
EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF
TITLE, FITNESS FOR A PARTICULAR PURPOSE, MERCHANTABLITY OR NON-INFRINGEMENT.
See the Apache Version 2.0 License for specific language governing permissions and
limitations under the License.
*/
using System;
using System.Collections;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Text;
using System.IO;
using System.Reflection;
using System.Threading;
using System.Linq;
using System.Linq.Expressions;
using System.CodeDom;
using System.Xml;
using System.Diagnostics;
using Microsoft.Research.DryadLinq.Internal;
using System.Globalization;
namespace Microsoft.Research.DryadLinq
{
/// <summary>
/// This class handles code generation for multiple queries and execution invocation.
/// </summary>
internal class DryadLinqQueryGen
{
internal const int StartPhaseId = -4;
private const string DryadLinqProgram = "DryadLinqProgram__.xml";
private const string QueryGraph = "QueryGraph__.txt";
private const string VertexHostExe = "VertexHost.exe";
private static int s_uniqueProgId = -1;
private static object s_queryGenLock = new Object();
private int m_currentPhaseId = StartPhaseId;
private int m_nextVertexId = 0;
private bool m_codeGenDone = false;
private DryadLinqCodeGen m_codeGen;
private Expression[] m_queryExprs;
private DLinqQueryNode[] m_queryPlan1;
private DLinqQueryNode[] m_queryPlan2;
private DLinqQueryNode[] m_queryPlan3;
private Uri[] m_outputTableUris;
private bool[] m_isTempOutput;
private Uri[] m_outputDatapaths;
private Type[] m_outputTypes;
private QueryNodeInfo[] m_queryNodeInfos;
private DryadLinqQuery[] m_outputTables;
private string m_DryadLinqProgram;
private string m_queryGraph;
private Dictionary<Expression, QueryNodeInfo> m_exprNodeInfoMap;
private Dictionary<Expression, QueryNodeInfo> m_referencedQueryMap;
private Dictionary<string, DLinqInputNode> m_inputUriMap;
private Dictionary<string, DLinqOutputNode> m_outputUriMap;
private DryadLinqJobExecutor m_queryExecutor;
private DryadLinqContext m_context;
// This constructor is specifically to support enumeration of a query.
// It assumes that the Expressions all terminate with a ToStore node.
internal DryadLinqQueryGen(DryadLinqContext context,
VertexCodeGen vertexCodeGen,
Expression queryExpr,
Uri tableUri,
bool isTempOutput)
{
this.m_queryExprs = new Expression[] { queryExpr };
Uri fullTableUri = tableUri;
this.m_outputTableUris = new Uri[] { fullTableUri };
this.m_isTempOutput = new bool[] { isTempOutput };
this.m_context = context;
this.Initialize(vertexCodeGen);
}
// This constructor is specifically to support SubmitAndWait() calls.
// It assumes that the Expressions all terminate with a ToStore node.
internal DryadLinqQueryGen(DryadLinqContext context,
VertexCodeGen vertexCodeGen,
Expression[] qlist)
{
this.m_queryExprs = new Expression[qlist.Length];
this.m_outputTableUris = new Uri[qlist.Length];
this.m_isTempOutput = new bool[qlist.Length];
this.m_context = context;
for (int i = 0; i < this.m_queryExprs.Length; i++)
{
MethodCallExpression mcExpr = (MethodCallExpression)qlist[i];
this.m_queryExprs[i] = mcExpr.Arguments[0];
Uri tableUri;
if (mcExpr.Method.Name == ReflectedNames.DLQ_ToStore)
{
ExpressionSimplifier<Uri> e2 = new ExpressionSimplifier<Uri>();
tableUri = e2.Eval(mcExpr.Arguments[1]);
ExpressionSimplifier<bool> e3 = new ExpressionSimplifier<bool>();
this.m_isTempOutput[i] = e3.Eval(mcExpr.Arguments[2]);
}
else
{
throw new DryadLinqException("Internal error: The method must be " + ReflectedNames.DLQ_ToStore);
}
this.m_outputTableUris[i] = tableUri;
}
this.Initialize(vertexCodeGen);
}
private void Initialize(VertexCodeGen vertexCodeGen)
{
this.m_codeGen = new DryadLinqCodeGen(this.m_context, vertexCodeGen);
this.m_queryPlan1 = null;
this.m_queryPlan2 = null;
this.m_queryPlan3 = null;
this.m_DryadLinqProgram = null;
this.m_queryPlan1 = null;
this.m_exprNodeInfoMap = new Dictionary<Expression, QueryNodeInfo>();
this.m_referencedQueryMap = new Dictionary<Expression, QueryNodeInfo>();
this.m_inputUriMap = new Dictionary<string, DLinqInputNode>();
this.m_outputUriMap = new Dictionary<string, DLinqOutputNode>();
this.m_queryExecutor = new DryadLinqJobExecutor(this.m_context);
// Initialize the data structures for the output tables
this.m_outputTypes = new Type[this.m_queryExprs.Length];
this.m_outputDatapaths = new Uri[this.m_queryExprs.Length];
this.m_queryNodeInfos = new QueryNodeInfo[this.m_queryExprs.Length];
for (int i = 0; i < this.m_queryExprs.Length; i++)
{
this.m_queryNodeInfos[i] = this.BuildNodeInfoGraph(this.m_queryExprs[i]);
this.m_queryNodeInfos[i] = new DummyQueryNodeInfo(this.m_queryExprs[i], false, this.m_queryNodeInfos[i]);
this.m_outputDatapaths[i] = this.m_outputTableUris[i];
if (!DataPath.IsValidDataPath(this.m_outputDatapaths[i]))
{
throw new DryadLinqException(DryadLinqErrorCode.UnrecognizedDataSource,
String.Format(SR.UnrecognizedDataSource,
this.m_outputTableUris[i].AbsoluteUri));
}
}
}
internal DryadLinqContext Context
{
get { return this.m_context; }
}
internal DryadLinqCodeGen CodeGen
{
get { return this.m_codeGen; }
}
internal Dictionary<Expression, QueryNodeInfo> ReferencedQueryMap
{
get { return this.m_referencedQueryMap; }
}
/// <summary>
/// Probes the running assembly and its dependencies, and throws an exception
/// if any of them is targetted to x86. Returns silently if all managed assemblies
/// in the list are x64 or AnyCPU. Native or unloadable binaries are ignored.
/// </summary>
private void CheckAssemblyArchitectures()
{
// First create the list of assemblies to probe
// We use a stripped down version of the resource discovery logic in GenerateDryadProgram.
// i) We start with the same set of currently loaded binaries
// (== client app + its dependencies + dynamically loaded assemblies)
// ii) We take out user specified resource exclusions (this enables a workaround
// for x86 assemblies that must be loaded on the client side, think UI plugins,
// but aren't needed by the vertex code)
//
// The difference is we don't add the vertex DLL, or user resources.
List<string> resourcesToExclude = new List<string>();
resourcesToExclude.AddRange(this.m_context.ResourcesToRemove.Select(x => x.ToLower(CultureInfo.InvariantCulture)));
IEnumerable<string> loadedAssemblyPaths
= TypeSystem.GetLoadedNonSystemAssemblyPaths().Select(x => x.ToLower(CultureInfo.InvariantCulture));
var asembliesToCheck = loadedAssemblyPaths.Where(path => !resourcesToExclude.Contains(path));
foreach (string path in asembliesToCheck)
{
Assembly asm = null;
try
{
asm = Assembly.ReflectionOnlyLoadFrom(path);
}
catch
{
// silently ignore load errors
}
if (asm != null)
{
PortableExecutableKinds peKind;
ImageFileMachine machine;
asm.ManifestModule.GetPEKind(out peKind, out machine);
// machine will always be reported as "I386" for both true x86 and AnyCPU assemblies
// peKind will have the "Required32Bit" flag set only for x86 binaries. Therefore we
// use peKind to make our decision.
if ((peKind & PortableExecutableKinds.Required32Bit) != 0)
{
string offendingAssemblyName = Path.GetFileName(path);
throw new DryadLinqException(DryadLinqErrorCode.Binaries32BitNotSupported,
String.Format(SR.Binaries32BitNotSupported, offendingAssemblyName));
}
}
}
}
internal DryadLinqQuery[] Execute()
{
lock (s_queryGenLock)
{
this.GenerateDryadProgram();
this.CheckAssemblyArchitectures();
// Invoke the background execution
this.m_queryExecutor.ExecuteAsync(this.m_DryadLinqProgram);
// Create the resulting partitioned tables
this.m_outputTables = new DryadLinqQuery[this.m_outputTableUris.Length];
MethodInfo minfo = typeof(Microsoft.Research.DryadLinq.DataProvider).GetMethod(
ReflectedNames.DataProvider_GetPartitionedTable,
BindingFlags.NonPublic | BindingFlags.Static);
for (int i = 0; i < this.m_outputTableUris.Length; i++)
{
MethodInfo minfo1 = minfo.MakeGenericMethod(this.m_outputTypes[i]);
object[] args = new object[] { this.m_context, this.m_outputTableUris[i] };
this.m_outputTables[i] = (DryadLinqQuery)minfo1.Invoke(null, args);
this.m_outputTables[i].QueryExecutor = this.m_queryExecutor;
}
return this.m_outputTables;
}
}
// Phase 1 of the query optimization
internal void GenerateQueryPlanPhase1()
{
if (this.m_queryPlan1 != null) return;
// Apply some simple rewrite rules
SimpleRewriter rewriter = new SimpleRewriter(this.m_exprNodeInfoMap.Values.ToList());
rewriter.Rewrite();
// Generate the query plan of phase1
var referencedNodes = this.m_referencedQueryMap.Values;
this.m_queryPlan1 = new DLinqQueryNode[this.m_queryExprs.Length + referencedNodes.Count];
for (int i = 0; i < this.m_queryExprs.Length; i++)
{
this.m_queryPlan1[i] = this.Visit(this.m_queryNodeInfos[i].Children[0].Child);
}
int idx = this.m_queryExprs.Length;
foreach (DummyQueryNodeInfo nodeInfo in referencedNodes)
{
if (nodeInfo.NeedsMerge)
{
// Add a Tee'd Merge
this.m_queryPlan1[idx] = this.Visit(nodeInfo.Children[0].Child);
DLinqQueryNode mergeNode = new DLinqMergeNode(true,
nodeInfo.QueryExpression,
this.m_queryPlan1[idx]);
this.m_queryPlan1[idx] = new DLinqTeeNode(mergeNode.OutputTypes[0], true,
mergeNode.QueryExpression, mergeNode);
}
else
{
this.m_queryPlan1[idx] = this.Visit(nodeInfo.Children[0].Child);
}
nodeInfo.QueryNode = this.m_queryPlan1[idx];
idx++;
}
// Finally, add the output nodes.
Dictionary<DLinqQueryNode, int> forkCounts = new Dictionary<DLinqQueryNode, int>();
for (int i = 0; i < this.m_queryExprs.Length; i++)
{
DLinqQueryNode queryNode = this.m_queryPlan1[i];
int cnt;
if (!forkCounts.TryGetValue(queryNode, out cnt))
{
cnt = queryNode.Parents.Count;
}
forkCounts[queryNode] = cnt + 1;
}
for (int i = 0; i < this.m_queryExprs.Length; i++)
{
DryadLinqClientLog.Add("Query " + i + " Output: " + this.m_outputDatapaths[i]);
DLinqQueryNode queryNode = this.m_queryPlan1[i];
if (TypeSystem.IsAnonymousType(queryNode.OutputTypes[0]))
{
throw new DryadLinqException(DryadLinqErrorCode.OutputTypeCannotBeAnonymous,
SR.OutputTypeCannotBeAnonymous);
}
// Add dummy Apply to make Dryad happy (it doesn't like to hook inputs straight to outputs)
if ((queryNode is DLinqInputNode) || (forkCounts[queryNode] > 1))
{
// Add a dummy Apply
Type paramType = typeof(IEnumerable<>).MakeGenericType(queryNode.OutputTypes[0]);
ParameterExpression param = Expression.Parameter(paramType, "x");
Type type = typeof(Func<,>).MakeGenericType(paramType, paramType);
LambdaExpression applyExpr = Expression.Lambda(type, param, param);
DLinqQueryNode applyNode = new DLinqApplyNode(applyExpr, this.m_queryExprs[i], queryNode);
applyNode.OutputDataSetInfo = queryNode.OutputDataSetInfo;
queryNode = applyNode;
}
if (queryNode is DLinqConcatNode)
{
// Again, we add dummy Apply in certain cases to make Dryad happy
((DLinqConcatNode)queryNode).FixInputs();
}
// Add the output node
CompressionScheme outputScheme = this.m_context.OutputDataCompressionScheme;
DLinqOutputNode outputNode = new DLinqOutputNode(this.m_context,
this.m_outputDatapaths[i],
this.m_isTempOutput[i],
outputScheme,
this.m_queryExprs[i],
queryNode);
this.m_queryPlan1[i] = outputNode;
if (this.m_outputUriMap.ContainsKey(this.m_outputDatapaths[i].AbsoluteUri.ToLower()))
{
throw new DryadLinqException(DryadLinqErrorCode.MultipleOutputsWithSameDscUri,
String.Format(SR.MultipleOutputsWithSameUri, this.m_outputDatapaths[i]));
}
this.m_outputUriMap.Add(this.m_outputDatapaths[i].AbsoluteUri.ToLower(), outputNode);
this.m_outputTypes[i] = this.m_queryPlan1[i].OutputTypes[0];
// Remove useless Tees to make Dryad happy
if ((queryNode is DLinqTeeNode) && (forkCounts[queryNode] == 1))
{
DLinqQueryNode teeChild = queryNode.Children[0];
teeChild.UpdateParent(queryNode, outputNode);
outputNode.UpdateChildren(queryNode, teeChild);
}
}
}
// Phase 2 of the query optimization
internal DLinqQueryNode[] GenerateQueryPlanPhase2()
{
if (this.m_queryPlan2 == null)
{
this.GenerateQueryPlanPhase1();
this.m_queryPlan2 = new DLinqQueryNode[this.m_queryPlan1.Length];
for (int i = 0; i < this.m_queryPlan1.Length; i++)
{
this.m_queryPlan2[i] = this.VisitPhase2(this.m_queryPlan1[i]);
}
this.m_currentPhaseId++;
}
return this.m_queryPlan2;
}
private DLinqQueryNode VisitPhase2(DLinqQueryNode node)
{
DLinqQueryNode resNode = node;
if (node.m_uniqueId == this.m_currentPhaseId)
{
if (node is DLinqForkNode)
{
// For now, we require every branch of a fork be used:
DLinqForkNode forkNode = (DLinqForkNode)node;
for (int i = 0; i < forkNode.Parents.Count; i++)
{
if ((forkNode.Parents[i] is DLinqTeeNode) &&
(forkNode.Parents[i].Parents.Count == 0))
{
throw DryadLinqException.Create(DryadLinqErrorCode.BranchOfForkNotUsed,
string.Format(SR.BranchOfForkNotUsed, i),
node.QueryExpression);
}
}
}
resNode = node.SuperNode;
if (resNode == null)
{
for (int i = 0; i < node.Children.Length; i++)
{
node.Children[i] = this.VisitPhase2(node.Children[i]);
}
resNode = node.PipelineReduce();
resNode.m_uniqueId++;
// Special treatment for DoWhile
DLinqDoWhileNode doWhile = resNode as DLinqDoWhileNode;
if (doWhile != null)
{
doWhile.Body = this.VisitPhase2(doWhile.Body);
doWhile.Cond = this.VisitPhase2(doWhile.Cond);
}
// Insert a Tee node if needed:
DLinqQueryNode outputNode = resNode.OutputNode;
if (outputNode.IsForked &&
!(outputNode is DLinqForkNode) &&
!(outputNode is DLinqTeeNode))
{
resNode = resNode.InsertTee(true);
}
}
}
return resNode;
}
// Phase 3 of the query optimization
internal DLinqQueryNode[] GenerateQueryPlanPhase3()
{
if (this.m_queryPlan3 == null)
{
this.GenerateQueryPlanPhase2();
this.m_queryPlan3 = this.m_queryPlan2;
for (int i = 0; i < this.m_queryPlan2.Length; i++)
{
this.VisitPhase3(this.m_queryPlan2[i]);
}
this.m_currentPhaseId++;
}
return this.m_queryPlan3;
}
private void VisitPhase3(DLinqQueryNode node)
{
if (node.m_uniqueId == this.m_currentPhaseId)
{
node.m_uniqueId++;
// Remove some useless Tee nodes
foreach (DLinqQueryNode child in node.Children)
{
if ((child is DLinqTeeNode) && !child.IsForked)
{
DLinqQueryNode teeChild = child.Children[0];
teeChild.UpdateParent(child, node);
node.UpdateChildren(child, teeChild);
}
}
// Remove some useless Merge nodes
if ((node is DLinqMergeNode) &&
!node.IsForked &&
!(node.Parents[0] is DLinqOutputNode) &&
!node.Children[0].IsDynamic &&
node.Children[0].PartitionCount == 1)
{
node.Children[0].UpdateParent(node, node.Parents[0]);
node.Parents[0].UpdateChildren(node, node.Children[0]);
}
// Add dynamic managers for tee nodes.
if ((StaticConfig.DynamicOptLevel & StaticConfig.NoDynamicOpt) != 0 &&
node is DLinqTeeNode &&
node.DynamicManager.ManagerType == DynamicManagerType.None)
{
// insert a dynamic broadcast manager on Tee
node.DynamicManager = DynamicManager.Broadcast;
}
// Recurse on the children of node
foreach (DLinqQueryNode child in node.Children)
{
this.VisitPhase3(child);
}
if (node is DLinqDoWhileNode)
{
this.VisitPhase3(((DLinqDoWhileNode)node).Body);
this.VisitPhase3(((DLinqDoWhileNode)node).Cond);
}
}
}
// The main method that generates the query plan.
internal DLinqQueryNode[] QueryPlan()
{
return this.GenerateQueryPlanPhase3();
}
// Generate the vertex code for all the query nodes.
internal void CodeGenVisit()
{
if (!this.m_codeGenDone)
{
DLinqQueryNode[] optimizedPlan = this.QueryPlan();
// make sure none of the outputs share a URI with inputs
foreach (var kvp in this.m_outputUriMap)
{
string outputPath = kvp.Key;
if(m_inputUriMap.ContainsKey(outputPath))
{
throw new DryadLinqException(DryadLinqErrorCode.OutputUriAlsoQueryInput,
String.Format(SR.OutputUriAlsoQueryInput, outputPath));
}
}
foreach (DLinqQueryNode node in optimizedPlan)
{
this.CodeGenVisit(node);
}
this.m_currentPhaseId++;
this.m_codeGenDone = true;
}
}
private void CodeGenVisit(DLinqQueryNode node)
{
if (node.m_uniqueId == this.m_currentPhaseId)
{
node.m_uniqueId++;
// We process the types first so that children will also know about all
// proxies/mappings that should be used.
node.CreateCodeAndMappingsForVertexTypes(false);
// Recurse on the children
foreach (DLinqQueryNode child in node.Children)
{
this.CodeGenVisit(child);
}
switch (node.NodeType)
{
case QueryNodeType.InputTable:
{
// vertex with no code
string t = ((DLinqInputNode)node).Table.DataSourceUri.AbsolutePath;
int index = t.LastIndexOf('/');
int bk = t.LastIndexOf('\\');
if (index < bk) index = bk;
node.m_vertexEntryMethod = t.Substring(index + 1);
break;
}
case QueryNodeType.OutputTable:
{
// vertex with no code
string t = ((DLinqOutputNode)node).OutputUri.AbsolutePath;
int index = t.LastIndexOf('/');
int bk = t.LastIndexOf('\\');
if (index < bk) index = bk;
int len = Math.Min(8, t.Length - index - 1);
node.m_vertexEntryMethod = t.Substring(index + 1, len);
break;
}
case QueryNodeType.Tee:
{
// vertex with no code
node.m_vertexEntryMethod = DryadLinqCodeGen.MakeUniqueName("Tee");
// broadcast manager code generation
if (node.DynamicManager.ManagerType != DynamicManagerType.None)
{
node.DynamicManager.CreateVertexCode();
}
break;
}
case QueryNodeType.Concat:
case QueryNodeType.Dummy:
{
// vertex with no code
node.m_vertexEntryMethod = DryadLinqCodeGen.MakeUniqueName(node.NodeType.ToString());
break;
}
case QueryNodeType.DoWhile:
{
// vertex with no code
node.m_vertexEntryMethod = DryadLinqCodeGen.MakeUniqueName(node.NodeType.ToString());
this.CodeGenVisit(((DLinqDoWhileNode)node).Body);
this.CodeGenVisit(((DLinqDoWhileNode)node).Cond);
break;
}
default:
{
CodeMemberMethod vertexMethod = this.m_codeGen.AddVertexMethod(node);
node.m_vertexEntryMethod = vertexMethod.Name;
node.DynamicManager.CreateVertexCode();
break;
}
}
}
}
// Assign unique ids to all the query nodes
private void AssignUniqueId()
{
if (this.m_currentPhaseId != -1)
{
//@@TODO: this should not be reachable. could change to Assert/InvalidOpEx
throw new DryadLinqException(DryadLinqErrorCode.Internal,
"Internal error: Optimization phase should be -1, not " +
this.m_currentPhaseId);
}
foreach (DLinqQueryNode node in this.QueryPlan())
{
this.AssignUniqueId(node);
}
}
private void AssignUniqueId(DLinqQueryNode node)
{
if (node.m_uniqueId == this.m_currentPhaseId)
{
foreach (Pair<ParameterExpression, DLinqQueryNode> refChild in node.GetReferencedQueries())
{
this.AssignUniqueId(refChild.Value);
}
foreach (DLinqQueryNode child in node.Children)
{
this.AssignUniqueId(child);
}
if (node.m_uniqueId == this.m_currentPhaseId)
{
node.m_uniqueId = this.m_nextVertexId++;
// Special treatment for DoWhile
DLinqDoWhileNode doWhileNode = node as DLinqDoWhileNode;
if (doWhileNode != null)
{
this.AssignUniqueId(doWhileNode.Body);
this.AssignUniqueId(doWhileNode.Cond);
this.AssignUniqueId(doWhileNode.BodySource);
this.AssignUniqueId(doWhileNode.CondSource1);
this.AssignUniqueId(doWhileNode.CondSource2);
}
// Special treatment for Fork
if (node.OutputNode is DLinqForkNode)
{
foreach (DLinqQueryNode pnode in node.Parents)
{
if (pnode.m_uniqueId == this.m_currentPhaseId)
{
pnode.m_uniqueId = this.m_nextVertexId++;
}
}
}
}
}
}
private void CreateQueryPlan(XmlDocument queryDoc, XmlElement queryPlan)
{
this.AssignUniqueId();
HashSet<int> seen = new HashSet<int>();
foreach (DLinqQueryNode node in this.QueryPlan())
{
node.AddToQueryPlan(queryDoc, queryPlan, seen);
}
}
/// <summary>
/// Find the pdb file associated with a given filename.
/// </summary>
/// <param name="filename">Filename with debugging information.</param>
/// <returns>The associated pdb.</returns>
internal static string FindPDB(string filename)
{
string basename = Path.GetFileNameWithoutExtension(filename);
string directory = Path.GetDirectoryName(filename);
string pdbname = directory + Path.DirectorySeparatorChar + basename + ".pdb";
return (File.Exists(pdbname)) ? pdbname : null;
}
/// <summary>
/// Add a resource to the Xml plan.
/// </summary>
/// <param name="queryDoc">Document holding the xml plan.</param>
/// <param name="parent">Parent node.</param>
/// <param name="resource">Resource to add.</param>
/// <returns>Handle to the inserted node.</returns>
private void AddResourceToPlan(XmlDocument queryDoc,
XmlElement parent,
string resource,
IEnumerable<string> resourcesToExclude)
{
AddResourceToPlan_Core(queryDoc, parent, resource, resourcesToExclude);
if (this.m_context.CompileForVertexDebugging)
{
string pdb = FindPDB(resource);
if (pdb != null)
{
AddResourceToPlan_Core(queryDoc, parent, pdb, resourcesToExclude);
}
}
}
// Add a resource to the plan unless it was specifically in the exclusions list.
private void AddResourceToPlan_Core(XmlDocument queryDoc,
XmlElement parent,
string resource,
IEnumerable<string> resourcesToExclude)
{
if (resourcesToExclude.Contains(resource, StringComparer.OrdinalIgnoreCase))
{
return;
}
XmlElement resourceElem = queryDoc.CreateElement("Resource");
resourceElem.InnerText = m_queryExecutor.AddResource(resource);
parent.AppendChild(resourceElem);
}
private void GenerateAppConfigResource(string appConfigPath)
{
// Generates an app config XML for the VertexHost which
//
// 1) specifies the server GC mode
//
// 2) requests a .NET runtime version equal to that of the client
// submitting the job (if the client has a .NET version higher than 3.5)
// also specifies <supportedRuntime version="v2.0.50727"/> as a fallback
// in case the cluster nodes don't have the client's newer .Net version
// (v2.0.50727 corresponds to .Net 3.5, both Sp1 and Sp2 pointing to this
// version ID, since it's actually the underlying CLR's version). This rule
// only kicks in if the query config has the MatchClientNetFrameworkVersion flag set.
//
// 3) disables Authenticode checks for the VH by specifying
// <generatePublisherEvidence enabled="false"/>. This is necessary
// because some cluster nodes don't have an Internet connection, in which
// case which the Authenticode check leads to a 20sec delay during process startup.
//
// NOTE: useLegacyV2RuntimeActivationPolicy="true" is needed becuase the VH binary is mixed mode.
string clientVersionString = "";
if (Environment.Version.Major > 2 && this.m_context.MatchClientNetFrameworkVersion)
{
clientVersionString = String.Format(CultureInfo.InvariantCulture,
@" <supportedRuntime version=""v{0}.{1}""/>",
Environment.Version.Major,
Environment.Version.Minor);
// NOTE: We use the "v4.0" syntax instead of the longer "v4.0.30319" because as of .NET4
// the format of this app config tag has been simplified to "major.minor"
}
string appConfigBody =
@"<?xml version =""1.0""?>
<configuration>
<runtime>
<gcServer enabled=""true""/>
<gcConcurrent enabled=""false""/>
<generatePublisherEvidence enabled=""false""/>
<gcAllowVeryLargeObjects enabled=""true"" />
</runtime>
<startup useLegacyV2RuntimeActivationPolicy=""true"">
";
appConfigBody += clientVersionString; // add the client specific version string if we generated one.
appConfigBody +=
@"
<supportedRuntime version=""v2.0.50727""/>
</startup>
</configuration>
";
File.WriteAllText(appConfigPath, appConfigBody);
}
/// <summary>
/// Generate the executable code.
/// </summary>
/// <returns>The path to the queryPlanXml location</returns>
internal string GenerateDryadProgram()
{
DryadLinqObjectStore.Clear();
// Any resource that we try to add will be tested against this list.
IEnumerable<string> resourcesToExclude = this.m_context.ResourcesToRemove;
// BuildDryadLinqAssembly:
// 1. Performs query optimizations
// 2. Generate vertex code for all the nodes in the query plan
this.m_codeGen.BuildDryadLinqAssembly(this);
// DryadQueryExplain explain = new DryadQueryExplain();
// DryadLinqClientLog.Add("{0}", explain.Explain(this));
// Finally, write out the query plan.
if (this.m_DryadLinqProgram == null)
{
int progId = Interlocked.Increment(ref s_uniqueProgId);
this.m_DryadLinqProgram = DryadLinqCodeGen.GetPathForGeneratedFile(DryadLinqProgram, progId);
this.m_queryGraph = DryadLinqCodeGen.GetPathForGeneratedFile(QueryGraph, progId);
}
XmlDocument queryDoc = new XmlDocument();
queryDoc.LoadXml("<Query></Query>");
// Write the assembly version information
Assembly dryadlinqassembly = Assembly.GetExecutingAssembly();
AssemblyName asmName = dryadlinqassembly.GetName();
XmlElement elem = queryDoc.CreateElement("DryadLinqVersion");
elem.InnerText = asmName.Version.ToString();
queryDoc.DocumentElement.AppendChild(elem);
// Add the cluster name element
elem = queryDoc.CreateElement("ClusterName");
elem.InnerText = this.m_context.HeadNode;
queryDoc.DocumentElement.AppendChild(elem);
// Add the minimum number of nodes
int minComputeNodes = 1;
if (this.m_context.JobMinNodes.HasValue)
{
minComputeNodes = this.m_context.JobMinNodes.Value;
}
elem = queryDoc.CreateElement("MinimumComputeNodes");
elem.InnerText = minComputeNodes.ToString();
queryDoc.DocumentElement.AppendChild(elem);
// Add the maximum number of nodes
int maxComputeNodes = -1;
if (this.m_context.JobMaxNodes.HasValue)
{
maxComputeNodes = this.m_context.JobMaxNodes.Value;
}
elem = queryDoc.CreateElement("MaximumComputeNodes");
elem.InnerText = maxComputeNodes.ToString();
queryDoc.DocumentElement.AppendChild(elem);
// intermediate data compression
elem = queryDoc.CreateElement("IntermediateDataCompression");
elem.InnerText = ((int)this.m_context.IntermediateDataCompressionScheme).ToString();
queryDoc.DocumentElement.AppendChild(elem);
// Speculative Duplication Node
elem = queryDoc.CreateElement("EnableSpeculativeDuplication");
elem.InnerText = this.m_context.EnableSpeculativeDuplication.ToString();
queryDoc.DocumentElement.AppendChild(elem);
// Add the visualization element
//@@TODO[p2]: remove this element from the queryXML.
elem = queryDoc.CreateElement("Visualization");
elem.InnerText = "none";
queryDoc.DocumentElement.AppendChild(elem);
// Add the query name element
elem = queryDoc.CreateElement("QueryName");
if (String.IsNullOrEmpty(this.m_context.JobFriendlyName))
{
elem.InnerText = asmName.Name;
}
else
{
elem.InnerText = this.m_context.JobFriendlyName;
}
queryDoc.DocumentElement.AppendChild(elem);
// Add the XmlExecHostArgs element
elem = queryDoc.CreateElement("XmlExecHostArgs");
queryDoc.DocumentElement.AppendChild(elem);
// Add an element for each argument
string[] args = StaticConfig.XmlExecHostArgs.Split(new char[] {' '}, StringSplitOptions.RemoveEmptyEntries);
List<string> xmlExecResources = new List<string>();
for (int i = 0; i < args.Length; ++i)
{
string arg = args[i];
XmlElement argElem = queryDoc.CreateElement("Argument");
argElem.InnerText = arg;
elem.AppendChild(argElem);
if (arg.Equals("-bw") || arg.Equals("-r"))
{
xmlExecResources.Add(args[i+1]);
}
}
// Add the resources element
elem = queryDoc.CreateElement("Resources");
queryDoc.DocumentElement.AppendChild(elem);
// Add resource item for this LINQ DLL
AddResourceToPlan(queryDoc, elem, this.m_codeGen.GetTargetLocation(), resourcesToExclude);
// Add resource item for each loaded DLL that isn't a system DLL
IEnumerable<string> loadedAssemblyPaths = TypeSystem.GetLoadedNonSystemAssemblyPaths();
foreach(string assemblyPath in loadedAssemblyPaths)
{
AddResourceToPlan(queryDoc, elem, assemblyPath, resourcesToExclude);
}
// Add the xmlExec resources
foreach (string resourcePath in xmlExecResources)
{
AddResourceToPlan(queryDoc, elem, resourcePath, resourcesToExclude);
}
// Add codegen resources
foreach (string resourcePath in this.m_codeGen.VertexCodeGen.ResourcesToAdd())
{
AddResourceToPlan(queryDoc, elem, resourcePath, resourcesToExclude);
}
// Create an app config file for the VertexHost process, and add it to the resources
string vertexHostAppConfigPath = DryadLinqCodeGen.GetPathForGeneratedFile(Path.ChangeExtension(VertexHostExe, "exe.config"), null);
GenerateAppConfigResource(vertexHostAppConfigPath);
AddResourceToPlan(queryDoc, elem, vertexHostAppConfigPath, resourcesToExclude);
// Save and add the object store as a resource
if (!DryadLinqObjectStore.IsEmpty)
{
DryadLinqObjectStore.Save();
AddResourceToPlan(queryDoc, elem, DryadLinqObjectStore.GetClientSideObjectStorePath(), resourcesToExclude);
}
// Add resource item for user-added resources
foreach (string userResource in this.m_context.ResourcesToAdd.Distinct(StringComparer.OrdinalIgnoreCase))
{
AddResourceToPlan(queryDoc, elem, userResource, resourcesToExclude);
}
// Add the query plan element
XmlElement queryPlanElem = queryDoc.CreateElement("QueryPlan");
queryDoc.DocumentElement.AppendChild(queryPlanElem);
// Add the query tree as a sequence of nodes
this.CreateQueryPlan(queryDoc, queryPlanElem);
// Finally, save the DryadQuery doc to a file
queryDoc.Save(this.m_DryadLinqProgram);
return this.m_DryadLinqProgram;
}
private void BuildReferencedQuery(Expression expr)
{
ExpressionQuerySet querySet = new ExpressionQuerySet();
querySet.Visit(expr);
foreach (Expression qexpr in querySet.QuerySet)
{
QueryNodeInfo nodeInfo = this.BuildNodeInfoGraph(qexpr);
this.m_referencedQueryMap[qexpr] = new DummyQueryNodeInfo(qexpr, true, nodeInfo);
}
}
private void BuildReferencedQuery(int startIdx, ReadOnlyCollection<Expression> exprs)
{
ExpressionQuerySet querySet = new ExpressionQuerySet();
for (int i = startIdx; i < exprs.Count; i++)
{
querySet.Visit(exprs[i]);
}
foreach (Expression qexpr in querySet.QuerySet)
{
QueryNodeInfo nodeInfo = this.BuildNodeInfoGraph(qexpr);
this.m_referencedQueryMap[qexpr] = new DummyQueryNodeInfo(qexpr, true, nodeInfo);
}
}
// Document what the 'NodeInfo' and 'ReferencedQuery' system does
private QueryNodeInfo BuildNodeInfoGraph(Expression expression)
{
QueryNodeInfo resNodeInfo = null;
if (this.m_exprNodeInfoMap.TryGetValue(expression, out resNodeInfo))
{
return resNodeInfo;
}
MethodCallExpression mcExpr = expression as MethodCallExpression;
if (mcExpr != null && mcExpr.Method.IsStatic && TypeSystem.IsQueryOperatorCall(mcExpr))
{
switch (mcExpr.Method.Name)
{
case "Join":
case "GroupJoin":
case "Union":
case "Intersect":
case "Except":
case "Zip":
case "SequenceEqual":
case "SequenceEqualAsQuery":
{
QueryNodeInfo child1 = this.BuildNodeInfoGraph(mcExpr.Arguments[0]);
QueryNodeInfo child2 = this.BuildNodeInfoGraph(mcExpr.Arguments[1]);
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1, child2);
this.BuildReferencedQuery(2, mcExpr.Arguments);
break;
}
case "Concat":
{
NewArrayExpression others = mcExpr.Arguments[1] as NewArrayExpression;
if (others == null)
{
QueryNodeInfo child1 = this.BuildNodeInfoGraph(mcExpr.Arguments[0]);
QueryNodeInfo child2 = this.BuildNodeInfoGraph(mcExpr.Arguments[1]);
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1, child2);
}
else
{
QueryNodeInfo[] infos = new QueryNodeInfo[others.Expressions.Count + 1];
infos[0] = this.BuildNodeInfoGraph(mcExpr.Arguments[0]);
for (int i = 0; i < others.Expressions.Count; ++i)
{
infos[i + 1] = this.BuildNodeInfoGraph(others.Expressions[i]);
}
resNodeInfo = new QueryNodeInfo(mcExpr, true, infos);
}
break;
}
case "Apply":
case "ApplyPerPartition":
{
QueryNodeInfo child1 = this.BuildNodeInfoGraph(mcExpr.Arguments[0]);
if (mcExpr.Arguments.Count == 2)
{
// Apply with only one input source
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1);
this.BuildReferencedQuery(mcExpr.Arguments[1]);
}
else
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(mcExpr.Arguments[2]);
NewArrayExpression others = mcExpr.Arguments[1] as NewArrayExpression;
if (lambda.Parameters.Count == 2 && others == null)
{
// Apply with two input sources
QueryNodeInfo child2 = this.BuildNodeInfoGraph(mcExpr.Arguments[1]);
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1, child2);
}
else
{
// Apply with multiple sources
QueryNodeInfo[] infos = new QueryNodeInfo[others.Expressions.Count + 1];
infos[0] = child1;
for (int i = 0; i < others.Expressions.Count; ++i)
{
infos[i + 1] = this.BuildNodeInfoGraph(others.Expressions[i]);
}
resNodeInfo = new QueryNodeInfo(mcExpr, true, infos);
}
this.BuildReferencedQuery(mcExpr.Arguments[2]);
}
break;
}
case "RangePartition":
{
QueryNodeInfo child1 = this.BuildNodeInfoGraph(mcExpr.Arguments[0]);
if (mcExpr.Arguments.Count == 6)
{
// This is a key part of handling for RangePartition( ... , IQueryable<> keysQuery, ...)
// The keys expression is established as a child[1] nodeInfo for the rangePartition node.
QueryNodeInfo child2 = this.BuildNodeInfoGraph(mcExpr.Arguments[2]);
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1, child2);
}
else
{
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1);
}
this.BuildReferencedQuery(mcExpr.Arguments[1]);
break;
}
case "DoWhile":
{
QueryNodeInfo child1 = BuildNodeInfoGraph(mcExpr.Arguments[0]);
Expression bodyExpr = mcExpr.Arguments[1];
QueryNodeInfo bodyNodeInfo = this.BuildNodeInfoGraph(bodyExpr);
Expression condExpr = mcExpr.Arguments[2];
QueryNodeInfo condNodeInfo = this.BuildNodeInfoGraph(condExpr);
Expression bodySourceExpr = mcExpr.Arguments[3];
QueryNodeInfo bodySourceNodeInfo = this.BuildNodeInfoGraph(bodySourceExpr);
Expression condSourceExpr1 = mcExpr.Arguments[4];
QueryNodeInfo condSourceNodeInfo1 = this.BuildNodeInfoGraph(condSourceExpr1);
Expression condSourceExpr2 = mcExpr.Arguments[5];
QueryNodeInfo condSourceNodeInfo2 = this.BuildNodeInfoGraph(condSourceExpr2);
resNodeInfo = new DoWhileQueryNodeInfo(mcExpr, bodyNodeInfo,
condNodeInfo,
bodySourceNodeInfo,
condSourceNodeInfo1, condSourceNodeInfo2,
child1);
break;
}
case "Dummy":
{
// Just return a QueryNodeInfo with no child.
resNodeInfo = new QueryNodeInfo(mcExpr, true);
break;
}
default:
{
QueryNodeInfo child1 = BuildNodeInfoGraph(mcExpr.Arguments[0]);
resNodeInfo = new QueryNodeInfo(mcExpr, true, child1);
this.BuildReferencedQuery(1, mcExpr.Arguments);
break;
}
}
}
if (resNodeInfo == null)
{
resNodeInfo = new QueryNodeInfo(expression, false);
}
this.m_exprNodeInfoMap.Add(expression, resNodeInfo);
return resNodeInfo;
}
private static bool IsMergeNodeNeeded(DLinqQueryNode node)
{
return node.IsDynamic || node.PartitionCount > 1;
}
internal DLinqQueryNode Visit(QueryNodeInfo nodeInfo)
{
Expression expression = nodeInfo.QueryExpression;
if (expression.NodeType == ExpressionType.Call)
{
MethodCallExpression mcExpr = (MethodCallExpression)expression;
if (mcExpr.Method.IsStatic && TypeSystem.IsQueryOperatorCall(mcExpr))
{
return this.VisitQueryOperatorCall(nodeInfo);
}
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, mcExpr.Method.Name),
expression);
}
else if (expression.NodeType == ExpressionType.Constant)
{
DLinqInputNode inputNode = new DLinqInputNode(this, (ConstantExpression)expression);
if (!this.m_inputUriMap.ContainsKey(inputNode.Table.DataSourceUri.AbsoluteUri.ToLower()))
{
this.m_inputUriMap.Add(inputNode.Table.DataSourceUri.AbsoluteUri.ToLower(), inputNode);
}
return inputNode;
}
else
{
string errMsg = "Can't handle expression of type " + expression.NodeType;
throw DryadLinqException.Create(DryadLinqErrorCode.UnsupportedExpressionsType,
String.Format(SR.UnsupportedExpressionsType,expression.NodeType),
expression);
}
}
private static bool IsLambda(Expression expr, int n)
{
LambdaExpression lambdaExpr = DryadLinqExpression.GetLambda(expr);
return (lambdaExpr != null && lambdaExpr.Parameters.Count == n);
}
// Checks if the child of the source node is a groupby node without any result selectors.
private static bool IsGroupByWithoutResultSelector(Expression source)
{
bool isGroupBy = false;
if (source.NodeType == ExpressionType.Call)
{
MethodCallExpression expression = (MethodCallExpression)source;
if (expression.Method.IsStatic &&
TypeSystem.IsQueryOperatorCall(expression) &&
expression.Method.Name == "GroupBy")
{
if (expression.Arguments.Count == 2)
{
isGroupBy = true;
}
else if (expression.Arguments.Count == 3)
{
isGroupBy = !IsLambda(expression.Arguments[2], 2);
}
else if (expression.Arguments.Count == 4)
{
isGroupBy = !(IsLambda(expression.Arguments[2], 2) ||
IsLambda(expression.Arguments[3], 2));
}
}
}
return isGroupBy;
}
private DLinqQueryNode CreateOffset(bool isLong, Expression queryExpr, DLinqQueryNode child)
{
// Count node
DLinqQueryNode countNode = new DLinqBasicAggregateNode(null, AggregateOpType.LongCount,
true, false, queryExpr, child);
// Apply node for x => Offsets(x)
Type paramType = typeof(IEnumerable<>).MakeGenericType(typeof(long));
ParameterExpression param = Expression.Parameter(paramType, "x");
MethodInfo minfo = typeof(DryadLinqEnumerable).GetMethod("Offsets");
Expression body = Expression.Call(minfo, param, Expression.Constant(isLong, typeof(bool)));
Type type = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(type, body, param);
DLinqQueryNode mergeCountNode = new DLinqMergeNode(true, queryExpr, countNode);
DLinqQueryNode offsetsNode = new DLinqApplyNode(procFunc, queryExpr, mergeCountNode);
// HashPartition
LambdaExpression keySelectExpr = IdentityFunction.Instance(typeof(IndexedValue<long>));
int pcount = child.OutputPartition.Count;
DLinqQueryNode hdistNode = new DLinqHashPartitionNode(keySelectExpr, null, null, pcount,
false, queryExpr, offsetsNode);
DLinqQueryNode resNode = new DLinqMergeNode(false, queryExpr, hdistNode);
return resNode;
}
private DLinqQueryNode PromoteConcat(QueryNodeInfo source,
DLinqQueryNode sourceNode,
Func<DLinqQueryNode, DLinqQueryNode> func)
{
DLinqQueryNode resNode = sourceNode;
if ((resNode is DLinqConcatNode) && !source.IsForked)
{
DLinqQueryNode[] children = resNode.Children;
DLinqQueryNode[] newChildren = new DLinqQueryNode[children.Length];
for (int i = 0; i < children.Length; i++)
{
children[i].Parents.Remove(resNode);
newChildren[i] = func(children[i]);
}
resNode = new DLinqConcatNode(source.QueryExpression, newChildren);
}
else
{
resNode = func(resNode);
}
return resNode;
}
private DLinqQueryNode VisitSelect(QueryNodeInfo source,
QueryNodeType nodeType,
LambdaExpression selector,
LambdaExpression resultSelector,
MethodCallExpression queryExpr)
{
DLinqQueryNode selectNode;
if (selector.Type.GetGenericArguments().Length == 2)
{
// If this select's child is a groupby node, push this select into its child, if
// 1. The groupby node is not tee'd, and
// 2. The groupby node has no result selector, and
// 3. The selector is decomposable
if (!source.IsForked &&
IsGroupByWithoutResultSelector(source.QueryExpression) &&
Decomposition.GetDecompositionInfoList(selector, m_codeGen) != null)
{
MethodCallExpression expr = (MethodCallExpression)source.QueryExpression;
LambdaExpression keySelectExpr = DryadLinqExpression.GetLambda(expr.Arguments[1]);
// Figure out elemSelectExpr and comparerExpr
LambdaExpression elemSelectExpr = null;
Expression comparerExpr = null;
if (expr.Arguments.Count == 3)
{
elemSelectExpr = DryadLinqExpression.GetLambda(expr.Arguments[2]);
if (elemSelectExpr == null)
{
comparerExpr = expr.Arguments[2];
}
}
else if (expr.Arguments.Count == 4)
{
elemSelectExpr = DryadLinqExpression.GetLambda(expr.Arguments[2]);
comparerExpr = expr.Arguments[3];
}
// Construct new query expression by building result selector expression
// and pushing it to groupby node.
selectNode = VisitGroupBy(source.Children[0].Child, keySelectExpr,
elemSelectExpr, selector, comparerExpr, queryExpr);
if (nodeType == QueryNodeType.SelectMany)
{
Type selectorRetType = selector.Type.GetGenericArguments()[1];
LambdaExpression id = IdentityFunction.Instance(selectorRetType);
selectNode = new DLinqSelectNode(nodeType, id, resultSelector, queryExpr, selectNode);
}
}
else
{
DLinqQueryNode child = this.Visit(source);
selectNode = this.PromoteConcat(
source, child,
x => new DLinqSelectNode(nodeType, selector, resultSelector, queryExpr, x));
}
}
else
{
// The "indexed" version
DLinqQueryNode child = this.Visit(source);
if (!child.IsDynamic && child.OutputPartition.Count == 1)
{
selectNode = this.PromoteConcat(
source, child,
x => new DLinqSelectNode(nodeType, selector, resultSelector, queryExpr, x));
}
else
{
child.IsForked = true;
// Create (x, y) => Select(x, y, selector)
Type ptype1 = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
Type ptype2 = typeof(IEnumerable<>).MakeGenericType(typeof(IndexedValue<long>));
ParameterExpression param1 = Expression.Parameter(ptype1, DryadLinqCodeGen.MakeUniqueName("x"));
ParameterExpression param2 = Expression.Parameter(ptype2, DryadLinqCodeGen.MakeUniqueName("y"));
string methodName = queryExpr.Method.Name;
Type[] selectorTypeArgs = selector.Type.GetGenericArguments();
Type typeArg2 = selectorTypeArgs[selectorTypeArgs.Length - 1];
if (nodeType == QueryNodeType.SelectMany)
{
if (resultSelector != null)
{
methodName += "Result";
}
typeArg2 = typeArg2.GetGenericArguments()[0];
}
string targetMethodName = methodName + "WithStartIndex";
MethodInfo minfo = typeof(DryadLinqEnumerable).GetMethod(targetMethodName);
Expression body;
if (resultSelector == null)
{
minfo = minfo.MakeGenericMethod(child.OutputTypes[0], typeArg2);
body = Expression.Call(minfo, param1, param2, selector);
}
else
{
minfo = minfo.MakeGenericMethod(child.OutputTypes[0], typeArg2, resultSelector.Body.Type);
body = Expression.Call(minfo, param1, param2, selector, resultSelector);
}
Type type = typeof(Func<,,>).MakeGenericType(ptype1, ptype2, body.Type);
LambdaExpression procFunc = Expression.Lambda(type, body, param1, param2);
bool isLong = methodName.StartsWith("Long", StringComparison.Ordinal);
DLinqQueryNode offsetNode = this.CreateOffset(isLong, queryExpr, child);
selectNode = new DLinqApplyNode(procFunc, queryExpr, child, offsetNode);
}
}
return selectNode;
}
private DLinqQueryNode VisitWhere(QueryNodeInfo source,
LambdaExpression predicate,
MethodCallExpression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
DLinqQueryNode whereNode;
if (predicate.Type.GetGenericArguments().Length == 2 ||
(!child.IsDynamic && child.OutputPartition.Count == 1))
{
whereNode = this.PromoteConcat(source, child, x => new DLinqWhereNode(predicate, queryExpr, x));
}
else
{
// The "indexed" version
// Create (x, y) => DryadWhere(x, y, predicate)
Type ptype1 = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
Type ptype2 = typeof(IEnumerable<>).MakeGenericType(typeof(IndexedValue<long>));
ParameterExpression param1 = Expression.Parameter(ptype1, DryadLinqCodeGen.MakeUniqueName("x"));
ParameterExpression param2 = Expression.Parameter(ptype2, DryadLinqCodeGen.MakeUniqueName("y"));
string targetMethod = queryExpr.Method.Name + "WithStartIndex";
MethodInfo minfo = typeof(DryadLinqEnumerable).GetMethod(targetMethod);
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
Expression body = Expression.Call(minfo, param1, param2, predicate);
Type type = typeof(Func<,,>).MakeGenericType(ptype1, ptype2, body.Type);
LambdaExpression procFunc = Expression.Lambda(type, body, param1, param2);
child.IsForked = true;
bool isLong = (queryExpr.Method.Name == "LongWhere");
DLinqQueryNode offsetNode = this.CreateOffset(isLong, queryExpr, child);
whereNode = new DLinqApplyNode(procFunc, queryExpr, child, offsetNode);
}
return whereNode;
}
private DLinqQueryNode VisitJoin(QueryNodeInfo outerSource,
QueryNodeInfo innerSource,
QueryNodeType nodeType,
LambdaExpression outerKeySelector,
LambdaExpression innerKeySelector,
LambdaExpression resultSelector,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode outerChild = this.Visit(outerSource);
DLinqQueryNode innerChild = this.Visit(innerSource);
DLinqQueryNode joinNode = null;
Type keyType = outerKeySelector.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
// The comparer object:
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
if (outerChild.IsDynamic || innerChild.IsDynamic)
{
// Well, let us do the simplest thing for now
outerChild = new DLinqHashPartitionNode(outerKeySelector,
comparerExpr,
StaticConfig.DefaultPartitionCount,
queryExpr,
outerChild);
if (IsMergeNodeNeeded(outerChild))
{
outerChild = new DLinqMergeNode(false, queryExpr, outerChild);
}
innerChild = new DLinqHashPartitionNode(innerKeySelector,
comparerExpr,
StaticConfig.DefaultPartitionCount,
queryExpr,
innerChild);
if (IsMergeNodeNeeded(innerChild))
{
innerChild = new DLinqMergeNode(false, queryExpr, innerChild);
}
joinNode = new DLinqJoinNode(nodeType,
"Hash" + nodeType,
outerKeySelector,
innerKeySelector,
resultSelector,
comparerExpr,
queryExpr,
outerChild,
innerChild);
return joinNode;
}
bool isOuterDescending = outerChild.OutputDataSetInfo.orderByInfo.IsDescending;
bool isInnerDescending = innerChild.OutputDataSetInfo.orderByInfo.IsDescending;
// Partition outer and inner if needed
if (outerChild.OutputPartition.ParType == PartitionType.Range &&
outerChild.OutputPartition.HasKeys &&
outerChild.OutputPartition.IsPartitionedBy(outerKeySelector, comparer))
{
if (innerChild.OutputPartition.ParType != PartitionType.Range ||
!innerChild.OutputPartition.IsPartitionedBy(innerKeySelector, comparer) ||
!outerChild.OutputPartition.IsSamePartition(innerChild.OutputPartition))
{
// Range distribute inner using outer's partition.
innerChild = outerChild.OutputPartition.CreatePartitionNode(innerKeySelector, innerChild);
if (IsMergeNodeNeeded(innerChild))
{
innerChild = new DLinqMergeNode(false, queryExpr, innerChild);
}
}
}
else if (innerChild.OutputPartition.ParType == PartitionType.Range &&
innerChild.OutputPartition.HasKeys &&
innerChild.OutputPartition.IsPartitionedBy(innerKeySelector, comparer))
{
// Range distribute outer using inner's partition.
outerChild = innerChild.OutputPartition.CreatePartitionNode(outerKeySelector, outerChild);
if (IsMergeNodeNeeded(outerChild))
{
outerChild = new DLinqMergeNode(false, queryExpr, outerChild);
}
}
else if (outerChild.OutputPartition.ParType == PartitionType.Hash &&
outerChild.OutputPartition.IsPartitionedBy(outerKeySelector, comparer))
{
if (innerChild.OutputPartition.ParType != PartitionType.Hash ||
!innerChild.OutputPartition.IsPartitionedBy(innerKeySelector, comparer) ||
!outerChild.OutputPartition.IsSamePartition(innerChild.OutputPartition))
{
innerChild = new DLinqHashPartitionNode(innerKeySelector,
comparerExpr,
outerChild.OutputPartition.Count,
queryExpr,
innerChild);
if (IsMergeNodeNeeded(innerChild))
{
innerChild = new DLinqMergeNode(false, queryExpr, innerChild);
}
}
}
else if (innerChild.OutputPartition.ParType == PartitionType.Hash &&
innerChild.OutputPartition.IsPartitionedBy(innerKeySelector, comparer))
{
outerChild = new DLinqHashPartitionNode(outerKeySelector,
comparerExpr,
innerChild.OutputPartition.Count,
queryExpr,
outerChild);
if (IsMergeNodeNeeded(outerChild))
{
outerChild = new DLinqMergeNode(false, queryExpr, outerChild);
}
}
else
{
// No luck. Hash partition both outer and inner
int parCnt = Math.Max(outerChild.OutputPartition.Count, innerChild.OutputPartition.Count);
if (parCnt > 1)
{
outerChild = new DLinqHashPartitionNode(outerKeySelector,
comparerExpr,
parCnt,
queryExpr,
outerChild);
if (IsMergeNodeNeeded(outerChild))
{
outerChild = new DLinqMergeNode(false, queryExpr, outerChild);
}
innerChild = new DLinqHashPartitionNode(innerKeySelector,
comparerExpr,
parCnt,
queryExpr,
innerChild);
if (IsMergeNodeNeeded(innerChild))
{
innerChild = new DLinqMergeNode(false, queryExpr, innerChild);
}
}
}
// Perform either merge or hash join
string opName = "Hash";
if (outerChild.IsOrderedBy(outerKeySelector, comparer))
{
if (!innerChild.IsOrderedBy(innerKeySelector, comparer) ||
isOuterDescending != isInnerDescending)
{
// Sort inner if unsorted
innerChild = new DLinqOrderByNode(innerKeySelector, comparerExpr,
isOuterDescending, queryExpr, innerChild);
}
opName = "Merge";
}
else if (innerChild.IsOrderedBy(innerKeySelector, comparer))
{
if (!outerChild.IsOrderedBy(outerKeySelector, comparer) ||
isOuterDescending != isInnerDescending)
{
// Sort outer if unsorted
outerChild = new DLinqOrderByNode(outerKeySelector, comparerExpr,
isInnerDescending, queryExpr, outerChild);
}
opName = "Merge";
}
joinNode = new DLinqJoinNode(nodeType,
opName + nodeType,
outerKeySelector,
innerKeySelector,
resultSelector,
comparerExpr,
queryExpr,
outerChild,
innerChild);
return joinNode;
}
private DLinqQueryNode VisitDistinct(QueryNodeInfo source,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = child.OutputTypes[0];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
LambdaExpression keySelectExpr = IdentityFunction.Instance(keyType);
if (!child.OutputPartition.IsPartitionedBy(keySelectExpr, comparer))
{
if (child.IsDynamic || child.OutputPartition.Count > 1)
{
child = new DLinqDistinctNode(true, comparerExpr, queryExpr, child);
bool isDynamic = (StaticConfig.DynamicOptLevel & StaticConfig.DynamicHashPartitionLevel) != 0;
child = new DLinqHashPartitionNode(keySelectExpr,
comparerExpr,
child.OutputPartition.Count,
isDynamic,
queryExpr,
child);
child = new DLinqMergeNode(false, queryExpr, child);
}
}
DLinqQueryNode resNode = new DLinqDistinctNode(false, comparerExpr, queryExpr, child);
return resNode;
}
private DLinqQueryNode VisitConcat(QueryNodeInfo source, MethodCallExpression queryExpr)
{
DLinqQueryNode[] childs = new DLinqQueryNode[source.Children.Count];
for (int i = 0; i < source.Children.Count; ++i)
{
childs[i] = this.Visit(source.Children[i].Child);
}
DLinqQueryNode resNode = new DLinqConcatNode(queryExpr, childs);
int parCount = resNode.OutputPartition.Count;
if (!resNode.IsDynamic && parCount > StaticConfig.MaxPartitionCount)
{
// Too many partitions, need to repartition
int newParCount = parCount / 2;
DLinqQueryNode countNode = new DLinqBasicAggregateNode(null, AggregateOpType.LongCount,
true, false, queryExpr, resNode);
DLinqQueryNode mergeCountNode = new DLinqMergeNode(true, queryExpr, countNode);
// Apply node for s => IndexedCount(s)
Type paramType = typeof(IEnumerable<>).MakeGenericType(typeof(long));
ParameterExpression param = Expression.Parameter(paramType, "s");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("IndexedCount");
minfo = minfo.MakeGenericMethod(typeof(long));
Expression body = Expression.Call(minfo, param);
Type funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression indexedCountFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode indexedCountNode = new DLinqApplyNode(indexedCountFunc, queryExpr, mergeCountNode);
// HashPartition(x => x.index, parCount)
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
Expression keySelectBody = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, keySelectBody.Type);
LambdaExpression keySelectExpr = Expression.Lambda(funcType, keySelectBody, param);
DLinqQueryNode distCountNode = new DLinqHashPartitionNode(keySelectExpr,
null,
parCount,
queryExpr,
indexedCountNode);
// Apply node for (x, y) => AddPartitionIndex(x, y, newParCount)
ParameterExpression param1 = Expression.Parameter(body.Type, "x");
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(resNode.OutputTypes[0]);
ParameterExpression param2 = Expression.Parameter(paramType2, "y");
minfo = typeof(DryadLinqHelper).GetMethod("AddPartitionIndex");
minfo = minfo.MakeGenericMethod(resNode.OutputTypes[0]);
body = Expression.Call(minfo, param1, param2, Expression.Constant(newParCount));
funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression addIndexFunc = Expression.Lambda(funcType, body, param1, param2);
DLinqQueryNode addIndexNode = new DLinqApplyNode(addIndexFunc, queryExpr, distCountNode, resNode);
// HashPartition(x => x.index, x => x.value, newParCount)
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
body = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
keySelectExpr = Expression.Lambda(funcType, body, param);
body = Expression.Property(param, "Value");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression resultSelectExpr = Expression.Lambda(funcType, body, param);
resNode = new DLinqHashPartitionNode(keySelectExpr,
resultSelectExpr,
null,
newParCount,
false,
queryExpr,
addIndexNode);
resNode = new DLinqMergeNode(true, queryExpr, resNode);
}
return resNode;
}
private DLinqQueryNode VisitSetOperation(QueryNodeInfo source1,
QueryNodeInfo source2,
QueryNodeType nodeType,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode child1 = this.Visit(source1);
DLinqQueryNode child2 = this.Visit(source2);
DLinqQueryNode resNode = null;
Type keyType = child1.OutputTypes[0];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
// The comparer object:
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
LambdaExpression keySelectExpr = IdentityFunction.Instance(keyType);
if (child1.IsDynamic || child2.IsDynamic)
{
// Well, let us do the simplest thing for now
child1 = new DLinqHashPartitionNode(keySelectExpr,
null,
StaticConfig.DefaultPartitionCount,
queryExpr,
child1);
if (IsMergeNodeNeeded(child1))
{
child1 = new DLinqMergeNode(false, queryExpr, child1);
}
child2 = new DLinqHashPartitionNode(keySelectExpr,
null,
StaticConfig.DefaultPartitionCount,
queryExpr,
child2);
if (IsMergeNodeNeeded(child2))
{
child2 = new DLinqMergeNode(false, queryExpr, child2);
}
resNode = new DLinqSetOperationNode(nodeType, nodeType.ToString(), comparerExpr,
queryExpr, child1, child2);
return resNode;
}
bool isDescending1 = child1.OutputDataSetInfo.orderByInfo.IsDescending;
bool isDescending2 = child2.OutputDataSetInfo.orderByInfo.IsDescending;
// Partition child1 and child2 if needed
if (child1.OutputPartition.ParType == PartitionType.Range &&
child1.OutputPartition.HasKeys &&
child1.OutputPartition.IsPartitionedBy(keySelectExpr, comparer))
{
if (child2.OutputPartition.ParType != PartitionType.Range ||
!child2.OutputPartition.IsPartitionedBy(keySelectExpr, comparer) ||
child1.OutputPartition.IsSamePartition(child2.OutputPartition))
{
// Range distribute child2 using child1's partition
child2 = child1.OutputPartition.CreatePartitionNode(keySelectExpr, child2);
if (IsMergeNodeNeeded(child2))
{
child2 = new DLinqMergeNode(false, queryExpr, child2);
}
}
}
else if (child2.OutputPartition.ParType == PartitionType.Range &&
child2.OutputPartition.HasKeys &&
child2.OutputPartition.IsPartitionedBy(keySelectExpr, comparer))
{
// Range distribute child1 using child2's partition
child1 = child2.OutputPartition.CreatePartitionNode(keySelectExpr, child1);
if (IsMergeNodeNeeded(child1))
{
child1 = new DLinqMergeNode(false, queryExpr, child1);
}
}
else if (child1.OutputPartition.ParType == PartitionType.Hash &&
child1.OutputPartition.IsPartitionedBy(keySelectExpr, comparer))
{
if (child2.OutputPartition.ParType != PartitionType.Hash ||
!child2.OutputPartition.IsPartitionedBy(keySelectExpr, comparer) ||
!child1.OutputPartition.IsSamePartition(child2.OutputPartition))
{
// Hash distribute child2:
child2 = new DLinqHashPartitionNode(keySelectExpr,
comparerExpr,
child1.OutputPartition.Count,
queryExpr,
child2);
if (IsMergeNodeNeeded(child2))
{
child2 = new DLinqMergeNode(false, queryExpr, child2);
}
}
}
else if (child2.OutputPartition.ParType == PartitionType.Hash &&
child2.OutputPartition.IsPartitionedBy(keySelectExpr, comparer))
{
child1 = new DLinqHashPartitionNode(keySelectExpr,
comparerExpr,
child2.OutputPartition.Count,
queryExpr,
child1);
if (IsMergeNodeNeeded(child1))
{
child1 = new DLinqMergeNode(false, queryExpr, child1);
}
}
else
{
// No luck. Hash distribute both child1 and child2, then perform hash operation
int parCnt = Math.Max(child1.OutputPartition.Count, child2.OutputPartition.Count);
if (parCnt > 1)
{
child1 = new DLinqHashPartitionNode(keySelectExpr, comparerExpr, parCnt, queryExpr, child1);
if (IsMergeNodeNeeded(child1))
{
child1 = new DLinqMergeNode(false, queryExpr, child1);
}
child2 = new DLinqHashPartitionNode(keySelectExpr, comparerExpr, parCnt, queryExpr, child2);
if (IsMergeNodeNeeded(child2))
{
child2 = new DLinqMergeNode(false, queryExpr, child2);
}
}
}
// Perform either hash or ordered operation
string opName = "";
if (child1.IsOrderedBy(keySelectExpr, comparer))
{
if (!child1.IsOrderedBy(keySelectExpr, comparer) ||
isDescending1 != isDescending2)
{
// Sort inner if unsorted
child2 = new DLinqOrderByNode(keySelectExpr, comparerExpr, isDescending1, queryExpr, child2);
}
opName = "Ordered";
}
else if (child2.IsOrderedBy(keySelectExpr, comparer))
{
if (!child1.IsOrderedBy(keySelectExpr, comparer) ||
isDescending1 != isDescending2)
{
// Sort outer if unsorted
child1 = new DLinqOrderByNode(keySelectExpr, comparerExpr, isDescending2, queryExpr, child1);
}
opName = "Ordered";
}
resNode = new DLinqSetOperationNode(nodeType, opName + nodeType, comparerExpr, queryExpr, child1, child2);
return resNode;
}
private DLinqQueryNode VisitContains(QueryNodeInfo source,
Expression valueExpr,
Expression comparerExpr,
bool isQuery,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = child.OutputTypes[0];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
DLinqQueryNode resNode = this.PromoteConcat(
source, child,
x => new DLinqContainsNode(valueExpr, comparerExpr, queryExpr, x));
resNode = new DLinqBasicAggregateNode(null, AggregateOpType.Any, false, isQuery, queryExpr, resNode);
return resNode;
}
private DLinqQueryNode VisitQuantifier(QueryNodeInfo source,
LambdaExpression lambda,
AggregateOpType aggType,
bool isQuery,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
DLinqQueryNode resNode = this.PromoteConcat(
source, child,
x => new DLinqBasicAggregateNode(
lambda, aggType, true, isQuery, queryExpr, x));
resNode = new DLinqBasicAggregateNode(null, aggType, false, isQuery, queryExpr, resNode);
return resNode;
}
private DLinqQueryNode VisitAggregate(QueryNodeInfo source,
Expression seed,
LambdaExpression funcLambda,
LambdaExpression resultLambda,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
DLinqQueryNode resNode = child;
if (DryadLinqExpression.IsAssociative(funcLambda))
{
LambdaExpression combinerLambda = DryadLinqExpression.GetAssociativeCombiner(funcLambda);
ResourceAttribute attrib = AttributeSystem.GetResourceAttrib(funcLambda);
bool funcIsExpensive = (attrib != null && attrib.IsExpensive);
resNode = this.PromoteConcat(
source, child,
x => new DLinqAggregateNode("AssocAggregate", seed,
funcLambda, combinerLambda, null,
1, queryExpr, x, false));
resNode = new DLinqAggregateNode("AssocAggregate", null, combinerLambda, combinerLambda, resultLambda,
3, queryExpr, resNode, funcIsExpensive);
}
else
{
resNode = new DLinqAggregateNode("Aggregate", seed, funcLambda, null, resultLambda, 3,
queryExpr, child, false);
}
return resNode;
}
private DLinqQueryNode VisitBasicAggregate(QueryNodeInfo source,
LambdaExpression lambda,
AggregateOpType aggType,
bool isQuery,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
if (aggType == AggregateOpType.Min || aggType == AggregateOpType.Max)
{
Type elemType = child.OutputTypes[0];
if (lambda != null)
{
elemType = lambda.Body.Type;
}
if (!TypeSystem.HasDefaultComparer(elemType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.AggregationOperatorRequiresIComparable,
String.Format(SR.AggregationOperatorRequiresIComparable, aggType ),
queryExpr);
}
}
DLinqQueryNode resNode = this.PromoteConcat(
source, child,
x => new DLinqBasicAggregateNode(lambda, aggType, true, isQuery, queryExpr, x));
switch (aggType)
{
case AggregateOpType.Count:
case AggregateOpType.LongCount:
{
resNode = new DLinqBasicAggregateNode(null, AggregateOpType.Sum, false,
isQuery, queryExpr, resNode);
break;
}
case AggregateOpType.Sum:
case AggregateOpType.Min:
case AggregateOpType.Max:
case AggregateOpType.Average:
{
resNode = new DLinqBasicAggregateNode(null, aggType, false,
isQuery, queryExpr, resNode);
break;
}
default:
{
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, aggType),
queryExpr);
}
}
return resNode;
}
private DLinqQueryNode VisitGroupBy(QueryNodeInfo source,
LambdaExpression keySelectExpr,
LambdaExpression elemSelectExpr,
LambdaExpression resultSelectExpr,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
ExpressionInfo einfo = new ExpressionInfo(keySelectExpr);
if (einfo.IsExpensive)
{
// Any method call that is not tagged as "expensive=false" will be deemed expensive.
// if the keySelector is expensive, we rewrite the query so that the key-function is invoked only once
// and the record key passed around via a Pair<TKey,TRecord>.
// keyFunc becomes pair=>pair.Key
// elementSelector must be rewritten so that references to (record) become (pair.Value)
Type[] vkTypes = keySelectExpr.Type.GetGenericArguments();
Type pairType = typeof(Pair<,>).MakeGenericType(vkTypes[1], vkTypes[0]);
ParameterExpression pairParam = Expression.Parameter(pairType, "e");
// Add Select(x => new Pair<K,S>(key(x), x))
ParameterExpression valueParam = keySelectExpr.Parameters[0];
Expression body = Expression.New(pairType.GetConstructors()[0], keySelectExpr.Body, valueParam);
Type delegateType = typeof(Func<,>).MakeGenericType(valueParam.Type, body.Type);
LambdaExpression selectExpr = Expression.Lambda(delegateType, body, valueParam);
child = new DLinqSelectNode(QueryNodeType.Select, selectExpr, null, queryExpr, child);
// Change keySelector to e => e.Key
PropertyInfo keyInfo = pairParam.Type.GetProperty("Key");
body = Expression.Property(pairParam, keyInfo);
delegateType = typeof(Func<,>).MakeGenericType(pairParam.Type, body.Type);
keySelectExpr = Expression.Lambda(delegateType, body, pairParam);
// Add or change elementSelector with e.Value
PropertyInfo valueInfo = pairParam.Type.GetProperty("Value");
body = Expression.Property(pairParam, valueInfo);
if (elemSelectExpr != null)
{
ParameterSubst subst = new ParameterSubst(elemSelectExpr.Parameters[0], body);
body = subst.Visit(elemSelectExpr.Body);
}
delegateType = typeof(Func<,>).MakeGenericType(pairParam.Type, body.Type);
elemSelectExpr = Expression.Lambda(delegateType, body, pairParam);
}
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
Type elemType;
if (elemSelectExpr == null)
{
elemType = keySelectExpr.Type.GetGenericArguments()[0];
}
else
{
elemType = elemSelectExpr.Type.GetGenericArguments()[1];
}
// The comparer object:
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
LambdaExpression keySelectExpr1 = keySelectExpr;
LambdaExpression elemSelectExpr1 = elemSelectExpr;
LambdaExpression resultSelectExpr1 = resultSelectExpr;
LambdaExpression seedExpr1 = null;
LambdaExpression accumulateExpr1 = null;
List<DecompositionInfo> dInfoList = null;
if (resultSelectExpr != null)
{
dInfoList = Decomposition.GetDecompositionInfoList(resultSelectExpr, this.m_codeGen);
}
String groupByOpName = "GroupBy";
DLinqQueryNode groupByNode = child;
bool isPartitioned = child.OutputPartition.IsPartitionedBy(keySelectExpr, comparer);
if (dInfoList != null)
{
// ** Decomposable GroupBy-Reduce
// This block creates the first GroupByNode and does some preparation for subsequent nodes.
if (child.IsOrderedBy(keySelectExpr, comparer))
{
groupByOpName = "OrderedGroupBy";
}
int dcnt = dInfoList.Count;
ParameterExpression keyParam;
if (resultSelectExpr.Parameters.Count == 1)
{
keyParam = Expression.Parameter(keyType, DryadLinqCodeGen.MakeUniqueName("k"));
}
else
{
keyParam = resultSelectExpr.Parameters[0];
}
// Seed:
ParameterExpression param2 = Expression.Parameter(
elemType, DryadLinqCodeGen.MakeUniqueName("e"));
Expression zeroExpr = Expression.Constant(0, typeof(int));
Expression seedBody = zeroExpr;
if (dcnt != 0)
{
LambdaExpression seed = dInfoList[dcnt-1].Seed;
ParameterSubst subst = new ParameterSubst(seed.Parameters[0], param2);
seedBody = subst.Visit(seed.Body);
for (int i = dcnt - 2; i >= 0; i--)
{
seed = dInfoList[i].Seed;
subst = new ParameterSubst(seed.Parameters[0], param2);
Expression firstExpr = subst.Visit(seed.Body);
Type newPairType = typeof(Pair<,>).MakeGenericType(firstExpr.Type, seedBody.Type);
seedBody = Expression.New(newPairType.GetConstructors()[0], firstExpr, seedBody);
}
}
LambdaExpression seedExpr = Expression.Lambda(seedBody, param2);
// Accumulate:
ParameterExpression param1 = Expression.Parameter(
seedBody.Type, DryadLinqCodeGen.MakeUniqueName("a"));
Expression accumulateBody = zeroExpr;
if (dcnt != 0)
{
accumulateBody = Decomposition.AccumulateList(param1, param2, dInfoList, 0);
}
LambdaExpression accumulateExpr = Expression.Lambda(accumulateBody, param1, param2);
// Now prepare for the merge-aggregator and/or in the secondary group-by.
// keySelectExpr1: e => e.Key
Type reducerResType = typeof(Pair<,>).MakeGenericType(keyParam.Type, accumulateBody.Type);
ParameterExpression reducerResParam = Expression.Parameter(reducerResType, "e");
PropertyInfo keyInfo = reducerResParam.Type.GetProperty("Key");
Expression body = Expression.Property(reducerResParam, keyInfo);
Type delegateType = typeof(Func<,>).MakeGenericType(reducerResParam.Type, body.Type);
keySelectExpr1 = Expression.Lambda(delegateType, body, reducerResParam);
// elemSelectExpr1: e => e.Value
PropertyInfo valueInfo = reducerResParam.Type.GetProperty("Value");
body = Expression.Property(reducerResParam, valueInfo);
delegateType = typeof(Func<,>).MakeGenericType(reducerResParam.Type, body.Type);
elemSelectExpr1 = Expression.Lambda(delegateType, body, reducerResParam);
// SeedExpr1
param2 = Expression.Parameter(elemSelectExpr1.Body.Type,
DryadLinqCodeGen.MakeUniqueName("e"));
seedExpr1 = Expression.Lambda(param2, param2);
// AccumulateExpr1
Expression recursiveAccumulateBody = zeroExpr;
if (dcnt != 0)
{
recursiveAccumulateBody = Decomposition.RecursiveAccumulateList(param1, param2, dInfoList, 0);
}
accumulateExpr1 = Expression.Lambda(recursiveAccumulateBody, param1, param2);
// resultSelectExpr1
resultSelectExpr1 = null;
// The first groupByNode.
// If the input was already correctly partitioned, this will be the only groupByNode.
bool isPartial = StaticConfig.GroupByLocalAggregationIsPartial && !isPartitioned;
groupByNode = new DLinqGroupByNode(
groupByOpName, keySelectExpr, elemSelectExpr, null,
seedExpr, accumulateExpr, accumulateExpr1, comparerExpr,
isPartial, queryExpr, child);
}
else
{
// Can't do partial aggregation.
// Use sort, mergesort, and ordered groupby, if TKey implements IComparable.
if ((comparer != null && TypeSystem.IsComparer(comparer, keyType)) ||
(comparer == null && TypeSystem.HasDefaultComparer(keyType)))
{
if (!child.IsOrderedBy(keySelectExpr, comparer))
{
groupByNode = new DLinqOrderByNode(keySelectExpr, comparerExpr, true, queryExpr, child);
}
groupByOpName = "OrderedGroupBy";
}
// Add a GroupByNode if it is partitioned or has elementSelector.
// If the input was already correctly partitioned, this will be the only groupByNode.
if (isPartitioned)
{
groupByNode = new DLinqGroupByNode(groupByOpName,
keySelectExpr,
elemSelectExpr,
resultSelectExpr,
null, // seed
null, // accumulate
null, // recursiveAccumulate
comparerExpr,
false, // isPartial
queryExpr,
groupByNode);
}
else if (elemSelectExpr != null)
{
// Local GroupBy without resultSelector:
groupByNode = new DLinqGroupByNode(groupByOpName,
keySelectExpr,
elemSelectExpr,
null, // resultSelect
null, // seed
null, // accumulate
null, // recursiveAccumulate
comparerExpr,
StaticConfig.GroupByLocalAggregationIsPartial, // isPartial
queryExpr,
groupByNode);
// keySelectExpr1: g => g.Key
ParameterExpression groupParam = Expression.Parameter(groupByNode.OutputTypes[0], "g");
PropertyInfo keyInfo = groupParam.Type.GetProperty("Key");
Expression body = Expression.Property(groupParam, keyInfo);
Type delegateType = typeof(Func<,>).MakeGenericType(groupParam.Type, body.Type);
keySelectExpr1 = Expression.Lambda(delegateType, body, groupParam);
// No elementSelector
elemSelectExpr1 = null;
// resultSelectExpr1
ParameterExpression keyParam;
Type groupType = typeof(IEnumerable<>).MakeGenericType(groupByNode.OutputTypes[0]);
groupParam = Expression.Parameter(groupType, DryadLinqCodeGen.MakeUniqueName("g"));
if (resultSelectExpr == null)
{
// resultSelectExpr1: (k, g) => MakeDryadLinqGroup(k, g)
keyParam = Expression.Parameter(keySelectExpr1.Body.Type, DryadLinqCodeGen.MakeUniqueName("k"));
MethodInfo groupingInfo = typeof(DryadLinqEnumerable).GetMethod("MakeDryadLinqGroup");
groupingInfo = groupingInfo.MakeGenericMethod(keyParam.Type, elemType);
body = Expression.Call(groupingInfo, keyParam, groupParam);
}
else
{
// resultSelectExpr1: (k, g) => resultSelectExpr(k, FlattenGroups(g))
keyParam = resultSelectExpr.Parameters[0];
MethodInfo flattenInfo = typeof(DryadLinqEnumerable).GetMethod("FlattenGroups");
flattenInfo = flattenInfo.MakeGenericMethod(keyParam.Type, elemType);
Expression groupExpr = Expression.Call(flattenInfo, groupParam);
ParameterSubst subst = new ParameterSubst(resultSelectExpr.Parameters[1], groupExpr);
body = subst.Visit(resultSelectExpr.Body);
}
delegateType = typeof(Func<,,>).MakeGenericType(keyParam.Type, groupParam.Type, body.Type);
resultSelectExpr1 = Expression.Lambda(delegateType, body, keyParam, groupParam);
}
}
// At this point, the first GroupByNode has been created.
DLinqQueryNode groupByNode1 = groupByNode;
DLinqMergeNode mergeNode = null;
if (!isPartitioned)
{
// Create HashPartitionNode, MergeNode, and second GroupByNode
// Note, if we are doing decomposable-GroupByReduce, there is still some work to go after this
// - attach the combiner to the first merge-node
// - attach the combiner to the merge-node
// - attach finalizer to second GroupBy
int parCount = (groupByNode.IsDynamic) ? StaticConfig.DefaultPartitionCount : groupByNode.OutputPartition.Count;
bool isDynamic = (StaticConfig.DynamicOptLevel & StaticConfig.DynamicHashPartitionLevel) != 0;
DLinqQueryNode hdistNode = new DLinqHashPartitionNode(keySelectExpr1, comparerExpr, parCount,
isDynamic, queryExpr, groupByNode);
// Create the Merge Node
if (groupByOpName == "OrderedGroupBy")
{
// Mergesort with the same keySelector of the hash partition
mergeNode = new DLinqMergeNode(keySelectExpr1, comparerExpr, true, queryExpr, hdistNode);
}
else
{
// Random merge
mergeNode = new DLinqMergeNode(false, queryExpr, hdistNode);
}
groupByNode1 = new DLinqGroupByNode(groupByOpName, keySelectExpr1, elemSelectExpr1,
resultSelectExpr1, seedExpr1, accumulateExpr1,
accumulateExpr1, comparerExpr, false, queryExpr,
mergeNode);
}
// Final tidy-up for decomposable GroupBy-Reduce pattern.
// - attach combiner to first GroupByNode
// - attache combiner to MergeNode as an aggregator
// - build a SelectNode to project out results and call finalizer on them.
if (dInfoList != null)
{
// Add dynamic aggregator to the merge-node, if applicable
if (StaticConfig.GroupByDynamicReduce && !isPartitioned)
{
mergeNode.AddAggregateNode(groupByNode1);
}
// Add the final Select node
Type keyResultPairType = typeof(Pair<,>).MakeGenericType(keyType, seedExpr1.Body.Type);
ParameterExpression keyResultPairParam = Expression.Parameter(keyResultPairType,
DryadLinqCodeGen.MakeUniqueName("e"));
PropertyInfo valuePropInfo_1 = keyResultPairType.GetProperty("Value");
Expression combinedValueExpr = Expression.Property(keyResultPairParam, valuePropInfo_1);
// First, build the combinerList
int dcnt = dInfoList.Count;
Expression[] combinerList = new Expression[dcnt];
for (int i = 0; i < dcnt; i++)
{
if (i + 1 == dcnt)
{
combinerList[i] = combinedValueExpr;
}
else
{
PropertyInfo keyPropInfo = combinedValueExpr.Type.GetProperty("Key");
combinerList[i] = Expression.Property(combinedValueExpr, keyPropInfo);
PropertyInfo valuePropInfo = combinedValueExpr.Type.GetProperty("Value");
combinedValueExpr = Expression.Property(combinedValueExpr, valuePropInfo);
}
LambdaExpression finalizerExpr = dInfoList[i].FinalReducer;
if (finalizerExpr != null)
{
ParameterSubst subst = new ParameterSubst(finalizerExpr.Parameters[0], combinerList[i]);
combinerList[i] = subst.Visit(finalizerExpr.Body);
}
}
// Build the funcList
Expression[] funcList = new Expression[dcnt];
for (int i = 0; i < dcnt; i++)
{
funcList[i] = dInfoList[i].Func;
}
// Apply the substitutions
CombinerSubst combinerSubst = new CombinerSubst(resultSelectExpr, keyResultPairParam, funcList, combinerList);
Expression finalizerSelectBody = combinerSubst.Visit();
// Finally, the Select node
Type delegateType = typeof(Func<,>).MakeGenericType(keyResultPairType, finalizerSelectBody.Type);
LambdaExpression selectExpr = Expression.Lambda(delegateType, finalizerSelectBody, keyResultPairParam);
groupByNode1 = new DLinqSelectNode(QueryNodeType.Select, selectExpr, null, queryExpr, groupByNode1);
}
return groupByNode1;
}
// Creates an "auto-sampling range-partition sub-query"
private DLinqQueryNode CreateRangePartition(bool isDynamic,
LambdaExpression keySelectExpr,
LambdaExpression resultSelectExpr,
Expression comparerExpr,
Expression isDescendingExpr,
Expression queryExpr,
Expression partitionCountExpr,
DLinqQueryNode child)
{
// Make child a Tee node
child.IsForked = true;
// The partition count
Expression countExpr = null;
if (isDescendingExpr == null)
{
isDescendingExpr = Expression.Constant(false, typeof(bool)); //default for isDescending is false.
}
// NOTE: for MayRTM, isDynamic should never be true
if (!isDynamic)
{
if (partitionCountExpr != null)
{
countExpr = partitionCountExpr;
}
else
{
// If partitionCount was not explicitly set, use the child's partition count.
countExpr = Expression.Constant(child.OutputPartition.Count);
}
}
Type recordType = child.OutputTypes[0];
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
// Create x => Phase1Sampling(x_1, keySelector, denv)
Type lambdaParamType1 = typeof(IEnumerable<>).MakeGenericType(recordType);
ParameterExpression lambdaParam1 = Expression.Parameter(lambdaParamType1, "x_1");
ParameterExpression denvParam = Expression.Parameter(typeof(VertexEnv), "denv");
MethodInfo minfo = typeof(DryadLinqSampler).GetMethod("Phase1Sampling");
Expression body = Expression.Call(minfo.MakeGenericMethod(recordType, keyType),
lambdaParam1, keySelectExpr, denvParam);
Type type = typeof(Func<,>).MakeGenericType(lambdaParam1.Type, body.Type);
LambdaExpression samplingExpr = Expression.Lambda(type, body, lambdaParam1);
// Create the Sampling node
DLinqApplyNode samplingNode = new DLinqApplyNode(samplingExpr, queryExpr, child);
// Create x => RangeSampler(x, keySelectExpr, comparer, isDescendingExpr)
Type lambdaParamType = typeof(IEnumerable<>).MakeGenericType(keyType);
ParameterExpression lambdaParam = Expression.Parameter(lambdaParamType, "x_2");
//For RTM, isDynamic should never be true.
//string methodName = (isDynamic) ? "RangeSampler_Dynamic" : "RangeSampler_Static";
Debug.Assert(isDynamic == false, "Internal error: isDynamic is true.");
string methodName = "RangeSampler_Static";
minfo = typeof(DryadLinqSampler).GetMethod(methodName);
minfo = minfo.MakeGenericMethod(keyType);
Expression comparerArgExpr = comparerExpr;
if (comparerExpr == null)
{
if (!TypeSystem.HasDefaultComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable, keyType),
queryExpr);
}
comparerArgExpr = Expression.Constant(null, typeof(IComparer<>).MakeGenericType(keyType));
}
Expression lastArg;
if (isDynamic)
{
lastArg = denvParam;
}
else
{
lastArg = countExpr;
}
body = Expression.Call(minfo, lambdaParam, comparerArgExpr, isDescendingExpr, lastArg);
type = typeof(Func<,>).MakeGenericType(lambdaParam.Type, body.Type);
LambdaExpression samplerExpr = Expression.Lambda(type, body, lambdaParam);
// Create the sample node
DLinqQueryNode sampleDataNode = new DLinqMergeNode(false, queryExpr, samplingNode);
DLinqQueryNode sampleNode = new DLinqApplyNode(samplerExpr, queryExpr, sampleDataNode);
sampleNode.IsForked = true;
// Create the range distribute node
DLinqQueryNode resNode = new DLinqRangePartitionNode(keySelectExpr,
resultSelectExpr,
null,
comparerExpr,
isDescendingExpr,
countExpr,
queryExpr,
child,
sampleNode);
resNode = new DLinqMergeNode(false, queryExpr, resNode);
// Set the dynamic manager for sampleNode
if (isDynamic)
{
sampleDataNode.DynamicManager = new DynamicRangeDistributor(resNode);
}
return resNode;
}
private DLinqQueryNode VisitOrderBy(QueryNodeInfo source,
LambdaExpression keySelectExpr,
Expression comparerExpr,
bool isDescending,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable, keyType),
queryExpr);
}
DLinqQueryNode resNode = child;
if (child.OutputPartition.ParType == PartitionType.Range &&
child.OutputPartition.IsPartitionedBy(keySelectExpr, comparerExpr, isDescending))
{
// Only need to sort each partition
resNode = new DLinqOrderByNode(keySelectExpr, comparerExpr, isDescending, queryExpr, child);
}
else
{
Expression isDescendingExpr = Expression.Constant(isDescending);
bool dynamicOptEnabled = (StaticConfig.DynamicOptLevel & StaticConfig.DynamicRangePartitionLevel) != 0;
if (dynamicOptEnabled || child.IsDynamic)
{
// NOTE: for MayRTM, this path should not be taken.
resNode = this.CreateRangePartition(true, keySelectExpr, null, comparerExpr,
isDescendingExpr, queryExpr, null, child);
resNode = new DLinqOrderByNode(keySelectExpr, comparerExpr, isDescending, queryExpr, resNode);
}
else
{
if (child.OutputPartition.Count > 1)
{
resNode = this.CreateRangePartition(false, keySelectExpr, null, comparerExpr,
isDescendingExpr, queryExpr, null, child);
}
resNode = new DLinqOrderByNode(keySelectExpr, comparerExpr, isDescending, queryExpr, resNode);
}
}
return resNode;
}
private DLinqQueryNode FirstStagePartitionOp(string opName,
QueryNodeType nodeType,
Expression controlExpr,
MethodCallExpression queryExpr,
DLinqQueryNode child)
{
if (nodeType == QueryNodeType.TakeWhile)
{
Type ptype = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param = Expression.Parameter(ptype, DryadLinqCodeGen.MakeUniqueName("x"));
MethodInfo minfo = typeof(DryadLinqEnumerable).GetMethod("GroupTakeWhile");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
Expression body = Expression.Call(minfo, param, controlExpr);
Type type = typeof(Func<,>).MakeGenericType(ptype, body.Type);
LambdaExpression procFunc = Expression.Lambda(type, body, param);
return new DLinqApplyNode(procFunc, queryExpr, child);
}
else
{
return new DLinqPartitionOpNode(opName, nodeType, controlExpr, true, queryExpr, child);
}
}
private DLinqQueryNode VisitPartitionOp(string opName,
QueryNodeInfo source,
QueryNodeType nodeType,
Expression controlExpr,
MethodCallExpression queryExpr)
{
DLinqQueryNode resNode;
if (nodeType == QueryNodeType.TakeWhile &&
controlExpr.Type.GetGenericArguments().Length != 2)
{
// The "indexed" version.
resNode = this.Visit(source);
// The following block used to be skipped for resNode.OutputPartition.Count == 1,
// which causes compilation error (bug 13593)
// @@TODO[p3] : implement a working optimization for nPartition==1 that calls
// directly to Linq TakeWhile.
// Note: the test is: if (resNode.IsDynamic || resNode.OutputPartition.Count > 1)
{
resNode.IsForked = true;
bool isLong = (queryExpr.Method.Name == "LongTakeWhile");
DLinqQueryNode offsetNode = this.CreateOffset(isLong, queryExpr, resNode);
// Create (x, y) => GroupIndexedTakeWhile(x, y, controlExpr)
Type ptype1 = typeof(IEnumerable<>).MakeGenericType(resNode.OutputTypes[0]);
Type ptype2 = typeof(IEnumerable<>).MakeGenericType(typeof(IndexedValue<long>));
ParameterExpression param1 = Expression.Parameter(ptype1, DryadLinqCodeGen.MakeUniqueName("x"));
ParameterExpression param2 = Expression.Parameter(ptype2, DryadLinqCodeGen.MakeUniqueName("y"));
string methodName = "GroupIndexed" + queryExpr.Method.Name;
MethodInfo minfo = typeof(DryadLinqEnumerable).GetMethod(methodName);
minfo = minfo.MakeGenericMethod(resNode.OutputTypes[0]);
Expression body = Expression.Call(minfo, param1, param2, controlExpr);
Type type = typeof(Func<,,>).MakeGenericType(ptype1, ptype2, body.Type);
LambdaExpression procFunc = Expression.Lambda(type, body, param1, param2);
resNode = new DLinqApplyNode(procFunc, queryExpr, resNode, offsetNode);
}
}
else if (!source.IsForked &&
(nodeType == QueryNodeType.Take || nodeType == QueryNodeType.TakeWhile) &&
(source.OperatorName == "OrderBy" || source.OperatorName == "OrderByDescending"))
{
resNode = this.Visit(source.Children[0].Child);
bool isDescending = (source.OperatorName == "OrderByDescending");
MethodCallExpression sourceQueryExpr = (MethodCallExpression)source.QueryExpression;
LambdaExpression keySelectExpr = DryadLinqExpression.GetLambda(sourceQueryExpr.Arguments[1]);
Expression comparerExpr = null;
if (sourceQueryExpr.Arguments.Count == 3)
{
comparerExpr = sourceQueryExpr.Arguments[2];
}
resNode = this.PromoteConcat(
source.Children[0].Child,
resNode,
delegate(DLinqQueryNode x) {
DLinqQueryNode y = new DLinqOrderByNode(keySelectExpr, comparerExpr,
isDescending, sourceQueryExpr, x);
return FirstStagePartitionOp(opName, nodeType, controlExpr, queryExpr, y);
});
if (resNode.IsDynamic || resNode.OutputPartition.Count > 1)
{
// Need a mergesort
resNode = new DLinqMergeNode(keySelectExpr, comparerExpr, isDescending, sourceQueryExpr, resNode);
}
}
else
{
resNode = this.Visit(source);
if (nodeType == QueryNodeType.Take || nodeType == QueryNodeType.TakeWhile)
{
resNode = this.PromoteConcat(
source, resNode,
x => FirstStagePartitionOp(opName, nodeType, controlExpr, queryExpr, x));
}
}
resNode = new DLinqPartitionOpNode(opName, nodeType, controlExpr, false, queryExpr, resNode);
return resNode;
}
private DLinqQueryNode VisitZip(QueryNodeInfo first,
QueryNodeInfo second,
LambdaExpression resultSelector,
MethodCallExpression queryExpr)
{
DLinqQueryNode child1 = this.Visit(first);
DLinqQueryNode child2 = this.Visit(second);
if (child1.IsDynamic || child2.IsDynamic)
{
// Well, let us for now do it on a single machine
child1 = new DLinqMergeNode(true, queryExpr, child1);
child2 = new DLinqMergeNode(true, queryExpr, child2);
// Apply node for (x, y) => Zip(x, y, resultSelector)
Type paramType1 = typeof(IEnumerable<>).MakeGenericType(child1.OutputTypes[0]);
ParameterExpression param1 = Expression.Parameter(paramType1, "s1");
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(child2.OutputTypes[0]);
ParameterExpression param2 = Expression.Parameter(paramType2, "s2");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("Zip");
minfo = minfo.MakeGenericMethod(child1.OutputTypes[0]);
Expression body = Expression.Call(minfo, param1, param2, resultSelector);
Type funcType = typeof(Func<,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(funcType, body, param1, param2);
return new DLinqApplyNode(procFunc, queryExpr, child1, child2);
}
else
{
int parCount1 = child1.OutputPartition.Count;
int parCount2 = child2.OutputPartition.Count;
// Count nodes
DLinqQueryNode countNode1 = new DLinqBasicAggregateNode(null, AggregateOpType.LongCount,
true, false, queryExpr, child1);
DLinqQueryNode countNode2 = new DLinqBasicAggregateNode(null, AggregateOpType.LongCount,
true, false, queryExpr, child2);
countNode1 = new DLinqMergeNode(true, queryExpr, countNode1);
countNode2 = new DLinqMergeNode(true, queryExpr, countNode2);
// Apply node for (x, y) => ZipCount(x, y)
Type paramType1 = typeof(IEnumerable<>).MakeGenericType(typeof(long));
ParameterExpression param1 = Expression.Parameter(paramType1, "x");
ParameterExpression param2 = Expression.Parameter(paramType1, "y");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("ZipCount");
Expression body = Expression.Call(minfo, param1, param2);
Type funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression zipCount = Expression.Lambda(funcType, body, param1, param2);
DLinqQueryNode indexedCountNode = new DLinqApplyNode(zipCount, queryExpr, countNode1, countNode2);
// HashPartition(x => x.index, parCount2)
ParameterExpression param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
Expression keySelectBody = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, keySelectBody.Type);
LambdaExpression keySelectExpr = Expression.Lambda(funcType, keySelectBody, param);
DLinqQueryNode distCountNode = new DLinqHashPartitionNode(keySelectExpr,
null,
parCount2,
queryExpr,
indexedCountNode);
// Apply node for (x, y) => AssignPartitionIndex(x, y)
param1 = Expression.Parameter(body.Type, "x");
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(child2.OutputTypes[0]);
param2 = Expression.Parameter(paramType2, "y");
minfo = typeof(DryadLinqHelper).GetMethod("AssignPartitionIndex");
minfo = minfo.MakeGenericMethod(child2.OutputTypes[0]);
body = Expression.Call(minfo, param1, param2);
funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression assignIndex = Expression.Lambda(funcType, body, param1, param2);
DLinqQueryNode addIndexNode = new DLinqApplyNode(assignIndex, queryExpr, distCountNode, child2);
// HashPartition(x => x.index, x => x.value, parCount1)
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
body = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
keySelectExpr = Expression.Lambda(funcType, body, param);
body = Expression.Property(param, "Value");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression resultSelectExpr = Expression.Lambda(funcType, body, param);
DLinqQueryNode newChild2 = new DLinqHashPartitionNode(keySelectExpr,
resultSelectExpr,
null,
parCount1,
false,
queryExpr,
addIndexNode);
newChild2 = new DLinqMergeNode(true, queryExpr, newChild2);
// Finally the zip node
return new DLinqZipNode(resultSelector, queryExpr, child1, newChild2);
}
}
// Basic plan: (reverse all partitions) then (reverse data in each partition)
// The main complication is to perform the first step.
// Approach:
// - tee the input.
// - have a dummy apply node that produces the singleton {0} at each partition
// - merge to get a seq {0,0,..} whose length = nPartition.
// - convert that seq to { (0,n), (1,n), ...}
// - hash-partition to send one item to each of the n workers.
// - use binary-apply to attach targetIndex to each source item
// Apply( seq1 = indexCountPair, seq2 = original data) => ({tgt, item0}, {tgt, item1}, .. )
// - hash-partition to move items to target partition.
// - use local LINQ reverse to do the local data reversal.
private DLinqQueryNode VisitReverse(QueryNodeInfo source, Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
if (child.IsDynamic)
{
throw new DryadLinqException("Reverse is only supported for static partition count");
}
child.IsForked = true;
// Apply node for s => ValueZero(s)
Type paramType = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param = Expression.Parameter(paramType, "s");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("ValueZero");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
Expression body = Expression.Call(minfo, param);
Type funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode valueZeroNode = new DLinqApplyNode(procFunc, queryExpr, child);
// Apply node for s => ReverseIndex(s)
paramType = typeof(IEnumerable<>).MakeGenericType(typeof(int));
param = Expression.Parameter(paramType, "s");
minfo = typeof(DryadLinqHelper).GetMethod("MakeIndexCountPairs");
body = Expression.Call(minfo, param);
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode mergeZeroNode = new DLinqMergeNode(true, queryExpr, valueZeroNode);
DLinqQueryNode indexCountNode = new DLinqApplyNode(procFunc, queryExpr, mergeZeroNode);
// HashPartition to distribute the indexCounts -- one to each partition.
// each partition will receive (myPartitionID, pcount).
int pcount = child.OutputPartition.Count;
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
Expression keySelectBody = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, keySelectBody.Type);
LambdaExpression keySelectExpr = Expression.Lambda(funcType, keySelectBody, param);
DLinqQueryNode hdistNode = new DLinqHashPartitionNode(keySelectExpr,
null,
pcount,
queryExpr,
indexCountNode);
// Apply node for (x, y) => AddIndexForReverse(x, y)
ParameterExpression param1 = Expression.Parameter(body.Type, "x");
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param2 = Expression.Parameter(paramType2, "y");
minfo = typeof(DryadLinqHelper).GetMethod("AddIndexForReverse");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
body = Expression.Call(minfo, param1, param2);
funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression addIndexFunc = Expression.Lambda(funcType, body, param1, param2);
DLinqQueryNode addIndexNode = new DLinqApplyNode(addIndexFunc, queryExpr, hdistNode, child);
// HashPartition(x => x.index, x => x.value, pcount)
// Moves all data to correct target partition. (each worker will direct all its items to one target partition)
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
body = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
keySelectExpr = Expression.Lambda(funcType, body, param);
body = Expression.Property(param, "Value");
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression resultSelectExpr = Expression.Lambda(funcType, body, param);
DLinqQueryNode reversePartitionNode = new DLinqHashPartitionNode(
keySelectExpr, resultSelectExpr, null,
pcount, false, queryExpr, addIndexNode);
// Reverse node
paramType = typeof(IEnumerable<>).MakeGenericType(reversePartitionNode.OutputTypes[0]);
param = Expression.Parameter(paramType, "x");
minfo = typeof(DryadLinqVertex).GetMethod("Reverse");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
body = Expression.Call(minfo, param);
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode resNode = new DLinqMergeNode(true, queryExpr, reversePartitionNode);
resNode = new DLinqApplyNode(procFunc, queryExpr, resNode);
return resNode;
}
private DLinqQueryNode VisitSequenceEqual(QueryNodeInfo source1,
QueryNodeInfo source2,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode child1 = this.Visit(source1);
DLinqQueryNode child2 = this.Visit(source2);
Type elemType = child1.OutputTypes[0];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(elemType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerExpressionMustBeSpecifiedOrElementTypeMustBeIEquatable,
String.Format(SR.ComparerExpressionMustBeSpecifiedOrElementTypeMustBeIEquatable, elemType),
queryExpr);
}
// Well, let us do it on a single machine for now
child1 = new DLinqMergeNode(true, queryExpr, child1);
child2 = new DLinqMergeNode(true, queryExpr, child2);
// Apply node for (x, y) => SequenceEqual(x, y, c)
Type paramType = typeof(IEnumerable<>).MakeGenericType(elemType);
ParameterExpression param1 = Expression.Parameter(paramType, "s1");
ParameterExpression param2 = Expression.Parameter(paramType, "s2");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("SequenceEqual");
minfo = minfo.MakeGenericMethod(elemType);
if (comparerExpr == null)
{
comparerExpr = Expression.Constant(null, typeof(IEqualityComparer<>).MakeGenericType(elemType));
}
Expression body = Expression.Call(minfo, param1, param2, comparerExpr);
Type funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(funcType, body, param1, param2);
return new DLinqApplyNode(procFunc, queryExpr, child1, child2);
}
private DLinqQueryNode VisitHashPartition(QueryNodeInfo source,
LambdaExpression keySelectExpr,
LambdaExpression resultSelectExpr,
Expression comparerExpr,
Expression countExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType),
queryExpr);
}
bool isDynamic = (StaticConfig.DynamicOptLevel & StaticConfig.DynamicHashPartitionLevel) != 0;
int nOutputPartitions;
if (countExpr != null)
{
ExpressionSimplifier<int> evaluator = new ExpressionSimplifier<int>();
nOutputPartitions = evaluator.Eval(countExpr);
isDynamic = false;
}
else
{
// Note: For MayRTM, isDynamic will never be true.
nOutputPartitions = (isDynamic) ? 1 : child.OutputPartition.Count;
}
DLinqQueryNode resNode = new DLinqHashPartitionNode(
keySelectExpr, resultSelectExpr, comparerExpr, nOutputPartitions,
isDynamic, queryExpr, child);
resNode = new DLinqMergeNode(false, queryExpr, resNode);
return resNode;
}
private DLinqQueryNode VisitRangePartition(QueryNodeInfo source,
LambdaExpression keySelectExpr,
Expression keysExpr,
Expression comparerExpr,
Expression isDescendingExpr,
Expression partitionCountExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable,
string.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable, keyType),
queryExpr);
}
DLinqQueryNode resNode;
if (keysExpr == null)
{
// case: no keys are provided -- create range partitioner with auto-separator-selection
bool dynamicOptEnabled = (StaticConfig.DynamicOptLevel & StaticConfig.DynamicRangePartitionLevel) != 0;
bool useDynamic = (dynamicOptEnabled || child.IsDynamic);
// NOTE: for MayRTM, useDynamic should always be false
resNode = this.CreateRangePartition(useDynamic, keySelectExpr, null, comparerExpr, isDescendingExpr,
queryExpr, partitionCountExpr, child);
}
else
{
// case: keys are local enum (eg an array) -- create range partitioner with keys input via Object-store.
resNode = new DLinqRangePartitionNode(keySelectExpr, null, keysExpr, comparerExpr,
isDescendingExpr, null, queryExpr, child);
resNode = new DLinqMergeNode(false, queryExpr, resNode);
}
return resNode;
}
private DLinqQueryNode VisitApply(QueryNodeInfo source1,
QueryNodeInfo source2,
LambdaExpression procLambda,
bool perPartition,
bool isFirstOnly,
Expression queryExpr)
{
DLinqQueryNode child1 = this.Visit(source1);
DLinqQueryNode applyNode;
if (source2 == null)
{
// Unary-apply case:
if (perPartition)
{
// homomorphic
applyNode = this.PromoteConcat(source1, child1, x => new DLinqApplyNode(procLambda, queryExpr, x));
}
else
{
// non-homomorphic
if (child1.IsDynamic || child1.OutputPartition.Count > 1)
{
child1 = new DLinqMergeNode(true, queryExpr, child1);
}
applyNode = new DLinqApplyNode(procLambda, queryExpr, child1);
}
}
else
{
// Binary-apply case:
DLinqQueryNode child2 = this.Visit(source2);
if (perPartition && isFirstOnly)
{
// The function is left homomorphic:
if (child1.IsDynamic || child1.OutputPartition.Count > 1)
{
// The normal cases..
if (IsMergeNodeNeeded(child2))
{
if (child1.IsDynamic)
{
child2 = new DLinqMergeNode(true, queryExpr, child2);
child2.IsForked = true;
}
else
{
// Rather than do full merge and broadcast, which has lots of data-movement
// 1. Tee output2 with output cross-product
// 2. Do a merge-stage which will have input1.nPartition nodes each performing a merge.
// This acheives a distribution of the entire input2 to the Apply nodes with least data-movement.
child2 = new DLinqTeeNode(child2.OutputTypes[0], true, queryExpr, child2);
child2.ConOpType = ConnectionOpType.CrossProduct;
child2 = new DLinqMergeNode(child1.OutputPartition.Count, queryExpr, child2);
}
}
else
{
// the right-data is alread a single partition, so just tee it.
// this will provide a copy to each of the apply nodes.
child2 = new DLinqTeeNode(child2.OutputTypes[0], true, queryExpr, child2);
}
}
else
{
// a full merge of the right-data may be necessary.
if (child2.IsDynamic || child2.OutputPartition.Count > 1)
{
child2 = new DLinqMergeNode(true, queryExpr, child2);
if (child1.IsDynamic || child1.OutputPartition.Count > 1)
{
child2.IsForked = true;
}
}
}
applyNode = new DLinqApplyNode(procLambda, queryExpr, child1, child2);
}
else if (perPartition && !isFirstOnly && !child1.IsDynamic && !child2.IsDynamic)
{
// Full homomorphic
// No merging occurs.
// NOTE: We generally expect that both the left and right datasets have matching partitionCount.
// however, we don't test for it yet as users might know what they are doing, and it makes
// LocalDebug inconsistent as LocalDebug doesn't throw in that situation.
applyNode = new DLinqApplyNode(procLambda, queryExpr, child1, child2);
}
else
{
// Non-homomorphic
// Full merges of both data sets is necessary.
if (child1.IsDynamic || child1.OutputPartition.Count > 1)
{
child1 = new DLinqMergeNode(true, queryExpr, child1);
}
if (child2.IsDynamic || child2.OutputPartition.Count > 1)
{
child2 = new DLinqMergeNode(true, queryExpr, child2);
}
applyNode = new DLinqApplyNode(procLambda, queryExpr, child1, child2);
}
}
return applyNode;
}
private DLinqQueryNode VisitMultiApply(QueryNodeInfo source,
LambdaExpression procLambda,
bool perPartition,
bool isFirstOnly,
MethodCallExpression queryExpr)
{
DLinqQueryNode[] childs = new DLinqQueryNode[source.Children.Count];
for (int i = 0; i < source.Children.Count; ++i)
{
childs[i] = this.Visit(source.Children[i].Child);
}
bool isDynamic = childs.Any(x => x.IsDynamic);
if (perPartition && !isDynamic)
{
// Homomorphic case.
if (isFirstOnly)
{
for (int i = 1; i < childs.Length; ++i)
{
childs[i] = new DLinqTeeNode(childs[i].OutputTypes[0], true, queryExpr, childs[i]);
childs[i].ConOpType = ConnectionOpType.CrossProduct;
childs[i] = new DLinqMergeNode(childs[0].OutputPartition.Count, queryExpr, childs[i]);
}
}
else
{
int count = childs[0].OutputPartition.Count;
for (int i = 1; i < childs.Length; ++i)
{
if (childs[i].OutputPartition.Count != count)
{
throw DryadLinqException.Create(DryadLinqErrorCode.HomomorphicApplyNeedsSamePartitionCount,
SR.HomomorphicApplyNeedsSamePartitionCount,
queryExpr);
}
}
}
}
else
{
// Non-homomorphic case.
for (int i = 0; i < childs.Length; ++i)
{
if (childs[i].IsDynamic || childs[i].OutputPartition.Count > 1)
{
childs[i] = new DLinqMergeNode(true, queryExpr, childs[i]);
}
}
}
DLinqQueryNode applyNode = new DLinqApplyNode(procLambda, true, queryExpr, childs);
return applyNode;
}
private DLinqQueryNode VisitFork(QueryNodeInfo source,
LambdaExpression forkLambda,
Expression keysExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
return new DLinqForkNode(forkLambda, keysExpr, queryExpr, child);
}
private DLinqQueryNode VisitForkChoose(QueryNodeInfo source,
Expression indexExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
ExpressionSimplifier<int> evaluator = new ExpressionSimplifier<int>();
int index = evaluator.Eval(indexExpr);
return ((DLinqForkNode)child).Parents[index];
}
private DLinqQueryNode VisitAssumeHashPartition(QueryNodeInfo source,
LambdaExpression keySelectExpr,
Expression keysExpr,
Expression comparerExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultEqualityComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable,
String.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIEquatable, keyType.FullName),
queryExpr);
}
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
DataSetInfo outputInfo = new DataSetInfo(child.OutputDataSetInfo);
outputInfo.partitionInfo = PartitionInfo.CreateHash(keySelectExpr,
child.OutputPartition.Count,
comparer,
keyType);
child.OutputDataSetInfo = outputInfo;
return child;
}
private DLinqQueryNode VisitAssumeRangePartition(QueryNodeInfo source,
LambdaExpression keySelectExpr,
Expression keysExpr,
Expression comparerExpr,
Expression isDescendingExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable,
String.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable, keyType),
queryExpr);
}
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
object keys = null;
if (keysExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
keys = evaluator.Eval(keysExpr);
}
//count the number of keys provided.
if (keys != null)
{
int nSeparators = 0;
var ie = ((IEnumerable)keys).GetEnumerator();
while (ie.MoveNext())
{
nSeparators++;
}
if (!child.IsDynamic && nSeparators != child.PartitionCount - 1)
{
throw DryadLinqException.Create(
DryadLinqErrorCode.BadSeparatorCount,
String.Format(SR.BadSeparatorCount, nSeparators, child.PartitionCount - 1),
queryExpr);
}
}
bool? isDescending = null;
if (isDescendingExpr != null)
{
ExpressionSimplifier<bool> evaluator = new ExpressionSimplifier<bool>();
isDescending = evaluator.Eval(isDescendingExpr);
}
DataSetInfo outputInfo = new DataSetInfo(child.OutputDataSetInfo);
outputInfo.partitionInfo = PartitionInfo.CreateRange(keySelectExpr,
keys,
comparer,
isDescending,
child.OutputPartition.Count,
keyType);
child.OutputDataSetInfo = outputInfo;
return child;
}
private DLinqQueryNode VisitAssumeOrderBy(QueryNodeInfo source,
LambdaExpression keySelectExpr,
Expression comparerExpr,
Expression isDescendingExpr,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
Type keyType = keySelectExpr.Type.GetGenericArguments()[1];
if (comparerExpr == null && !TypeSystem.HasDefaultComparer(keyType))
{
throw DryadLinqException.Create(DryadLinqErrorCode.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable,
String.Format(SR.ComparerMustBeSpecifiedOrKeyTypeMustBeIComparable, keyType),
queryExpr);
}
object comparer = null;
if (comparerExpr != null)
{
ExpressionSimplifier<object> evaluator = new ExpressionSimplifier<object>();
comparer = evaluator.Eval(comparerExpr);
}
ExpressionSimplifier<bool> bevaluator = new ExpressionSimplifier<bool>();
bool isDescending = bevaluator.Eval(isDescendingExpr);
DataSetInfo outputInfo = new DataSetInfo(child.OutputDataSetInfo);
outputInfo.orderByInfo = OrderByInfo.Create(keySelectExpr, comparer, isDescending, keyType);
child.OutputDataSetInfo = outputInfo;
return child;
}
private DLinqQueryNode VisitSlidingWindow(QueryNodeInfo source,
LambdaExpression procLambda,
Expression windowSizeExpr,
Expression queryExpr)
{
// var windows = source.Apply(s => DryadLinqHelper.Last(s, windowSize));
// var slided = windows.Apply(s => DryadLinqHelper.Slide(s)).HashPartition(x => x.Index);
// slided.Apply(source, (x, y) => DryadLinqHelper.ProcessWindows(x, y, procFunc, windowSize));
DLinqQueryNode child = this.Visit(source);
if (child.IsDynamic)
{
throw new DryadLinqException("SlidingWindow is only supported for static partition count");
}
ExpressionSimplifier<int> evaluator = new ExpressionSimplifier<int>();
Expression windowSize = Expression.Constant(evaluator.Eval(windowSizeExpr), typeof(int));
child.IsForked = true;
// Apply node for s => Last(s, windowSize)
Type paramType = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param = Expression.Parameter(paramType, DryadLinqCodeGen.MakeUniqueName("s"));
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("Last");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
Expression body = Expression.Call(minfo, param, windowSize);
Type funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode lastNode = new DLinqApplyNode(procFunc, queryExpr, child);
lastNode = new DLinqMergeNode(true, queryExpr, lastNode);
// Apply node for s => Slide(s)
param = Expression.Parameter(body.Type, DryadLinqCodeGen.MakeUniqueName("s"));
minfo = typeof(DryadLinqHelper).GetMethod("Slide");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
body = Expression.Call(minfo, param);
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode slideNode = new DLinqApplyNode(procFunc, queryExpr, lastNode);
// Hash partition to distribute from partition i to i+1
int pcount = child.OutputPartition.Count;
param = Expression.Parameter(body.Type.GetGenericArguments()[0], "x");
Expression keySelectBody = Expression.Property(param, "Index");
funcType = typeof(Func<,>).MakeGenericType(param.Type, keySelectBody.Type);
LambdaExpression keySelectExpr = Expression.Lambda(funcType, keySelectBody, param);
DLinqQueryNode hdistNode = new DLinqHashPartitionNode(keySelectExpr,
null,
pcount,
queryExpr,
slideNode);
// Apply node for (x, y) => ProcessWindows(x, y, proclambda, windowSize)
Type paramType1 = typeof(IEnumerable<>).MakeGenericType(body.Type);
ParameterExpression param1 = Expression.Parameter(paramType1, DryadLinqCodeGen.MakeUniqueName("x"));
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param2 = Expression.Parameter(paramType2, DryadLinqCodeGen.MakeUniqueName("y"));
minfo = typeof(DryadLinqHelper).GetMethod("ProcessWindows");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0], procLambda.Body.Type);
body = Expression.Call(minfo, param1, param2, procLambda, windowSize);
funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param1, param2);
return new DLinqApplyNode(procFunc, queryExpr, hdistNode, child);
}
private DLinqQueryNode
VisitApplyWithPartitionIndex(QueryNodeInfo source,
LambdaExpression procLambda,
Expression queryExpr)
{
// The computation looks like this:
// var indices = source.Apply(s => ValueZero(s)).Apply(s => AssignIndex(s))
// .HashPartition(x => x)
// indices.Apply(source, (x, y) => ApplyWithPartitionIndex(x, y, procFunc));
DLinqQueryNode child = this.Visit(source);
if (child.IsDynamic)
{
throw new DryadLinqException("ApplyWithPartitionIndex is only supported for static partition count");
}
child.IsForked = true;
// Apply node for s => ValueZero(s)
Type paramType = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param = Expression.Parameter(paramType, "s");
MethodInfo minfo = typeof(DryadLinqHelper).GetMethod("ValueZero");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0]);
Expression body = Expression.Call(minfo, param);
Type funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
LambdaExpression procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode valueZeroNode = new DLinqApplyNode(procFunc, queryExpr, child);
valueZeroNode = new DLinqMergeNode(true, queryExpr, valueZeroNode);
// Apply node for s => AssignIndex(s)
paramType = typeof(IEnumerable<>).MakeGenericType(typeof(int));
param = Expression.Parameter(paramType, "s");
minfo = typeof(DryadLinqHelper).GetMethod("AssignIndex");
body = Expression.Call(minfo, param);
funcType = typeof(Func<,>).MakeGenericType(param.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param);
DLinqQueryNode assignIndexNode = new DLinqApplyNode(procFunc, queryExpr, valueZeroNode);
// HashPartition to distribute the indices -- one to each partition.
int pcount = child.OutputPartition.Count;
param = Expression.Parameter(body.Type, "x");
funcType = typeof(Func<,>).MakeGenericType(param.Type, param.Type);
LambdaExpression keySelectExpr = Expression.Lambda(funcType, param, param);
DLinqQueryNode hdistNode = new DLinqHashPartitionNode(keySelectExpr,
null,
pcount,
queryExpr,
assignIndexNode);
// Apply node for (x, y) => ApplyWithPartitionIndex(x, y, procLambda));
Type paramType1 = typeof(IEnumerable<>).MakeGenericType(child.OutputTypes[0]);
ParameterExpression param1 = Expression.Parameter(paramType1, DryadLinqCodeGen.MakeUniqueName("x"));
Type paramType2 = typeof(IEnumerable<>).MakeGenericType(typeof(int));
ParameterExpression param2 = Expression.Parameter(paramType2, DryadLinqCodeGen.MakeUniqueName("y"));
minfo = typeof(DryadLinqHelper).GetMethod("ProcessWithIndex");
minfo = minfo.MakeGenericMethod(child.OutputTypes[0], procLambda.Body.Type.GetGenericArguments()[0]);
body = Expression.Call(minfo, param1, param2, procLambda);
funcType = typeof(Func<,,>).MakeGenericType(param1.Type, param2.Type, body.Type);
procFunc = Expression.Lambda(funcType, body, param1, param2);
return new DLinqApplyNode(procFunc, queryExpr, child, hdistNode);
}
private DLinqQueryNode VisitDoWhile(QueryNodeInfo source,
QueryNodeInfo body,
QueryNodeInfo cond,
QueryNodeInfo bodySource,
QueryNodeInfo condSource1,
QueryNodeInfo condSource2,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
DLinqQueryNode bodyNode = this.Visit(body);
DLinqQueryNode condNode = this.Visit(cond);
DLinqQueryNode bodySourceNode = this.Visit(bodySource);
DLinqQueryNode condSourceNode1 = this.Visit(condSource1);
DLinqQueryNode condSourceNode2 = this.Visit(condSource2);
return new DLinqDoWhileNode(bodyNode, condNode,
bodySourceNode,
condSourceNode1, condSourceNode2,
queryExpr, child);
}
private DLinqQueryNode VisitDummy(Expression queryExpr)
{
MethodCallExpression mcExpr = queryExpr as MethodCallExpression;
Expression sourceExpr = mcExpr.Arguments[0];
QueryNodeInfo source = this.m_exprNodeInfoMap[sourceExpr];
DLinqQueryNode child = this.Visit(source);
return new DLinqDummyNode(queryExpr, child);
}
private DLinqQueryNode VisitFirst(QueryNodeInfo source,
LambdaExpression lambda,
AggregateOpType aggType,
bool isQuery,
Expression queryExpr)
{
DLinqQueryNode child = this.Visit(source);
DLinqQueryNode resNode = this.PromoteConcat(
source, child,
x => new DLinqBasicAggregateNode(lambda, aggType, true, isQuery, queryExpr, x));
return new DLinqBasicAggregateNode(null, aggType, false, isQuery, queryExpr, resNode);
}
private DLinqQueryNode VisitThenBy(QueryNodeInfo source,
LambdaExpression keySelectExpr,
bool isDescending,
Expression queryExpr)
{
// YY: This makes it hard to maintain OrderByInfo.
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, "ThenBy"),
queryExpr);
}
private DLinqQueryNode VisitDefaultIfEmpty(QueryNodeInfo source,
Expression defaultValueExpr,
MethodCallExpression queryExpr)
{
// YY: Not very useful. We could add it later.
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, "DefaultIfEmpty"),
queryExpr);
}
private DLinqQueryNode VisitElementAt(string opName,
QueryNodeInfo source,
Expression indexExpr,
Expression queryExpr)
{
// YY: Not very useful. We could add it later.
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, opName),
queryExpr);
}
private DLinqQueryNode VisitOfType(QueryNodeInfo source,
Type ofType,
Expression queryExpr)
{
// YY: Not very useful.
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, "OfType"),
queryExpr);
}
private DLinqQueryNode VisitQueryOperatorCall(QueryNodeInfo nodeInfo)
{
DLinqQueryNode resNode = nodeInfo.QueryNode;
if (resNode != null) return resNode;
MethodCallExpression expression = (MethodCallExpression)nodeInfo.QueryExpression;
string methodName = expression.Method.Name;
#region LINQMETHODS
switch (methodName)
{
case "Aggregate":
case "AggregateAsQuery":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression funcLambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (funcLambda != null && funcLambda.Parameters.Count == 2)
{
resNode = this.VisitAggregate(nodeInfo.Children[0].Child,
null,
funcLambda,
null,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression funcLambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (funcLambda != null && funcLambda.Parameters.Count == 2)
{
resNode = this.VisitAggregate(nodeInfo.Children[0].Child,
expression.Arguments[1],
funcLambda,
null,
expression);
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression funcLambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
LambdaExpression resultLambda = DryadLinqExpression.GetLambda(expression.Arguments[3]);
if (funcLambda != null && funcLambda.Parameters.Count == 2 &&
resultLambda != null && resultLambda.Parameters.Count == 1)
{
resNode = this.VisitAggregate(nodeInfo.Children[0].Child,
expression.Arguments[1],
funcLambda,
resultLambda,
expression);
}
}
break;
}
case "Select":
case "LongSelect":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && (lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitSelect(nodeInfo.Children[0].Child,
QueryNodeType.Select,
lambda,
null,
expression);
}
}
break;
}
case "SelectMany":
case "LongSelectMany":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count <= 2)
{
resNode = this.VisitSelect(nodeInfo.Children[0].Child,
QueryNodeType.SelectMany,
lambda,
null,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda1 = DryadLinqExpression.GetLambda(expression.Arguments[1]);
LambdaExpression lambda2 = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (lambda1 != null && (lambda1.Parameters.Count == 1 || lambda1.Parameters.Count == 2) &&
lambda2 != null && lambda2.Parameters.Count == 2)
{
resNode = this.VisitSelect(nodeInfo.Children[0].Child,
QueryNodeType.SelectMany,
lambda1,
lambda2,
expression);
}
}
break;
}
case "Join":
case "GroupJoin":
{
QueryNodeType nodeType = (methodName == "Join") ? QueryNodeType.Join : QueryNodeType.GroupJoin;
if (expression.Arguments.Count == 5)
{
LambdaExpression lambda2 = DryadLinqExpression.GetLambda(expression.Arguments[2]);
LambdaExpression lambda3 = DryadLinqExpression.GetLambda(expression.Arguments[3]);
LambdaExpression lambda4 = DryadLinqExpression.GetLambda(expression.Arguments[4]);
if (lambda2 != null && lambda2.Parameters.Count == 1 &&
lambda3 != null && lambda3.Parameters.Count == 1 &&
lambda4 != null && lambda4.Parameters.Count == 2)
{
resNode = this.VisitJoin(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
nodeType,
lambda2,
lambda3,
lambda4,
null,
expression);
}
}
else if (expression.Arguments.Count == 6)
{
LambdaExpression lambda2 = DryadLinqExpression.GetLambda(expression.Arguments[2]);
LambdaExpression lambda3 = DryadLinqExpression.GetLambda(expression.Arguments[3]);
LambdaExpression lambda4 = DryadLinqExpression.GetLambda(expression.Arguments[4]);
if (lambda2 != null && lambda2.Parameters.Count == 1 &&
lambda3 != null && lambda3.Parameters.Count == 1 &&
lambda4 != null && lambda4.Parameters.Count == 2)
{
resNode = this.VisitJoin(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
nodeType,
lambda2,
lambda3,
lambda4,
expression.Arguments[5],
expression);
}
}
break;
}
case "OfType":
{
if (expression.Arguments.Count == 1)
{
Type ofType = expression.Method.GetGenericArguments()[0];
resNode = this.VisitOfType(nodeInfo.Children[0].Child, ofType, expression);
}
break;
}
case "Where":
case "LongWhere":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && (lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitWhere(nodeInfo.Children[0].Child,
lambda,
expression);
}
}
break;
}
case "First":
case "FirstOrDefault":
case "FirstAsQuery":
{
AggregateOpType aggType = (methodName == "FirstOrDefault") ? AggregateOpType.FirstOrDefault : AggregateOpType.First;
bool isQuery = (methodName == "FirstAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child, null, aggType, isQuery, expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child,
lambda,
aggType,
isQuery,
expression);
}
}
break;
}
case "Single":
case "SingleOrDefault":
case "SingleAsQuery":
{
AggregateOpType aggType = (methodName == "SingleOrDefault") ? AggregateOpType.SingleOrDefault : AggregateOpType.Single;
bool isQuery = (methodName == "SingleAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child, null, aggType, isQuery, expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child,
lambda,
aggType,
isQuery,
expression);
}
}
break;
}
case "Last":
case "LastOrDefault":
case "LastAsQuery":
{
AggregateOpType aggType = (methodName == "LastOrDefault") ? AggregateOpType.LastOrDefault : AggregateOpType.Last;
bool isQuery = (methodName == "LastAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child, null, aggType, isQuery, expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitFirst(nodeInfo.Children[0].Child,
lambda,
aggType,
isQuery,
expression);
}
}
break;
}
case "Distinct":
{
if (expression.Arguments.Count == 1)
{
resNode = this.VisitDistinct(nodeInfo.Children[0].Child, null, expression);
}
else if (expression.Arguments.Count == 2)
{
resNode = this.VisitDistinct(nodeInfo.Children[0].Child,
expression.Arguments[1],
expression);
}
break;
}
case "DefaultIfEmpty":
{
if (expression.Arguments.Count == 1)
{
resNode = this.VisitDefaultIfEmpty(nodeInfo.Children[0].Child,
null,
expression);
}
else if (expression.Arguments.Count == 2)
{
resNode = this.VisitDefaultIfEmpty(nodeInfo.Children[0].Child,
expression.Arguments[1],
expression);
}
break;
}
case "Concat":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitConcat(nodeInfo, expression);
}
break;
}
case "Union":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Union,
null,
expression);
}
else if (expression.Arguments.Count == 3)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Union,
expression.Arguments[2],
expression);
}
break;
}
case "Intersect":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Intersect,
null,
expression);
}
else if (expression.Arguments.Count == 3)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Intersect,
expression.Arguments[2],
expression);
}
break;
}
case "Except":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Except,
null,
expression);
}
else if (expression.Arguments.Count == 3)
{
resNode = this.VisitSetOperation(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
QueryNodeType.Except,
expression.Arguments[2],
expression);
}
break;
}
case "Any":
case "AnyAsQuery":
{
bool isQuery = (methodName == "AnyAsQuery");
if (expression.Arguments.Count == 1)
{
Type type = expression.Method.GetGenericArguments()[0];
ParameterExpression param = Expression.Parameter(type, "x");
Type delegateType = typeof(Func<,>).MakeGenericType(type, typeof(bool));
Expression body = Expression.Constant(true);
LambdaExpression lambda = Expression.Lambda(delegateType, body, param);
resNode = this.VisitQuantifier(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Any,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitQuantifier(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Any,
isQuery,
expression);
}
}
break;
}
case "All":
case "AllAsQuery":
{
bool isQuery = (methodName == "AllAsQuery");
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitQuantifier(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.All,
isQuery,
expression);
}
}
break;
}
case "Count":
case "CountAsQuery":
{
bool isQuery = (methodName == "CountAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.Count,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Count,
isQuery,
expression);
}
}
break;
}
case "LongCount":
case "LongCountAsQuery":
{
bool isQuery = (methodName == "LongCountAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.LongCount,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.LongCount,
isQuery,
expression);
}
}
break;
}
case "Sum":
case "SumAsQuery":
{
bool isQuery = (methodName == "SumAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.Sum,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Sum,
isQuery,
expression);
}
}
break;
}
case "Min":
case "MinAsQuery":
{
bool isQuery = (methodName == "MinAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.Min,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Min,
isQuery,
expression);
}
}
break;
}
case "Max":
case "MaxAsQuery":
{
bool isQuery = (methodName == "MaxAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.Max,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Max,
isQuery,
expression);
}
}
break;
}
case "Average":
case "AverageAsQuery":
{
bool isQuery = (methodName == "AverageAsQuery");
if (expression.Arguments.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
null,
AggregateOpType.Average,
isQuery,
expression);
}
else if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitBasicAggregate(nodeInfo.Children[0].Child,
lambda,
AggregateOpType.Average,
isQuery,
expression);
}
}
break;
}
case "GroupBy":
{
// groupby can take 2, 3, 4, or 5 arguments.
if (expression.Arguments.Count == 2)
{
//Supplied arguments are as follows:(source, key selector)
LambdaExpression keySelExpr = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (keySelExpr != null && keySelExpr.Parameters.Count == 1)
{
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
null,
null,
null,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
//Supplied arguments are as follows:(source, key selector, element selector/result selector/comparer)
LambdaExpression keySelExpr = DryadLinqExpression.GetLambda(expression.Arguments[1]);
LambdaExpression lambda2 = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (keySelExpr != null && lambda2 == null)
{
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
null,
null,
expression.Arguments[2],
expression);
}
else if (keySelExpr != null && keySelExpr.Parameters.Count == 1 && lambda2 != null)
{
LambdaExpression elemSelExpr = null;
LambdaExpression resSelExpr = null;
if (lambda2.Parameters.Count == 1)
{
elemSelExpr = lambda2;
}
else if (lambda2.Parameters.Count == 2)
{
resSelExpr = lambda2;
}
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
elemSelExpr,
resSelExpr,
null,
expression);
}
}
else if (expression.Arguments.Count == 4)
{
//Argument-0 is source and Argument-1 is key selector expression
LambdaExpression keySelExpr = DryadLinqExpression.GetLambda(expression.Arguments[1]);
LambdaExpression lambda2 = DryadLinqExpression.GetLambda(expression.Arguments[2]);
LambdaExpression lambda3 = DryadLinqExpression.GetLambda(expression.Arguments[3]);
if (keySelExpr != null && keySelExpr.Parameters.Count == 1 && lambda3 == null)
{
//Argument-2 can be either result selector, element selector and argument-3 is comparer
LambdaExpression elemSelExpr = null;
LambdaExpression resSelExpr = null;
if (lambda2.Parameters.Count == 1)
{
elemSelExpr = lambda2;
}
else if (lambda2.Parameters.Count == 2)
{
resSelExpr = lambda2;
}
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
elemSelExpr,
resSelExpr,
expression.Arguments[3],
expression);
}
else if (keySelExpr != null && keySelExpr.Parameters.Count == 1 &&
lambda2 != null && lambda2.Parameters.Count == 1 &&
lambda3 != null && lambda3.Parameters.Count == 2)
{
//Argument-2 is element selector and argument-3 is result selector
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
lambda2,
lambda3,
null,
expression);
}
}
else if (expression.Arguments.Count == 5)
{
//Supplied arguments are as follows:(source, key selector, element selector, result selector, comparer)
LambdaExpression keySelExpr = DryadLinqExpression.GetLambda(expression.Arguments[1]);
LambdaExpression elemSelExpr = DryadLinqExpression.GetLambda(expression.Arguments[2]);
LambdaExpression resSelExpr = DryadLinqExpression.GetLambda(expression.Arguments[3]);
if (keySelExpr != null && keySelExpr.Parameters.Count == 1 &&
elemSelExpr != null && elemSelExpr.Parameters.Count == 1 &&
resSelExpr != null && resSelExpr.Parameters.Count == 2)
{
resNode = this.VisitGroupBy(nodeInfo.Children[0].Child,
keySelExpr,
elemSelExpr,
resSelExpr,
expression.Arguments[4],
expression);
}
}
break;
}
case "OrderBy":
case "OrderByDescending":
{
bool isDescending = (methodName == "OrderByDescending");
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitOrderBy(nodeInfo.Children[0].Child,
lambda,
null,
isDescending,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitOrderBy(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
isDescending,
expression);
}
}
break;
}
case "ThenBy":
case "ThenByDescending":
{
bool isDescending = (methodName == "ThenByDescending");
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitThenBy(nodeInfo.Children[0].Child,
lambda,
isDescending,
expression);
}
}
break;
}
case "ElementAt":
case "ElementAtOrDefault":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitElementAt(methodName,
nodeInfo.Children[0].Child,
expression.Arguments[1],
expression);
}
break;
}
case "Take":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitPartitionOp(methodName,
nodeInfo.Children[0].Child,
QueryNodeType.Take,
expression.Arguments[1],
expression);
}
break;
}
case "TakeWhile":
case "LongTakeWhile":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null &&
(lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitPartitionOp("TakeWhile",
nodeInfo.Children[0].Child,
QueryNodeType.TakeWhile,
lambda,
expression);
}
}
break;
}
case "Skip":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitPartitionOp(methodName,
nodeInfo.Children[0].Child,
QueryNodeType.Skip,
expression.Arguments[1],
expression);
}
break;
}
case "SkipWhile":
case "LongSkipWhile":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null &&
(lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitPartitionOp(methodName,
nodeInfo.Children[0].Child,
QueryNodeType.SkipWhile,
lambda,
expression);
}
}
break;
}
case "Contains":
case "ContainsAsQuery":
{
bool isQuery = (methodName == "ContainsAsQuery");
if (expression.Arguments.Count == 2)
{
resNode = this.VisitContains(nodeInfo.Children[0].Child,
expression.Arguments[1],
null,
isQuery,
expression);
}
else if (expression.Arguments.Count == 3)
{
resNode = this.VisitContains(nodeInfo.Children[0].Child,
expression.Arguments[1],
expression.Arguments[2],
isQuery,
expression);
}
break;
}
case "Reverse":
{
resNode = this.VisitReverse(nodeInfo.Children[0].Child, expression);
break;
}
case "SequenceEqual":
case "SequenceEqualAsQuery":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitSequenceEqual(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
null,
expression);
}
else if (expression.Arguments.Count == 3)
{
resNode = this.VisitSequenceEqual(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
expression.Arguments[2],
expression);
}
break;
}
case "Zip":
{
if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (lambda != null && lambda.Parameters.Count == 2)
{
resNode = this.VisitZip(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
lambda,
expression);
}
}
break;
}
case "HashPartition":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
null,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[2].Type == typeof(int))
{
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
expression.Arguments[2],
expression);
}
else
{
LambdaExpression resLambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (resLambda == null)
{
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
null,
expression);
}
else
{
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
resLambda,
null,
null,
expression);
}
}
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[3].Type == typeof(int))
{
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression.Arguments[3],
expression);
}
else
{
LambdaExpression resLambda = DryadLinqExpression.GetLambda(expression.Arguments[3]);
resNode = this.VisitHashPartition(nodeInfo.Children[0].Child,
lambda,
resLambda,
expression.Arguments[2],
null,
expression);
}
}
}
break;
}
case "RangePartition":
{
//overloads:
//
// 2-param:
// (source, keySelector)
//
// 3-param:
// (source, keySelector, pcount)
// (source, keySelector, isDescending)
// (source, keySelector, rangeSeparators)
//
// 4-param:
// (source, keySelector, isDescending, pcount)
// (source, keySelector, keyComparer, isDescending)
// (source, keySelector, rangeSeparators, keyComparer)
//
// 5-param:
// (source, keySelector, keyComparer, isDescending, pcount)
// (source, keySelector, rangeSeparators, keyComparer, isDescending)
if (expression.Arguments.Count == 2)
{
// Case: (source, keySelector)
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
null,
null,
expression);
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[2].Type == typeof(int))
{
// Case: (source, keySelector, pcount)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
null,
expression.Arguments[2],
expression);
}
if (expression.Arguments[2].Type == typeof(bool))
{
// Case: (source, keySelector, isDescending)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
expression.Arguments[2],
null,
expression);
}
else if (expression.Arguments[2].Type.IsArray)
{
// Case: RangePartition(keySelector, TKey[] keys)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
null,
null,
null,
expression);
}
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[2].Type == typeof(bool))
{
//case: (source, keySelector, isDescending, pcount)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
expression.Arguments[2],
expression.Arguments[3],
expression);
}
else if (expression.Arguments[3].Type == typeof(bool))
{
//case: (source, keySelector, keyComparer, isDescending)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression.Arguments[3],
null,
expression);
}
else if (expression.Arguments[2].Type.IsArray)
{
//case: (source, keySelector, rangeSeparators, keyComparer)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression.Arguments[3],
null,
null,
expression);
}
}
}
else if (expression.Arguments.Count == 5)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[3].Type == typeof(bool))
{
// case: (source, keySelector, keyComparer, isDescending, pcount)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression.Arguments[3],
expression.Arguments[4],
expression);
}
else if (expression.Arguments[4].Type == typeof(bool))
{
// case: (source, keySelector, rangeSeparators, keyComparer, isDescending)
resNode = this.VisitRangePartition(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression.Arguments[3],
expression.Arguments[4],
null,
expression);
}
}
}
break;
}
case "Apply":
{
if (expression.Arguments.Count == 2)
{
// Apply with only one input source
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null &&
(lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitApply(nodeInfo.Children[0].Child,
null,
lambda,
false,
false,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
if (lambda != null)
{
if (nodeInfo.Children.Count == 2)
{
// Apply with two input sources
resNode = this.VisitApply(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
lambda,
false,
false,
expression);
}
else
{
// Apply with multiple sources
resNode = this.VisitMultiApply(nodeInfo, lambda, false, false, expression);
}
}
}
break;
}
case "ApplyPerPartition":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null &&
(lambda.Parameters.Count == 1 || lambda.Parameters.Count == 2))
{
resNode = this.VisitApply(nodeInfo.Children[0].Child,
null,
lambda,
true,
false,
expression);
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[2]);
ExpressionSimplifier<bool> evaluator = new ExpressionSimplifier<bool>();
bool isFirstOnly = evaluator.Eval(expression.Arguments[3]);
if (lambda != null)
{
if (nodeInfo.Children.Count == 2)
{
resNode = this.VisitApply(nodeInfo.Children[0].Child,
nodeInfo.Children[1].Child,
lambda,
true,
isFirstOnly,
expression);
}
else
{
// Apply with multiple sources of the same type
resNode = this.VisitMultiApply(nodeInfo, lambda, true, isFirstOnly, expression);
}
}
}
break;
}
case "Fork":
{
if (expression.Arguments.Count == 2) // ForkSelect and ForkApply
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitFork(nodeInfo.Children[0].Child,
lambda,
null,
expression);
}
}
else if (expression.Arguments.Count == 3) // ForkByKey
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitFork(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression);
}
}
break;
}
case "ForkChoose":
{
if (expression.Arguments.Count == 2)
{
resNode = this.VisitForkChoose(nodeInfo.Children[0].Child,
expression.Arguments[1],
expression);
}
break;
}
case "AssumeHashPartition":
{
if (expression.Arguments.Count == 2)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitAssumeHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
expression);
}
}
else if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitAssumeHashPartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression);
}
}
break;
}
case "AssumeRangePartition":
{
if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[2].Type.IsArray)
{
resNode = this.VisitAssumeRangePartition(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
null,
null,
expression);
}
else
{
resNode = this.VisitAssumeRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
null,
expression.Arguments[2],
expression);
}
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
if (expression.Arguments[2].Type.IsArray)
{
resNode = this.VisitAssumeRangePartition(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression.Arguments[3],
null,
expression);
}
else
{
resNode = this.VisitAssumeRangePartition(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression.Arguments[3],
expression);
}
}
}
break;
}
case "AssumeOrderBy":
{
if (expression.Arguments.Count == 3)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitAssumeOrderBy(nodeInfo.Children[0].Child,
lambda,
null,
expression.Arguments[2],
expression);
}
}
else if (expression.Arguments.Count == 4)
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitAssumeOrderBy(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression.Arguments[3],
expression);
}
}
break;
}
case "SlidingWindow":
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 1)
{
resNode = this.VisitSlidingWindow(nodeInfo.Children[0].Child,
lambda,
expression.Arguments[2],
expression);
}
break;
}
case "ApplyWithPartitionIndex":
{
LambdaExpression lambda = DryadLinqExpression.GetLambda(expression.Arguments[1]);
if (lambda != null && lambda.Parameters.Count == 2)
{
resNode = this.VisitApplyWithPartitionIndex(nodeInfo.Children[0].Child,
lambda,
expression);
}
break;
}
case "DoWhile":
{
DoWhileQueryNodeInfo doWhileNodeInfo = (DoWhileQueryNodeInfo)nodeInfo;
resNode = this.VisitDoWhile(nodeInfo.Children[0].Child,
doWhileNodeInfo.Body,
doWhileNodeInfo.Cond,
doWhileNodeInfo.BodySource,
doWhileNodeInfo.CondSource1,
doWhileNodeInfo.CondSource2,
expression);
break;
}
case "Dummy":
{
resNode = this.VisitDummy(expression);
break;
}
}
#endregion
if (resNode == null)
{
throw DryadLinqException.Create(DryadLinqErrorCode.OperatorNotSupported,
String.Format(SR.OperatorNotSupported, methodName),
expression);
}
resNode.IsForked = resNode.IsForked || nodeInfo.IsForked;
nodeInfo.QueryNode = resNode;
return resNode;
}
}
}
Reference in New Issue View Git Blame Copy Permalink