/*	table.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.
**
** ID = "$Id:"
**
*/


#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>

#ifdef OS2
#  include <io.h>
#  include <share.h>
#  include <types.h>
#  include <direct.h>
#  include <common/mman.h>
#else
#  include <arpa/inet.h>
#  include <string.h>
#  include <unistd.h>
#  include <sys/mman.h>
#endif

#ifdef HAVE_DIRENT_H
#  ifdef OS2
#    include <common/dirent.h>
#  else
#    include <dirent.h>
#  endif
#endif

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

#ifdef OS2
#  include "msql_yacc.h"
#else
#  include "y.tab.h"
#endif


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


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

#define REG             register


extern	char	errMsg[];
extern	cache_t tableCache[];
extern	char	*msqlHomeDir;


/* #define inline_bcopy(s,d,n)\
        { register int count;                                   \
        for(count=0;count<n;count++) *((char*)(d)+count) = *((char*)(s)+count);\
        }
*/

#define inline_bcopy bcopy


/****************************************************************************
** 	_openTable
**
**	Purpose	: Open the datafile for a given table
**	Args	: Database and Table names
**	Returns	: file descriptor for the data file
**	Notes	: 
*/

int openTable(table,db)
	char	*table;
	char	*db;
{
	char	path[MAXPATHLEN];

	(void)sprintf(path,"%s/msqldb/%s/%s.dat",msqlHomeDir,db,table);
#ifdef OS2
        return(_sopen(path,O_RDWR | O_BINARY, SH_DENYNO, S_IREAD | S_IWRITE));
#else
	return(open(path,O_RDWR));
#endif
}


int openOverflow(table,db)
	char	*table;
	char	*db;
{
	char	path[MAXPATHLEN];

	(void)sprintf(path,"%s/msqldb/%s/%s.ofl",msqlHomeDir,db,table);
#ifdef OS2
        return(_sopen(path,O_RDWR | O_BINARY, SH_DENYNO, S_IREAD | S_IWRITE));
#else
	return(open(path,O_RDWR));
#endif
}







int writeFreeList(cacheEntry, pos, entry)
	cache_t	*cacheEntry;
	u_int	pos,
		entry;
{
	off_t	seekPos;
	char	*buf;


	seekPos = pos * (cacheEntry->rowLen + HEADER_SIZE) + SBLK_SIZE +
		HEADER_SIZE;
	buf = ((char *)cacheEntry->dataMap) + seekPos;
	inline_bcopy(&entry,buf,sizeof(entry));
	return(0);
}


u_int readFreeList(cacheEntry,pos)
	cache_t	*cacheEntry;
	u_int	pos;
{
	off_t	seekPos;
	char	*buf;
	u_int	entry;


	seekPos = pos * (cacheEntry->rowLen + HEADER_SIZE) + SBLK_SIZE +
		HEADER_SIZE;
	buf = ((char *)cacheEntry->dataMap) + seekPos;
	inline_bcopy(buf, &entry, sizeof(entry));
	return(entry);
}



/****************************************************************************
** 	_popBlankPos
**
**	Purpose	: Pop the localtion of a hole from the table's free list
**	Args	: Database and table names
**	Returns	: Offset off hole, NO_POS if the free list is empty
**	Notes	: 
*/


u_int popBlankPos(cacheEntry,db,table)
	cache_t	*cacheEntry;
	char	*db,
		*table;
{
	u_int	pos;
	sblk_t	*sblkPtr;

	msqlTrace(TRACE_IN,"popBlankPos()");
	sblkPtr = (sblk_t *)cacheEntry->dataMap;
	if (!sblkPtr)
		return(NO_POS);
	pos = sblkPtr->freeList;
	if (pos != NO_POS)
	{
		sblkPtr->freeList = readFreeList(cacheEntry,sblkPtr->freeList);
	}
	return(pos);
}


/****************************************************************************
** 	_pushBlankPos
**
**	Purpose	: Store the location of a data hole 
**	Args	: database and table names
**		  offset for the hole.
**	Returns	: -1 on error
**	Notes	: 
*/

int pushBlankPos(cacheEntry,db,table,pos)
	cache_t	*cacheEntry;
	char	*db,
		*table;
	u_int	pos;
{
	sblk_t	 *sblkPtr;

	msqlTrace(TRACE_IN,"pushBlankPos()");

	sblkPtr = (sblk_t *)cacheEntry->dataMap;
	if (sblkPtr->freeList == NO_POS)
	{
		sblkPtr->freeList = pos;
	}
	else
	{
		if (writeFreeList(cacheEntry,sblkPtr->freeList,pos) < 0)
		{
			return(-1);
		}
	}
	if (writeFreeList(cacheEntry,pos,NO_POS) < 0)
	{
		return(-1);
	}
	return(0);
}







