/*	SORT.C	- 
**
**
** Copyright (c) 1996  Hughes Technologies Pty Ltd
**
** Permission to use, copy, and distribute for non-commercial purposes,
** is hereby granted without fee, providing that the above copyright
** notice appear in all copies and that both the copyright notice and this
** permission notice appear in supporting documentation.
**
** The software may be modified for your own purposes, but modified versions
** may not be distributed.
**
** This software is provided "as is" without any expressed or implied warranty.
**
**
*/


#include <stdio.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>

#  include <arpa/inet.h>
#  include <unistd.h>
#  include <stdlib.h>
#  include <string.h>

#  include <sys/mman.h>

#ifdef HAVE_DIRENT_H
#    include <dirent.h>
#endif

#ifdef HAVE_SYS_DIR_H
#  include <sys/dir.h>
#endif

#if defined(_OS_WIN32)
#  include <winsock.h>
#endif

#include <common/debug.h>
#include <common/site.h>
#include <common/portability.h>

#include "avl_tree.h"


#if defined(_OS_WIN32)
#  include "msql_yacc.h"
#else
#  include "y.tab.h"
#endif

#define _MSQL_SERVER_SOURCE

#include "msql_priv.h"
#include "msql.h"
#include "errmsg.h"

#define REG             register




extern	char	*msqlDbDir,
		*packet,
		errMsg[];
extern	int	outSock;




/* 
** The new sorting routine 
**
** This is an implementation of qSort
*/

static void swapRows(entry, low, high)
	cache_t	*entry;
	u_int	low,
		high;
{
	row_t	lowRow,
		highRow,
		*tmp;

	rowRead(entry,&lowRow,low);
	rowRead(entry,&highRow,high);
	tmp = dupRow(entry,&lowRow,&(entry->row));
	rowPlace(entry, &highRow, low);
	rowPlace(entry, tmp, high);
}


void qSort(entry, order, olist, lBound, uBound)
	cache_t	*entry;
	order_t	*order;
	int	*olist;
	u_int	lBound,
		uBound;
{
	int	lCount,
		uCount,
		res;
	row_t	curRow,
		pivotRow;

	res = rowRead(entry, &pivotRow, lBound);
	if (res < 0)
		return;
	lCount = lBound + 1;
	uCount = uBound;

	while(lCount <= uCount)
	{
		res = rowRead(entry,&curRow, lCount);
		if (res < 0)
			return;
		while(lCount <= uBound &&
			compareRows(entry,&curRow, &pivotRow, order, olist)<=0)
		{
			lCount++;
			res = rowRead(entry,&curRow, lCount);
			if (res < 0)
			{
				return;
			}
		}

		res =  rowRead(entry,&curRow, uCount);
		if (res < 0)
			return;
		while(uCount > lBound &&
			compareRows(entry,&curRow, &pivotRow, order, olist)>=0)
		{
			uCount--;
			res = rowRead(entry,&curRow, uCount);
			if (res < 0)
			{
				return;
			}
		}

		if (lCount < uCount)
		{
			swapRows(entry, lCount, uCount);
			lCount++;
			uCount--;
		}
	}

	/*
	** Move the pivot into place
	*/
	swapRows(entry, uCount, lBound);
	if (uCount != lBound)
	{
		qSort(entry, order, olist, lBound, uCount - 1);
	}
	if (uCount < uBound)
	{
		qSort(entry, order, olist, uCount+1, uBound);
	}
}



