/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.pig.test;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;

import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.builtin.COUNT;
import org.apache.pig.builtin.SIZE;
import org.apache.pig.impl.PigContext;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOLoad;
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.rules.FilterAboveForeach;
import org.apache.pig.newplan.logical.rules.LoadTypeCastInserter;
import org.apache.pig.newplan.optimizer.PlanOptimizer;
import org.apache.pig.newplan.optimizer.PlanTransformListener;
import org.apache.pig.newplan.optimizer.Rule;
import org.apache.pig.test.TestNewPlanPushDownForeachFlatten.MyFilterFunc;
import org.junit.Assert;
import org.junit.Test;


public class TestNewPlanFilterAboveForeach {
    PigContext pc = new PigContext(ExecType.LOCAL, new Properties());
    
    @Test
    public void testSimple() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines:bag{ t : ( cuisine ) } );" +
        "B = FOREACH A GENERATE name, flatten(cuisines);" +
        "C = FILTER B BY name == 'joe';" +
        "D = STORE C INTO 'empty';" ;  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe1 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }
    
    @Test
    public void testSimpleNotPossible() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines:bag{ t : ( cuisine ) } );" +
                "B = FOREACH A GENERATE name, flatten(cuisines) as cuisines;" +
                "C = FILTER B BY cuisines == 'pizza';" +
                "D = STORE C INTO 'empty';" ;  
        LogicalPlan newLogicalPlan = buildPlan( query );
        
        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1  = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
    }
    
    /**
     * Non-deterministic filters should not be pushed up (see PIG-2014).
     * In the example below, if Filter gets pushed above flatten, we might remove
     * whole bags of cuisines of random gets pushed up, while the intent is to sample from each bag.
     * @throws Exception
     */
    @Test
    public void testNondeterministicFilter() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines:bag{ t : ( cuisine ) }, num:int );" +
        "B = FOREACH A GENERATE name, flatten(cuisines), num;" +
        "C = FILTER B BY RANDOM(num) > 5;" +
        "D = STORE C INTO 'empty';" ;

        LogicalPlan newLogicalPlan = buildPlan( query );

        newLogicalPlan.explain(System.out, "text", true);

        // Expect Filter to not be pushed, so it should be load->foreach-> filter
        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
    }

    @Test
    public void testMultipleFilter() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines : bag{ t : ( cuisine ) } );" +
        "B = FOREACH A GENERATE name, flatten(cuisines);" +
        "C = FILTER B BY $1 == 'french';" +
        "D = FILTER C BY name == 'joe';" +
        "E = STORE D INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );
        
        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe1 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter2 = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter2 instanceof LOFilter );
    }
    
    @Test
    public void testMultipleFilter2() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, age, cuisines : bag{ t : ( cuisine ) } );" +
        "B = FOREACH A GENERATE name, age, flatten(cuisines);" +
        "C = FILTER B BY name == 'joe';" +
        "D = FILTER C BY age == 30;" +
        "E = STORE D INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );
        
        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator filter2 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( filter2 instanceof LOFilter );
        Operator fe1 = newLogicalPlan.getSuccessors( filter2 ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }
    
    @Test
    public void testMultipleFilterNotPossible() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines : bag{ t : ( cuisine, region ) } );" +
        "B = FOREACH A GENERATE name, flatten(cuisines);" +
        "C = FILTER B BY $1 == 'French';" +
        "D = FILTER C BY $2 == 'Europe';" +
        "E = STORE D INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator filter2 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( filter2 instanceof LOFilter );
    }
    
    @Test
    public void testNotPossibleFilter() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines:bag{ t : ( cuisine ) } );" +
        "B = FOREACH A GENERATE name, flatten(cuisines);" +
        "C = FILTER B BY cuisine == 'French';" +
        "D = STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
    }
    
    @Test
    public void testSimple2() throws Exception {
        String query = "A =LOAD 'file.txt' AS (name, cuisines:bag{ t : ( cuisine ) } );" +
        "B = FOREACH A GENERATE name, cuisines;" +
        "C = FILTER B BY name == 'joe';" +
        "D = STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe1 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }
    
    /**
     * Normal test case: all fields from Foreach are used by exhaustive list.
     * Optimization should kick in.
     * @throws Exception 
     */
    @Test
    public void test1() throws Exception {
        String query = "A =LOAD 'file.txt' AS (a:bag{(u,v)}, b, c);" +
        "B = FOREACH A GENERATE $0, b;" +
        "C = FILTER B BY " + COUNT.class.getName() +"($0) > 5;" +
        "STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe2 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }
    
    /**
     * Identical to test1() except that it use project *.
     * @throws Exception 
     */
    @Test
    public void test2() throws Exception {
        String query = "A =LOAD 'file.txt' AS (a:(u,v), b, c);" +
        "B = FOREACH A GENERATE $0, b;" +
        "C = FILTER B BY " + SIZE.class.getName() +"(TOTUPLE(*)) > 5;" +
        "STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe2 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }

    /**
     * No fields are used in filter condition at all.
     * @throws Exception 
     */
    @Test
    public void test3() throws Exception {
        String query = "A =LOAD 'file.txt' AS (a:(u,v), b, c);" +
        "B = FOREACH A GENERATE $0, b;" +
        "C = FILTER B BY 8 > 5;" +
        "STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe1 = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
    }
    
    /**
     * Similar to test2, but not all fields are available from the operator before foreach.
     * Optimziation doesn't kick in.
     * @throws Exception 
     */
    @Test
    public void test4() throws Exception {
        String query = "A =LOAD 'file.txt' AS (a:(u,v), b, c);" +
        "B = FOREACH A GENERATE $0, b, flatten(1);" +
        "C = FILTER B BY " + SIZE.class.getName() +"(TOTUPLE(*)) > 5;" +
        "STORE C INTO 'empty';";  
        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe1 = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe1 instanceof LOForEach );
        Operator fe2 = newLogicalPlan.getSuccessors( fe1 ).get( 0 );
        Assert.assertTrue( fe2 instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe2 ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
    }

    @Test
    public void testFilterForeach() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, $2;" +        
        "C = filter B by $0 < 18;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator store = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }

    @Test
    public void testFilterForeachAddedField() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, $2, COUNT({(1)});" +        
        "C = filter B by $2 < 18;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator store = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }

    @Test
    public void testFilterForeachCast() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate (int)$1, $2;" +        
        "C = filter B by $0 < 18;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator fe = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator filter = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator store = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }

    @Test
    public void testFilterCastForeach() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, $2;" +        
        "C = filter B by (int)$0 < 18;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator store = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }


    @Test
    public void testFilterConstantConditionForeach() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, $2;" +        
        "C = filter B by 1 == 1;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator store = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }

    @Test
    public void testFilterUDFForeach() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, $2;" +        
        "C = filter B by " + MyFilterFunc.class.getName() + "($1) ;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator store = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }
    
    @Test
    public void testFilterForeachFlatten() throws Exception {
        String query = "A =load 'myfile' as (name, age, gpa);" +
        "B = foreach A generate $1, flatten($2);" +        
        "C = filter B by $0 < 18;" +
         "D = STORE C INTO 'empty';"; 

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator load = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue( load instanceof LOLoad );
        Operator filter = newLogicalPlan.getSuccessors( load ).get( 0 );
        Assert.assertTrue( filter instanceof LOFilter );
        Operator fe = newLogicalPlan.getSuccessors( filter ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        fe = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( fe instanceof LOForEach );
        Operator store = newLogicalPlan.getSuccessors( fe ).get( 0 );
        Assert.assertTrue( store instanceof LOStore );
    }
    
    // See PIG-1669
    @Test
    public void testPushUpFilterWithScalar() throws Exception {
        String query = "a = load 'studenttab10k' as (name, age, gpa);" +
        "b = group a all;" +
        "c = foreach b generate AVG(a.age) as age;" +
        "d = foreach a generate name, age;" +
        "e = filter d by age > c.age;" +
        "f = store e into 'empty';";

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator store = newLogicalPlan.getSinks().get( 0 );
        Operator foreach = newLogicalPlan.getPredecessors(store).get(0);
        Assert.assertTrue( foreach instanceof LOForEach );
    }
    
    // See PIG-1935
    @Test
    public void testPushUpFilterAboveBinCond() throws Exception {
        String query = "data = LOAD 'data.txt' as (referrer:chararray, canonical_url:chararray, ip:chararray);" +
        "best_url = FOREACH data GENERATE ((canonical_url != '' and canonical_url is not null) ? canonical_url : referrer) AS url, ip;" +
        "filtered = FILTER best_url BY url == 'badsite.com';" +
        "store filtered into 'empty';";

        LogicalPlan newLogicalPlan = buildPlan( query );

        Operator store = newLogicalPlan.getSinks().get( 0 );
        Operator filter = newLogicalPlan.getPredecessors(store).get(0);
        Assert.assertTrue( filter instanceof LOFilter );
    }

    private LogicalPlan buildPlan(String query) throws Exception {
        PigServer pigServer = new PigServer( pc );
        LogicalPlan newLogicalPlan = Util.buildLp(pigServer, query);
        PlanOptimizer optimizer = new MyPlanOptimizer( newLogicalPlan, 3 );
        optimizer.optimize();
        return newLogicalPlan;
    }
    
    public class MyPlanOptimizer extends LogicalPlanOptimizer {
        protected MyPlanOptimizer(OperatorPlan p,  int iterations) {
            super(p, iterations, new HashSet<String>());
        }
        
        @Override
        public void addPlanTransformListener(PlanTransformListener listener) {
            super.addPlanTransformListener(listener);
        }
        
       @Override
       protected List<Set<Rule>> buildRuleSets() {            
            List<Set<Rule>> ls = new ArrayList<Set<Rule>>();
            
            Set<Rule> s = new HashSet<Rule>();
            // add split filter rule
            Rule r = new LoadTypeCastInserter( "TypeCastInserter" );
            s.add(r);
            ls.add(s);
             
            s = new HashSet<Rule>();
            r = new FilterAboveForeach( "FilterAboveForeach" );
            s.add(r);            
            ls.add(s);
            
            return ls;
        }
    }    
}