/****************************************************************************
** 	_freeTableDef
**
**	Purpose	: Free memory used by a table def
**	Args	: pointer to table def
**	Returns	: Nothing
**	Notes	: 
*/

void freeTableDef(tableDef)
	field_t	*tableDef;
{
	field_t	*curField,
		*prevField;

	msqlTrace(TRACE_IN,"freeTableDef()");
	curField = tableDef;
	while(curField)
	{
		if (curField->value)
		{
			msqlFreeValue(curField->value);
		}
		curField->value = NULL;
		prevField = curField;
		curField = curField->next;
		prevField->next = NULL;
		safeFree(prevField);
	}
	msqlTrace(TRACE_OUT,"freeTableDef()");
}



/****************************************************************************
** 	_readTableDef
**
**	Purpose	: Load a table definition from file
**	Args	: Database and Table names
**	Returns	: pointer to table definition
**	Notes	: 
*/

static field_t *readTableDef(table,alias,db)
	char	*table,
		*alias,
		*db;
{
	field_t	*headField,
		*tmpField,
		*prevField,
		*curField;
	char	path[MAXPATHLEN];
	int	numFields,
		numBytes,
		fieldCount,
		fd;
	static	char buf[MAX_FIELDS * sizeof(field_t)];

	msqlTrace(TRACE_IN,"readTableDef()");
	(void)sprintf(path,"%s/msqldb/%s/%s.def",msqlHomeDir,db,table);
#ifdef OS2
        fd = _sopen(path, O_RDONLY | O_BINARY, SH_DENYNO, S_IREAD | S_IWRITE);
#else
	fd = open(path,O_RDONLY,0);
#endif
	if (fd < 0)
	{
		sprintf(errMsg,BAD_TABLE_ERROR,table);
		msqlDebug(MOD_ERR,"Unknown table \"%s\"\n",table);
		msqlTrace(TRACE_OUT,"readTableDef()");
		return(NULL);
	}
	numBytes = read(fd,buf,sizeof(buf));
	if (numBytes < 1)
	{
		sprintf(errMsg,TABLE_READ_ERROR,table);
		msqlDebug(MOD_ERR,
			"Error reading table \"%s\" definition\n",
			table);
		close(fd);
		msqlTrace(TRACE_OUT,"readTableDef()");
		return(NULL);
	}
		
	numFields = numBytes / sizeof(field_t);
	fieldCount = 0;
	headField = NULL;
	while(fieldCount < numFields)
	{
		tmpField = (field_t *)(buf + (fieldCount * sizeof(field_t)));
		curField = (field_t *)fastMalloc(sizeof(field_t));
		if (!headField)
		{
			headField = prevField = curField;
		}
		else
		{
			prevField->next = curField;
			prevField = curField;
		}
		inline_bcopy(tmpField, curField, sizeof(field_t));
		if (alias)
		{
			strcpy(curField->table,alias);
		}
		curField->value = NULL;
		fieldCount++;
	}
	close(fd);
	msqlTrace(TRACE_OUT,"readTableDef()");
	return(headField);
}




void cleanTmpDir()
{
#ifdef HAVE_DIRENT
        struct  dirent *cur;
#else
        struct  direct *cur;
#endif
	DIR	*dirp;
	char	path[MAXPATHLEN];

	(void)sprintf(path,"%s/msqldb/.tmp",msqlHomeDir);
	dirp = opendir(path);
	if (!dirp)
	{
		/* Can't report this.  It'll be trapped later */
		return;
        }

        /*
        ** Blow away any files skipping . and ..
        */

	cur = readdir(dirp);
	while(cur)
	{
		if (*cur->d_name == '.')
		{
			cur = readdir(dirp);
			continue;
		}
		(void)sprintf(path,"%s/msqldb/.tmp/%s",msqlHomeDir,
			cur->d_name);
		unlink(path);
		cur = readdir(dirp);
	}
	closedir(dirp);
        return;
}



