src/org/apache/pig/parser/QueryParserDriver.java - pig - Git at Google

 /*
  * 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.parser;

 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.FileReader;
 import java.io.IOException;
 import java.io.StringReader;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.antlr.runtime.BaseRecognizer;
 import org.antlr.runtime.CharStream;
 import org.antlr.runtime.CommonTokenStream;
 import org.antlr.runtime.RecognitionException;
 import org.antlr.runtime.tree.CommonTree;
 import org.antlr.runtime.tree.CommonTreeNodeStream;
 import org.antlr.runtime.tree.Tree;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.pig.PigException;
 import org.apache.pig.PigServer;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.impl.PigContext;
 import org.apache.pig.impl.io.FileLocalizer;
 import org.apache.pig.impl.io.FileLocalizer.FetchFileRet;
 import org.apache.pig.impl.io.ResourceNotFoundException;
 import org.apache.pig.impl.logicalLayer.FrontendException;
 import org.apache.pig.newplan.Operator;
 import org.apache.pig.newplan.logical.relational.LogicalPlan;
 import org.apache.pig.newplan.logical.relational.LogicalSchema;
 import org.apache.pig.parser.QueryParser.literal_return;
 import org.apache.pig.parser.QueryParser.schema_return;
 import org.apache.pig.tools.pigstats.ScriptState;
 import org.apache.pig.validator.BlackAndWhitelistFilter;
 import org.apache.pig.validator.PigCommandFilter;

 public class QueryParserDriver {
     private static final Log LOG = LogFactory.getLog(QueryParserDriver.class);

     private static final String MACRO_DEF = "MACRO_DEF";
     private static final String MACRO_INLINE = "MACRO_INLINE";
     private static final String IMPORT_DEF = "import";
     private static final String REGISTER_DEF = "register";

     private PigContext pigContext;
     private PigServer pigServer;
     private String scope;
     private Map<String, String>fileNameMap;
     private Map<String, Operator> operators;
     private String lastRel;
     private Set<String> importSeen;
     private Set<String> macroSeen;

     private static Map<String, FetchFileRet> fnameMap = new HashMap<String, FetchFileRet>();

     public QueryParserDriver(PigContext pigContext, String scope, Map<String, String> fileNameMap) {
         this.pigContext = pigContext;
         this.pigServer = null; // lazily instantiated for register statements
         this.scope = scope;
         this.fileNameMap = fileNameMap;
         importSeen = new HashSet<String>();
         macroSeen = new HashSet<String>();
     }

     private static Tree parseSchema(CommonTokenStream tokens) throws ParserException {
         QueryParser parser = QueryParserUtils.createParser(tokens);

         schema_return result = null;
         try {
             result = parser.schema();
         } catch (RecognitionException e) {
             String msg = parser.getErrorHeader(e) + " "
                     + parser.getErrorMessage(e, parser.getTokenNames());
             throw new ParserException(msg);
         } catch(RuntimeException ex) {
             throw new ParserException( ex.getMessage() );
         }

         Tree ast = (Tree)result.getTree();
         checkError( parser );

         return ast;
     }

     public LogicalSchema parseSchema(String input) throws ParserException {
         CommonTokenStream tokenStream = tokenize( input, null );
         LogicalSchema schema = null;
         Tree ast = parseSchema( tokenStream );

         try{
             CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
             AstValidator walker = new AstValidator( nodes );
             ast = (Tree)walker.field_def_list().getTree();
             checkError( walker );

             LogicalPlanGenerator planGenerator =
                 new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
             schema = planGenerator.field_def_list().schema;
             checkError( planGenerator );
         } catch(RecognitionException ex) {
             throw new ParserException( ex );
         } catch(Exception ex) {
             throw new ParserException( ex.getMessage(), ex );
         }

         return schema;
     }

     private static Tree parseConstant(CommonTokenStream tokens) throws ParserException {
         QueryParser parser = QueryParserUtils.createParser(tokens);

         literal_return result = null;
         try {
             result = parser.literal();
         } catch (RecognitionException e) {
             String msg = parser.getErrorHeader(e) + " "
                     + parser.getErrorMessage(e, parser.getTokenNames());
             throw new ParserException(msg);
         } catch(RuntimeException ex) {
             throw new ParserException( ex.getMessage() );
         }

         Tree ast = (Tree)result.getTree();
         checkError( parser );

         return ast;
     }

     public Object parseConstant(String input) throws ParserException {
         CommonTokenStream tokenStream = tokenize( input, null );
         Object value = null;
         Tree ast = parseConstant( tokenStream );

         try{
             CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
             AstValidator walker = new AstValidator( nodes );
             ast = (Tree)walker.literal().getTree();
             checkError( walker );

             LogicalPlanGenerator planGenerator =
                 new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
             value = planGenerator.literal().value;
             checkError( planGenerator );
         } catch(RecognitionException ex) {
             throw new ParserException( ex );
         } catch(Exception ex) {
             throw new ParserException( ex.getMessage(), ex );
         }

         return value;
     }

     public LogicalPlan parse(String query) throws ParserException {
         LogicalPlan plan = null;

         ScriptState ss = ScriptState.get();
         CommonTokenStream tokenStream = tokenize(query, ss.getFileName());

         Tree ast = parse( tokenStream );
         ast = expandMacro( ast );

         try{
             ast = validateAst( ast );
             applyRegisters(ast);

             LogicalPlanGenerator planGenerator =
                 new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
             planGenerator.query();

             checkError( planGenerator );

             plan = planGenerator.getLogicalPlan();
             operators = planGenerator.getOperators();
             lastRel = planGenerator.getLastRel();
         } catch(RecognitionException ex) {
             throw new ParserException( ex );
         } catch(Exception ex) {
             throw new ParserException( ex.getMessage(), ex );
         }

         return plan;
     }

     public Map<String, Operator> getOperators() {
         return operators;
     }

     static CommonTokenStream tokenize(String query, String source)
             throws ParserException {
         CharStream input;
         try {
             input = new QueryParserStringStream(query, source);
         } catch (IOException ex) {
             throw new ParserException("Unexpected IOException: "
                     + ex.getMessage());
         }
         QueryLexer lexer = new QueryLexer(input);
         CommonTokenStream tokens = new CommonTokenStream(lexer);
         checkError(lexer);
         return tokens;
     }

     private static void checkError(BaseRecognizer recognizer)
             throws ParserException {
         int errorCount = recognizer.getNumberOfSyntaxErrors();
         if (0 < errorCount)
             throw new ParserException("Encountered " + errorCount
                     + " parsing errors in the query");
     }

     static Tree parse(CommonTokenStream tokens) throws ParserException {
         QueryParser parser = QueryParserUtils.createParser(tokens);

         QueryParser.query_return result = null;
         try {
             result = parser.query();
         } catch (RecognitionException e) {
             String msg = parser.getErrorHeader(e) + " "
                     + parser.getErrorMessage(e, parser.getTokenNames());
             SourceLocation location = new SourceLocation(null, e.line,e.charPositionInLine);
             throw new ParserException(msg, location);
         } catch(RuntimeException ex) {
             throw new ParserException( ex.getMessage() );
         }

         Tree ast = (Tree) result.getTree();
         checkError(parser);

         return ast;
     }

     private static Tree validateAst(Tree ast) throws RecognitionException, ParserException {
         CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
         AstValidator walker = new AstValidator( nodes );
         AstValidator.query_return newResult = walker.query();
         Tree newAst = (Tree)newResult.getTree();

         checkError( walker );

         return newAst;
     }

     Tree expandMacro(Tree ast) throws ParserException {
         LOG.debug("Original macro AST:\n" + ast.toStringTree() + "\n");

         // first insert the import files
         while (expandImport(ast))
             ;

         LOG.debug("macro AST after import:\n" + ast.toStringTree() + "\n");

         List<CommonTree> macroNodes = new ArrayList<CommonTree>();
         List<CommonTree> inlineNodes = new ArrayList<CommonTree>();

         // find all macro def/inline nodes
         traverse(ast, macroNodes, inlineNodes);

         Map<String, PigMacro> seen = new HashMap<String, PigMacro>();
         List<PigMacro> macroDefs = new ArrayList<PigMacro>();

         // gether all the def nodes
         for (CommonTree t : macroNodes) {
             macroDefs.add(makeMacroDef(t, seen));
         }

         // inline macros
         inlineMacro(inlineNodes, macroDefs);

         LOG.debug("Resulting macro AST:\n" + ast.toStringTree() + "\n");

         return ast;
     }

     private void inlineMacro(List<CommonTree> inlineNodes,
             List<PigMacro> macroDefs) throws ParserException {
         for (CommonTree t : inlineNodes) {
             Set<String> macroStack = new HashSet<String>();
             CommonTree newTree = PigMacro.macroInline(t, macroDefs, macroStack, pigContext);

             List<CommonTree> nodes = new ArrayList<CommonTree>();
             traverseInline(newTree, nodes);

             if (nodes.isEmpty()) {
                 QueryParserUtils.replaceNodeWithNodeList(t, newTree, null);
             } else {
                 inlineMacro(nodes, macroDefs);
             }
         }
     }

     private void applyRegisters(Tree t) throws ExecException, ParserException {
         if (t.getText().equalsIgnoreCase(REGISTER_DEF)) {
             String path = t.getChild(0).getText();
             path = path.substring(1, path.length()-1);
             try {
                 if (t.getChildCount() == 5) {
                     new RegisterResolver(getPigServer()).parseRegister(path, t.getChild(2).getText(), t.getChild(4).getText());
                 } else {
                     new RegisterResolver(getPigServer()).parseRegister(path, null, null);
                 }
             } catch (IOException ioe) {
                 throw new ParserException(ioe.getMessage());
             }
         } else {
             for (int i = 0; i < t.getChildCount(); i++) {
                 applyRegisters(t.getChild(i));
             }
         }
     }

     private PigServer getPigServer() throws ExecException {
         if (pigServer == null) {
             pigServer = new PigServer(pigContext, false);
         }
         return pigServer;
     }

     private void traverseInline(Tree t, List<CommonTree> nodes) {
         if (t.getText().equals(MACRO_INLINE)) {
             nodes.add((CommonTree)t);
         }
         int n = t.getChildCount();
         for (int i = 0; i < n; i++) {
             Tree t0 = t.getChild(i);
             traverseInline(t0, nodes);
         }
     }

     private boolean expandImport(Tree ast) throws ParserException {
         List<CommonTree> nodes = new ArrayList<CommonTree>();
         traverseImport(ast, nodes);
         if (nodes.isEmpty())
             return false;

         // Validate if imports are enabled/disabled
         final BlackAndWhitelistFilter filter = new BlackAndWhitelistFilter(
                 this.pigContext);
         try {
             filter.validate(PigCommandFilter.Command.IMPORT);
         } catch (FrontendException e) {
             throw new ParserException(e.getMessage());
         }
         for (CommonTree t : nodes) {
             macroImport(t);
         }

         return true;
     }

     static void traverseImport(Tree t, List<CommonTree> nodes) {
         if (t.getText().equalsIgnoreCase(IMPORT_DEF)) {
             nodes.add((CommonTree)t);
         }
         int n = t.getChildCount();
         for (int i = 0; i < n; i++) {
             Tree t0 = t.getChild(i);
             traverseImport(t0, nodes);
         }
     }

     static void traverse(Tree t, List<CommonTree> macroNodes,
             List<CommonTree> inlineNodes) {
         if (t.getText().equals(MACRO_DEF)) {
             macroNodes.add((CommonTree) t.getParent());
         } else if (t.getText().equals(MACRO_INLINE)) {
             inlineNodes.add((CommonTree) t);
         }
         int n = t.getChildCount();
         for (int i = 0; i < n; i++) {
             Tree t0 = t.getChild(i);
             traverse(t0, macroNodes, inlineNodes);
         }
     }

     private FetchFileRet getMacroFile(String fname) {
         FetchFileRet localFileRet = null;
         try {
             if (fnameMap.get(fname) != null) {
                 localFileRet = fnameMap.get(fname);
             } else {
               try {
                 File localFile = QueryParserUtils.getFileFromImportSearchPath(fname);
                 localFileRet = localFile == null ?
                         FileLocalizer.fetchFile(pigContext.getProperties(), fname)
                         : new FetchFileRet(localFile.getCanonicalFile(), false);
               } catch (FileNotFoundException e) {
                 // ignore this since we'll attempt to load as a resource before failing
                 LOG.debug(String.format("Macro file %s was not found", fname));
               }

               // try loading the macro file as a resource in case it is packaged in a registered jar
               if (localFileRet == null) {
                 LOG.debug(String.format("Attempting to load macro file %s as a resource", fname));

                 try
                 {
                   localFileRet = FileLocalizer.fetchResource(fname);
                   LOG.debug(String.format("Found macro file %s as resource", fname));
                 }
                 catch (ResourceNotFoundException e)
                 {
                   LOG.debug(String.format("Macro file %s was not found as resource either", fname));
                   LOG.error(String.format("Failed to find macro file %s", fname));
                   throw new ExecException("file '" + fname + "' does not exist.", 101, PigException.INPUT);
                 }
               }

               fnameMap.put(fname, localFileRet);
             }
         } catch (IOException e) {
             throw new RuntimeException("Unable to fetch macro file '" + fname + "'", e);
         }
         return localFileRet;
     }

     /*
      * MacroDef node has two child nodes:
      *      1. name
      *      2. MACRO_DEF (PARAMS, RETURN_VAL, MACRO_BODY)
      */
     private PigMacro makeMacroDef(CommonTree t, Map<String, PigMacro> seen)
             throws ParserException {
         String mn = t.getChild(0).getText();

         if (!macroSeen.add(mn)) {
             String msg = getErrorMessage(null, t, null,
                     "Duplicated macro name '" + mn + "'");
             throw new ParserException(msg);
         }

         if (seen != null) {
             for (String s : seen.keySet()) {
                 macroSeen.add(s);
             }
         }

         String fname = ((PigParserNode)t).getFileName();

         Tree defNode = t.getChild(1);

         // get parameter markers
         ArrayList<String> params = new ArrayList<String>();
         Tree paramNode = defNode.getChild(0);
         int n = paramNode.getChildCount();
         for (int i = 0; i < n; i++) {
             params.add(paramNode.getChild(i).getText());
         }

         // get return alias markers
         ArrayList<String> returns = new ArrayList<String>();
         Tree retNode = defNode.getChild(1);
         int m = retNode.getChildCount();
         for (int i = 0; i < m; i++) {
             returns.add(retNode.getChild(i).getText());
         }

         // get macro body
         Tree bodyNode = defNode.getChild(2);
         String body = bodyNode.getChild(0).getText();

         body = body.substring(1, body.length() - 1);

         // sometimes the script has no filename, like when a string is passed to PigServer for
         // example. See PIG-2866.
         if (!fname.isEmpty()) {
             FetchFileRet localFileRet = getMacroFile(fname);
             fname = localFileRet.file.getAbsolutePath();
         }

         PigMacro pm = new PigMacro(mn, fname, params, returns, body, seen);

         try {
             pm.validate();
         } catch (IOException e) {
             String msg = getErrorMessage(null, t,
                     "Invalid macro definition: ", e.getMessage());
             throw new ParserException(msg);
         }

         // set the starting line number of the macro
         PigParserNode pnode = (PigParserNode)bodyNode.getChild(0);
         pm.setStartLine(pnode.getStartLine());

         seen.put(mn, pm);

         // delete this node
         Tree defineNode = t.getParent();
         Tree stmtNode = defineNode.getParent();
         stmtNode.deleteChild(defineNode.getChildIndex());

         return pm;
     }

     private void macroImport(CommonTree t) throws ParserException {
         // remove quote
         String fname = t.getChild(0).getText();
         fname = QueryParserUtils.removeQuotes(fname);
         if (!importSeen.add(fname)) {
             // we've already imported this file, so just skip this import statement
             LOG.debug("Ignoring duplicated import " + fname);
             t.getParent().deleteChild(t.getChildIndex());
             return;
         }

         Tree macroAST = null;
         if (pigContext.macros.containsKey(fname)) {
             macroAST = pigContext.macros.get(fname);
         } else {
             FetchFileRet localFileRet = getMacroFile(fname);

             BufferedReader in = null;
             try {
                 in = new BufferedReader(new FileReader(localFileRet.file));
             } catch (FileNotFoundException e) {
                 String msg = getErrorMessage(fname, t,
                         "Failed to import file '" + fname + "'", e.getMessage());
                 throw new ParserException(msg);
             }

             StringBuilder sb = new StringBuilder();
             String line = null;
             try {
                 line = in.readLine();
                 while (line != null) {
                     sb.append(line).append("\n");
                     line = in.readLine();
                 }
             } catch (IOException e) {
                 String msg = getErrorMessage(fname, t,
                         "Failed to read file '" + fname + "'", e.getMessage());
                 throw new ParserException(msg);
             }

             String macroText = null;
             try {
                 in.close();
                 in = new BufferedReader(new StringReader(sb.toString()));
                 macroText = pigContext.doParamSubstitution(in);
             } catch (IOException e) {
                 String msg = getErrorMessage(fname, t,
                     "Parameter sustitution failed for macro.", e.getMessage());
                 throw new ParserException(msg);
             }

             // parse
             CommonTokenStream tokenStream = tokenize(macroText, fname);

             try {
                 macroAST = parse( tokenStream );
                 pigContext.macros.put(fname, macroAST);
             } catch(RuntimeException ex) {
                 throw new ParserException( ex.getMessage() );
             }
         }

         QueryParserUtils.replaceNodeWithNodeList(t, (CommonTree)macroAST, fname);
     }

     private String getErrorMessage(String importFile,
             CommonTree t, String header, String reason) {
         StringBuilder sb = new StringBuilder();
         PigParserNode node = (PigParserNode)t;
         String file = node.getFileName();
         sb.append("<");
         if (file == null) {
             ScriptState ss = ScriptState.get();
             if (ss != null) file = ss.getFileName();
         }
         if (!file.isEmpty() && !file.equals(importFile)) {
             sb.append("at ").append(file).append(", ");
         }
         sb.append("line ").append(t.getLine()).append("> ").append(header);
         if (reason != null) {
             sb.append(". Reason: ").append(reason);
         }
         return sb.toString();
     }

     public String getLastRel() {
         return lastRel;
     }
 }
	/*
	* 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.parser;

	import java.io.BufferedReader;
	import java.io.File;
	import java.io.FileNotFoundException;
	import java.io.FileReader;
	import java.io.IOException;
	import java.io.StringReader;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.antlr.runtime.BaseRecognizer;
	import org.antlr.runtime.CharStream;
	import org.antlr.runtime.CommonTokenStream;
	import org.antlr.runtime.RecognitionException;
	import org.antlr.runtime.tree.CommonTree;
	import org.antlr.runtime.tree.CommonTreeNodeStream;
	import org.antlr.runtime.tree.Tree;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.pig.PigException;
	import org.apache.pig.PigServer;
	import org.apache.pig.backend.executionengine.ExecException;
	import org.apache.pig.impl.PigContext;
	import org.apache.pig.impl.io.FileLocalizer;
	import org.apache.pig.impl.io.FileLocalizer.FetchFileRet;
	import org.apache.pig.impl.io.ResourceNotFoundException;
	import org.apache.pig.impl.logicalLayer.FrontendException;
	import org.apache.pig.newplan.Operator;
	import org.apache.pig.newplan.logical.relational.LogicalPlan;
	import org.apache.pig.newplan.logical.relational.LogicalSchema;
	import org.apache.pig.parser.QueryParser.literal_return;
	import org.apache.pig.parser.QueryParser.schema_return;
	import org.apache.pig.tools.pigstats.ScriptState;
	import org.apache.pig.validator.BlackAndWhitelistFilter;
	import org.apache.pig.validator.PigCommandFilter;

	public class QueryParserDriver {
	private static final Log LOG = LogFactory.getLog(QueryParserDriver.class);

	private static final String MACRO_DEF = "MACRO_DEF";
	private static final String MACRO_INLINE = "MACRO_INLINE";
	private static final String IMPORT_DEF = "import";
	private static final String REGISTER_DEF = "register";

	private PigContext pigContext;
	private PigServer pigServer;
	private String scope;
	private Map<String, String>fileNameMap;
	private Map<String, Operator> operators;
	private String lastRel;
	private Set<String> importSeen;
	private Set<String> macroSeen;

	private static Map<String, FetchFileRet> fnameMap = new HashMap<String, FetchFileRet>();

	public QueryParserDriver(PigContext pigContext, String scope, Map<String, String> fileNameMap) {
	this.pigContext = pigContext;
	this.pigServer = null; // lazily instantiated for register statements
	this.scope = scope;
	this.fileNameMap = fileNameMap;
	importSeen = new HashSet<String>();
	macroSeen = new HashSet<String>();
	}

	private static Tree parseSchema(CommonTokenStream tokens) throws ParserException {
	QueryParser parser = QueryParserUtils.createParser(tokens);

	schema_return result = null;
	try {
	result = parser.schema();
	} catch (RecognitionException e) {
	String msg = parser.getErrorHeader(e) + " "
	+ parser.getErrorMessage(e, parser.getTokenNames());
	throw new ParserException(msg);
	} catch(RuntimeException ex) {
	throw new ParserException( ex.getMessage() );
	}

	Tree ast = (Tree)result.getTree();
	checkError( parser );

	return ast;
	}

	public LogicalSchema parseSchema(String input) throws ParserException {
	CommonTokenStream tokenStream = tokenize( input, null );
	LogicalSchema schema = null;
	Tree ast = parseSchema( tokenStream );

	try{
	CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
	AstValidator walker = new AstValidator( nodes );
	ast = (Tree)walker.field_def_list().getTree();
	checkError( walker );

	LogicalPlanGenerator planGenerator =
	new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
	schema = planGenerator.field_def_list().schema;
	checkError( planGenerator );
	} catch(RecognitionException ex) {
	throw new ParserException( ex );
	} catch(Exception ex) {
	throw new ParserException( ex.getMessage(), ex );
	}

	return schema;
	}

	private static Tree parseConstant(CommonTokenStream tokens) throws ParserException {
	QueryParser parser = QueryParserUtils.createParser(tokens);

	literal_return result = null;
	try {
	result = parser.literal();
	} catch (RecognitionException e) {
	String msg = parser.getErrorHeader(e) + " "
	+ parser.getErrorMessage(e, parser.getTokenNames());
	throw new ParserException(msg);
	} catch(RuntimeException ex) {
	throw new ParserException( ex.getMessage() );
	}

	Tree ast = (Tree)result.getTree();
	checkError( parser );

	return ast;
	}

	public Object parseConstant(String input) throws ParserException {
	CommonTokenStream tokenStream = tokenize( input, null );
	Object value = null;
	Tree ast = parseConstant( tokenStream );

	try{
	CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
	AstValidator walker = new AstValidator( nodes );
	ast = (Tree)walker.literal().getTree();
	checkError( walker );

	LogicalPlanGenerator planGenerator =
	new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
	value = planGenerator.literal().value;
	checkError( planGenerator );
	} catch(RecognitionException ex) {
	throw new ParserException( ex );
	} catch(Exception ex) {
	throw new ParserException( ex.getMessage(), ex );
	}

	return value;
	}

	public LogicalPlan parse(String query) throws ParserException {
	LogicalPlan plan = null;

	ScriptState ss = ScriptState.get();
	CommonTokenStream tokenStream = tokenize(query, ss.getFileName());

	Tree ast = parse( tokenStream );
	ast = expandMacro( ast );

	try{
	ast = validateAst( ast );
	applyRegisters(ast);

	LogicalPlanGenerator planGenerator =
	new LogicalPlanGenerator( new CommonTreeNodeStream( ast ), pigContext, scope, fileNameMap );
	planGenerator.query();

	checkError( planGenerator );

	plan = planGenerator.getLogicalPlan();
	operators = planGenerator.getOperators();
	lastRel = planGenerator.getLastRel();
	} catch(RecognitionException ex) {
	throw new ParserException( ex );
	} catch(Exception ex) {
	throw new ParserException( ex.getMessage(), ex );
	}

	return plan;
	}

	public Map<String, Operator> getOperators() {
	return operators;
	}

	static CommonTokenStream tokenize(String query, String source)
	throws ParserException {
	CharStream input;
	try {
	input = new QueryParserStringStream(query, source);
	} catch (IOException ex) {
	throw new ParserException("Unexpected IOException: "
	+ ex.getMessage());
	}
	QueryLexer lexer = new QueryLexer(input);
	CommonTokenStream tokens = new CommonTokenStream(lexer);
	checkError(lexer);
	return tokens;
	}

	private static void checkError(BaseRecognizer recognizer)
	throws ParserException {
	int errorCount = recognizer.getNumberOfSyntaxErrors();
	if (0 < errorCount)
	throw new ParserException("Encountered " + errorCount
	+ " parsing errors in the query");
	}

	static Tree parse(CommonTokenStream tokens) throws ParserException {
	QueryParser parser = QueryParserUtils.createParser(tokens);

	QueryParser.query_return result = null;
	try {
	result = parser.query();
	} catch (RecognitionException e) {
	String msg = parser.getErrorHeader(e) + " "
	+ parser.getErrorMessage(e, parser.getTokenNames());
	SourceLocation location = new SourceLocation(null, e.line,e.charPositionInLine);
	throw new ParserException(msg, location);
	} catch(RuntimeException ex) {
	throw new ParserException( ex.getMessage() );
	}

	Tree ast = (Tree) result.getTree();
	checkError(parser);

	return ast;
	}

	private static Tree validateAst(Tree ast) throws RecognitionException, ParserException {
	CommonTreeNodeStream nodes = new CommonTreeNodeStream( ast );
	AstValidator walker = new AstValidator( nodes );
	AstValidator.query_return newResult = walker.query();
	Tree newAst = (Tree)newResult.getTree();

	checkError( walker );

	return newAst;
	}

	Tree expandMacro(Tree ast) throws ParserException {
	LOG.debug("Original macro AST:\n" + ast.toStringTree() + "\n");

	// first insert the import files
	while (expandImport(ast))
	;

	LOG.debug("macro AST after import:\n" + ast.toStringTree() + "\n");

	List<CommonTree> macroNodes = new ArrayList<CommonTree>();
	List<CommonTree> inlineNodes = new ArrayList<CommonTree>();

	// find all macro def/inline nodes
	traverse(ast, macroNodes, inlineNodes);

	Map<String, PigMacro> seen = new HashMap<String, PigMacro>();
	List<PigMacro> macroDefs = new ArrayList<PigMacro>();

	// gether all the def nodes
	for (CommonTree t : macroNodes) {
	macroDefs.add(makeMacroDef(t, seen));
	}

	// inline macros
	inlineMacro(inlineNodes, macroDefs);

	LOG.debug("Resulting macro AST:\n" + ast.toStringTree() + "\n");

	return ast;
	}

	private void inlineMacro(List<CommonTree> inlineNodes,
	List<PigMacro> macroDefs) throws ParserException {
	for (CommonTree t : inlineNodes) {
	Set<String> macroStack = new HashSet<String>();
	CommonTree newTree = PigMacro.macroInline(t, macroDefs, macroStack, pigContext);

	List<CommonTree> nodes = new ArrayList<CommonTree>();
	traverseInline(newTree, nodes);

	if (nodes.isEmpty()) {
	QueryParserUtils.replaceNodeWithNodeList(t, newTree, null);
	} else {
	inlineMacro(nodes, macroDefs);
	}
	}
	}

	private void applyRegisters(Tree t) throws ExecException, ParserException {
	if (t.getText().equalsIgnoreCase(REGISTER_DEF)) {
	String path = t.getChild(0).getText();
	path = path.substring(1, path.length()-1);
	try {
	if (t.getChildCount() == 5) {
	new RegisterResolver(getPigServer()).parseRegister(path, t.getChild(2).getText(), t.getChild(4).getText());
	} else {
	new RegisterResolver(getPigServer()).parseRegister(path, null, null);
	}
	} catch (IOException ioe) {
	throw new ParserException(ioe.getMessage());
	}
	} else {
	for (int i = 0; i < t.getChildCount(); i++) {
	applyRegisters(t.getChild(i));
	}
	}
	}

	private PigServer getPigServer() throws ExecException {
	if (pigServer == null) {
	pigServer = new PigServer(pigContext, false);
	}
	return pigServer;
	}

	private void traverseInline(Tree t, List<CommonTree> nodes) {
	if (t.getText().equals(MACRO_INLINE)) {
	nodes.add((CommonTree)t);
	}
	int n = t.getChildCount();
	for (int i = 0; i < n; i++) {
	Tree t0 = t.getChild(i);
	traverseInline(t0, nodes);
	}
	}

	private boolean expandImport(Tree ast) throws ParserException {
	List<CommonTree> nodes = new ArrayList<CommonTree>();
	traverseImport(ast, nodes);
	if (nodes.isEmpty())
	return false;

	// Validate if imports are enabled/disabled
	final BlackAndWhitelistFilter filter = new BlackAndWhitelistFilter(
	this.pigContext);
	try {
	filter.validate(PigCommandFilter.Command.IMPORT);
	} catch (FrontendException e) {
	throw new ParserException(e.getMessage());
	}
	for (CommonTree t : nodes) {
	macroImport(t);
	}

	return true;
	}

	static void traverseImport(Tree t, List<CommonTree> nodes) {
	if (t.getText().equalsIgnoreCase(IMPORT_DEF)) {
	nodes.add((CommonTree)t);
	}
	int n = t.getChildCount();
	for (int i = 0; i < n; i++) {
	Tree t0 = t.getChild(i);
	traverseImport(t0, nodes);
	}
	}

	static void traverse(Tree t, List<CommonTree> macroNodes,
	List<CommonTree> inlineNodes) {
	if (t.getText().equals(MACRO_DEF)) {
	macroNodes.add((CommonTree) t.getParent());
	} else if (t.getText().equals(MACRO_INLINE)) {
	inlineNodes.add((CommonTree) t);
	}
	int n = t.getChildCount();
	for (int i = 0; i < n; i++) {
	Tree t0 = t.getChild(i);
	traverse(t0, macroNodes, inlineNodes);
	}
	}

	private FetchFileRet getMacroFile(String fname) {
	FetchFileRet localFileRet = null;
	try {
	if (fnameMap.get(fname) != null) {
	localFileRet = fnameMap.get(fname);
	} else {
	try {
	File localFile = QueryParserUtils.getFileFromImportSearchPath(fname);
	localFileRet = localFile == null ?
	FileLocalizer.fetchFile(pigContext.getProperties(), fname)
	: new FetchFileRet(localFile.getCanonicalFile(), false);
	} catch (FileNotFoundException e) {
	// ignore this since we'll attempt to load as a resource before failing
	LOG.debug(String.format("Macro file %s was not found", fname));
	}

	// try loading the macro file as a resource in case it is packaged in a registered jar
	if (localFileRet == null) {
	LOG.debug(String.format("Attempting to load macro file %s as a resource", fname));

	try
	{
	localFileRet = FileLocalizer.fetchResource(fname);
	LOG.debug(String.format("Found macro file %s as resource", fname));
	}
	catch (ResourceNotFoundException e)
	{
	LOG.debug(String.format("Macro file %s was not found as resource either", fname));
	LOG.error(String.format("Failed to find macro file %s", fname));
	throw new ExecException("file '" + fname + "' does not exist.", 101, PigException.INPUT);
	}
	}

	fnameMap.put(fname, localFileRet);
	}
	} catch (IOException e) {
	throw new RuntimeException("Unable to fetch macro file '" + fname + "'", e);
	}
	return localFileRet;
	}

	/*
	* MacroDef node has two child nodes:
	* 1. name
	* 2. MACRO_DEF (PARAMS, RETURN_VAL, MACRO_BODY)
	*/
	private PigMacro makeMacroDef(CommonTree t, Map<String, PigMacro> seen)
	throws ParserException {
	String mn = t.getChild(0).getText();

	if (!macroSeen.add(mn)) {
	String msg = getErrorMessage(null, t, null,
	"Duplicated macro name '" + mn + "'");
	throw new ParserException(msg);
	}

	if (seen != null) {
	for (String s : seen.keySet()) {
	macroSeen.add(s);
	}
	}

	String fname = ((PigParserNode)t).getFileName();

	Tree defNode = t.getChild(1);

	// get parameter markers
	ArrayList<String> params = new ArrayList<String>();
	Tree paramNode = defNode.getChild(0);
	int n = paramNode.getChildCount();
	for (int i = 0; i < n; i++) {
	params.add(paramNode.getChild(i).getText());
	}

	// get return alias markers
	ArrayList<String> returns = new ArrayList<String>();
	Tree retNode = defNode.getChild(1);
	int m = retNode.getChildCount();
	for (int i = 0; i < m; i++) {
	returns.add(retNode.getChild(i).getText());
	}

	// get macro body
	Tree bodyNode = defNode.getChild(2);
	String body = bodyNode.getChild(0).getText();

	body = body.substring(1, body.length() - 1);

	// sometimes the script has no filename, like when a string is passed to PigServer for
	// example. See PIG-2866.
	if (!fname.isEmpty()) {
	FetchFileRet localFileRet = getMacroFile(fname);
	fname = localFileRet.file.getAbsolutePath();
	}

	PigMacro pm = new PigMacro(mn, fname, params, returns, body, seen);

	try {
	pm.validate();
	} catch (IOException e) {
	String msg = getErrorMessage(null, t,
	"Invalid macro definition: ", e.getMessage());
	throw new ParserException(msg);
	}

	// set the starting line number of the macro
	PigParserNode pnode = (PigParserNode)bodyNode.getChild(0);
	pm.setStartLine(pnode.getStartLine());

	seen.put(mn, pm);

	// delete this node
	Tree defineNode = t.getParent();
	Tree stmtNode = defineNode.getParent();
	stmtNode.deleteChild(defineNode.getChildIndex());

	return pm;
	}

	private void macroImport(CommonTree t) throws ParserException {
	// remove quote
	String fname = t.getChild(0).getText();
	fname = QueryParserUtils.removeQuotes(fname);
	if (!importSeen.add(fname)) {
	// we've already imported this file, so just skip this import statement
	LOG.debug("Ignoring duplicated import " + fname);
	t.getParent().deleteChild(t.getChildIndex());
	return;
	}

	Tree macroAST = null;
	if (pigContext.macros.containsKey(fname)) {
	macroAST = pigContext.macros.get(fname);
	} else {
	FetchFileRet localFileRet = getMacroFile(fname);

	BufferedReader in = null;
	try {
	in = new BufferedReader(new FileReader(localFileRet.file));
	} catch (FileNotFoundException e) {
	String msg = getErrorMessage(fname, t,
	"Failed to import file '" + fname + "'", e.getMessage());
	throw new ParserException(msg);
	}

	StringBuilder sb = new StringBuilder();
	String line = null;
	try {
	line = in.readLine();
	while (line != null) {
	sb.append(line).append("\n");
	line = in.readLine();
	}
	} catch (IOException e) {
	String msg = getErrorMessage(fname, t,
	"Failed to read file '" + fname + "'", e.getMessage());
	throw new ParserException(msg);
	}

	String macroText = null;
	try {
	in.close();
	in = new BufferedReader(new StringReader(sb.toString()));
	macroText = pigContext.doParamSubstitution(in);
	} catch (IOException e) {
	String msg = getErrorMessage(fname, t,
	"Parameter sustitution failed for macro.", e.getMessage());
	throw new ParserException(msg);
	}

	// parse
	CommonTokenStream tokenStream = tokenize(macroText, fname);

	try {
	macroAST = parse( tokenStream );
	pigContext.macros.put(fname, macroAST);
	} catch(RuntimeException ex) {
	throw new ParserException( ex.getMessage() );
	}
	}

	QueryParserUtils.replaceNodeWithNodeList(t, (CommonTree)macroAST, fname);
	}

	private String getErrorMessage(String importFile,
	CommonTree t, String header, String reason) {
	StringBuilder sb = new StringBuilder();
	PigParserNode node = (PigParserNode)t;
	String file = node.getFileName();
	sb.append("<");
	if (file == null) {
	ScriptState ss = ScriptState.get();
	if (ss != null) file = ss.getFileName();
	}
	if (!file.isEmpty() && !file.equals(importFile)) {
	sb.append("at ").append(file).append(", ");
	}
	sb.append("line ").append(t.getLine()).append("> ").append(header);
	if (reason != null) {
	sb.append(". Reason: ").append(reason);
	}
	return sb.toString();
	}

	public String getLastRel() {
	return lastRel;
	}
	}