/*
 * JBoss, Home of Professional Open Source.
 * See the COPYRIGHT.txt file distributed with this work for information
 * regarding copyright ownership.  Some portions may be licensed
 * to Red Hat, Inc. under one or more contributor license agreements.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 * Modified by
 * @author  Data Virtuality GmbH
 * @date    $Date$
 *  
 */
// https://issues.jboss.org/browse/TEIID-2231
// https://issues.jboss.org/browse/TEIID-2235
// https://issues.jboss.org/browse/TEIID-247
// https://issues.jboss.org/browse/TEIID-2233
// https://issues.jboss.org/browse/TEIID-2244
// https://issues.jboss.org/browse/TEIID-2183
// https://issues.jboss.org/browse/TEIID-2111
// https://issues.jboss.org/browse/TEIID-2368
// https://issues.jboss.org/browse/TEIID-2181 (only the MetadataFactory was adapted from the issue)
// https://issues.jboss.org/browse/TEIID-4913 (without renameColumn method which will be introduced with TEIID-3349 issue and new ALTER COLUMN command)
// https://issues.jboss.org/browse/TEIID-5046
// https://issues.jboss.org/browse/TEIID-5106
// https://issues.jboss.org/browse/TEIID-2159
// https://issues.jboss.org/browse/TEIID-2904
// https://issues.jboss.org/browse/TEIID-2572
// https://issues.jboss.org/browse/TEIID-3520
// https://issues.jboss.org/browse/TEIID-5183
// https://issues.jboss.org/browse/TEIID-2974
// https://issues.jboss.org/browse/TEIID-2471
// https://issues.redhat.com/browse/TEIID-2195
// https://issues.jboss.org/browse/TEIID-2181
// https://issues.redhat.com/browse/TEIID-2384
// https://issues.redhat.com/browse/TEIID-3901
// https://issues.redhat.com/browse/TEIID-3044 (partially, only this class)
// https://issues.redhat.com/browse/TEIID-4274
package org.teiid.metadata;

import com.datavirtuality.dv.core.interfaces.DVCaller;
import org.teiid.CommandContext;
import org.teiid.UserDefinedAggregate;
import org.teiid.adminapi.Model;
import org.teiid.adminapi.impl.DataPolicyMetadata.PermissionMetaData;
import org.teiid.adminapi.impl.ModelMetaData;
import org.teiid.connector.DataPlugin;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.types.DataTypeManager;
import org.teiid.core.util.Assertion;
import org.teiid.core.util.PropertiesUtils;
import org.teiid.core.util.StringUtil;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.metadata.BaseColumn.NullType;
import org.teiid.metadata.FunctionMethod.PushDown;
import org.teiid.metadata.MetadataStore.Grant;
import org.teiid.metadata.ProcedureParameter.Type;
import org.teiid.translator.TranslatorException;
import org.teiid.translator.TypeFacility;

import java.io.Reader;
import java.io.Serializable;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.*;


/**
 * Allows connectors to build metadata for use by the engine.
 * 
 * TODO: add support for datatype import
 * TODO: add support for unique constraints
 */
public class MetadataFactory implements Serializable, Cloneable {

	private static final String TEIID_RESERVED = "teiid_"; //$NON-NLS-1$
	private static final String TEIID_SF = "teiid_sf"; //$NON-NLS-1$
	private static final String TEIID_RELATIONAL = "teiid_rel"; //$NON-NLS-1$
	private static final String TEIID_LDAP = "teiid_ldap"; //$NON-NLS-1$
	private static final String TEIID_ODATA = "teiid_odata"; //$NON-NLS-1$
	private static final String TEIID_EXCEL = "teiid_excel"; //$NON-NLS-1$
	private static final String TEIID_JPA = "teiid_jpa"; //$NON-NLS-1$
	private static final String TEIID_REST = "teiid_rest"; //$NON-NLS-1$
	private static final String TEIID_SPATIAL = "teiid_spatial"; //$NON-NLS-1$

    public static final String LAZY_METADATA = "importer.lazyMetadataLoading"; //$NON-NLS-1$

	private static final long serialVersionUID = 8590341087771685630L;
	
	private String vdbName;
	private int vdbVersion;
	private Map<String, Datatype> dataTypes;
	private Map<String, Datatype> builtinDataTypes;
	private boolean autoCorrectColumnNames = true;
	private boolean renameDuplicateColumns = true;
	private boolean renameDuplicateTables = true;
	private Map<String, String> namespaces;
	private String rawMetadata;
	private Properties modelProperties;
	private Map<String, String> modelPropertiesRaw = new HashMap<>();
	private Schema schema = new Schema();
	private String idPrefix;
	// we should use static count to guarantee uniqueness of uuid of different metadata but with the same type and name
	// + with newly created instance of MetadataFactory class
	private static long count;
	private transient Parser parser;
	private transient ModelMetaData model;
	private List<Grant> grants;