cache_t *createTmpTable(table1,table2,fields)
	cache_t	*table1,
		*table2;
	field_t	*fields;
{
	REG	cache_t *new;
	field_t	*curField,
		*newField,
		*tmpField;
	char	path[MAXPATHLEN],
		*tmpfile,
		*tmpStr,
		*cp;
	int	fd,
		foundField,
		curOffset = 0;
	sblk_t	sblk;


	/*
	** Create a name for this tmp table
	*/
	msqlTrace(TRACE_IN,"createTmpTable()");
	tmpStr = tmpfile = (char *)msql_tmpnam(NULL);
#ifdef OS2
        cp = (char *)rindex(tmpfile,'\\');
        if (cp)
        {
                tmpfile = cp+1;
        }
#endif
	cp = (char *)rindex(tmpfile,'/');
	if (cp)
	{
		tmpfile = cp+1;
	}
	(void)sprintf(path,"%s/msqldb/.tmp/%s.dat",msqlHomeDir,tmpfile);


	/*
	** start building the table cache entry
	*/
	new = (cache_t *)malloc(sizeof(cache_t));
	if (!new)
	{
		sprintf(errMsg,TMP_MEM_ERROR);
		msqlDebug(MOD_ERR,"Out of memory for temporary table (%s)\n"
		,path);
		msqlTrace(TRACE_OUT,"createTmpTable()");
		free(tmpStr);
		return(NULL);
	}
	(void)strcpy(new->table,tmpfile);
#ifdef OS2
        fd = _sopen(path, O_RDWR | O_CREAT | O_TRUNC | O_BINARY, SH_DENYNO, 
		S_IREAD | S_IWRITE);
#else
	fd = open(path,O_RDWR|O_CREAT|O_TRUNC, 0700);
#endif
	if (fd < 0)
	{
		sprintf(errMsg,TMP_CREATE_ERROR);
		msqlDebug(MOD_ERR,"Couldn't create temporary table (%s)\n",
			path);
		(void)free(new);
		msqlTrace(TRACE_OUT,"createTmpTable()");
		free(tmpStr);
		close(fd);
		return(NULL);
	}
	sblk.version = DB_VERSION;
	sblk.numRows = 0;
	sblk.freeList = NO_POS;
	if (write(fd, &sblk, SBLK_SIZE) < SBLK_SIZE)
	{
		sprintf(errMsg,TMP_CREATE_ERROR);
		msqlDebug(MOD_ERR,"Couldn't create temporary table (%s)\n",
			path);
		(void)free(new);
		msqlTrace(TRACE_OUT,"createTmpTable()");
		free(tmpStr);
		return(NULL);
	}
	new->dataFD = fd;
	new->result = 1;
	new->size = SBLK_SIZE;
	zeroIndices(new);

	/*
	** Map the tmp table and setup the superblock
	*/
	new->dataMap = (caddr_t)mmap(NULL, (size_t)new->size,
		(PROT_READ|PROT_WRITE), MAP_SHARED, new->dataFD, (off_t)0);
#ifdef HAVE_MADVISE
	madvise(new->dataMap, new->size, MADV_SEQUENTIAL);
#endif

	if (new->dataMap == (caddr_t)-1)
		return(NULL);
	new->sblk=(sblk_t*)new->dataMap;


	/*
	** Add the field definitions.  Ensure that any key fields are
	** not flagged as such as we can't access the key data of the
	** original table when accessing this.
	*/

	curField = table1->def;
	newField = NULL;
	while(curField)
	{
		/*
		** If we've been given a list of fields, only add this
		** field to the tmp table if it's in the list.
		*/
		if (fields)
		{
			foundField = 0;
			tmpField = fields;
			while(tmpField)
			{
				if(strcmp(tmpField->name,curField->name)==0 &&
				   strcmp(tmpField->table,curField->table)==0)
				{
					foundField = 1;
					break;
				}
				tmpField = tmpField->next;
			}
			if (!foundField)
			{
				curField = curField->next;
				continue;
			}
		}

		/*
		** O.k.  Add this field
		*/
		if (newField)
		{
			newField->next = (field_t *)fastMalloc(sizeof(field_t));
			newField = newField->next;
		}
		else
		{
			new->def = (field_t *)fastMalloc(sizeof(field_t));
			newField = new->def;
		}
		inline_bcopy(curField,newField,sizeof(field_t));
		if( *(newField->table) == 0)
		{
			(void)strcpy(newField->table,table1->table);
		}
		/* newField->flags=0; */
		new->rowLen += curField->dataLength + 1;
		newField->offset = curOffset;
		curOffset += curField->dataLength + 1;
		newField->value = NULL;
		newField->entry = table1;
		curField = curField->next;
	}
	if (table2)
	{
		curField = table2->def;
		while(curField)
		{
			/*
			** If we've been given a list of fields, only add this
			** field to the tmp table if it's in the list.
			*/
			if (fields)
			{
				foundField = 0;
				tmpField = fields;
				while(tmpField)
				{
					if(strcmp(tmpField->name,
						curField->name)==0 &&
				   	strcmp(tmpField->table,
						curField->table)==0)
					{
						foundField = 1;
						break;
					}
					tmpField = tmpField->next;
				}
				if (!foundField)
				{
					curField = curField->next;
					continue;
				}
			}

			/*
			** Add it.
			*/
			if (newField)
			{
				newField->next = (field_t *)fastMalloc(
					sizeof(field_t));
				newField = newField->next;
			}
			else
			{
				new->def=newField=(field_t *)fastMalloc(
					sizeof(field_t));
			}
			inline_bcopy(curField,newField,sizeof(field_t));
			if( *(newField->table) == 0)
			{
				(void)strcpy(newField->table,table2->table);
			}
			new->rowLen += curField->dataLength + 1;
			newField->offset = curOffset;
			newField->value = NULL;
			newField->entry = table2;
			curOffset += curField->dataLength + 1;
			curField = curField->next;
		}
	}

	if (newField)	
	{
		newField->next = NULL;
	}
	new->row.buf = (u_char *)malloc(new->rowLen + HEADER_SIZE + 2);
	new->row.header = (hdr_t *)new->row.buf;
	new->row.data = new->row.buf + HEADER_SIZE;
	msqlTrace(TRACE_OUT,"createTmpTable()");
	free(tmpStr);
	return(new);
}




