package profileEngine;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ListIterator;

import dataWarehousingTools.LogFile;

// TO DO:  Look at old version.  
//         Add escaping of single quotes within strings and deal with empty/null strings when writing to Postgres              

public class Profile
{
	private long rowCount;

	public Profile
	(
		String databaseProduct, 
		Connection connection, 
		String databaseName, 
		String schemaName,
		String tableName,
		String partsInName,
		ColumnList columnList, 
		Metadata metadata,
		LogFile logFile
	)
	{
		ListIterator<ColumnElement> iterator;
		ColumnElement columnElement;
		StringBuffer s;
		String sqlStatement;
		int batchCount;                 // Hive
		int batchSize;                  // Hive
		String tableSpecification;

		if (partsInName.equals("3"))
			tableSpecification = new String(databaseName + "." + schemaName + "." + tableName);
		else if (partsInName.equals("2"))
			tableSpecification = new String(schemaName + "." + tableName);
		else
			tableSpecification = new String(tableName);
		
        batchCount = 0;                 // Hive
        batchSize = 200;                // Hive

		s = new StringBuffer("count(*) as count_rows, ");
	    for (iterator = columnList.listIterator(); iterator.hasNext();)
		{
	    	if (batchCount > 0)
	    	{
	    		s.append(", ");
	    	}
			columnElement = (ColumnElement) iterator.next();
			addQueryElements(databaseProduct, columnElement, s);
			if (batchCount < batchSize)
			{
				batchCount++;
			}
			else
			{
			    sqlStatement = new String
	    	    (
	    	    	"select " +
	    	    	s.toString() + " " +
	    	    	"from " +
	    	    	tableSpecification
	    	    );					
				
//			    System.out.println(sqlStatement);
			    
			    interpretResultSet(columnList, databaseProduct,  connection, sqlStatement, metadata, logFile);		
			    s = new StringBuffer("");
			    batchCount = 0;
			}			
		}
//
	    if (!(s.equals("")) && !(s.length() == 0))	    
	    {
//	    	System.out.println("Column list: |" + s.toString() + "|, length: " + s.toString().length());
		    sqlStatement = new String
		    (
		    	"select " +
		    	s.toString() + " " +
		    	"from " +
		    	tableSpecification
		    );	    		
		    interpretResultSet(columnList, databaseProduct,  connection, sqlStatement, metadata, logFile);
	    }
	}

