/*
 * 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.newplan.logical.relational;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;

import org.apache.pig.PigException;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.newplan.logical.expression.LogicalExpression;

/**
 * Schema, from a logical perspective.
 */
public class LogicalSchema {

    public static class LogicalFieldSchema {
        public String alias;
        public byte type;
        public long uid;
        public LogicalSchema schema;

        public LogicalFieldSchema(String alias, LogicalSchema schema,  byte type) {
            this(alias, schema, type, -1);
        }
        
        public LogicalFieldSchema(LogicalFieldSchema fs) {
            this(fs.alias, fs.schema, fs.type, fs.uid);
        }
        
        public LogicalFieldSchema(String alias, LogicalSchema schema,  byte type, long uid) {
            this.alias = alias;
            this.type = type;
            this.schema = schema;
            this.uid = uid;
        }
        
        /**
         * Equality is defined as having the same type and either the same schema
         * or both null schema.  Alias and uid are not checked.
         */
        public boolean isEqual(Object other) {
            return isEqual(other, false);
        }
               
        /**
         * Equality is defined as having the same type and either the same schema
         * or both null schema. if compareAlias argument is set to true, alias
         * is also compared.
         * @param other 
         * @param compareAlias
         * @return true if equal
         */
        public boolean isEqual(Object other, boolean compareAlias) {
            if (other instanceof LogicalFieldSchema) {
                LogicalFieldSchema ofs = (LogicalFieldSchema)other;
                if(compareAlias){
                    if(alias != null && !alias.equals(ofs.alias))
                        return false;
                }
                if (type != ofs.type) return false;
                if (schema == null && ofs.schema == null) return true;
                if (schema == null) return false;
                else return schema.isEqual(ofs.schema, compareAlias);
            } else {
                return false;
            }
        }
        
        
        public String toString(boolean verbose) {
            String uidString = "";
            if (verbose)
                uidString="#" + uid;
            
            String aliasToPrint = "";
            if (alias!=null)
                aliasToPrint = alias;
            
            if( type == DataType.BAG ) {
                if( schema == null ) {
                    return ( aliasToPrint + uidString + ":bag{}" );
                }
                return ( aliasToPrint + uidString + ":bag{" + schema.toString(verbose) + "}" );
            } else if( type == DataType.TUPLE ) {
                if( schema == null ) {
                    return ( aliasToPrint + uidString + ":tuple()" );
                }
                return ( aliasToPrint + uidString + ":tuple(" + schema.toString(verbose) + ")" );
            } else if (type == DataType.MAP) {
                if (schema == null ) {
                    return (aliasToPrint + uidString + ":map");
                } else {
                    return (aliasToPrint + uidString + ":map(" + schema.toString(verbose) + ")");
                }
            }
            return ( aliasToPrint + uidString + ":" + DataType.findTypeName(type) );
        }
        
        public String toString() {
            return toString(true);
        }
        
        /**
         * Give new value for uid if uid of field schema or those in fields schema
         * of inner schema are -1.
         */
        public void stampFieldSchema() {
            if (uid==-1)
                uid = LogicalExpression.getNextUid();
            if (schema!=null) {
                for (LogicalFieldSchema fs : schema.getFields()) {
                    fs.stampFieldSchema();
                }
            }
        }
        
        private boolean compatible(LogicalFieldSchema uidOnlyFieldSchema) {
            if (uidOnlyFieldSchema==null)
                return false;
            if (this.schema==null && uidOnlyFieldSchema.schema!=null ||
                    this.schema!=null && uidOnlyFieldSchema.schema==null)
                return false;
            if (this.schema!=null) {
                if (this.schema.size()!=uidOnlyFieldSchema.schema.size())
                    return false;
                for (int i=0;i<this.schema.size();i++) {
                    boolean comp = schema.getField(i).compatible(uidOnlyFieldSchema.schema.getField(i));
                    if (!comp) return false;
                }
            }
            return true;
        }

