001/*
002 * Copyright (c) 2007-2017 Xplenty, Inc. All Rights Reserved.
003 *
004 * Project and contact information: http://www.cascading.org/
005 *
006 * This file is part of the Cascading project.
007 *
008 * Licensed under the Apache License, Version 2.0 (the "License");
009 * you may not use this file except in compliance with the License.
010 * You may obtain a copy of the License at
011 *
012 *     http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing, software
015 * distributed under the License is distributed on an "AS IS" BASIS,
016 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017 * See the License for the specific language governing permissions and
018 * limitations under the License.
019 */
020
021package cascading.flow.planner.iso.subgraph.iterator;
022
023import java.util.ArrayList;
024import java.util.Collection;
025import java.util.List;
026import java.util.NoSuchElementException;
027import java.util.Set;
028
029import cascading.flow.FlowElement;
030import cascading.flow.planner.PlannerContext;
031import cascading.flow.planner.graph.ElementGraph;
032import cascading.flow.planner.iso.ElementAnnotation;
033import cascading.flow.planner.iso.expression.ElementCapture;
034import cascading.flow.planner.iso.expression.ExpressionGraph;
035import cascading.flow.planner.iso.finder.GraphFinder;
036import cascading.flow.planner.iso.finder.Match;
037import cascading.flow.planner.iso.subgraph.SubGraphIterator;
038import cascading.flow.planner.iso.transformer.ContractedTransformer;
039import cascading.flow.planner.iso.transformer.Transformed;
040import cascading.util.EnumMultiMap;
041
042import static cascading.flow.planner.graph.ElementGraphs.asSubGraph;
043import static cascading.util.Util.createIdentitySet;
044
045/**
046 *
047 */
048public class ExpressionSubGraphIterator implements SubGraphIterator
049  {
050  private final PlannerContext plannerContext;
051  private final ElementGraph elementGraph;
052
053  private ContractedTransformer contractedTransformer;
054  private GraphFinder graphFinder;
055
056  private Set<FlowElement> elementExcludes = createIdentitySet();
057  private ElementGraph contractedGraph;
058  private Transformed<ElementGraph> contractedTransformed;
059
060  private boolean firstOnly = false; // false will continue to accumulate around the primary
061  private Match match;
062
063  private List<Match> matches = new ArrayList<>();
064
065  int count = 0;
066
067  public ExpressionSubGraphIterator( ExpressionGraph matchExpression, ElementGraph elementGraph )
068    {
069    this( new PlannerContext(), matchExpression, elementGraph );
070    }
071
072  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph matchExpression, ElementGraph elementGraph )
073    {
074    this( plannerContext, null, matchExpression, elementGraph );
075    }
076
077  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, ElementGraph elementGraph )
078    {
079    this( plannerContext, contractionExpression, matchExpression, false, elementGraph );
080    }
081
082  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, ElementGraph elementGraph, Collection<FlowElement> elementExcludes )
083    {
084    this( plannerContext, contractionExpression, matchExpression, false, elementGraph, elementExcludes );
085    }
086
087  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, boolean firstOnly, ElementGraph elementGraph )
088    {
089    this( plannerContext, contractionExpression, matchExpression, firstOnly, elementGraph, null );
090    }
091
092  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, boolean firstOnly, ElementGraph elementGraph, Collection<FlowElement> elementExcludes )
093    {
094    this.plannerContext = plannerContext;
095    this.firstOnly = firstOnly;
096    this.elementGraph = elementGraph;
097
098    if( elementExcludes != null )
099      this.elementExcludes.addAll( elementExcludes );
100
101    if( contractionExpression != null )
102      contractedTransformer = new ContractedTransformer( contractionExpression );
103    else
104      contractedGraph = elementGraph;
105
106    graphFinder = new GraphFinder( matchExpression );
107    }
108
109  @Override
110  public ElementGraph getElementGraph()
111    {
112    return elementGraph;
113    }
114
115  public List<Match> getMatches()
116    {
117    return matches;
118    }
119
120  public ElementGraph getContractedGraph()
121    {
122    if( contractedGraph == null )
123      {
124      contractedTransformed = contractedTransformer.transform( plannerContext, elementGraph );
125      contractedGraph = contractedTransformed.getEndGraph();
126      }
127
128    return contractedGraph;
129    }
130
131  public Match getLastMatch()
132    {
133    if( matches.isEmpty() )
134      return null;
135
136    return matches.get( count - 1 );
137    }
138
139  @Override
140  public EnumMultiMap getAnnotationMap( ElementAnnotation[] annotations )
141    {
142    EnumMultiMap annotationsMap = new EnumMultiMap();
143
144    if( annotations.length == 0 )
145      return annotationsMap;
146
147    Match match = getLastMatch();
148
149    for( ElementAnnotation annotation : annotations )
150      annotationsMap.addAll( annotation.getAnnotation(), match.getCapturedElements( annotation.getCapture() ) );
151
152    return annotationsMap;
153    }
154
155  @Override
156  public boolean hasNext()
157    {
158    if( match == null )
159      {
160      match = graphFinder.findMatchesOnPrimary( plannerContext, getContractedGraph(), firstOnly, elementExcludes );
161
162      if( match.foundMatch() )
163        {
164        matches.add( match );
165        elementExcludes.addAll( match.getCapturedElements( ElementCapture.Primary ) ); // idempotent
166        count++;
167        }
168      }
169
170    return match.foundMatch();
171    }
172
173  @Override
174  public ElementGraph next()
175    {
176    try
177      {
178      if( !hasNext() )
179        throw new NoSuchElementException();
180
181      ElementGraph contractedMatchedGraph = match.getMatchedGraph();
182
183      Set<FlowElement> excludes = getContractedGraph().vertexSetCopy();
184
185      excludes.removeAll( contractedMatchedGraph.vertexSet() );
186
187      return asSubGraph( elementGraph, contractedMatchedGraph, excludes );
188      }
189    finally
190      {
191      match = null;
192      }
193    }
194
195  @Override
196  public void remove()
197    {
198    throw new UnsupportedOperationException();
199    }
200  }