	private void addQueryElements(String databaseProduct, ColumnElement columnElement, StringBuffer s)
	{
		int    columnId;
		String columnIdAsString;
		
		columnId = columnElement.getColumnId();
		columnIdAsString = new String(new Integer(columnId).toString());
		if (columnElement.getProfileDataType().equals("Date"))
		{
			if (databaseProduct.equals("Apache Hive"))
			{
				s.append
				(
					"min(" + columnElement.getColumnName() + ") as min" + columnIdAsString + ", " +
					"max(" + columnElement.getColumnName() + ") as max" + columnIdAsString + ", "
				);
			}
			else if (databaseProduct.equals("Microsoft SQL Server"))
			{
				s.append
				(
					"convert(varchar, min(" + columnElement.getColumnName() + "), 13) as min" + columnIdAsString +  ", " +
					"convert(varchar, max(" + columnElement.getColumnName() + "), 13) as max" + columnIdAsString + ", "
				);
			}
			else  // Below works for Oracle? Netezza? Postgres?
			{
				s.append
				(
					"to_char(min(" + columnElement.getColumnName() + "), 'DD-Mon-YYYY HH24:MI:SS') as min" + columnIdAsString +  ", " +
					"to_char(max(" + columnElement.getColumnName() + "), 'DD-Mon-YYYY HH24:MI:SS') as max" + columnIdAsString + ", "
				);
			}
		}
        else if	(columnElement.getProfileDataType().equals("Number"))
		{
			s.append
			(
				"min(" + 
				columnElement.getColumnName() +
				") as min" +
				columnIdAsString  + ", " +
				"max(" +
				columnElement.getColumnName() +
				") as max" +
				columnIdAsString +
				", "
			);	
		}
        else if	(columnElement.getProfileDataType().equals("Time"))
		{
			s.append
			(
				"min(" + 
				columnElement.getColumnName() +
				") as min" +
				columnIdAsString + ", " +
				"max(" +
				columnElement.getColumnName() +
				") as max" +
				columnIdAsString + 
				", "
			);	
		}
        else if	(columnElement.getProfileDataType().equals("String"))
        {
        	if (columnElement.getDataLength() <= 64)
 			{
				s.append
				(
					"min(" + 
					columnElement.getColumnName() +
					") as min" +
					columnIdAsString + ", " +
					"max(" +
					columnElement.getColumnName() +
					") as max" +
					columnIdAsString + 
					", "
				);	
			}
        	else
 			{
    			if (databaseProduct.equals("Microsoft SQL Server")) 
    			{
    				if (!columnElement.getProfileDataType().equals("Boolean"))
	    			{
						s.append
						(
							"min(substring(" + 
							columnElement.getColumnName() +
							", 1, 64)) as min" +
							columnIdAsString + ", " +
							"max(substring(" +
							columnElement.getColumnName() +
							", 1, 64)) as max" +
							columnIdAsString + 
							", "
						);        				
	    			}
    			}
    			else if (databaseProduct.equals("Apache Hive"))
    			{
					s.append
					(
						"min(substr(" + 
						columnElement.getColumnName() +
						", 1, 64)) as min" +
						columnIdAsString + ", " +
						"max(substr(" +
						columnElement.getColumnName() +
						", 1, 64)) as max" +
						columnIdAsString + 
						", "
					);        				
    			}
    			else	// Below is PostgreSQL and Netezza version.  Oracle should be same as above, I think.
    			{
					s.append
					(
						"min(substring(" + 
						columnElement.getColumnName() +
						" for 64)) as min" +
						columnIdAsString + ", " +
						"max(substring(" +
						columnElement.getColumnName() +
						" for 64)) as max" +
						columnIdAsString + ", "
					);
    			}
			}
        }
		
		if (columnElement.getProfileDataType().equals("String"))
		{
			if (databaseProduct.equals("Microsoft SQL Server"))
			{
				if 
				(
					(columnElement.getDataType().equalsIgnoreCase("text")) ||
					(columnElement.getDataType().equalsIgnoreCase("longvarchar")) ||
					(columnElement.getDataType().equalsIgnoreCase("longvarbinary"))
				)
				{
					s.append
					(
						"max(datalength(" +
						columnElement.getColumnName() +
						")) as max_length" +
						columnIdAsString +
						", "
					);			
				}
				else
				{
					s.append
					(
						"max(len(" +
						columnElement.getColumnName() +
						")) as max_length" +
						columnIdAsString +
						", "
					);	
				}
			}
			else
			{
				s.append
				(
					"max(length(" +
					columnElement.getColumnName() +
					")) as max_length" +
					columnIdAsString +
					", "
				);
			}
		}

		if 
		(
			(databaseProduct.equals("Microsoft SQL Server")) &&
			(
				(columnElement.getDataType().equalsIgnoreCase("text")) ||
				(columnElement.getDataType().equalsIgnoreCase("longvarchar")) ||
				(columnElement.getDataType().equalsIgnoreCase("longvarbinary"))
			)
		)
		{
			s.append
			(
				"0 as distinct" +
				columnIdAsString
			);
		}
		else
		{
			s.append
			(
				"count(distinct " +
				columnElement.getColumnName() +
				") as distinct" +
				columnIdAsString
			);
		}
	} 