void freeTmpTable(entry)
	cache_t	*entry;
{
	char	path[MAXPATHLEN];

        msqlTrace(TRACE_IN,"freeTmpTable()");
	(void)sprintf(path,"%s/msqldb/.tmp/%s.dat",msqlHomeDir,entry->table);
	freeTableDef(entry->def);
	entry->def = NULL;
	*(entry->db) = 0;
	*(entry->table) = 0;
	*(entry->cname) = 0;
	entry->age = 0;
	safeFree(entry->row.buf);
	if (entry->dataMap != (caddr_t) NULL)
	{
		munmap(entry->dataMap,entry->size);
		entry->dataMap = NULL;
		entry->size = 0;
	}
	close(entry->dataFD);
	closeIndices(entry);
	(void)free(entry);
	unlink(path);
        msqlTrace(TRACE_OUT,"freeTmpTable()");
}



/****************************************************************************
** 	_findRowLen
**
**	Purpose	: Determine the on-disk size of a table's rows
**	Args	: None
**	Returns	: Row Length
**	Notes	: Uses global table definition pointer.
*/

static int findRowLen(cacheEntry)
	cache_t	*cacheEntry;
{
	int	rowLen;
	field_t	*fieldDef;

	rowLen = 0;
	fieldDef = cacheEntry->def;
	while(fieldDef)
	{
		rowLen += fieldDef->dataLength +1;  /* +1 for NULL indicator */
		fieldDef = fieldDef->next;
	}
	return(rowLen);
}




/****************************************************************************
** 	_loadTableDef
**
**	Purpose	: Locate a table definition
**	Args	: Database and Table names
**	Returns	: -1 on error
**	Notes	: Table description cache searched first.  If it's not
**		  there, the LRU entry is freed and the table def is
**		  loaded into the cache.  The tableDef, 
**		  cacheEntry and dataFD globals are set.
*/