        // Check if fs1 is equal to fs2 with regard to type
        public static boolean typeMatch(LogicalFieldSchema fs1, LogicalFieldSchema fs2) {
            if (fs1==null && fs2==null) {
                return true;
            }
            if (fs1==null || fs2==null) {
                return false;
            }
            if (fs1.type!=fs2.type) {
                return false;
            }
            if (DataType.isComplex(fs1.type)) {
                LogicalSchema s1 = fs1.schema;
                LogicalSchema s2 = fs2.schema;
                if (s1==null && s2==null) {
                    return true;
                }
                if (s1==null || s2==null) {
                    return false;
                }
                if (s1.size()!=s2.size()) {
                    return false;
                }
                for (int i=0;i<s1.size();i++) {
                    if (!typeMatch(s1.getField(i), s2.getField(i))) {
                        return false;
                    }
                }
            }
            return true;
        }

        /**
         * Adds the uid from FieldSchema argument to this FieldSchema
         * If the argument is null, it stamps this FieldSchema with uid
         * @param uidOnlyFieldSchema
         * @return FieldSchema 
         * @throws FrontendException
         */
        public LogicalSchema.LogicalFieldSchema mergeUid(LogicalFieldSchema uidOnlyFieldSchema) throws FrontendException {
            if (uidOnlyFieldSchema!=null && compatible(uidOnlyFieldSchema)) {
                this.uid = uidOnlyFieldSchema.uid;
                if (this.schema!=null) {
                    for (int i=0;i<this.schema.size();i++) {
                        schema.getField(i).mergeUid(uidOnlyFieldSchema.schema.getField(i));
                    }
                }
                return uidOnlyFieldSchema;
            }
            else {
                if (uidOnlyFieldSchema==null) {
                    stampFieldSchema();
                }
                else {
                    this.uid = uidOnlyFieldSchema.uid;
                    if (this.schema!=null) {
                        for (int i=0;i<this.schema.size();i++) {
                            schema.getField(i).stampFieldSchema();
                        }
                    }
                }
                LogicalFieldSchema clonedUidOnlyCopy = cloneUid();
                return clonedUidOnlyCopy;
            }
        }
        

        /**
         * Rest uid of this fieldschema and inner schema
         */
        public void resetUid(){
            uid = -1;
            if(schema != null){
                schema.resetUid();
            }
        }
        
        public LogicalFieldSchema cloneUid() {
            LogicalFieldSchema resultFs = null;
            if (schema==null) {
                resultFs = new LogicalFieldSchema(null, null, type, uid);
            }
            else {
                LogicalSchema newSchema = new LogicalSchema();
                resultFs = new LogicalFieldSchema(null, newSchema, type, uid);
                for (int i=0;i<schema.size();i++) {
                    LogicalFieldSchema fs = schema.getField(i).cloneUid();
                    newSchema.addField(fs);
                }
            }
            return resultFs;
        }
        
        public LogicalFieldSchema deepCopy() {
            LogicalFieldSchema newFs = new LogicalFieldSchema(alias!=null?alias:null, schema!=null?schema.deepCopy():null, 
                    type, uid);
            return newFs;
        }
        
        /***
         * Compare two field schema for equality

         * @param relaxInner If true, we don't check inner tuple schemas
         * @param relaxAlias If true, we don't check aliases
         * @return true if FieldSchemas are equal, false otherwise
         */
        public static boolean equals(LogicalFieldSchema fschema,
                                     LogicalFieldSchema fother,
                                     boolean relaxInner,
                                     boolean relaxAlias) {
            if( fschema == null || fother == null ) {
                return false ;
            }

            if( fschema.type != fother.type ) {
                return false ;
            }


            if (!relaxAlias) {
                if ( fschema.alias == null && fother.alias == null ) {
                    // good
                } else if ( fschema.alias == null ) {
                    return false ;
                } else if( !fschema.alias.equals( fother.alias ) ) {
                    return false ;
                }
            }

            if ( (!relaxInner) && (DataType.isSchemaType(fschema.type))) {
                // Don't do the comparison if both embedded schemas are
                // null.  That will cause Schema.equals to return false,
                // even though we want to view that as true.
                if (!(fschema.schema == null && fother.schema == null)) {
                    // compare recursively using schema
                    if (!LogicalSchema.equals(fschema.schema, fother.schema, false, relaxAlias)) {
                        return false ;
                    }
                }
            }

            return true ;
        }

