/*
 * 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$
 *  
 */
/*
 * EXTENSION
 * ISSUE: DVCORE-2521, DVCORE-2693
 * FIXED BY: Roman Pedchenko, Salvatore Raunich
 * DATE: 2014-12-05, 2015-02-20
 * ADOPTED FROM: TEIID-2153, 2171, 2359, 2149, 2358, 2410, 2271, 2467, 2429, 2363, 2160, 2461, 2797, 2326, 1557, 2628
 * END EXTENSION
 */
// https://issues.jboss.org/browse/TEIID-2141
// https://issues.jboss.org/browse/TEIID-4449
// https://issues.jboss.org/browse/TEIID-2897
// https://issues.jboss.org/browse/TEIID-2580
// https://issues.jboss.org/browse/TEIID-2980
// https://issues.jboss.org/browse/TEIID-3030
// https://issues.jboss.org/browse/TEIID-5233
// https://issues.jboss.org/browse/TEIID-4115
// https://issues.jboss.org/browse/TEIID-5135
// https://issues.jboss.org/browse/TEIID-3390
// https://issues.jboss.org/browse/TEIID-2112
// https://issues.jboss.org/browse/TEIID-2743
// https://issues.redhat.com/browse/TEIID-3248 (only partially, only streaming variable)
package org.teiid.query.processor.relational;

import java.sql.Timestamp;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;

import net.sf.saxon.tree.iter.EmptyIterator;
import org.teiid.api.exception.query.ExpressionEvaluationException;
import org.teiid.common.buffer.BlockedException;
import org.teiid.common.buffer.BufferManager.TupleSourceType;
import org.teiid.common.buffer.TupleBatch;
import org.teiid.common.buffer.TupleBuffer;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.types.ArrayImpl;
import org.teiid.core.types.DataTypeManager;
import org.teiid.core.types.TransformationException;
import org.teiid.core.types.XMLType;
import org.teiid.query.QueryPlugin;
import org.teiid.query.analysis.AnalysisRecord;
import org.teiid.query.eval.Evaluator;
import org.teiid.query.function.FunctionDescriptor;
import org.teiid.query.function.source.XMLSystemFunctions;
import org.teiid.query.sql.LanguageObject;
import org.teiid.query.sql.lang.XMLTable;
import org.teiid.query.sql.lang.XMLTable.XMLColumn;
import org.teiid.query.sql.symbol.XMLNamespaces;
import org.teiid.query.sql.symbol.XMLNamespaces.NamespaceItem;
import org.teiid.query.xquery.saxon.PushBackSequenceIterator;
import org.teiid.query.xquery.saxon.SaxonXQueryExpression.Result;
import org.teiid.query.xquery.saxon.SaxonXQueryExpression.RowProcessor;

import com.datavirtuality.dv.core.interfaces.DVCaller;

import org.teiid.query.xquery.saxon.XQueryEvaluator;

import net.sf.saxon.lib.ConversionRules;
import net.sf.saxon.om.Item;
import net.sf.saxon.om.NodeInfo;
import net.sf.saxon.om.SequenceIterator;
import net.sf.saxon.om.SequenceTool;
import net.sf.saxon.sxpath.XPathDynamicContext;
import net.sf.saxon.sxpath.XPathExpression;
import net.sf.saxon.trans.XPathException;
import net.sf.saxon.type.BuiltInAtomicType;
import net.sf.saxon.type.ConversionResult;
import net.sf.saxon.type.ValidationException;
import net.sf.saxon.type.Converter;
import net.sf.saxon.type.ValidationException;
import net.sf.saxon.value.AtomicValue;
import net.sf.saxon.value.CalendarValue;
import net.sf.saxon.value.StringValue;

/**
 * Handles xml table processing.
 * 
 * When streaming the results will be fully built and stored in a buffer
 * before being returned
 */
public class XMLTableNode extends SubqueryAwareRelationalNode implements RowProcessor {

	private static Map<Class<?>, BuiltInAtomicType> typeMapping = new HashMap<Class<?>, BuiltInAtomicType>();
	
	static {
		typeMapping.put(DataTypeManager.DefaultDataClasses.TIMESTAMP, BuiltInAtomicType.DATE_TIME);
		typeMapping.put(DataTypeManager.DefaultDataClasses.TIME, BuiltInAtomicType.TIME);
		typeMapping.put(DataTypeManager.DefaultDataClasses.DATE, BuiltInAtomicType.DATE);
		typeMapping.put(DataTypeManager.DefaultDataClasses.FLOAT, BuiltInAtomicType.FLOAT);
		typeMapping.put(DataTypeManager.DefaultDataClasses.DOUBLE, BuiltInAtomicType.DOUBLE);
		typeMapping.put(DataTypeManager.DefaultDataClasses.BLOB, BuiltInAtomicType.HEX_BINARY);
		typeMapping.put(DataTypeManager.DefaultDataClasses.VARBINARY, BuiltInAtomicType.HEX_BINARY);
	}
	
