001/*
002 * Copyright (c) 2007-2015 Concurrent, 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.transformer;
022
023import cascading.flow.planner.PlannerContext;
024import cascading.flow.planner.graph.ElementGraph;
025import cascading.flow.planner.graph.ElementGraphs;
026import cascading.flow.planner.graph.ElementSubGraph;
027import cascading.flow.planner.iso.expression.ExpressionGraph;
028import cascading.flow.planner.iso.finder.GraphFinder;
029import cascading.flow.planner.iso.finder.Match;
030
031/**
032 * Class SubGraphTransformer will return a bounded sub-graph after matching a sub-graph within a contracted graph.
033 */
034public class SubGraphTransformer extends GraphTransformer<ElementGraph, ElementSubGraph>
035  {
036  private GraphTransformer graphTransformer;
037  private ExpressionGraph subGraphMatcher;
038  private GraphFinder subGraphFinder;
039  private boolean findAllPrimaries = false;
040
041  public SubGraphTransformer( GraphTransformer graphTransformer, ExpressionGraph subGraphMatcher )
042    {
043    this.graphTransformer = graphTransformer;
044    this.subGraphMatcher = subGraphMatcher;
045    this.subGraphFinder = new GraphFinder( subGraphMatcher );
046    this.findAllPrimaries = subGraphMatcher.supportsNonRecursiveMatch();
047    }
048
049  public Transformed<ElementSubGraph> transform( PlannerContext plannerContext, ElementGraph rootGraph )
050    {
051    Transformed<ElementSubGraph> transformed = new Transformed<>( plannerContext, this, subGraphMatcher, rootGraph );
052
053    Transformed contractedTransformed = graphTransformer.transform( plannerContext, rootGraph ); // contracted graph transform
054
055    transformed.addChildTransform( contractedTransformed );
056
057    // apply contracted sub-graph matcher to get the bounded sub-graph of the original graph
058    ElementGraph contractedGraph = contractedTransformed.getEndGraph();
059
060    Match match = findAllPrimaries ? subGraphFinder.findAllMatches( plannerContext, contractedGraph ) : subGraphFinder.findFirstMatch( plannerContext, contractedGraph );
061
062    if( !match.foundMatch() )
063      return transformed;
064
065    ElementGraph contractedSubGraph = match.getMatchedGraph();
066
067    ElementSubGraph resultSubGraph = asSubGraphOf( rootGraph, contractedSubGraph ); // the bounded sub-graph of the rootGraph
068
069    transformed.setEndGraph( resultSubGraph );
070
071    return transformed;
072    }
073
074  protected ElementSubGraph asSubGraphOf( ElementGraph rootGraph, ElementGraph contractedSubGraph )
075    {
076    return ElementGraphs.asSubGraph( rootGraph, contractedSubGraph, null );
077    }
078  }