        /**
         * Check if FieldSchema inFs is castable to outFs
         * @param inFs
         * @param outFs
         * @return true if it is castable
         */
        public static boolean castable(LogicalFieldSchema inFs,
                LogicalFieldSchema outFs) {
            
            if(outFs == null && inFs == null) {
                return false;
            }
            
            if (outFs == null) {
                return false ;
            }
    
            if (inFs == null) {
                return false ;
            }
            byte inType = inFs.type;
            byte outType = outFs.type;
    
            if (DataType.isSchemaType(outFs.type)) {
                if(inType == DataType.BYTEARRAY) {
                    // good
                } else if (inType == outType) {
                    // Don't do the comparison if either input inner schema 
                    // is null/empty or  both inner schemas are
                    // null.  That will cause Schema.equals to return false,
                    // even though we want to view that as true.
                    if (!(inFs.schema == null || inFs.schema.size() == 0 || 
                            (outFs.schema == null && inFs.schema == null))) { 
                        // compare recursively using schema
                        if (!LogicalSchema.castable(inFs.schema, outFs.schema)) {
                            return false ;
                        }
                    }
                } else {
                    return false;
                }
            } else {
                if (inType == outType) {
                    // good
                }
                else if (inType == DataType.BOOLEAN && (outType == DataType.CHARARRAY
                        || outType == DataType.BYTEARRAY || DataType.isNumberType(outType))) {
                    // good
                }
                else if (DataType.isNumberType(inType) && (outType == DataType.CHARARRAY
                        || outType == DataType.BYTEARRAY || DataType.isNumberType(outType))
                        || outType == DataType.BOOLEAN) {
                    // good
                }
                else if (inType == DataType.CHARARRAY && (outType == DataType.BYTEARRAY
                        || DataType.isNumberType(outType)) || outType == DataType.BOOLEAN) {
                    // good
                }
                else if (inType == DataType.BYTEARRAY) {
                    // good
                }
                else {
                    return false;
                }
            }
    
            return true ;
        }
        