cache_t *loadTableDef(table,cname,db)
	char	*table,
		*cname,
		*db;
{
	int	maxAge,
		cacheIndex;
	field_t	*def,
		*curField;
	REG 	cache_t	*entry;
	REG 	int	count;
	char	 *tableName;


	/*
	** Look for the entry in the cache.  Keep track of the oldest
	** entry during the pass so that we can replace it if needed
	*/
	msqlTrace(TRACE_IN,"loadTableDef()");
	msqlDebug(MOD_CACHE,"Table cache search for %s:%s\n",table,db);
	count = cacheIndex = 0;
	maxAge = -1;
	if (cname)
	{
		if (!*cname)
		{
			cname = NULL;
		}
	}
	while(count < CACHE_SIZE)
	{
		entry = tableCache + count;
		msqlDebug(MOD_CACHE,"Cache entry %d = %s:%s, age = %d (%s)\n", 
			count,
			entry->table?entry->table:"NULL",
			entry->db?entry->db:"NULL", 
			entry->age,
			entry->def?"OK":"NULL Def!!!");
		if (entry->age > 0 &&
		    strcmp(entry->db,db)==0 && 
		    strcmp(entry->table,table)==0 &&
		    strcmp((cname)?cname:"",entry->cname)==0)
		{
			msqlDebug(MOD_CACHE,"Found cache entry at %d\n", count);
			entry->age = 1;
			msqlTrace(TRACE_OUT,"loadTableDef()");
			return(entry);
		}
		if (entry->age > 0)
			entry->age++;

		/*
		** Empty entries have an age of 0.  If we're marking
		** an empty cache position just keep the mark
		*/
		if ((entry->age == 0) && (maxAge != 0))
		{
			maxAge = entry->age;
			cacheIndex = count;
		}
		else
		{
			if ((entry->age > maxAge) && (maxAge != 0))
			{
				maxAge = entry->age;
				cacheIndex = count;
			}
		}
		count++;
	}

	/*
	** It wasn't in the cache.  Free up the oldest cache entry 
	*/

	entry = tableCache + cacheIndex;
	if(entry->def)
	{
		msqlDebug(MOD_CACHE,"Removing cache entry %d (%s:%s)\n", 
			cacheIndex, entry->db, entry->table);
		if (entry->dataMap != (caddr_t) NULL)
		{
			munmap(entry->dataMap,entry->size);
			entry->dataMap = NULL;
			entry->size = 0;
		}
		if (entry->overflowMap != (caddr_t) NULL)
		{
			munmap(entry->overflowMap,entry->overflowSize);
			entry->overflowMap = NULL;
			entry->size = 0;
		}
		(void)close(entry->dataFD);
		(void)close(entry->overflowFD);
		closeIndices(entry);
		freeTableDef(entry->def);
		entry->def = NULL;
		*(entry->table)=0;
		*(entry->db)=0;
		entry->age=0;
	}
	safeFree(entry->row.buf);

	/*
	** Now load the new entry
	*/
	if (cname)
	{
		tableName = cname;
		def = readTableDef(cname,table,db);
	}
	else
	{
		tableName = table;
		def = readTableDef(table,NULL,db);
	}
	if (!def)
	{
		sprintf(errMsg,TABLE_READ_ERROR,table);
		msqlDebug(MOD_ERR,"Couldn't read table definition for %s\n",table);
		msqlTrace(TRACE_OUT,"loadTableDef()");
		return(NULL);
	}
	entry->indices = loadIndices(tableName,db);
	entry->def = def;
	entry->age = 1;
	entry->result = 0;
	zeroIndices(entry);
	strcpy(entry->db,db);
	strcpy(entry->table,table);
	if (cname)
	{
		strcpy(entry->cname,cname);
	}
	else
	{
		*(entry->cname) = 0;
	}
	
	msqlDebug(MOD_CACHE,"Loading cache entry %d (%s:%s)\n", cacheIndex, 
		entry->db, entry->table);
	if((entry->dataFD = openTable(tableName,db)) < 0)
	{
		closeIndices(entry);
		freeTableDef(entry->def);
		entry->def = NULL;
		*(entry->table)=0;
		*(entry->db)=0;
		entry->age=0;
		sprintf(errMsg,DATA_OPEN_ERROR,tableName);
		msqlTrace(TRACE_OUT,"loadTableDef()");
		return(NULL);
	}
	if((entry->overflowFD = openOverflow(tableName,db)) < 0)
	{
		close(entry->dataFD);
		closeIndices(entry);
		freeTableDef(entry->def);
		entry->def = NULL;
		*(entry->table)=0;
		*(entry->db)=0;
		entry->age=0;
		sprintf(errMsg,DATA_OPEN_ERROR,tableName);
		msqlTrace(TRACE_OUT,"loadTableDef()");
		return(NULL);
	}
	curField = entry->def;
	while(curField)
	{
		curField->entry = entry;
		curField = curField->next;
	}

	/*
	** Setup for Mapping the data file
	*/
	entry->dataMap = NULL;
	entry->remapData = 1;
	entry->overflowMap = NULL;
	entry->remapOverflow = 1;
	initTable(entry,FULL_REMAP);

	/*
	** Set the globals and bail.  We need rowLen + 2 (one for the
	** active byte and also one for regexp over-run protection) and
	** keyLen + 1 (one for the active byte) buffers for performance.
	** Ensure we have a rowLen of at least 4 bytes.  We use the
	** first 4 bytes for the free list entries.
	*/
	entry->rowLen = findRowLen(entry);
	if (entry->rowLen < 4)
		entry->rowLen = 4;
	entry->row.buf = (u_char *)malloc(entry->rowLen + HEADER_SIZE + 2);
	entry->row.header = (hdr_t *)entry->row.buf;
        entry->row.data = entry->row.buf + HEADER_SIZE;
	entry->sblk = (sblk_t *)entry->dataMap;
	msqlTrace(TRACE_OUT,"loadTableDef()");
	return(entry);
}







/****************************************************************************
** 	_initTable
**
**	Purpose	: Reset table pointers used during query processing
**	Args	: None
**	Returns	: Nothing
**	Notes	: This just puts the file into a known state, particular
**		  the current seek pointers.
*/

int initTable(cacheEntry,mapFlag)
	cache_t	*cacheEntry;
	int	mapFlag;
{
	struct	stat sbuf;

	msqlTrace(TRACE_IN,"initTable()");
	if (mapFlag && FULL_REMAP)
	{
		if (cacheEntry->remapData)
		{
			fstat(cacheEntry->dataFD, &sbuf);
			cacheEntry->size = sbuf.st_size;
			if (cacheEntry->size)
			{
				cacheEntry->dataMap = (caddr_t)mmap(NULL, 
					(size_t)cacheEntry->size, 
					(PROT_READ | PROT_WRITE), 
					MAP_SHARED, cacheEntry->dataFD, 
					(off_t)0);
				if (cacheEntry->dataMap == (caddr_t)-1)
					return(-1);
#ifdef HAVE_MADVISE
				madvise(cacheEntry->dataMap, cacheEntry->size, 
					MADV_SEQUENTIAL);
#endif
				cacheEntry->sblk=(sblk_t*)cacheEntry->dataMap;
			}
			cacheEntry->remapData = 0;
		}
		if (cacheEntry->remapOverflow)
		{
			if (cacheEntry->overflowMap)
			{
				munmap(cacheEntry->overflowMap,
					cacheEntry->overflowSize);
			}
			fstat(cacheEntry->overflowFD, &sbuf);
			cacheEntry->overflowSize = sbuf.st_size;
			if (cacheEntry->overflowSize)
			{
				cacheEntry->overflowMap = (caddr_t)mmap(NULL, 
					(size_t)cacheEntry->overflowSize, 
					(PROT_READ | PROT_WRITE), 
					MAP_SHARED, cacheEntry->overflowFD, 
					(off_t)0);
				if (cacheEntry->overflowMap == (caddr_t)-1)
					return(-1);
#ifdef HAVE_MADVISE
				madvise(cacheEntry->overflowMap, 
					cacheEntry->overflowSize, 
					MADV_SEQUENTIAL);
#endif
			}
			cacheEntry->remapOverflow = 0;
		}
	}
	msqlTrace(TRACE_OUT,"initTable()");
	return(0);
}

	