	public static final String SF_URI = "{http://www.teiid.org/translator/salesforce/2012}"; //$NON-NLS-1$
	public static final String LDAP_URI = "{http://www.teiid.org/translator/ldap/2015}"; //$NON-NLS-1$
	public static final String ODATA_URI = "{http://www.jboss.org/teiiddesigner/ext/odata/2012}"; //$NON-NLS-1$
	public static final String EXCEL_URI = "{http://www.teiid.org/translator/excel/2014}"; //$NON-NLS-1$
	public static final String JPA_URI = "{http://www.teiid.org/translator/jpa/2014}"; //$NON-NLS-1$
	public static final String REST_URI = "{http://teiid.org/rest}"; //$NON-NLS-1$
	public static final String SPATIAL_URI = "{http://www.teiid.org/translator/spatial/2015}"; //$NON-NLS-1$
	public static final String ODATA_PREFIX = ODATA_URI;//TEIID_ODATA+":"; //$NON-NLS-1$
	
	public static final Map<String, String> BUILTIN_NAMESPACES;
	static {
		Map<String, String> map = new TreeMap<String, String>(StringUtil.CASE_INSENSITIVE_ORDER);
		map.put(TEIID_RELATIONAL, AbstractMetadataRecord.RELATIONAL_URI.substring(1, AbstractMetadataRecord.RELATIONAL_URI.length()-1));
		map.put(TEIID_SF, SF_URI.substring(1, SF_URI.length()-1));
		map.put(TEIID_LDAP, LDAP_URI.substring(1, LDAP_URI.length()-1));
		map.put(TEIID_ODATA, ODATA_URI.substring(1, ODATA_URI.length()-1));
		map.put(TEIID_EXCEL, EXCEL_URI.substring(1, EXCEL_URI.length()-1));
		map.put(TEIID_JPA, JPA_URI.substring(1, JPA_URI.length()-1));
		map.put(TEIID_REST, REST_URI.substring(1, REST_URI.length()-1));
		map.put(TEIID_SPATIAL, SPATIAL_URI.substring(1, SPATIAL_URI.length()-1));
		BUILTIN_NAMESPACES = Collections.unmodifiableMap(map);
	}
	
	public MetadataFactory(String vdbName, int vdbVersion, Map<String, Datatype> runtimeTypes, ModelMetaData model) {
		this(vdbName, vdbVersion, model.getName(), runtimeTypes, model.getProperties(), model.getSchemaText());
		this.model = model;
		this.modelPropertiesRaw = (Map<String, String>) DVCaller.call(261, model.getModelPropertiesRaw());
	}

	public MetadataFactory(String vdbName, int vdbVersion, String schemaName, Map<String, Datatype> runtimeTypes, Properties modelProperties, String rawMetadata) {
		this.vdbName = vdbName;
		this.vdbVersion = vdbVersion;
		this.dataTypes = runtimeTypes;
		this.builtinDataTypes = runtimeTypes;
		schema.setName(schemaName);
		long msb = longHash(vdbName, 0);
		msb = 31*msb + vdbVersion;
		msb = longHash(schemaName, msb);
		idPrefix = "tid:" + hex(msb, 12); //$NON-NLS-1$
		setUUID(schema);
		if (modelProperties != null) {
			this.autoCorrectColumnNames = PropertiesUtils.getBooleanProperty(modelProperties, "importer.autoCorrectColumnNames", this.autoCorrectColumnNames); //$NON-NLS-1$
			this.renameDuplicateColumns = PropertiesUtils.getBooleanProperty(modelProperties, "importer.renameDuplicateColumns", this.renameDuplicateColumns); //$NON-NLS-1$
			this.renameDuplicateTables = PropertiesUtils.getBooleanProperty(modelProperties, "importer.renameDuplicateTables", this.renameDuplicateTables); //$NON-NLS-1$
		}
		this.modelProperties = modelProperties;
		this.rawMetadata = rawMetadata;
	}

	private long longHash(String s, long h) {
		if (s == null) {
			return h;
		}
		for (int i = 0; i < s.length(); i++) {
            h = 31*h + s.charAt(i);
        }
		return h;
	}
	
	public static String hex(long val, int hexLength) {
		long hi = 1L << (Math.min(63, hexLength * 4));
		return Long.toHexString(hi | (val & (hi - 1))).substring(1);
    }

	/**
	 * @deprecated
	 * @return
	 * @see #getModelProperties()
	 */
	public Properties getImportProperties() {
		return modelProperties;
	}
	
	public Properties getModelProperties() {
		return modelProperties;
	}
	
	public String getRawMetadata() {
		return this.rawMetadata;
	}
	
	public Model getModel() {
		return model;
	}
	
	protected void setUUID(AbstractMetadataRecord record) {
		int lsb = 0;
		if (record.getParent() != null) {
			lsb  = record.getParent().getUUID().hashCode();
		}
		lsb = 31*lsb + record.getName().hashCode();
		String uuid = idPrefix+"-"+hex(lsb, 8) + "-" + hex(count++, 8); //$NON-NLS-1$ //$NON-NLS-2$
		record.setUUID(uuid);
	}
	
	public String getName() {
		return this.schema.getName();
	}
	
	public Schema getSchema() {
		return this.schema;
	}