        /***
         * Merge two LogicalFieldSchema, the behavior of merge depends on mode. 
         * If mode==MergeType.LoadForEach or MergeType.LoadForEachInner, take left side if compatible, otherwise, throw exception.
         * If mode==MergeType.UnionInner, if not same type, throw exception, end up with null inner schema.
         * If mode==MergeType.Union, take more specific type.
         * @param fs1 In Load/Foreach, fs1 is user declared schema; in Union, fs1 is left side
         * @param fs2 In Load/Foreach, fs1 is inferred schema; in Union, fs1 is left side
         * @param mode merge mode
         */
        public static LogicalFieldSchema merge(LogicalFieldSchema fs1, LogicalFieldSchema fs2, MergeMode mode) throws FrontendException {
            // deal with null schema
            if (mode==MergeMode.LoadForEach) {
                if (fs1==null) throw new FrontendException("We cannot cast into null", 1031);
                if (fs2==null) return fs1.deepCopy();
            } else if (mode==MergeMode.LoadForEachInner) {
                if (fs1==null)
                    return null;
                if (fs2==null)
                    return fs1.deepCopy();
            } else { // Union/UnionInner
                if(fs1==null||fs2==null)
                    return null;
            }
            
            String mergedAlias;
            byte mergedType = DataType.UNKNOWN;
            LogicalSchema mergedSubSchema = null;
            
            // Infer merged data type
            if (mode==MergeMode.UnionInner) {
                if (fs1.type!=fs2.type)
                    // We don't merge inner schema of different type for union, throw exception
                    throw new FrontendException("Incompatible field schema: left is \"" + fs1.toString(false) + "\", right is \"" + fs2.toString(false) + "\"", 1031);
                else
                    mergedType = fs1.type;
            }
            else if (mode==MergeMode.LoadForEach||mode==MergeMode.LoadForEachInner) {
                if (fs1.type==DataType.NULL||fs1.type==DataType.BYTEARRAY)  // If declared schema does not have type part
                    mergedType = fs2.type;
                else if (!DataType.castable(fs1.type, fs2.type))
                    throw new FrontendException("Incompatible field schema: declared is \"" + fs1.toString(false) + "\", infered is \"" + fs2.toString(false) + "\"", 1031);
                else mergedType = fs1.type; // If compatible type, we take the declared type
            }
            else {
                // Union schema
                if (fs1.type==DataType.BYTEARRAY) {
                    mergedType=fs2.type;
                } else if (fs2.type==DataType.BYTEARRAY) {
                    mergedType = fs1.type;
                }
                else {
                    // Take the more specific type
                    mergedType = DataType.mergeType(fs1.type, fs2.type);
                    if (mergedType == DataType.ERROR) {
                        // True incompatible, set to bytearray
                        mergedType = DataType.BYTEARRAY;
                    }
                }
            }
            
            if (fs1.alias==null)
                mergedAlias = fs2.alias;
            else if (fs2.alias==null)
                mergedAlias = fs1.alias;
            else {
                mergedAlias = mergeNameSpacedAlias(fs1.alias, fs2.alias);
                if (mergedAlias==null)
                    mergedAlias = fs1.alias;
            }

            if (DataType.isSchemaType(mergedType)) {
                if (mode==MergeMode.Union) {
                    try {
                        if (fs1.type==DataType.BYTEARRAY) {
                            if (fs2.schema!=null)
                                mergedSubSchema = fs2.schema.deepCopy();
                        }
                        else if (fs2.type==DataType.BYTEARRAY) {
                            if (fs1.schema!=null)
                                mergedSubSchema = fs1.schema.deepCopy();
                        }
                        else {
                            mergedSubSchema = LogicalSchema.merge(fs1.schema, fs2.schema, MergeMode.UnionInner);
                        }
                    } catch (FrontendException e) {
                        if(fs1.type == DataType.BAG && fs2.type == DataType.BAG){
                            //create an empty tuple as subschema
                            mergedSubSchema = new LogicalSchema();
                            mergedSubSchema.addField(new LogicalFieldSchema(null, new LogicalSchema(), DataType.TUPLE));
                        }else if(fs1.type == DataType.TUPLE && fs2.type == DataType.TUPLE){
                            mergedSubSchema = new LogicalSchema();
                        }
                        // If inner schema is not compatible, mergedSubSchema set to null
                    }
                }
                else {
                    if (mode==MergeMode.UnionInner)
                        mergedSubSchema = LogicalSchema.merge(fs1.schema, fs2.schema, MergeMode.UnionInner);
                    else {
                        // LoadForEach/LoadForEachInner
                        if (fs1.type==DataType.BYTEARRAY)
                            mergedSubSchema = fs2.schema;
                        else {
                            try {
                                // Only check compatibility
                                mergedSubSchema = LogicalSchema.merge(fs1.schema, fs2.schema, MergeMode.LoadForEachInner);
                            } catch (FrontendException e) {
                                throw new FrontendException("Incompatible field schema: left is \"" + fs1.toString(false) + "\", right is \"" + fs2.toString(false) + "\"", 1031);
                            }
                        }
                    }
                }
            }
            LogicalFieldSchema mergedFS = new LogicalFieldSchema(mergedAlias, mergedSubSchema, mergedType);
            return mergedFS;
        }

        public static boolean isEqualUnlessUnknown(LogicalFieldSchema fs1, LogicalFieldSchema fs2) throws FrontendException {
            if (fs1.type == DataType.BYTEARRAY) {
                return true;
            } else if (fs2.type == DataType.BYTEARRAY) {
                return true;
            } else if (fs1.type == fs2.type) {
                if (DataType.isComplex(fs1.type)) {
                    return LogicalSchema.isEqualUnlessUnknown(fs1.schema, fs2.schema);
                } else {
                    return true;
                }
            } else {
                return false;
            }
        }

        /***
         * Old Pig field schema does not require a tuple schema inside a bag;
         * Now it is required to have that; this method is to fill the gap
         */
        public void normalize() {
            if (type==DataType.BAG) {
                if (schema!=null) {
                    // Check if the BAG has a tuple field 
                    if (schema.size()!=1 || schema.getField(0).type!=DataType.TUPLE) {
                        LogicalSchema tupleSchema = new LogicalSchema();
                        for (LogicalFieldSchema innerFs : schema.getFields()) {
                            tupleSchema.addField(innerFs);
                        }
                        schema = new LogicalSchema();
                        schema.addField(new LogicalFieldSchema(null, tupleSchema, DataType.TUPLE));
                    }
                }
            }
            if (schema!=null) {
                for (LogicalFieldSchema fs : schema.getFields()) {
                    fs.normalize();
                }
            }
        }
    }
    