int createSortedTable(entry,order)
	cache_t	*entry;
	order_t	*order;
{
	int	olist[MAX_FIELDS],
		intBuf;
	u_int	rowNum,
		uintBuf;
	double	realBuf;
	char	idxPath[255],
		dataPath[255],
		oldPath[255],
		*bufPtr;
	avltree *tree;
	avl_nod	*node;
	row_t	row;
	cache_t	tmp;
	sblk_t	sblock;
	avl_cur	cursor;


	msqlTrace(TRACE_IN,"createSortedTable()");

	if(initTable(entry,FULL_REMAP) < 0)
	{
		msqlTrace(TRACE_OUT,"createSortedTable()");
		return(-1);
	}
	if (entry->sblk->numRows == 0)
	{
		return(0);
	}
	if (msqlSetupOrder(entry,olist,order) < 0)
	{
		msqlTrace(TRACE_OUT,"createSortedTable()");
		return(-1);
	}


	/*
	** OK, this may look wierd.  Sorting a table using any
	** sorting algorithm sucks for performance.  The engine
	** can create a new, indexed table at a rate of over 800
	** rows per second on a Pentium yet a sort of 1200 row
	** table will take 20 to 30 seconds.  So, a quick way to
	** handle it is to create a dummy index containing the
	** sort field.  The rowID is stored in the index also.  A
	** left to right parse of the index tree once generated will
	** give us the sorted rowID's.
	**
	** Note, this only works for a single field.  For multi-fields
	** we still do this for the first order field to reduce the
	** work of the qSort implementation (should be mainly sorted by
	** the first effort).
	*/

	switch(order->type)
	{
		case CHAR_TYPE:
			tree=avlMemCreate(order->length,AVL_CHAR,AVL_DUP);
			break;

		case INT_TYPE:
		case UINT_TYPE:
		case MONEY_TYPE:
		case TIME_TYPE:
		case DATE_TYPE:
			tree=avlMemCreate(order->length,AVL_INT,AVL_DUP);
			break;

		case REAL_TYPE:
			tree=avlMemCreate(order->length,AVL_REAL,AVL_DUP);
			break;

		default:
			snprintf(errMsg, MAX_ERR_MSG,
				"Invalid field type '%s' (type %d) in sort",
				msqlTypeName(order->type), order->type);
			return(-1);
			break;
	}

	/* 
	** Scan the table creating the index on the fly 
	*/
	rowNum = 0;
	while(rowNum < entry->sblk->numRows)
	{
		if (rowRead(entry,&row, rowNum) < 0)
			return(-1);
		if (row.header->active == 0)
		{
			rowNum++;
			continue;
		}
		switch(order->type)
		{
			case CHAR_TYPE:
			bufPtr = (u_char *)row.data + *olist + 1;
			break;

			case INT_TYPE:
			case UINT_TYPE:
			case MONEY_TYPE:
			case TIME_TYPE:
			case DATE_TYPE:
				bcopy4(row.data + *olist + 1,&intBuf);
				bufPtr = (char *)&intBuf;
				break;

			case REAL_TYPE:
				bcopy8(row.data + *olist + 2,&realBuf);
				bufPtr = (char *)&realBuf;
				break;

			default:
				snprintf(errMsg, MAX_ERR_MSG,
					"Unknown type %s in sort", order->type);
				return(-1);
				break;
		}

		avlInsert(tree, (char *)bufPtr, (off_t)rowNum);
		rowNum++;
	}

	/*
	** Create the dummy output table 
	*/
	snprintf(dataPath,MAXPATHLEN,"%s/.tmp/%s.tmp", msqlDbDir,
		entry->table);
	tmp.dataFD = open(dataPath, O_CREAT|O_RDWR|O_BINARY, 0600);
        bzero(&sblock, sizeof(sblock));
        sblock.version = DB_VERSION;
        sblock.numRows = sblock.activeRows = 0;
        sblock.freeList = NO_POS;
	write(tmp.dataFD,&sblock,SBLK_SIZE);

	tmp.indices = NULL;
	tmp.def = entry->def;
	tmp.overflowFD = -1;
	tmp.dataMap = NULL;
	tmp.remapData = 1;
	tmp.remapOverflow = 0;
	initTable(&tmp, FULL_REMAP);
	tmp.rowDataLen = entry->rowDataLen;
	tmp.rowLen = entry->rowDataLen + (8 -
                ((entry->rowDataLen + HEADER_SIZE) % 8));
        tmp.row.buf = (u_char *)malloc(entry->rowLen + HEADER_SIZE + 2);
        tmp.row.header = (hdr_t *)tmp.row.buf;
        tmp.row.data = tmp.row.buf + HEADER_SIZE;
        tmp.sblk = (sblk_t *)tmp.dataMap;


	/*
	** Parse the index creating a new data table
	*/
	if (order->dir == DESC)
		node = avlGetLast(tree);
	else
		node = avlGetFirst(tree);
	avlSetCursor(tree, &cursor);
	while(node)
	{
		rowNum = (u_int)node->data;
		rowRead(entry, &row, rowNum);
		rowWrite(&tmp, &row, NO_POS);
		if (order->dir == DESC)
			node = avlGetPrev(tree, &cursor);
		else
			node = avlGetNext(tree, &cursor);
	}
	avlClose(tree);

	/*
	** Swap the new table into place
	*/
	snprintf(oldPath,MAXPATHLEN,"%s/.tmp/%s.dat",msqlDbDir,
		entry->table);
	munmap(entry->dataMap, entry->size);
	close(entry->dataFD);
	unlink(oldPath);
	entry->dataFD = tmp.dataFD;
	entry->size = tmp.size;
	entry->dataMap = tmp.dataMap;
	entry->sblk = (sblk_t *)entry->dataMap;
#if defined(_OS_OS2)
           // under OS/2, is not possible to rename an open file!
           // and also the file is not deleted after use.
           // The code works, but files aren't renamed/deleted.
           // .tmp directory gets filled from tmp files.
           // Code could be hacked to allow rename, but it not
           // possible to delete it after close().
#endif
	rename(dataPath, oldPath);
	free(tmp.row.buf);


	/*
	** If it's a Multi-field sort, kick in the qSort
	*/

	if (order->next)
	{
		qSort(entry, order, olist, 0, entry->sblk->numRows - 1);
	}
	msqlTrace(TRACE_OUT,"createSortedTable()");
	return(0);
}