package profileEngine;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.DateFormat;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.ListIterator;

import dataWarehousingTools.LogFile;

public class Frequency
{
	
	public Frequency
	(
		String databaseProduct,
		Connection connection, 
		String databaseName, 
		String schemaName,
		String tableName, 
		String partsInName,
		long rowCount,
		ColumnList columnList, 
		Metadata metadata,
		LogFile logFile
	)
	{
		ListIterator<ColumnElement> iterator;
		ColumnElement columnElement;
		Statement statement = null;
		ResultSet resultSet = null;
		int i;
		Double number = null;
		Calendar dateValue = null;
		String displayValue = null;
		String query;
		String orderedQuery;
		DecimalFormat decimalFormat;
		DateFormat dateFormat;
		boolean bool;
		Integer year;
		Integer month;
		Integer day;
		String tableSpecification;
		String sqlServerLimit;
		String oracleLimit;
		String postgresLimit;
		
		String[] order = {"desc", "asc"};
		
		decimalFormat = new DecimalFormat("###,###,###,###,###,###.######");
		dateFormat = new SimpleDateFormat("dd-MMM-yyyy HH:mm:ss.SSSSSS");

		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);
		
		for (iterator = columnList.listIterator(); iterator.hasNext();)
		{
			columnElement = (ColumnElement) iterator.next(); 

			if (databaseProduct.equals("Microsoft SQL Server"))
			{
				sqlServerLimit = new String ("top 100 ");
				oracleLimit = new String ("");
				postgresLimit = new String ("");
			}
			else if (databaseProduct.equals("Oracle"))
			{
				sqlServerLimit = new String ("");
				oracleLimit = new String ("where rownum <= 100 ");
				postgresLimit = new String ("");
			}
			else            // may need to add other databases here, but for now assume PostgreSQL
			{
				sqlServerLimit = new String ("");
				oracleLimit = new String ("");
				postgresLimit = new String ("limit 100 ");
			}
				
			try
			{
				statement = connection.createStatement();
			} 
			catch (SQLException e)
			{
				e.printStackTrace();
			}
			
			if (columnElement.getProfileDataType().equals("Date"))
			{
				query = new String
				(
					"select " + 
				        sqlServerLimit +
						columnElement.getColumnName() + ", " +
					    "count(*) as frequency " +
					"from " +
						tableSpecification + " " + 
						oracleLimit + 
					"group by " +
					    columnElement.getColumnName() + " "
				);
			}
			else if
			(
				(columnElement.getProfileDataType().equals("String")) &&
				(columnElement.getDataLength() > 64)
			)
			{
				if (databaseProduct.equals("Microsoft SQL Server"))
				{
					query = new String
					(
						"select " +
					        sqlServerLimit +								
							"substring(" +
						    columnElement.getColumnName() + ", 1, 64) as " + columnElement.getColumnName() + ", " +
						    "count(*) as frequency " +
						"from " +
						    tableSpecification + " " +
						"group by " +
						    "substring(" + columnElement.getColumnName() + ", 1, 64) "
					);	
				}
				else if (databaseProduct.equals("Apache Hive"))
				{
					query = new String
					(
						"select " +
						     sqlServerLimit +
					         "substr(" +
						     columnElement.getColumnName() + ", 1, 64) as " + columnElement.getColumnName() + ", " +
						     "count(*) as frequency " +
						"from " +
							tableSpecification + " " +
						"group by " +
						    "substr(" + columnElement.getColumnName() + ", 1, 64) "
					);	
				}
				else 
				{
					query = new String
					(
						"select " +
						     sqlServerLimit +
							 "substring(" +
						     columnElement.getColumnName() + " for 64) as " + columnElement.getColumnName() + ", " +
						     "count(*) as frequency " +
						"from " +
							tableSpecification + " " +
						"group by " +
						    "substring(" + columnElement.getColumnName() + " for 64) "
					);							
				}
			}
			else
			{
				query = new String
				(
					"select " +
					    sqlServerLimit +
					    columnElement.getColumnName() + ", " +
					    "count(*) as frequency " +
					"from " +
						tableSpecification + " " +
					"group by " +
					    columnElement.getColumnName() + " "
				);
			}

			for (String topBottom : order)
			{
				orderedQuery = new String
				(
					query +
					"order by " +
					    "frequency " + topBottom + ", " +
					    columnElement.getColumnName() + " asc " +
					    postgresLimit
				);
				logFile.logMessage(orderedQuery);
				
				try 
				{
					statement.setMaxRows(100);
				} 
				catch (SQLException e1) 
				{
					e1.printStackTrace();
				}
				
				try
				{
					resultSet = statement.executeQuery(orderedQuery);
				} 
				catch (SQLException e)
				{
					e.printStackTrace();
				}
				
				i = 0;
				try
				{
					while(resultSet.next())
					{
						if (i > 100)
							break;
						i++;
						if (columnElement.getProfileDataType().equals("Number"))
						{
							number = resultSet.getDouble(columnElement.getColumnName());
							if (resultSet.wasNull())
								displayValue = new String("<null>");
							else
								displayValue = decimalFormat.format(number);
						}
						else if (columnElement.getProfileDataType().equals("Date"))
						{
							dateValue = Calendar.getInstance();

							try
							{
//								Hive returns dates as strings
								if (databaseProduct.equals("Apache Hive"))
								{
									displayValue = resultSet.getString(columnElement.getColumnName());
			            			if (resultSet.wasNull())
			            			{
			            				dateValue = null;
			            				displayValue = new String("<null>");
			            			}
			            			else
			            			{
				            			year = new Integer(displayValue.substring(0, 4));
				            			month = new Integer(displayValue.substring(5, 7));
				            			day = new Integer(displayValue.substring(8, 10));
										dateValue.set(year.intValue(), month.intValue(), day.intValue(), 0, 0, 0);
			            			}
								}
								else
								{
									dateValue.setTime(resultSet.getDate(columnElement.getColumnName()));
									displayValue = dateFormat.format(dateValue.getTime());
								}
							}
		            		catch (NullPointerException e)
		            		{
		            			if (resultSet.wasNull())
		            				displayValue = new String("<null>");
		            			else
		            				e.printStackTrace();
		            		}
						}
						else if (columnElement.getProfileDataType().equals("Boolean"))
						{
							try
							{
								bool = resultSet.getBoolean(columnElement.getColumnName());
								if (resultSet.wasNull())
									displayValue = new String("<null>");
								else
								{
									if (bool)
										displayValue = new String("True");
									else
										displayValue = new String("False");
								}
							}
							catch (NullPointerException e)
		            		{
		            			if (resultSet.wasNull())
		            				displayValue = new String("<null>");
		            			else
		            				e.printStackTrace();
		            		}
						}
						else 
						{
		            		try
		            		{
		            			displayValue = new String(resultSet.getString(columnElement.getColumnName()));
		            		}
		            		catch (NullPointerException e)
		            		{
		            			if (resultSet.wasNull())
		            				displayValue = new String("<null>");
		            			else
		            				e.printStackTrace();
		            		}
						}
						
						metadata.insertColumnValueFrequency
						(
							columnElement.getColumnProfileId(),
							columnElement.getProfileDataType(),
							displayValue,
							resultSet.getLong("frequency")
						);
					}
				} 
				catch (SQLException e)
				{
					e.printStackTrace();
				}
			}
		}
	}
}