    private List<LogicalFieldSchema> fields;
    
    public LogicalSchema() {
        fields = new ArrayList<LogicalFieldSchema>();
    }


    /**
     * Reset uids of all fieldschema that the schema contains
     */
    public void resetUid() {
        for(LogicalFieldSchema fs : fields){
            fs.resetUid();
        }
    }

    /**
     * Recursively compare two schemas to check if the input schema 
     * can be cast to the cast schema
     * @param inSch schema of the cast input
     * @param outSch schema of the cast operator
     * @return true if castable
     */
    public static boolean castable(LogicalSchema inSch, LogicalSchema outSch) {
        // If both of them are null, they are castable
        if ((outSch == null) && (inSch == null)) {
            return false ;
        }

        // otherwise
        if (outSch == null) {
            return false ;
        }

        // Cast to a more specific type is good
        if (inSch == null) {
            return true ;
        }

        if (outSch.size() > inSch.size()) return false;

        
        Iterator<LogicalFieldSchema> i = outSch.fields.iterator();
        Iterator<LogicalFieldSchema> j = inSch.fields.iterator();

        while (i.hasNext()) {
        //iterate only for the number of fields in cast

            LogicalFieldSchema outFs = i.next() ;
            LogicalFieldSchema inFs = j.next() ;

            // Compare recursively using field schema
            if (!LogicalFieldSchema.castable(inFs, outFs)) {
                return false ;
            }

        }
        return true;

    }

    /**
     * Add a field to this schema.
     * @param field to be added to the schema
     */
    public void addField(LogicalFieldSchema field) {
        fields.add(field);
    }
    
    /**
     * Fetch a field by alias
     * @param alias
     * @return field associated with alias, or null if no such field
     * @throws FrontendException 
     */
    public LogicalFieldSchema getField(String alias) throws FrontendException {
        LogicalFieldSchema result = null;
        //first look for an exact match
        for (LogicalFieldSchema fs : fields) {
            if (fs.alias!=null && fs.alias.equals(alias) ) {
                if (result==null) {
                    result = fs;
                }
                else {
                    StringBuilder sb = new StringBuilder("Found more than one match: " + result.alias + ", " + fs.alias);
                    throw new FrontendException(sb.toString(), 1025);
                }
            }
        }
        if(result != null){
            return result;
        }
        //if no exact match is found, look for matches for scoped aliases
        for (LogicalFieldSchema fs : fields) {
            if (fs.alias!=null && fs.alias.matches(".*::"+alias+"$") ) {
                if (result==null) {
                    result = fs;
                }
                else {
                    StringBuilder sb = new StringBuilder("Found more than one match: " + result.alias + ", " + fs.alias);
                    throw new FrontendException(sb.toString(), 1025);
                }
            }
        }
        
        return result;
    }
    
    /**
     * Given an alias name, find the associated LogicalFieldSchema. If exact name is 
     * not found see if any field matches the part of the 'namespaced' alias.
     * eg. if given alias is nm::a , and schema is (a,b). It will return 
     * FieldSchema of a.
     * if given alias is nm::a and schema is (nm2::a, b), it will return null
     * @param alias Alias to look up.
     * @return LogicalFieldSchema, or null if no such alias is in this tuple.
     */
    public LogicalFieldSchema getFieldSubNameMatch(String alias) throws FrontendException {
        if(alias == null)
            return null;
        LogicalFieldSchema fs = getField(alias);
        if(fs != null){
            return fs;
        }
        //fs is null
        final String sep = "::";
        ArrayList<LogicalFieldSchema> matchedFieldSchemas = new ArrayList<LogicalFieldSchema>();
        if(alias.contains(sep)){
            for(LogicalFieldSchema field : fields) {
                if(alias.endsWith(sep + field.alias)){
                    matchedFieldSchemas.add(field);
                }
            }
        }
        if(matchedFieldSchemas.size() > 1){
            boolean hasNext = false;
            StringBuilder sb = new StringBuilder("Found more than one " +
            "sub alias name match: ");
            for (LogicalFieldSchema matchFs : matchedFieldSchemas) {
                if(hasNext) {
                    sb.append(", ");
                } else {
                    hasNext = true;
                }
                sb.append(matchFs.alias);
            }
            int errCode = 1116;
            throw new FrontendException(sb.toString(), errCode, PigException.INPUT);
        }else if(matchedFieldSchemas.size() == 1){
            fs = matchedFieldSchemas.get(0);
        }

        return fs;
    }
    