	/**
	 * Add a table with the given name to the model.  
	 * @param name
	 * @return
	 * @throws MetadataException 
	 */
	public Table addTable(String name) {
		Table table = new Table();
		table.setTableType(Table.Type.Table);
        if (renameDuplicateTables && this.schema.getTable(name) != null) {
           int suffix = 1;
           String newName = name + "_" + suffix; //$NON-NLS-1$
           while (this.schema.getTable(newName) != null) {
        suffix++;
		newName = name + "_" + suffix; //$NON-NLS-1$
   }
  LogManager.logInfo(LogConstants.CTX_CONNECTOR, DataPlugin.Util.gs(DataPlugin.Event.TEIID60040, name, newName));
  name = newName;
    }
		table.setName(name);
		setUUID(table);
		schema.addTable(table);
		// BUG: Migrate to the Teiid 8.1 release
		// ISSUE: DV-34
		// SOLUTION: MetadataValidator uses this attachment to actually add a table
		// FIXED BY: Roman Pedchenko
		// DATE: 2012/10/05
		table.addAttchment(MetadataFactory.class, this);
		// BUG END
		return table;
	}
	
	/**
	 * Adds a column to the table with the given name and type.
	 * @param name
	 * @param type should be one of {@link TypeFacility.RUNTIME_NAMES}
	 * @param table
	 * @return
	 * @throws MetadataException
	 */
	public Column addColumn(String name, String type, ColumnSet<?> table) {
	// BUG: DVCORE-122
	// ISSUE: some column parameters are not correctly set for views defined in the VDB 
	// SOLUTION: if the projected symbol is an ElementSymbol, copy properties values from the corresponding Column object
	// FIXED by Roman Pedchenko
	// date 02.11.2012
		int arrayDimensions = 0;
		while (DataTypeManager.isArrayType(type)) {
			arrayDimensions++;
			type = type.substring(0, type.length()-2);
		}
		return addColumn(name, setDataType(type, null, this.dataTypes, false), arrayDimensions, table);
	}

	public Column addColumn(String name, Datatype type, int arrayDimensions, ColumnSet<?> table) {
	// BUG END
		if (autoCorrectColumnNames) {
			String newName = name.replace(AbstractMetadataRecord.NAME_DELIM_CHAR, '_');
			if (!newName.equals(name)) {
				LogManager.logInfo(LogConstants.CTX_CONNECTOR, DataPlugin.Util.gs(DataPlugin.Event.TEIID60038, name, newName));
			}
			name = newName;
		} else if (name.indexOf(AbstractMetadataRecord.NAME_DELIM_CHAR) != -1) {
			//TODO: for now this is not used
			throw new MetadataException(DataPlugin.Event.TEIID60008, DataPlugin.Util.gs(DataPlugin.Event.TEIID60008, name));
		}
		if (table.getColumnByName(name) != null) {
			if (!renameDuplicateColumns) {
				throw new DuplicateRecordException(DataPlugin.Event.TEIID60016, DataPlugin.Util.gs(DataPlugin.Event.TEIID60016, table.getFullName() + AbstractMetadataRecord.NAME_DELIM_CHAR + name));
			}
			int suffix = 1;
			String newName = name + "_" + suffix; //$NON-NLS-1$
			while (table.getColumnByName(newName) != null) {
				suffix++;
				newName = name + "_" + suffix; //$NON-NLS-1$
			}
			LogManager.logInfo(LogConstants.CTX_CONNECTOR, DataPlugin.Util.gs(DataPlugin.Event.TEIID60039, name, newName));
			name = newName;
		}
		Column column = new Column();
		column.setName(name);
		column.setParent(table);
		// BUG: DVCORE-122
		// ISSUE: some column parameters are not correctly set for views defined in the VDB
		// SOLUTION: if the projected symbol is an ElementSymbol, copy properties values from the corresponding Column object
		// FIXED by Roman Pedchenko
		// date 02.11.2012
		column.setDatatype(type, true, arrayDimensions);
		// BUG END
		table.addColumn(column);
		column.setPosition(table.getColumns().size()); //1 based indexing
		setUUID(column);
		return column;
	}

	public static Datatype setDataType(String type, BaseColumn column, Map<String, Datatype> dataTypes, boolean allowNull) {
		int arrayDimensions = 0;
		while (DataTypeManager.isArrayType(type)) {
			arrayDimensions++;
			type = type.substring(0, type.length()-2);
		}
		Datatype datatype = dataTypes.get(type);
		if (datatype == null && (!allowNull || !DataTypeManager.DefaultDataTypes.NULL.equals(type))) {
			 throw new MetadataException(DataPlugin.Event.TEIID60009, DataPlugin.Util.gs(DataPlugin.Event.TEIID60009, type));
		}
		// BUG: DVCORE-122
		// ISSUE: some column parameters are not correctly set for views defined in the VDB 
		// SOLUTION: if the projected symbol is an ElementSymbol, copy properties values from the corresponding Column object
		// FIXED by Roman Pedchenko
		// date 02.11.2012
//		column.setDatatype(datatype, true);
		if (column != null)
			column.setDatatype(datatype, true, arrayDimensions);
		// BUG END
		return datatype;
	}
	