	private static TeiidRuntimeException EARLY_TERMINATION = new TeiidRuntimeException();
	
	private XMLTable table;
	private List<XMLColumn> projectedColumns;
	
	private Result result;
	private int rowCount = 0;
	private Item item;
	
	private TupleBuffer buffer;
	
	private enum State {
		BUILDING,
		AVAILABLE,
		DONE
	};
	
	private State state = State.BUILDING;
	private volatile TeiidRuntimeException asynchException;
	private int outputRow = 1;
	private boolean usingOutput;

	private int rowLimit = -1;
	private boolean allowTextNodeWhitespace;

	private boolean streaming;

	public XMLTableNode(int nodeID) {
		super(nodeID);
	}
	
	@Override
	public synchronized void closeDirect() {
		super.closeDirect();
		if(this.buffer != null) {
			if (!usingOutput) {
				this.buffer.remove();
			}
        	this.buffer = null;
        }
		reset();
	}
	
	@Override
	public synchronized void reset() {
		super.reset();
		if (this.result != null) {
			result.close();
			result = null;
		}
		item = null;
		rowCount = 0;
		outputRow = 1;
		usingOutput = false;
		this.buffer = null;
		this.state = State.BUILDING;
		this.asynchException = null;
		this.rowLimit = -1;
		this.table.reset();
	}
	
	public void setTable(XMLTable table) {
		this.table = table;
	}

	public List<XMLColumn> getProjectedColumns() {
		return projectedColumns;
	}

	public void setProjectedColumns(List<XMLColumn> projectedColumns) {
		this.projectedColumns = projectedColumns;
	}
	
	@Override
	public XMLTableNode clone() {
		XMLTableNode clone = new XMLTableNode(getID());
		this.copyTo(clone);
		final XMLTable tableClone = (XMLTable) this.table.clone();
		clone.setTable(tableClone);
		// correctly refer filtered columns
		final List<XMLColumn> projColsClone = new ArrayList<>(this.projectedColumns.size());
		for (XMLColumn column : this.projectedColumns) {
			for (XMLColumn column1 : tableClone.getColumns()) {
				if (column.equals(column1)) {
					projColsClone.add(column1);
					break;
				}
			}
		}
		clone.setProjectedColumns(projColsClone);
		clone.setAllowTextNodeWhitespace(this.allowTextNodeWhitespace);
		return clone;
	}

	@Override
	public void open() throws TeiidComponentException, TeiidProcessingException {
		super.open();
		if (getParent() instanceof LimitNode) {
			LimitNode parent = (LimitNode)getParent();
			if (parent.getLimit() > 0) {
				rowLimit = parent.getLimit() + parent.getOffset();
			}
		}
	}