/****************************************************************************
** 	_rowWrite
**
**	Purpose	: Store a table row in the database
**	Args	: datafile FD, row data, length of data, target offset
**	Returns	: -1 on error
**	Notes	: If the rowNum is NO_POS then append the row
*/


static int __rowWrite(cacheEntry,row,rowNum, active)
	cache_t	*cacheEntry;
	row_t	*row;
	u_int	rowNum;
	int	active;
{
	REG	off_t	seekPos;
	char	*buf;
	struct stat sbuf;
	extern	time_t queryTime;


	if (rowNum == NO_POS)
	{
		msqlDebug(MOD_ACCESS,"rowWrite() : append to %s\n",
			(cacheEntry->result)?cacheEntry->resInfo:
			cacheEntry->table);

	}
	else
	{
		msqlDebug(MOD_ACCESS,"rowWrite() : write at row %u of %s\n",
			rowNum, (cacheEntry->result)?cacheEntry->resInfo:
			cacheEntry->table);
	}


	/*
	** Write the row to the table
	*/
	if (rowNum == NO_POS)  /* append */
	{
		seekPos = cacheEntry->sblk->numRows * 
			(cacheEntry->rowLen + HEADER_SIZE) + SBLK_SIZE;
		cacheEntry->sblk->numRows++;
	}
	else
	{
		seekPos = rowNum * (cacheEntry->rowLen + HEADER_SIZE) +
			SBLK_SIZE;
	}

	if (seekPos + cacheEntry->rowLen + HEADER_SIZE >= cacheEntry->size)
	{
               	munmap(cacheEntry->dataMap, cacheEntry->size);
		cacheEntry->dataMap = NULL;
		cacheEntry->size = 0;
		lseek(cacheEntry->dataFD, seekPos + 
			(48 * cacheEntry->rowLen + HEADER_SIZE),
			SEEK_SET);
		write(cacheEntry->dataFD, "\0", 1);
		fstat(cacheEntry->dataFD, &sbuf);
		cacheEntry->size = sbuf.st_size;
		if (cacheEntry->size)
		{
			cacheEntry->dataMap = (caddr_t)mmap(NULL, 
				(size_t)cacheEntry->size,(PROT_READ|PROT_WRITE),
				MAP_SHARED, cacheEntry->dataFD, (off_t)0);
			if (cacheEntry->dataMap == (caddr_t)-1)
					return(-1);
#ifdef HAVE_MADVISE
			madvise(cacheEntry->dataMap, cacheEntry->size, 
					MADV_SEQUENTIAL);
#endif
			cacheEntry->sblk=(sblk_t*)cacheEntry->dataMap;
		}
	}
	buf = ((char *)cacheEntry->dataMap) + seekPos;
	if (row)
	{
		if (active)
			row->header->active = 1;
		bcopy(&queryTime,&(row->header->timestamp),sizeof(queryTime));
		inline_bcopy(row->buf, buf, HEADER_SIZE +
			cacheEntry->rowLen);
	}
	else
	{
		if (active)
			cacheEntry->row.header->active = 1;
		bcopy(&queryTime,&(cacheEntry->row.header->timestamp),
			sizeof(queryTime));
		inline_bcopy(cacheEntry->row.buf, buf,
			cacheEntry->rowLen + HEADER_SIZE);
	}
	return(0);
}



int rowWrite(cacheEntry,row,rowNum)
	cache_t	*cacheEntry;
	row_t	*row;
	u_int	rowNum;
{
	return(__rowWrite(cacheEntry,row,rowNum, 1));
}


int rowPlace(cacheEntry,row,rowNum)
	cache_t	*cacheEntry;
	row_t	*row;
	u_int	rowNum;
{
	return(__rowWrite(cacheEntry,row,rowNum, 0));
}