	/**
	 * Adds a primary key to the given table.  The column names should be in key order.
	 * @param name
	 * @param columnNames
	 * @param table
	 * @return
	 * @throws MetadataException
	 */
	public KeyRecord addPrimaryKey(String name, List<String> columnNames, Table table) {
		KeyRecord primaryKey = new KeyRecord(KeyRecord.Type.Primary);
		primaryKey.setParent(table);
		primaryKey.setupColumns(columnNames.size());
		primaryKey.setName(name);
		setUUID(primaryKey);
		assignColumns(columnNames, table, primaryKey);
		table.setPrimaryKey(primaryKey);
		return primaryKey;
	}
	
	/**
	 * Adds an access pattern to the given table.
	 * @param name
	 * @param columnNames
	 * @param table
	 * @return
	 * @throws MetadataException
	 */
	public KeyRecord addAccessPattern(String name, List<String> columnNames, Table table) {
		KeyRecord ap = new KeyRecord(KeyRecord.Type.AccessPattern);
		ap.setParent(table);
		ap.setupColumns(columnNames.size());
		ap.setName(name);
		setUUID(ap);
		assignColumns(columnNames, table, ap);
		table.getAccessPatterns().add(ap);
		return ap;
	}	
	
	/**
	 * Adds an index to the given table.
	 * @param name
	 * @param nonUnique true indicates that an index is being added.
	 * @param columnNames
	 * @param table
	 * @return
	 * @throws MetadataException
	 */
	public KeyRecord addIndex(String name, boolean nonUnique, List<String> columnNames, Table table) {
		KeyRecord index = new KeyRecord(nonUnique?KeyRecord.Type.Index:KeyRecord.Type.Unique);
		index.setParent(table);
		index.setupColumns(columnNames.size());
		index.setName(name);
		assignColumns(columnNames, table, index);
		setUUID(index);
		if (nonUnique) {
			table.getIndexes().add(index);
		}
		else {
			table.getUniqueKeys().add(index);
		}
		return index;
	}

	/**
	 * Adds a function based index on the given expressions.
	 * @param name
	 * @param expressions
	 * @param nonColumnExpressions a Boolean list indicating when expressions are non-column expressions
	 * @param table
	 * @return
	 * @throws TranslatorException
	 */
 	public KeyRecord addFunctionBasedIndex(String name, List<String> expressions, List<Boolean> nonColumnExpressions, Table table) throws TranslatorException {
		KeyRecord index = new KeyRecord(KeyRecord.Type.Index);
		index.setParent(table);
		index.setupColumns(expressions.size());
		index.setName(name);
		setUUID(index);
		boolean functionBased = false;
		for (int i = 0; i < expressions.size(); i++) {
			String expr = expressions.get(i);
			if (nonColumnExpressions.get(i)) {
				Column c = new Column();
				//TODO: we could choose a derived name at this point, but we delay that to get a single unique name across all index expressions
				c.setName(expr);
				c.setNameInSource(expr);
				setUUID(c);
				c.setParent(index);
				c.setPosition(i + 1); //position is temporarily relative to the index, but the validator changes this
				index.getColumns().add(c);
				functionBased = true;
			} else {
				assignColumn(table, index, expr);
			}
		}
		if (!functionBased) {
			table.getIndexes().add(index);
		} else {
			table.getFunctionBasedIndexes().add(index);
		}
		return index;
 	}

	private void assignColumn(Table table, ColumnSet<?> columns, String columnName)
 	                throws TranslatorException {
		Column column = table.getColumnByName(columnName);
		if (column == null) {
			throw new TranslatorException(DataPlugin.Event.TEIID60011, DataPlugin.Util.gs(DataPlugin.Event.TEIID60011, columnName));
		}
		columns.getColumns().add(column);
 	}
	
	/**
	 * Adds a foreign key to the given table.  The referenced primary key must already exist.  The column names should be in key order.
	 * @param name
	 * @param columnNames
	 * @param referenceTable - schema qualified reference table name
	 * @param table
	 * @return
	 * @throws MetadataException
	 */
	public ForeignKey addForiegnKey(String name, List<String> columnNames, String referenceTable, Table table) {
		return addForeignKey(name, columnNames, null, referenceTable, null, table);
	}

	public ForeignKey addForeignKey(String name, List<String> columnNames, String referenceTable, String referenceTableSchema, Table table) {
		return addForeignKey(name, columnNames, null, referenceTable, referenceTableSchema, table);
	}

	/**
	 * Adds a foreign key to the given table.  The referenced key may be automatically created if addUniqueConstraint is true. The column names should be in key order.
	 * if reference table is is another schema, they will be resolved during validation.
	 * @param name
	 * @param columnNames
	 * @param referencedColumnNames, may be null to indicate that the primary key should be used.
	 * @param referenceTable - schema qualified reference table name, can be from another schema
	 * @param table
	 * @return
	 * @throws MetadataException
	 */	
	public ForeignKey addForiegnKey(String name, List<String> columnNames, List<String> referencedColumnNames, String referenceTable, Table table) {
		return addForeignKey(name,  columnNames, referencedColumnNames, referenceTable, null, table);
	}