    public int getFieldPosition(String alias) {
        LogicalFieldSchema fs = null;
        try {
            fs = getField(alias);
        } catch (FrontendException e) {
        }
        if( fs == null ) {
            return -1;
        }

        return fields.indexOf(fs);
    }

    /**
     * Fetch a field by field number
     * @param fieldNum field number to fetch
     * @return field
     */
    public LogicalFieldSchema getField(int fieldNum) {
        return fields.get(fieldNum);
    }
    
    /**
     * Get all fields
     * @return list of all fields
     */
    public List<LogicalFieldSchema> getFields() {
        return fields;
    }

    /**
     * Get the size of the schema.
     * @return size
     */
    public int size() {
       return fields.size();
    }
    
    /**
     * Two schemas are equal if they are of equal size and their fields
     * schemas considered in order are equal. This function does 
     * not compare the alias of the fields.
     */
    public boolean isEqual(Object other) {
        return isEqual(other, false);
    }
    
    
    /**
     * Two schemas are equal if they are of equal size and their fields
     * schemas considered in order are equal. If compareAlias argument is 
     * set to true, the alias of the fields are also compared.
     * @param other
     * @param compareAlias
     * @return true if equal
     */
    public boolean isEqual(Object other, boolean compareAlias) {
        if (other != null && other instanceof LogicalSchema) {
            LogicalSchema os = (LogicalSchema)other;
            if (size() != os.size()) return false;
            for (int i = 0; i < size(); i++) {
                if (!getField(i).isEqual(os.getField(i), compareAlias)) return false;
            }
            return true;
        } else {
            return false;
        }
    }
    
    /**
     * Look for the index of the field that contains the specified uid
     * @param uid the uid to look for
     * @return the index of the field, -1 if not found
     */
    public int findField(long uid) {            
        
        for(int i=0; i< size(); i++) {
            LogicalFieldSchema f = getField(i);
            // if this field has the same uid, then return this field
            if (f.uid == uid) {
                return i;
            } 
            
            // if this field has a schema, check its schema
            if (f.schema != null) {
                if (f.schema.findField(uid) != -1) {
                    return i;
                }
            }
        }
        
        return -1;
    }
    
    public static enum MergeMode {
        LoadForEach,
        LoadForEachInner,
        Union,
        UnionInner
    }
    
    /**
     * Merge two schemas.
     * @param s1 In Load/ForEach, s1 is user declared schema; In Union, s1 is left side.
     * @param s2 In Load/ForEach, s2 is infered schema; In Union, s2 is right side.
     * @param mode We merge schema in Load/Foreach/Union. In Load/Foreach, we always take s1 if compatible (s1 is set to be user defined schema),
     * In union, we take more specific type (between numeric and string, we take string). In the case type mismatch in s1/s2, 
     * we expect TypeCheckingVisitor will fill the gap later. 
     * @return a merged schema, or null if the merge fails
     */
    public static LogicalSchema merge(LogicalSchema s1, LogicalSchema s2, MergeMode mode) throws FrontendException {
        // If any of the schema is null, take the other party
        if (s1==null || s2==null) {
            if (mode==MergeMode.LoadForEach||mode==MergeMode.LoadForEachInner) {
                if (s1!=null) return s1.deepCopy();
                else if (s2!=null) return s2.deepCopy();
                else return null;
            }
            else // Union/UnionInner, return null
                return null;
        }
        
        
        if (s1.size()!=s2.size()) {
            if (mode==MergeMode.Union) // In union, incompatible type result a null schema
                return null;
            else
                throw new FrontendException("Incompatible schema: left is \"" + s1.toString(false) + "\", right is \"" + s2.toString(false) + "\"", 1031);    
        }
            
        LogicalSchema mergedSchema = new LogicalSchema();
        for (int i=0;i<s1.size();i++) {
            LogicalFieldSchema fs1 = s1.getField(i);
            LogicalFieldSchema fs2 = s2.getField(i);
            
            LogicalFieldSchema mergedFS = LogicalFieldSchema.merge(fs1, fs2, mode);
            mergedSchema.addField(mergedFS);
        }
        return mergedSchema;
    }