/****************************************************************************
** 	_rowRead
**
**	Purpose	: Grab a row from a datafile
**	Args	: datafile FD, length the row, pointer to active flag buf
**	Returns	: pointer to static row buffer
**	Notes	: Some boxes (e.g. early SPARCs etc) don't do multiply
**		  in hardware.  As rowRead() is the most called function
**		  in the entire code we do a hack to reduce the hit
**		  on boxes without hardware math.
*/

int rowRead(cacheEntry,row,rowNum)
	cache_t	*cacheEntry;
	row_t	*row;
	u_int	rowNum;
{
	REG 	off_t	seekPos;
	static	u_int	lastRow,
			lastLen;
	static	off_t	lastPos;


	if (rowNum == lastRow + 1 && lastLen == cacheEntry->rowLen)
	{
		seekPos = lastPos + cacheEntry->rowLen + HEADER_SIZE;
	}
	else
	{
		seekPos = (rowNum * (cacheEntry->rowLen + HEADER_SIZE)) + 
			SBLK_SIZE;
	}
	lastPos = seekPos;
	lastRow = rowNum;
	lastLen = cacheEntry->rowLen;
	if ((seekPos >= cacheEntry->size) || (!cacheEntry->dataMap))
	{
		msqlDebug(MOD_ACCESS,"rowRead() : %u of %s - No Such Row \n",
			rowNum, (cacheEntry->result)?cacheEntry->resInfo:
			cacheEntry->table);
		row->header->active = 0;
		return(-1);
	}
	row->buf = ((u_char *)cacheEntry->dataMap) + seekPos;
	row->header = (hdr_t *)row->buf;
	row->data = row->buf + HEADER_SIZE;
	row->rowID = rowNum;
	msqlDebug(MOD_ACCESS,"rowRead() : %u of %s - %s\n",
		rowNum, (cacheEntry->result)?cacheEntry->resInfo:
		cacheEntry->table,(row->header->active)?"Active":"Inactive");
	return(1);
}




/****************************************************************************
** 	_deleteRow
**
**	Purpose	: Invalidate a row in the table
**	Args	: datafile FD, rowlength, desired row location
**	Returns	: -1 on error
**	Notes	: This just sets the row header byte to 0 indicating
**		  that it's no longer in use 
*/

int deleteRow(cacheEntry,rowNum)
	cache_t	*cacheEntry;
	u_int	rowNum;
{
	int	rowLen;
	hdr_t	*hdrPtr;

	msqlTrace(TRACE_IN,"deleteRow()");
	msqlDebug(MOD_ACCESS,"deleteRow() : row %u of %s\n",
		rowNum, (cacheEntry->result)?cacheEntry->resInfo:
		cacheEntry->table);
	rowLen = cacheEntry->rowLen;


	hdrPtr = (hdr_t *)((char *)cacheEntry->dataMap + 
		(rowNum * (rowLen + HEADER_SIZE) + SBLK_SIZE));
	hdrPtr->active = 0;
	msqlTrace(TRACE_OUT,"deleteRow()");
	return(0);
}






/****************************************************************************
** 	_fillRow
**
**	Purpose	: Create a new row-buf using the info given
**	Args	: 
**	Returns	: 
**	Notes	: 
*/


int fillRow(entry,row,fields,flist)
	cache_t	*entry;
	row_t	*row;
	field_t	*fields;
	int	flist[];
{
	int	*offset,
		length;
	field_t	*curField;
	u_char	*data,
		*cp;
	u_int	overflow;

	msqlTrace(TRACE_IN,"fillRow()");
	data = row->data;
	curField = fields;
	offset = flist;
	while(curField)
	{
		if (!curField->value->nullVal)
		{
			cp = data + *offset;
			*cp = '\001';
			cp++;
			switch(curField->type)
			{
				case INT_TYPE:
#ifndef _CRAY
					bcopy4(&(curField->value->val.intVal),
						cp);
#else
					packInt32(curField->value->val.intVal,
						cp);
#endif
					break;
		
				case CHAR_TYPE:
					length=strlen((char *)
						curField->value->val.charVal);
					if (length > curField->length)
					{
						sprintf(errMsg,VALUE_SIZE_ERROR,
							curField->name);
						return(-1);
					}
					inline_bcopy(
						curField->value->val.charVal,cp,
						length);
					break;

				case TEXT_TYPE:
					length=strlen((char *)
						curField->value->val.charVal);
					if (curField->overflow != NO_POS)
					{
						overflow = curField->overflow;
					}
					else
					{
						overflow = NO_POS;
						if (length > curField->length)
					    	  overflow = writeVarChar(entry,
						  curField->value->val.charVal,
						  curField->length);
					}
					bcopy(&length,cp,sizeof(int));
					cp += sizeof(int);
					bcopy(&overflow,cp,sizeof(u_int));
					cp += sizeof(u_int);
					inline_bcopy(
						curField->value->val.charVal,
						cp, curField->length);
					break;

				case REAL_TYPE:
					bcopy8(&(curField->value->val.realVal),
						cp);
					break;
			}
		}
		offset++;
		curField = curField->next;
	}
	msqlTrace(TRACE_OUT,"fillRow()");
	return(0);
}