	/**
	 * Adds a foreign key to the given table.  The referenced key may be automatically created if addUniqueConstraint is true. The column names should be in key order.
	 * if reference table is is another schema, they will be resolved during validation.
	 * @param name name of the foreign key
	 * @param columnNames
	 * @param referencedColumnNames, may be null to indicate that the primary key should be used.
	 * @param referenceTable - schema qualified reference table name, can be from another schema
	 * @param referenceTableSchema - schema name, can be from another schema
	 * @param table current table not reference table
	 * @return
	 * @throws MetadataException
	 */
	public ForeignKey addForeignKey(String name, List<String> columnNames, List<String> referencedColumnNames,
									String referenceTable, String referenceTableSchema, Table table) {
		ForeignKey foreignKey = new ForeignKey();
		foreignKey.setParent(table);
		foreignKey.setupColumns(columnNames.size());
		assignColumns(columnNames, table, foreignKey);
		foreignKey.setReferenceTableName(referenceTable);
		foreignKey.setReferenceTableSchemaName(referenceTableSchema);
		foreignKey.setReferenceColumns(referencedColumnNames);
		foreignKey.setName(name);
		setUUID(foreignKey);
		table.getForeignKeys().add(foreignKey);
		return foreignKey;
	}
	
	/**
	 * Add a procedure with the given name to the model.  
	 * @param name
	 * @return
	 * @throws MetadataException 
	 */
	public Procedure addProcedure(String name) {
		Assertion.isNotNull(name, "name cannot be null"); //$NON-NLS-1$
		Procedure procedure = new Procedure();
		procedure.setName(name);
		setUUID(procedure);
		procedure.setParameters(new LinkedList<ProcedureParameter>());
		schema.addProcedure(procedure);
		return procedure;
	}
	
	/**
	 * Add a procedure parameter.
	 * @param name
	 * @param type should be one of {@link TypeFacility.RUNTIME_NAMES}
	 * @param parameterType should be one of {@link ProcedureParameter.Type}
	 * @param procedure
	 * @return
	 * @throws MetadataException 
	 */
	public ProcedureParameter addProcedureParameter(String name, String type, ProcedureParameter.Type parameterType, Procedure procedure) {
		ProcedureParameter param = new ProcedureParameter();
		param.setName(name);
		setUUID(param);
		param.setType(parameterType);
		param.setProcedure(procedure);
		setDataType(type, param, this.dataTypes, false);
		if (parameterType == Type.ReturnValue) {
			procedure.getParameters().add(0, param);
			for (int i = 0; i < procedure.getParameters().size(); i++) {
				procedure.getParameters().get(i).setPosition(i+1); //1 based indexing
			}
		} else {
			procedure.getParameters().add(param);
			param.setPosition(procedure.getParameters().size()); //1 based indexing
		}
		return param;
	}
	
	/**
	 * Add a procedure resultset column to the given procedure.
	 * @param name
	 * @param type should be one of {@link TypeFacility.RUNTIME_NAMES}
	 * @param procedure
	 * @return
	 * @throws MetadataException 
	 */
	public Column addProcedureResultSetColumn(String name, String type, Procedure procedure) {
		if (procedure.getResultSet() == null) {
			ColumnSet<Procedure> resultSet = new ColumnSet<Procedure>();
			resultSet.setParent(procedure);
			resultSet.setName("RSParam"); //$NON-NLS-1$
			setUUID(resultSet);
			procedure.setResultSet(resultSet);
		}
		return addColumn(name, type, procedure.getResultSet());
	}

	private void assignColumns(List<String> columnNames, Table table,
			ColumnSet<?> columns) {
		for (String columnName : columnNames) {
			Column column = table.getColumnByName(columnName);
			if (column == null) {
				// EXTENSION
				// ISSUE: DVCORE-2122
				// FIXED BY: Roman Pedchenko
				// DATE: 2014.06.17
//				throw new TranslatorException(DataPlugin.Event.TEIID60011, DataPlugin.Util.gs(DataPlugin.Event.TEIID60011, columnName));
				continue;
				// END EXTENSION
			}
			columns.getColumns().add(column);
		}
	}
	
	/**
	 * Add a function with the given name to the model.  
	 * @param name
	 * @return
	 * @throws MetadataException 
	 */
	public FunctionMethod addFunction(String name) {
		Assertion.isNotNull(name, "name cannot be null");
		FunctionMethod function = new FunctionMethod();
		function.setName(name);
		setUUID(function);
		schema.addFunction(function);
		return function;
	}

	/**
	 * Add a function with the given name to the model.
	 * @param name
	 * @return
	 * @throws MetadataException
	 */
 	public FunctionMethod addFunction(String name, String returnType, String... paramTypes) {
		FunctionMethod function = FunctionMethod.createFunctionMethod(name, null, null, returnType, paramTypes);
	    setFunctionMethodTypes(function);
		function.setPushdown(PushDown.MUST_PUSHDOWN);
		setUUID(function);
		schema.addFunction(function);
		return function;
 	}