	@Override
	protected synchronized TupleBatch nextBatchDirect() throws BlockedException,
			TeiidComponentException, TeiidProcessingException {
		
		evaluate(false);
		
		if (streaming) {
			while (state == State.BUILDING) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					 throw new TeiidRuntimeException(QueryPlugin.Event.TEIID30169, e);
				}
			}
			unwrapException(asynchException);
			TupleBatch batch = this.buffer.getBatch(outputRow);
			outputRow = batch.getEndRow() + 1;
			if (state != State.DONE && !batch.getTerminationFlag()) {
				state = hasNextBatch()?State.AVAILABLE:State.BUILDING;
			}
			return batch;
		}

		if (result == null) {
			terminateBatches();
		} else
		while (!isBatchFull() && !isLastBatch()) {
			if (item == null) {
				try {
					item = result.iter.next();
				} catch (XPathException e) {
					 throw new TeiidProcessingException(QueryPlugin.Event.TEIID30170, e, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30170, e.getMessage()));
				}
				rowCount++;
				if (item == null) {
					terminateBatches();
					break;
				}
			}
			addBatchRow(processRow());
			if (rowCount == rowLimit) {
				terminateBatches();
				break;
			}
		}
		return pullBatch();
	}

	private void evaluate(final boolean useFinalBuffer) throws TeiidComponentException,
			ExpressionEvaluationException, BlockedException,
			TeiidProcessingException {
		if (result != null || this.buffer != null) {
			return;
		}
		setReferenceValues(this.table);
		final HashMap<String, Object> parameters = new HashMap<String, Object>();
		Evaluator eval = getEvaluator(Collections.emptyMap());
		// we must compile XQuery expression only once as we need it indeed only once and as several threads can work on the same XMLTableNode
		if (this.table.getXQueryExpression() == null) {
			final Object xqueryString = eval.evaluate(table.getXquery(), null);
			XMLNamespaces ns = table.getNamespaces();
			if (ns != null)
				for (NamespaceItem item : ns.getNamespaceItems())
					eval.evaluateNamespaceItem(ns, item, null);
			table.setXqueryString((String) xqueryString);
			for (XMLColumn col : table.getColumns()) {
				if (col.getPath() != null)
					col.setPathString((String) eval.evaluate(col.getPath(), null));
			}
			table.compileXqueryExpression();
			if (table.getXQueryExpression().usesContextItem() && !table.hasContext()) {
				throw new ExpressionEvaluationException(QueryPlugin.Util.getString("ValidationVisitor.context_required") + table); //$NON-NLS-1$
			}
			table.getXQueryExpression().useDocumentProjection(this.getProjectedColumns(), new AnalysisRecord(false, false));
			streaming = this.table.getXQueryExpression().isStreaming();
		}
		final Object contextItem = eval.evaluateParameters(this.table.getPassing(), null, parameters);

		if (!this.table.getPassing().isEmpty() && contextItem == null && parameters.isEmpty()) {
			// when passing array isn't empty, but evaluator didn't produce neither contextItem nor parameters
			// then a null passing passed to XmlTable
			result = new Result();
			result.iter = EmptyIterator.emptyIterator();
			streaming = false;
			return;
		}

		if (this.table.getXQueryExpression().isStreaming()) {
			if (this.buffer == null) {
				this.buffer = this.getBufferManager().createTupleBuffer(getOutputElements(), getConnectionID(), TupleSourceType.PROCESSOR);
				if (!useFinalBuffer) {
					this.buffer.setForwardOnly(true);
				}
			}
			Runnable r = new Runnable() {
				TupleBuffer b = buffer;
				@Override
				public void run() {
					try {
						XQueryEvaluator.evaluateXQuery(table.getXQueryExpression(), contextItem, parameters, new RowProcessor() {
							@Override
							public void processRow(NodeInfo row) {
								synchronized (XMLTableNode.this) {
									if (b != buffer) {
										//if the buffer has changed we've been reset
										throw EARLY_TERMINATION;
									}
									XMLTableNode.this.processRow(row);
								}
							}

							@Override
							public boolean isAllowTextNodeWhitespace() {
								return XMLTableNode.this.isAllowTextNodeWhitespace();
							}

							@Override
							public void setAllowTextNodeWhitespace(boolean allowTextNodeWhitespace) {
								XMLTableNode.this.setAllowTextNodeWhitespace(allowTextNodeWhitespace);
							}
						}, getContext());
					} catch (TeiidRuntimeException e) {
						if (e != EARLY_TERMINATION) {
							asynchException = e;
						}
					} catch (Throwable e) {
						asynchException = new TeiidRuntimeException(e);
					} finally {
						synchronized (XMLTableNode.this) {
							if (buffer != null && asynchException == null) {
								try {
									buffer.close();
								} catch (TeiidComponentException e) {
									asynchException = new TeiidRuntimeException(e);
								}
							}
							state = State.DONE;
							XMLTableNode.this.notifyAll();
						}
					}
				}
			};
			this.getContext().getExecutor().execute(r);
			return;
		}
		try {
			result = XQueryEvaluator.evaluateXQuery(this.table.getXQueryExpression(), contextItem, parameters, null, this.getContext());
		} catch (TeiidRuntimeException e) {
			unwrapException(e);
		}
	}
	
	private List<?> processRow() throws ExpressionEvaluationException, BlockedException,
			TeiidComponentException, TeiidProcessingException {
		List<Object> tuple = new ArrayList<Object>(projectedColumns.size());
		for (XMLColumn proColumn : projectedColumns) {
			if (proColumn.isOrdinal()) {
				tuple.add(rowCount);
			} else {
				try {
					XPathExpression path = proColumn.getPathExpression();
					XPathDynamicContext dynamicContext = path.createDynamicContext(item);
					final SequenceIterator pathIter = path.iterate(dynamicContext);
					Item colItem = pathIter.next();
					if (colItem == null) {
						if (proColumn.getDefaultExpression() != null) {
							tuple.add(getEvaluator(Collections.emptyMap()).evaluate(proColumn.getDefaultExpression(), null));
						} else {
							tuple.add(null);
						}
						continue;
					}
					if (proColumn.getSymbol().getType() == DataTypeManager.DefaultDataClasses.XML) {
						SequenceIterator pushBack = new PushBackSequenceIterator(pathIter, colItem);
						XMLType value = table.getXQueryExpression().createXMLType(pushBack, this.getBufferManager(), false, getContext());
						tuple.add(value);
						continue;
					}
						if (proColumn.getSymbol().getType().isArray()) {
							ArrayList<Object> vals = new ArrayList<Object>();
							vals.add(getValue(proColumn.getSymbol().getType().getComponentType(), colItem, proColumn.getName()));
						Item next = null;
							while ((next = pathIter.next()) != null) {
								vals.add(getValue(proColumn.getSymbol().getType().getComponentType(), next, proColumn.getName()));
							}
							Object value = new ArrayImpl(vals.toArray((Object[]) Array.newInstance(proColumn.getSymbol().getType().getComponentType(), vals.size())));
							tuple.add(value);
							continue;
					} else if (pathIter.next() != null) {
						 throw new TeiidProcessingException(QueryPlugin.Event.TEIID30171, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30171, proColumn.getName()));
					}
					Object value = getValue(proColumn.getSymbol().getType(), colItem, proColumn.getName());
					tuple.add(value);
				} catch (XPathException e) {
					 throw new TeiidProcessingException(QueryPlugin.Event.TEIID30172, e, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30172, proColumn.getName()));
				}
			}
		}
		item = null;
		return tuple;
	}

	private Object getValue(Class<?> type,
 	                Item colItem, String proColumnName) throws XPathException,
 	                ValidationException, TransformationException {
		Object value = colItem;
		if (value instanceof AtomicValue) {
			value = getValue((AtomicValue)colItem);
		} else if (value instanceof Item) {
			Item i = (Item)value;
			if (XMLSystemFunctions.isNull(i)) {
				return null;
			}
			BuiltInAtomicType bat = typeMapping.get(type);
			if (bat != null) {
				AtomicValue av = new StringValue(i.getStringValueCS());
				ConversionResult cr = Converter.convert(av, bat, new ConversionRules());
				value = cr.asAtomic();
				value = getValue((AtomicValue)value);
				if (value instanceof Item) {
					value = ((Item)value).getStringValue();
				}
			} else {
				value = i.getStringValue();
			}
		}
		try {
			Object importedValue = FunctionDescriptor.importValue(value, type, getContext());
			return importedValue;
		} catch (TransformationException e) {
			DVCaller.call(140, e.getClass().getName() + ":column", proColumnName);
			throw e;
		}
	}

	private Object getValue(AtomicValue value) throws XPathException {
		if (value instanceof CalendarValue) {
			CalendarValue cv = (CalendarValue)value;
			if (!cv.hasTimezone()) {
				TimeZone tz = getContext().getServerTimeZone();
				int tzMin = tz.getRawOffset()/60000;
				if (tz.getDSTSavings() > 0) {
					tzMin = tz.getOffset(cv.getCalendar().getTimeInMillis())/60000;
				}
				cv.setTimezoneInMinutes(tzMin);
				Calendar cal = cv.getCalendar();
				return new Timestamp(cal.getTime().getTime());
			}
		}
		return SequenceTool.convertToJava(value);
	}

	@Override
	public void processRow(NodeInfo row) {
		if (isClosed()) {
			throw EARLY_TERMINATION;
		}
		assert this.state != State.DONE;
		this.item = row;
		rowCount++;
		try {
			this.buffer.addTuple(processRow());
			if (this.buffer.getRowCount() == rowLimit) {
				throw EARLY_TERMINATION;
			}
			if (state == State.BUILDING && hasNextBatch()) {
				this.state = State.AVAILABLE;
				this.notifyAll();
			}
		} catch (TeiidException e) {
			 throw new TeiidRuntimeException(e);
		}
	}

	private boolean hasNextBatch() {
		return this.outputRow + this.buffer.getBatchSize() <= rowCount + 1;
	}
	
	@Override
	public Collection<? extends LanguageObject> getObjects() {
		return this.table.getPassing();
	}

	@Override
	public boolean isAllowTextNodeWhitespace() {
		return this.allowTextNodeWhitespace;
	}

	@Override
	public void setAllowTextNodeWhitespace(boolean allowTextNodeWhitespace) {
		this.allowTextNodeWhitespace = allowTextNodeWhitespace;
	}
}