/****************************************************************************
** 	_updateValues
**
**	Purpose	: Modify a row-buf to reflect the contents of the field list
**	Args	: 
**	Returns	: 
**	Notes	: 
*/

int updateValues(entry,row,fields,flist)
	cache_t	*entry;
	row_t	*row;
	field_t	*fields;
	int	flist[];
{
	int	*offset,
		length;
	field_t	*curField;
	u_char	*data,
		*cp;
	u_int	pos;

	msqlTrace(TRACE_IN,"updateValues()");
	data = row->data;
	curField = fields;
	offset = flist;
	while(curField)
	{
		cp = data + *offset;
		if (!curField->value->nullVal)
                {
                        *cp = '\001';
                        cp++;
			switch(curField->type)
			{
				case INT_TYPE:
#ifndef _CRAY
					bcopy4(&(curField->value->val.intVal),
						cp);
#else
				 	packInt32(curField->value->val.intVal, 
						cp);
#endif
					break;
		
				case CHAR_TYPE:
					length = strlen((char *) 
						curField->value->val.charVal);
                                        if (length > curField->length)
                                        {
						sprintf(errMsg,VALUE_SIZE_ERROR,
                                                        curField->name);
                                               return(-1);
                                        }
					strncpy(cp,curField->value->val.charVal,
						length);
					if (length < curField->length)
						*(cp + length) = 0;
					break;

				case TEXT_TYPE:
					bcopy(cp + sizeof(int), &pos,
						sizeof(u_int));
					deleteVarChar(entry, pos);

					length=strlen((char *)
						curField->value->val.charVal);
					pos = NO_POS;
					if (length > curField->length)
					    pos = writeVarChar(entry,
						curField->value->val.charVal,
						curField->length);
					bcopy(&length,cp,sizeof(int));
					cp += sizeof(int);
					bcopy(&pos,cp,sizeof(u_int));
					cp += sizeof(u_int);
					inline_bcopy(
						curField->value->val.charVal,
						cp, curField->length);
					break;

				case REAL_TYPE:
					bcopy8(&(curField->value->val.realVal),
						cp);
					break;
			}
		}
		else
		{
			*cp = '\000';
		}
		offset++;
		curField = curField->next;
	}
	msqlTrace(TRACE_OUT,"updateValues()");
	return(0);
}





/****************************************************************************
** 	_extractValues
**
**	Purpose	: Rip the required data from a row-buf
**	Args	: 
**	Returns	: 
**	Notes	: 
*/

void extractValues(entry,row,fields,flist)
	cache_t	*entry;
	row_t	*row;
	field_t	*fields;
	int	flist[];
{
	field_t	*curField;
	u_char	*cp,
		*data;
	int	ip,
		*offset;
	double	*fp;
	char	buf[8];

	msqlTrace(TRACE_IN,"extractValues()");
	if (row)
		data = row->data;
	curField = fields;
	offset = flist;
	while(curField)
	{
		if (curField->value)
		{
			msqlFreeValue(curField->value);
			curField->value = NULL;
		}
		if (curField->sysvar)
		{
			getSysVar(entry,row,curField);
			curField = curField->next;
			continue;
		}
		if ( * (data + *offset)) 
		{
			curField->value=NULL;
			switch(curField->type)
			{
				case INT_TYPE:
#ifndef _CRAY
					bcopy4(data + *offset + 1,&ip);
					curField->value =(val_t *)
						fillValue((char *)&ip,INT_TYPE, 
							4);
#else
					curField->value = (val_t*)fillValue(
						data + *offset + 1, INT_TYPE,4);
#endif
					break;

				case CHAR_TYPE:
					cp = (u_char *)data + *offset + 1;
					curField->value = (val_t *)
						fillValue((char *)cp, CHAR_TYPE,
						curField->length);
					break;

				case TEXT_TYPE:
					cp = (u_char *)data + *offset + 1;
					curField->value = (val_t *)
						malloc(sizeof(val_t));
					bcopy(cp + sizeof(u_int), 
						&(curField->overflow),
						sizeof(u_int));
					curField->value->type = TEXT_TYPE;
        				curField->value->nullVal = 0;
					curField->value->val.charVal = 
						readVarChar(curField->entry, cp,
						curField->length);
					break;

				case REAL_TYPE:
					bcopy8(data + *offset + 1,buf);
					fp = (double *)buf;
					curField->value =(val_t *)
						fillValue((char *)fp,REAL_TYPE,
							8);
					break;
			}
		} 
		else 
		{
			curField->value = (val_t *)nullValue();
		}
		curField = curField->next;
		offset++;
	}
	msqlTrace(TRACE_OUT,"extractValues()");
}