	private void setFunctionMethodTypes(FunctionMethod function) {
		FunctionParameter outputParameter = function.getReturnParameter();
		if (outputParameter != null) {
			setDataType(outputParameter.getRuntimeType(), outputParameter, builtinDataTypes, outputParameter.getNullType() == NullType.Nullable);
		}
		for (FunctionParameter param : function.getInputParameters()) {
			setDataType(param.getRuntimeType(), param, builtinDataTypes, param.getNullType() == NullType.Nullable);
		}
	}
	
	/**
	 * Adds a non-pushdown function based upon the given {@link Method}.
	 * @param name
	 * @param method
	 * @return
	 * @throws MetadataException
	 */
	public FunctionMethod addFunction(String name, Method method) {
		FunctionMethod func = createFunctionFromMethod(name, method);
		setFunctionMethodTypes(func);
		setUUID(func);
		getSchema().addFunction(func);
		return func;
	}

	public static FunctionMethod createFunctionFromMethod(String name, Method method) {
		Class<?> returnTypeClass = method.getReturnType();
		AggregateAttributes aa = null;
		//handle user defined aggregates
		if ((method.getModifiers() & Modifier.STATIC) == 0 && UserDefinedAggregate.class.isAssignableFrom(method.getDeclaringClass())) {
			aa = new AggregateAttributes();
			Method m;
			try {
				m = method.getDeclaringClass().getMethod("getResult", CommandContext.class); //$NON-NLS-1$
			} catch (NoSuchMethodException e) {
				throw new TeiidRuntimeException(e);
			} catch (SecurityException e) {
				throw new TeiidRuntimeException(e);
			}
			returnTypeClass = m.getReturnType();
		}
		if (returnTypeClass.isPrimitive()) {
			returnTypeClass = TypeFacility.convertPrimitiveToObject(returnTypeClass);
		}
		String returnType = DataTypeManager.getDataTypeName(DataTypeManager.getRuntimeType(returnTypeClass));
		Class<?>[] params = method.getParameterTypes();
		String[] paramTypes = new String[params.length];
		boolean nullOnNull = false;
		for (int i = 0; i < params.length; i++) {
			Class<?> clazz = params[i];
			if (clazz.isPrimitive()) {
				nullOnNull = true;
				if (clazz == Boolean.TYPE) clazz = Boolean.class;
				else if (clazz == Character.TYPE) clazz = Character.class;
				else if (clazz == Byte.TYPE) clazz = Byte.class;
				else if (clazz == Short.TYPE) clazz = Short.class;
				else if (clazz == Integer.TYPE) clazz = Integer.class;
				else if (clazz == Long.TYPE) clazz = Long.class;
				else if (clazz == Float.TYPE) clazz = Float.class;
				else if (clazz == Double.TYPE) clazz = Double.class;
			}
			if (method.isVarArgs() && i == params.length - 1) {
				paramTypes[i] = DataTypeManager.getDataTypeName(DataTypeManager.getRuntimeType(clazz.getComponentType()));
			} else {
				paramTypes[i] = DataTypeManager.getDataTypeName(DataTypeManager.getRuntimeType(clazz));
			}
		}
		if (params.length > 0 && CommandContext.class.isAssignableFrom(params[0])) {
			paramTypes = Arrays.copyOfRange(paramTypes, 1, paramTypes.length);
		}
		FunctionMethod func = FunctionMethod.createFunctionMethod(name, null, null, returnType, paramTypes);
		func.setMethod(method);
		func.setAggregateAttributes(aa);
		func.setInvocationMethod(method.getName());
		func.setPushdown(PushDown.CANNOT_PUSHDOWN);
		func.setInvocationClass(method.getDeclaringClass().getName());
		func.setNullOnNull(nullOnNull);
		if (method.isVarArgs()) {
			func.setVarArgs(method.isVarArgs());
		}
		return func;
	}

	public FunctionParameter addFunctionParameter(String name, String type, FunctionMethod function) {
		FunctionParameter param = functionParameter(name, type, function);
		function.getInputParameters().add(param);
		param.setPosition(function.getInputParameters().size()); //1 based indexing
		
		return param;
	}

	public FunctionParameter addFunctionOutputParameter(String name, String type, FunctionMethod function) {
		FunctionParameter param = functionParameter(name, type, function);
		function.setReturnParameter(param);

		return param;
	}
	
	private FunctionParameter functionParameter(String name, String type, FunctionMethod function) {
		FunctionParameter param = new FunctionParameter();
		param.setName(name);
		setUUID(param);
		setDataType(type, param, this.dataTypes, false);
		param.setFunction(function);

		return param;
	}
	
	/**
	 * Set to false to disable correcting column and other names to be valid Teiid names.
	 * @param autoCorrectColumnNames
	 */
	public void setAutoCorrectColumnNames(boolean autoCorrectColumnNames) {
		this.autoCorrectColumnNames = autoCorrectColumnNames;
	}
	