    public static boolean isEqualUnlessUnknown(LogicalSchema s1, LogicalSchema s2) throws FrontendException {
        if (s1 == null) {
            return true;
        } else if (s2 == null) {
            return true;
        } else if (s1.size() != s2.size()) {
            return false;
        } else {
            for (int i=0;i<s1.size();i++) {
                if (!LogicalFieldSchema.isEqualUnlessUnknown(s1.getField(i), s1.getField(i))) {
                    return false;
                }
            }
            return true;
        }
    }

    public String toString(boolean verbose) {
        StringBuilder str = new StringBuilder();
        
        for( LogicalFieldSchema field : fields ) {
            str.append( field.toString(verbose) + "," );
        }
        if( fields.size() != 0 ) {
            str.deleteCharAt( str.length() -1 );
        }
        return str.toString();
    }
    
    public String toString() {
        return toString(true);
    }

    public LogicalSchema mergeUid(LogicalSchema uidOnlySchema) throws FrontendException {
        if (uidOnlySchema!=null) {
            if (size()!=uidOnlySchema.size()) {
                throw new FrontendException("Structure of schema change. Original: " + uidOnlySchema + " Now: " + this, 2239);
                }
            for (int i=0;i<size();i++) {
                getField(i).mergeUid(uidOnlySchema.getField(i));
                }
            return uidOnlySchema;
        }
        else {
            LogicalSchema clonedUidOnlyCopy = new LogicalSchema();
            for (int i=0;i<size();i++) {
                getField(i).stampFieldSchema();
                clonedUidOnlyCopy.addField(getField(i).cloneUid());                    
                }
            return clonedUidOnlyCopy;
        }
    }
    
    public LogicalSchema deepCopy()  {
        LogicalSchema newSchema = new LogicalSchema();
        for (int i=0;i<size();i++)
            newSchema.addField(getField(i).deepCopy());
        return newSchema;
    }

    /**
     * Merges collection of schemas using their column aliases 
     * (unlike mergeSchema(..) functions which merge using positions)
     * Schema will not be merged if types are incompatible, 
     * as per DataType.mergeType(..)
     * For Tuples and Bags, SubSchemas have to be equal be considered compatible
     * @param schemas - list of schemas to be merged using their column alias
     * @return merged schema
     */
    public static LogicalSchema mergeSchemasByAlias(List<LogicalSchema> schemas)
    throws FrontendException{
        LogicalSchema mergedSchema = null;

        // list of schemas that have currently been merged, used in error message
        ArrayList<LogicalSchema> mergedSchemas = new ArrayList<LogicalSchema>(schemas.size());
        for(LogicalSchema sch : schemas){
            if(mergedSchema == null){
                mergedSchema = sch.deepCopy();
                mergedSchemas.add(sch);
                continue;
            }
            try{
                mergedSchema = mergeSchemaByAlias( mergedSchema, sch );
                mergedSchemas.add(sch);
            }catch(FrontendException e){
                String msg = "Error merging schema: ("  + sch + ") with " 
                    + "merged schema: (" + mergedSchema + ")" + " of schemas : "
                    + mergedSchemas;
                throw new FrontendException(msg, e);
            }
        }
        return mergedSchema;
    }
    