	private void interpretResultSet
	(
		ColumnList columnList, 
		String databaseProduct, 
		Connection connection, 
		String sqlStatement,
		Metadata metadata,
		LogFile logFile
	)
	{
		Statement statement = null;
		ResultSet resultSet = null;
		ResultSetMetaData resultSetMetaData = null;
		int numberOfColumns = 0;
		int i;
		int columnId;
		int previousColumnId;
		String columnName;
		Integer columnIdAsInteger;
		ColumnElement columnElement = null;
		String minDateString;
		String maxDateString;
		String minString;
		String maxString;
		
		
		logFile.logMessage(sqlStatement);
	    
		try
		{
			statement = connection.createStatement();
		}
		catch(SQLException sqle)
		{
			System.out.println(sqle);
		}

		try
		{
			resultSet = statement.executeQuery(sqlStatement);

		}
		catch(SQLException sqle)
		{
			System.out.println(sqle);
		}
		
		try 
		{
			resultSetMetaData = resultSet.getMetaData();
		} 
		catch (SQLException e) 
		{
//			System.out.println(sqlStatement);
			e.printStackTrace();
		}
   	 
	   	try 
	   	{
			numberOfColumns = resultSetMetaData.getColumnCount();
		} 
	   	catch (SQLException e) 
	   	{
			e.printStackTrace();
		}

	   	try 
	   	{
			while (resultSet.next())
			{
				previousColumnId = 0;
			   	for (i = 1; i <= numberOfColumns; i++)
			   	{  
			   		try 
			   		{
						columnName = new String(resultSetMetaData.getColumnName(i));
						
						if (columnName.equals("count_rows"))
						{
							try
							{
								rowCount = resultSet.getLong("count_rows");
							}
						   	catch (SQLException e) 
						   	{
								e.printStackTrace();
							}
							metadata.updateTableProfile(rowCount);
						} 
						else
						{
							if (columnName.startsWith("min"))
								columnIdAsInteger = new Integer(columnName.substring(3));
							else if (columnName.startsWith("max_length")) 
								columnIdAsInteger = new Integer(columnName.substring(10)); 
							else if (columnName.startsWith("max")) 
								columnIdAsInteger = new Integer(columnName.substring(3));
							else if (columnName.startsWith("distinct")) 
								columnIdAsInteger = new Integer(columnName.substring(8));
							else
							{
								System.out.println("You missed one! " + columnName);
								columnIdAsInteger = new Integer(0);
							}
							columnId = columnIdAsInteger.intValue();
				
							if (columnId != previousColumnId)
							{
								if (previousColumnId != 0)
								{
//									System.out.println
//									(
//										columnElement.getColumnName() +
//										columnElement.getColumnProfileId(),
//										columnElement.getProfileDataType(),
//										": Distinct " +
//										columnElement.getUniqueValueCount() +
//										", Max Length " +
//										columnElement.getMaxLength() +
//										", Min " +
//										columnElement.getMinString() +
//										", Max " +
//										columnElement.getMaxString() +
//										", Min " +
//									    columnElement.getMinNumber() +
//										", Min " +
//									    columnElement.getMinDate() +
//										", Max " +
//									    columnElement.getMaxNumber() +
//										", Max " +
//									    columnElement.getMaxDate()
//									);
									metadata.updateColumnProfile
									(
										columnElement.getColumnProfileId(),
										columnElement.getProfileDataType(),
									    columnElement.getUniqueValueCount(),
									    columnElement.getMinNumber(),
									    columnElement.getMinDate(),
									    columnElement.getMinString(),
									    columnElement.getMaxNumber(),
									    columnElement.getMaxDate(),
									    columnElement.getMaxString(),
									    columnElement.getMaxLength()
									);
								}
								columnElement = getColumnElement(columnList, columnId);
								previousColumnId = columnId;
							}
							if (columnName.startsWith("min")) 
							{				
					            if (columnElement.getProfileDataType().equals("Number"))
					            {
					            	try
									{
										columnElement.setMinNumber(resultSet.getDouble(columnName));
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if (columnElement.getProfileDataType().equals("Date"))
					            {
					            	try
									{
	//
	//		      For Apache Hive.  Although we asked for a date, Hive returns a string, in the format 
	//		      YYYY-MM-DD, which will be OK for storing in the Postgres Data Quality database.
	//		                    		
					            		minDateString = resultSet.getString(columnName);
					            		if (resultSet.wasNull())
					            			minDateString = "<null>";
										columnElement.setMinDate(minDateString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if	(columnElement.getProfileDataType().equals("Time"))
					            {
					            	try
									{
					            		minString = resultSet.getString(columnName);
					            		if (resultSet.wasNull())
					            			minString = "<null>";                   		
										columnElement.setMinString(minString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if	(columnElement.getProfileDataType().equals("String"))
					            {
					            	try
									{
					            		minString = resultSet.getString(columnName);
					            		if (resultSet.wasNull())
					            			minString = "<null>";                   		
										columnElement.setMinString(minString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
							}
							else if (columnName.startsWith("max_length")) 
							{
					            if	(columnElement.getProfileDataType().equals("String"))
					            {
					            	try
									{
										columnElement.setMaxLength(resultSet.getInt(columnName));							
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
							}
							else if (columnName.startsWith("max")) 
							{
					            if (columnElement.getProfileDataType().equals("Number"))
					            {
					            	try
									{
										columnElement.setMaxNumber(resultSet.getDouble(columnName));
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if (columnElement.getProfileDataType().equals("Date"))
					            {
					            	try
									{
	//
	//		      For Apache Hive.  Although we asked for a date, Hive returns a string, in the format 
	//		      YYYY-MM-DD, which will be OK for storing in the Postgres Data Quality database.
	//		                    		
										maxDateString = resultSet.getString(columnName);
										if (resultSet.wasNull())
											maxDateString = "<null>";
//										System.out.println("Max. Date String: " + maxDateString);
	
										columnElement.setMaxDate(maxDateString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if	(columnElement.getProfileDataType().equals("Time"))
					            {
					            	try
									{
					            		maxString = resultSet.getString(columnName);
					            		if (resultSet.wasNull())
					            			maxString = "<null>";                   		
										columnElement.setMaxString(maxString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
					            else if	(columnElement.getProfileDataType().equals("String"))
					            {
					            	try
									{
					            		maxString = resultSet.getString(columnName);
					            		if (resultSet.wasNull())
					            			maxString = "<null>";                   		
										columnElement.setMaxString(maxString);
									} 
					            	catch (SQLException e)
									{
										e.printStackTrace();
									}
					            }
							}
							else if (columnName.startsWith("distinct")) 
							{
				            	try
								{
									columnElement.setUniqueValueCount(resultSet.getLong(columnName));             		
								} 
				            	catch (SQLException e)
								{
									e.printStackTrace();
								}
							}
							else
							{
								System.out.println("You missed one! " + columnName);
							}
						}
			   		}
			   		catch (SQLException e) 
			   		{
						e.printStackTrace();
					}
				} 
			}
//			System.out.println
//			(
//				columnElement.getColumnName() +
//					columnElement.getColumnProfileId(),
//					columnElement.getProfileDataType(),
//				": Distinct " +
//				columnElement.getUniqueValueCount() +
//				", Max Length " +
//				columnElement.getMaxLength() +
//				", Min " +
//				columnElement.getMinString() +
//				", Max " +
//				columnElement.getMaxString() +
//				", Min " +
//			    columnElement.getMinNumber() +
//				", Min " +
//			    columnElement.getMinDate() +
//				", Max " +
//			    columnElement.getMaxNumber() +
//				", Max " +
//			    columnElement.getMaxDate()
//			);
			metadata.updateColumnProfile
			(
				columnElement.getColumnProfileId(),
				columnElement.getProfileDataType(),
			    columnElement.getUniqueValueCount(),
			    columnElement.getMinNumber(),
			    columnElement.getMinDate(),
			    columnElement.getMinString(),
			    columnElement.getMaxNumber(),
			    columnElement.getMaxDate(),
			    columnElement.getMaxString(),
			    columnElement.getMaxLength()
			);
		} 
	   	catch (SQLException e) 
	   	{
			e.printStackTrace();
		}
	   	
	   	try 
	   	{
			statement.close();
		} 
	   	catch (SQLException e) 
	   	{
			e.printStackTrace();
		}
	}
	
	private ColumnElement getColumnElement(ColumnList columnList, int columnId)
	{
		ListIterator<ColumnElement> iterator;
        ColumnElement columnElement;
		
	    for (iterator = columnList.listIterator(); iterator.hasNext();)
		{
			columnElement = (ColumnElement) iterator.next();
			if (columnElement.getColumnId() == columnId)
				return columnElement;
		}
	    System.out.println("Internal error: Column " + columnId + "not found");
	    return new ColumnElement();		
	}
	
	public long getRowCount()
	{
		return rowCount;
	}
}