	public void addNamespace(String prefix, String uri) {
		if (uri == null || uri.indexOf('}') != -1) {
			throw new MetadataException(DataPlugin.Event.TEIID60018, DataPlugin.Util.gs(DataPlugin.Event.TEIID60018, uri));
		}

		if (StringUtil.startsWithIgnoreCase(prefix, TEIID_RESERVED)) {
			String validURI = BUILTIN_NAMESPACES.get(prefix);
			if (validURI == null || !uri.equals(validURI)) {
				throw new MetadataException(DataPlugin.Event.TEIID60017, DataPlugin.Util.gs(DataPlugin.Event.TEIID60017, prefix));
			}
		}

		if (this.namespaces == null) {
			 this.namespaces = new TreeMap<String, String>(StringUtil.CASE_INSENSITIVE_ORDER);
		}
		this.namespaces.put(prefix, uri);
	}
	
	public void mergeInto (MetadataStore store) {
		store.addSchema(this.schema);
		store.addDataTypes(this.builtinDataTypes.values());
		store.addGrants(this.grants);
	}
	
	public MetadataStore asMetadataStore() {
		MetadataStore store = new MetadataStore();
		mergeInto(store);
		return store;
	}
	
	/**
	 * Set the {@link Schema} to a different instance.  This is typically called
	 * in special situations where the {@link MetadataFactory} logic is not used
	 * to construct the {@link Schema}.
	 * @param schema
	 */
	public void setSchema(Schema schema) {
		this.schema = schema;
	}

	/**
	 * get runtime types keyed by runtime name, which is 
	 * a type name known to the Teiid engine
	 * @return
	 */
	public Map<String, Datatype> getDataTypes() {
		return dataTypes;
	}
	
	/**
	 * To be called if the MetadataFactory is deserialized to set the canonical system
	 * type value.
	 * @param dt
	 * @param builtin
	 */
	public void correctDatatypes(Map<String, Datatype> dt, Map<String, Datatype> builtin) {
		this.dataTypes = dt;
		this.builtinDataTypes = builtin;
		for (Table t : this.schema.getTables().values()) {
			correctDataTypes(t.getColumns());
		}
		for (Procedure p : this.schema.getProcedures().values()) {
			correctDataTypes(p.getParameters());
			if (p.getResultSet() != null) {
				correctDataTypes(p.getResultSet().getColumns());
			}
		}
		for (FunctionMethod p : this.schema.getFunctions().values()) {
			correctDataTypes(p.getInputParameters());
	}
	}

	private void correctDataTypes(List<? extends BaseColumn> cols) {
		if (cols == null) {
			return;
		}
		for (BaseColumn c : cols) {
			Datatype datatype = c.getDatatype();
			String name = null;
			if (datatype == null) {
				name = c.getRuntimeType();
			} else {
				name = datatype.getName();
			}
			Datatype dt = this.builtinDataTypes.get(name);
			if (dt != null) {
				c.setDatatype(dt);
			} else if (datatype != null) {
				//must be an enterprise type
				//if it's used in a single schema, we're ok, but when used in multiple there's an issue
				addDatatype(dt);
			}
		}
	}
	
	public void addDatatype(Datatype datatype) {
		this.builtinDataTypes.put(datatype.getName(), datatype);
	}

	public String getVdbName() {
		return vdbName;
	}

	public int getVdbVersion() {
		return vdbVersion;
	}	
	
	/**
	 * Get the namespace map.  Will be an unmodifiable empty map if {@link #addNamespace(String, String)}
	 * has not been called successfully.
	 * @return
	 */
	public Map<String, String> getNamespaces() {
		if (this.namespaces == null) {
			return Collections.emptyMap();
		}
		return namespaces;
	}
	
	public void setBuiltinDataTypes(Map<String, Datatype> builtinDataTypes) {
		this.builtinDataTypes = builtinDataTypes;
	}

	/**
	 * get all built-in types, known to Designer and defined in the system metadata.
	 * The entries are keyed by type name, which is typically the xsd type name. 
	 * @see #getDataTypes() for run-time types
	 * @return
	 */
	public Map<String, Datatype> getBuiltinDataTypes() {
		return builtinDataTypes;
	}

	/**
	 * Parses, but does not close, the given {@link Reader} into this {@link MetadataFactory}
	 * @param ddl
	 * @throws MetadataException
	 */
	public void parse(Reader ddl) throws MetadataException {
		this.parser.parseDDL(this, ddl);
	}
	
	public void setParser(Parser parser) {
		this.parser = parser;
	}
	
	public void setModel(ModelMetaData model) {
		this.model = model;
	}
	
	public Parser getParser() {
		return parser;
	}

	/**
	 * Add a permission for a {@link Table} or {@link Procedure}
	 * @param role
	 * @param resource
	 * @param allowAlter
	 * @param allowCreate
	 * @param allowRead
	 * @param allowUpdate
	 * @param allowDelete
	 * @param allowExecute
	 * @param condition
	 * @param constraint
	 */
	public void addPermission(String role, AbstractMetadataRecord resource, Boolean allowAlter, Boolean allowCreate, Boolean allowRead, Boolean allowUpdate,
	                          Boolean allowDelete, Boolean allowExecute, String condition, Boolean constraint) {
		PermissionMetaData pmd = new PermissionMetaData();
		pmd.setResourceName(resource.getFullName());
		pmd.setAllowAlter(allowAlter);
		pmd.setAllowCreate(allowCreate);
		pmd.setAllowDelete(allowDelete);
		pmd.setAllowExecute(allowExecute);
		pmd.setAllowRead(allowRead);
		pmd.setAllowUpdate(allowUpdate);
		pmd.setCondition(condition);
		pmd.setConstraint(constraint);
		addPermission(pmd, role);
	}