    /**
     * Merges two schemas using their column aliases 
     * (unlike mergeSchema(..) functions which merge using positions)
     * Schema will not be merged if types are incompatible, 
     * as per DataType.mergeType(..)
     * For Tuples and Bags, SubSchemas have to be equal be considered compatible
     */
    public static LogicalSchema mergeSchemaByAlias(LogicalSchema schema1, LogicalSchema schema2)
    throws FrontendException{
        LogicalSchema mergedSchema = new LogicalSchema();
        HashSet<LogicalFieldSchema> schema2colsAdded = new HashSet<LogicalFieldSchema>();
        // add/merge fields present in first schema 
        for(LogicalFieldSchema fs1 : schema1.getFields()){
            checkNullAlias(fs1, schema1);
            LogicalFieldSchema fs2 = schema2.getFieldSubNameMatch( fs1.alias );
            if(fs2 != null){
                if(schema2colsAdded.contains(fs2)){
                    // alias corresponds to multiple fields in schema1,
                    // just do a lookup on
                    // schema1 , that will throw the appropriate error.
                    schema1.getFieldSubNameMatch( fs2.alias );
                }
                schema2colsAdded.add(fs2);
                LogicalFieldSchema mergedFs = LogicalFieldSchema.merge(fs1,fs2, MergeMode.Union);
                mergedFs.alias = mergeNameSpacedAlias(fs1.alias, fs2.alias);
                if (mergedFs.alias==null)
                    mergedFs.alias = fs1.alias;
                mergedSchema.addField(mergedFs);
            }
            else
                mergedSchema.addField(new LogicalFieldSchema(fs1));
        }

        //add schemas from 2nd schema, that are not already present in
        // merged schema
        for(LogicalFieldSchema fs2 : schema2.getFields()){
            checkNullAlias(fs2, schema2);
            if(! schema2colsAdded.contains(fs2)){
                mergedSchema.addField( new LogicalFieldSchema( fs2 ) );
            }
        }
        return mergedSchema;
    }

    private static void checkNullAlias(LogicalFieldSchema fs, LogicalSchema schema)
    throws FrontendException {
        if(fs.alias == null){
            throw new FrontendException(
                    "Schema having field with null alias cannot be merged " +
                    "using alias. Schema :" + schema
            );
        }
    }

    /**
     * If one of the aliases is of form 'nm::str1', and other is of the form
     * 'str1', this returns str1
     */
    private static String mergeNameSpacedAlias(String alias1, String alias2)
    throws FrontendException {
        if(alias1.equals(alias2)){
            return alias1;
        }
        if(alias1.endsWith("::" + alias2)){
            return alias2;
        }
        if(alias2.endsWith("::" + alias1)){
            return alias1;
        }
        //the aliases are different, alias cannot be merged
        return null;
    }

    /**
     * Recursively compare two schemas for equality
     * @param schema
     * @param other
     * @param relaxInner if true, inner schemas will not be checked
     * @param relaxAlias if true, aliases will not be checked
     * @return true if schemas are equal, false otherwise
     */
    public static boolean equals(LogicalSchema schema,
                                 LogicalSchema other,
                                 boolean relaxInner,
                                 boolean relaxAlias) {
        // If both of them are null, they are equal
        if ((schema == null) && (other == null)) {
            return true ;
        }

        // otherwise
        if (schema == null || other == null ) {
            return false ;
        }

        if (schema.size() != other.size()) return false;

        Iterator<LogicalFieldSchema> i = schema.fields.iterator();
        Iterator<LogicalFieldSchema> j = other.fields.iterator();

        while (i.hasNext()) {
            LogicalFieldSchema myFs = i.next() ;
            LogicalFieldSchema otherFs = j.next() ;

            if (!relaxAlias) {
                if( myFs.alias == null && otherFs.alias == null ) {
                    // good
                } else if( myFs.alias == null ) {
                    return false ;
                } else if( !myFs.alias.equals(otherFs.alias) ) {
                    return false ;
                }
            }

            if (myFs.type != otherFs.type) {
                return false ;
            }

            if (!relaxInner && !LogicalFieldSchema.equals( myFs, otherFs, false, relaxAlias ) ) {
                // Compare recursively using field schema
                return false ;
            }
        }
        return true;
    }
    /***
     * Old Pig schema does not require a tuple schema inside a bag;
     * Now it is required to have that; this method is to fill the gap
     */
    public void normalize() {
        for (LogicalFieldSchema fs : getFields()) {
            fs.normalize();
        }
    }
}