	/**
	 * Add a permission for the current {@link Schema} which will typically act as a default for all child objects.
	 * @param role
	 * @param allowAlter
	 * @param allowCreate
	 * @param allowRead
	 * @param allowUpdate
	 * @param allowDelete
	 * @param allowExecute
	 */
 	public void addSchemaPermission(String role, Boolean allowAlter, Boolean allowCreate, Boolean allowRead, Boolean allowUpdate,
 	                Boolean allowDelete, Boolean allowExecute) {
		PermissionMetaData pmd = new PermissionMetaData();
		pmd.setResourceName(this.schema.getFullName());
		pmd.setAllowAlter(allowAlter);
		pmd.setAllowCreate(allowCreate);
		pmd.setAllowDelete(allowDelete);
		pmd.setAllowExecute(allowExecute);
		pmd.setAllowRead(allowRead);
		pmd.setAllowUpdate(allowUpdate);
		addPermission(pmd, role);
 	}

	/**
	 * Add a permission for a {@link Column}
	 * @param role
	 * @param resource
	 * @param allowCreate
	 * @param allowRead
	 * @param allowUpdate
	 * @param condition
	 * @param mask
	 * @param order
	 */
	public void addColumnPermission(String role, Column resource, Boolean allowCreate, Boolean allowRead, Boolean allowUpdate,
	                                String condition, String mask, Integer order) {
		PermissionMetaData pmd = new PermissionMetaData();
		String resourceName = null;
		if (resource.getParent() != null && resource.getParent().getParent() instanceof Procedure) {
			resourceName = resource.getParent().getParent().getFullName() + '.' + resource.getName();
		} else {
			resourceName = resource.getFullName();
		}
		pmd.setResourceName(resourceName);
		pmd.setAllowCreate(allowCreate);
		pmd.setAllowRead(allowRead);
		pmd.setAllowUpdate(allowUpdate);
		pmd.setCondition(condition);
		pmd.setMask(mask);
		pmd.setOrder(order);
		addPermission(pmd, role);
	}

	private void addPermission(PermissionMetaData pmd, String role) {
		if (this.grants == null) {
			this.grants = new ArrayList<Grant>();
		}
		this.grants.add(new Grant(role, pmd));
	}

	public void addFunction(FunctionMethod function) {
		function.setParent(this.schema);
		setUUID(function);
		for (FunctionParameter param : function.getInputParameters()) {
			setUUID(param);
		}
		setUUID(function.getReturnParameter());
		this.schema.addFunction(function);
	}

	public void setRenameDuplicateColumns(boolean renameDuplicateColumns) {
		this.renameDuplicateColumns = renameDuplicateColumns;
	}

	public String getCatalog(boolean raw) {
		if (raw && this.modelPropertiesRaw != null) {
			String catalogKey = getImporterCatalogKey(this.modelPropertiesRaw.keySet());
			return this.modelPropertiesRaw.get(catalogKey);
		}
		String catalogKey = getImporterCatalogKey(this.modelProperties.stringPropertyNames());
		return this.modelProperties.getProperty(catalogKey);
	}

	private String getImporterCatalogKey(Set<String> keys) {
		String property = "importer.catalog";
		for (String key : keys) {
			if (property.equalsIgnoreCase(key)) {
				return key;
			}
		}
		return "";
	}

    public boolean isLazyMetadataLoading() {
        return modelProperties.getProperty(LAZY_METADATA, "false").equalsIgnoreCase("true");
    }

    @Override
    public MetadataFactory clone() {
        try {
            MetadataFactory clone = (MetadataFactory) super.clone();
            clone.vdbName = this.vdbName;
            clone.vdbVersion = this.vdbVersion;
            clone.dataTypes = new HashMap<>(this.dataTypes);
            clone.builtinDataTypes = new HashMap<>(this.builtinDataTypes);
            clone.autoCorrectColumnNames = this.autoCorrectColumnNames;
            clone.renameDuplicateColumns = this.renameDuplicateColumns;
            clone.renameDuplicateTables = this.renameDuplicateTables;
            clone.namespaces = this.namespaces != null ? new TreeMap<>(this.namespaces) : null;
            clone.rawMetadata = this.rawMetadata;
            clone.modelProperties = this.modelProperties != null ? (Properties) this.modelProperties.clone() : null;
            clone.modelPropertiesRaw = this.modelPropertiesRaw != null ? new HashMap<>(this.modelPropertiesRaw) : null;
            clone.schema = new Schema();
            clone.schema.setName(this.schema.getName());
            clone.schema.setPhysical(this.schema.isPhysical());
            clone.idPrefix = this.idPrefix;
            clone.parser = this.parser;
            clone.model = this.model;
            clone.grants = this.grants != null ? new ArrayList<>(this.grants) : null;
            return clone;
        } catch (CloneNotSupportedException e) {
            throw new AssertionError();
        }
    }
}
