firebird-mirror/src/journal/rebuild.epp

/*
 *	PROGRAM:	JRD Journalling Subsystem
 *	MODULE:		rebuild.epp
 *	DESCRIPTION:
 *
 * The contents of this file are subject to the Interbase Public
 * License Version 1.0 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy
 * of the License at http://www.Inprise.com/IPL.html
 *
 * Software distributed under the License is distributed on an
 * "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either express
 * or implied. See the License for the specific language governing
 * rights and limitations under the License.
 *
 * The Original Code was created by Inprise Corporation
 * and its predecessors. Portions created by Inprise Corporation are
 * Copyright (C) Inprise Corporation.
 *
 * All Rights Reserved.
 * Contributor(s): ______________________________________.
 *
 * 2002.10.29 Sean Leyne - Removed obsolete "Netware" port
 *
 */

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "../jrd/jrd_time.h"

// MOVE_FAST & MOVE_CLEAR have to be defined before #include common.h

#define MOVE_FAST(from,to,length)	memcpy (to, from, (int) (length))
#define MOVE_CLEAR(to,length)		memset (to, 0, (int) (length))

#ifndef MAX_PATH_LENGTH
#define MAX_PATH_LENGTH			512
#endif

#include "firebird.h"
#include "../jrd/ib_stdio.h"
#include "../jrd/common.h"

#include "../jrd/ods.h"
#include "../jrd/y_ref.h"
#include "../jrd/ibase.h"
#include "../jrd/license.h"
#include "../jrd/jrn.h"
#include "../journal/journal.h"
#include "../jrd/thd.h"
#include "../jrd/os/pio.h"
#include "../wal/wal.h"
#include "../jrd/jrd.h"
#include "../jrd/dsc.h"
#include "../jrd/exe.h"
#include "../jrd/btr.h"
#include "../jrd/old.h"
#include "../jrd/llio.h"
#include "../wal/walr_proto.h"
#include "../journal/conso_proto.h"
#include "../journal/gjrn_proto.h"
#include "../jrd/misc_proto.h"
#include "../journal/miscj_proto.h"
#include "../journal/oldr_proto.h"
#include "../journal/rebui_proto.h"
#include "../jrd/gds_proto.h"
#include "../jrd/isc_f_proto.h"
#include "../jrd/llio_proto.h"

// UNIX Stuff

#ifdef UNIX
#define UNIX_JOURNALLING
#define LIBRARY_IO
#define SYS_ERROR	isc_arg_unix
#define ERRNO		errno
#include <sys/types.h>
extern int errno;
#endif

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif

// VMS Stuff

#ifdef VMS
#include <rms.h>
#include <iodef.h>
#include <descrip.h>
#include <ssdef.h>
#define SYS_ERROR	isc_arg_vms
#endif

// Windows NT Stuff

#ifdef WIN_NT
#include <sys/types.h>
#include <sys/timeb.h>
#include <winsock2.h>
#define SYS_ERROR	isc_arg_win32
#define ERRNO		GetLastError()
#endif

#define HIGH_WATER(x)		((int) &((DPG) NULL)->dpg_rpt [x])
#define MOVE_BYTE(x_from,x_to)	*x_to++ = *x_from++;

DATABASE DB =
	STATIC COMPILETIME FILENAME "journal.fdb" RUNTIME FILENAME journal_dir;
DATABASE DB_NEW =
	STATIC COMPILETIME FILENAME "journal.fdb" RUNTIME FILENAME db_name;

// Page cache

typedef struct cache {
	cache* cache_next;
	SSHORT cache_flags;
	SSHORT cache_use_count;
	SLONG cache_page_number;
	ULONG cache_age;
	PAG cache_page;
} *CACHE;

// cache_flags

#define	PAGE_CLEAN	0
#define	PAGE_DIRTY	1
#define	PAGE_HEADER	2
#define PAGE_CORRUPT	-1

#define	MARK_PAGE(cch)		cch->cache_flags |= PAGE_DIRTY
#define MARK_HEADER(cch)	cch->cache_flags |= PAGE_DIRTY | PAGE_HEADER

// Database recovery block

#define CACHE_SIZE	100
#define INITIAL_ALLOC	16

typedef struct drb {
	drb* drb_next;				// Next database in chain
	USHORT drb_page_size;		// Database page size
	fil* drb_file;				// list of file pointers
	ULONG drb_max_page;			// max page read
	ULONG drb_age;
	UCHAR *drb_buffers;
	CACHE drb_cache;
	TEXT drb_filename[1];
} *DRB;

extern FILE *msg_file;

static USHORT add_file(DRB, FIL, UCHAR*, SSHORT, SLONG, bool);
static void add_time(timeval*, timeval*);
static void apply_data(DRB, DPG, JRND *);
static void apply_header(HDR, JRND *);
static void apply_ids(PPG, JRND *);
static void apply_index(BTR, JRND *);
static void apply_log(LIP, JRND *);
static void apply_pip(PIP, JRND *);
static void apply_pointer(PPG, JRND *);
static void apply_root(IRT, JRND *);
static void apply_transaction(TIP, JRND *);
static USHORT checksum(DRB, PAG);
static void close_database(DRB);
static bool commit(DRB, LTJC *, USHORT);
static SSHORT compress(DRB, DPG);
static void disable(DRB, LTJC *);
static int error(const TEXT*, ISC_STATUS, const TEXT*, ISC_STATUS);
static void expand_num_alloc(SCHAR ***, SSHORT *);
static void fixup_header(DRB);
static void format_time(SLONG[2], TEXT*);
static CACHE get_free_buffer(DRB, SLONG);
static void get_log_files(SLONG, SSHORT, SCHAR ***, SLONG **, SSHORT *,
						  SLONG);
static void get_old_files(SLONG, SCHAR ***, SSHORT *, SLONG *, SLONG *);
static CACHE get_page(DRB, SLONG, bool);
static JRNP *next_clump(JRND *, void *);
static void open_all_files(void);
static DRB open_database(const TEXT*, USHORT, SSHORT);
static bool open_database_file(DRB, const TEXT*, bool, SLONG*);
static int open_journal(const SCHAR*, SCHAR**, SSHORT, SLONG, SLONG*);
static int process_online_dump(const SCHAR*, SCHAR**, SSHORT);
static int process_journal(const SCHAR*, SCHAR**, SSHORT, SLONG, SLONG, SLONG*);
static SLONG process_new_file(DRB, JRNF *);
static bool process_old_record(JRND *, USHORT);
static void process_page(DRB, JRND *, ULONG, ULONG, bool);
static bool process_partial(SLONG, SCHAR *);
static bool process_record(JRNH *, USHORT, ULONG, ULONG, bool);
static void quad_move(register UCHAR *, register UCHAR *);
static void read_page(DRB, CACHE);
static void rebuild_abort(SLONG);
static void rebuild_partial(SLONG, SLONG, SCHAR *);
static void rec_add_clump_entry(DRB, HDR, USHORT, USHORT, UCHAR *);
static bool rec_add_hdr_entry(DRB, USHORT, USHORT, UCHAR *, USHORT);
static bool rec_delete_hdr_entry(DRB, USHORT);
static void rec_find_space(DRB, CACHE *, HDR *, USHORT, USHORT, UCHAR *);
static bool rec_find_type(DRB, CACHE *, HDR *, USHORT, UCHAR **, UCHAR **);
#ifdef NOT_USED_OR_REPLACED
static bool rec_get_hdr_entry(DRB, USHORT, USHORT *, UCHAR *);
#endif
static void rec_restore(SCHAR *, SCHAR *);
static void rec_restore_manual(SCHAR *);
static void release_db(DRB);
static void release_page(DRB, CACHE);
static FIL seek_file(DRB, FIL, CACHE, SLONG *);
static FIL setup_file(const UCHAR*, USHORT, int);
static bool test_partial_db(SLONG);
static void update_rebuild_seqno(SLONG);
static void write_page(DRB, CACHE);

static DRB databases;
static bool sw_verbose, sw_debug, sw_interact, sw_disable;
static bool sw_partial, sw_activate, sw_first_recover;
static bool sw_trace;
static SLONG until[2];
static SCHAR journal_dir[MAX_PATH_LENGTH], db_name[MAX_PATH_LENGTH];
static bool sw_journal = false;
static bool sw_db = false;
static UCHAR sec_file[MAX_PATH_LENGTH];
static bool sec_added = false;
static SSHORT sec_len = 0;
static SSHORT sw_until = 0;
static SCHAR *partial_db;
static SSHORT max_seqno;
static SSHORT dump_id = 0;
static bool end_reached = false;

// Time, size statistics of log records processed

static SLONG elapsed_time;
static SLONG bytes_processed;
static SLONG bytes_applied;
static SLONG et_sec, et_msec;
static SLONG *tr1 = 0;

struct counter {
	SLONG num_recs;
	SLONG num_bytes;
};

static counter totals[JRNP_MAX + 1 - JRN_PAGE];

// WAL defines

static UCHAR wal_buff[MAX_WALBUFLEN];
static WALRS WALR_handle;
static ISC_STATUS_ARRAY wal_status;


#ifndef FILE_OPEN_WRITE
#define FILE_OPEN_WRITE 	"w"
#endif

// if new journal records are added in jrn.h, they should be added here

static char *table[] = {
	"JRN_PAGE                ",	// A full page WAL record
	"JRNP_DATA_SEGMENT       ",	// Add segment to data page
	"JRNP_POINTER_SLOT       ",	// Add pointer page to relation
	"JRNP_TRANSACTION        ",	// Journal transaction state
	"JRNP_NULL               ",	// Odd byte
	"JRNP_PIP                ",	// Page allocation/deallocation
	"JRNP_BTREE_NODE         ",	// Add/delete a node to BTREE
	"JRNP_BTREE_SEGMENT      ",	// B Tree split - valid part
	"JRNP_BTREE_DELETE       ",	// B Tree node delete - logical
	"JRNP_INDEX_ROOT         ",	// Add/drop an index
	"JRNP_DB_HEADER          ",	// Modify db header page
	"JRNP_GENERATOR          ",	// generator		
	"JRNP_ROOT_PAGE          ",	// Index root page	 
	"JRNP_DB_ATTACHMENT      ",	// next attachment
	"JRNP_DB_HDR_PAGES       ",	// Header pages	
	"JRNP_DB_HDR_FLAGS       ",	// Header flags
	"JRNP_DB_HDR_SDW_COUNT   ",	// Header shadow count
	"JRNP_LOG_PAGE           ",	// log page information
	"JRNP_NEXT_TIP           ",	// next tip page
	"JRNP_MAX                "
};


bool REBUILD_start_restore(int argc,
						   char ** argv)
{
/**************************************
 *
 *	R E B U I L D _ s t a r t _ r e s t o r e
 *
 **************************************
 *
 * Functional description
 *	Parse switches and do work.
 *
 **************************************/
	UCHAR *db, string[512];
	UCHAR *journal, jrn[JOURNAL_PATH_LENGTH + 1];
	SSHORT i;
	SCHAR *p;
	SCHAR *msg;

// Start by parsing switches

	sw_trace = sw_disable = sw_interact = false;
	sw_debug = sw_verbose = sw_partial = sw_activate = false;
	sw_first_recover = true;

	until[0] = until[1] = 0;

	for (i = 0; i < (JRNP_MAX - JRN_PAGE); i++) {
		totals[i].num_bytes = 0;
		totals[i].num_recs = 0;
	}

	db = NULL;
	databases = NULL;
	journal = NULL;
	partial_db = NULL;

	argv++;
	while (--argc > 0) {
		if ((*argv)[0] != '-') {
			if (db) {
				GJRN_printf(12, (char*) db, NULL, NULL, NULL);
				exit(FINI_ERROR);
			}
			db = (UCHAR*) *argv++;
			continue;
		}

		MISC_down_case((UCHAR*) *argv++, (UCHAR*) string);
		switch (string[1]) {
		case 'v':
			sw_verbose = true;
			break;

		case 'd':
			sw_verbose = sw_debug = true;
			break;

		case 't':
			sw_trace = true;
			break;

		case 'i':
			sw_interact = true;
			break;

		case 'r':
			break;

		case 'j':
			if (--argc > 0) {
				strcpy((char*) jrn, *argv++);
				journal = jrn;
			}
			else
				MISC_print_journal_syntax();
			break;

		case 'm':
			if (--argc > 0) {
				msg = (SCHAR *) * argv++;
				if (!(msg_file = fopen(msg, FILE_OPEN_WRITE)))
					rebuild_abort(141);
			}
			else
				MISC_print_journal_syntax();
			break;

		case 'a':
			sw_activate = true;
			break;

		case 'p':
			if (--argc > 0) {
				sw_partial = true;
				partial_db = (SCHAR *) * argv++;
			}
			else
				MISC_print_journal_syntax();
			break;

		case 'u':
			sw_until = true;

			if (!(p = (SCHAR *) * argv++))
				rebuild_abort(18);

			argc--;

			for (i = strlen(p); i; i--) {
				if (p[i - 1] == '/')
					p[i - 1] = ' ';
			}

			if (MISC_time_convert(p, strlen(p), until) == FB_FAILURE)
				rebuild_abort(19);
			break;

		default:
			MISC_print_journal_syntax();
			break;
		}
	}

	if (!journal || !db)
		sw_interact = true;

	rec_restore((SCHAR*) journal, (SCHAR*) db);

	if (sw_trace) {
		GJRN_printf(206, NULL, NULL, NULL, NULL);
		GJRN_printf(207, NULL, NULL, NULL, NULL);

		for (i = 0; i < (JRNP_MAX - JRN_PAGE); i++)
			GJRN_printf(208, table[i], (TEXT*) totals[i].num_recs,
						(TEXT*) totals[i].num_bytes, NULL);
	}

	if (msg_file != stdout)
		fclose(msg_file);

	return false;
}


static USHORT add_file(DRB database,
					   FIL main_file,
					   UCHAR* file_name,
					   SSHORT length,
					   SLONG start,
					   bool new_file)
{
/**************************************
 *
 *	a d d _ f i l e
 *
 **************************************
 *
 * Functional description
 *	Add a file to an existing database.  Return the sequence
 *	number of the new file.  If anything goes wrong, return a
 *	sequence of 0.
 *
 **************************************/
	USHORT sequence;
	FIL file, next_file;
	SLONG file_handle;

	file_name[length] = 0;

	if (!(open_database_file(database, reinterpret_cast<const SCHAR*>(file_name),
		new_file, &file_handle)))
	{
		return 0;
	}

	next_file = setup_file(file_name, strlen((const char*) file_name), (int) file_handle);

	next_file->fil_min_page = start;
	sequence = 1;

	for (file = main_file; file->fil_next; file = file->fil_next)
		++sequence;

	file->fil_max_page = start - 1;
	file->fil_next = next_file;

	return sequence;
}


static void add_time(timeval* first,
					 timeval* next)
{
/**************************************
 *
 *	a d d _ t i m e
 *
 **************************************
 *
 * Functional description
 *	Add time to elapsed time.
 *
 **************************************/

	if (first->tv_usec > next->tv_usec) {
		next->tv_usec += 1000000;
		next->tv_sec--;
	}
	et_msec += next->tv_usec - first->tv_usec;
	et_sec += next->tv_sec - first->tv_sec;
}


static void apply_data(DRB database,
					   DPG page,
					   JRND * record)
{
/**************************************
 *
 *	a p p l y _ d a t a
 *
 **************************************
 *
 * Functional description
 *	Apply incremental changes to a data page.
 *
 **************************************/
	JRNP temp, *clump;
	SSHORT space, l, top, used;
	UCHAR *p, *q;
	dpg::dpg_repeat* index;
	dpg::dpg_repeat* end;

	if (sw_debug)
		GJRN_printf(21, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

// Process clumps

	for (clump = NULL; clump = next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNP_SIZE);

		if (temp.jrnp_type != JRNP_DATA_SEGMENT) {
			GJRN_printf(55, (TEXT*)(ULONG) temp.jrnp_type, NULL, NULL, NULL);
			rebuild_abort(72);
		}

		if (sw_trace)
			continue;

		// Handle segment deletion

		if (!temp.jrnp_length) {
			index = page->dpg_rpt + temp.jrnp_index;
			index->dpg_offset = 0;
			index->dpg_length = 0;
		}

		// Re-compute page high water mark

		index = page->dpg_rpt;
		end = index + page->dpg_count;
		page->dpg_count = 0;
		space = database->drb_page_size;

		for (l = 1, used = 0; index < end; index++, l++)
			if (index->dpg_length) {
				page->dpg_count = l;
				space = MIN(space, index->dpg_offset);
				used += ROUNDUP(index->dpg_length, ODS_ALIGNMENT);
			}

		if (!temp.jrnp_length)
			continue;

		// Handle segment addition

		index = page->dpg_rpt + temp.jrnp_index;
		q = clump->jrnp_data;
		l = temp.jrnp_length;

		if (index < end && l <= index->dpg_length) {
			index->dpg_length = l;
			p = (UCHAR *) page + index->dpg_offset;
			do
				*p++ = *q++;
			while (--l);
			continue;
		}

		page->dpg_count = MAX(page->dpg_count, temp.jrnp_index + 1);
		top = HIGH_WATER(page->dpg_count);
		l = ROUNDUP(l, ODS_ALIGNMENT);
		space -= l;

		if (space < top) {
			index->dpg_length = 0;
			space = compress(database, page);
			space -= l;
			if (space < top)
				rebuild_abort(56);
		}

		if (space + l > database->drb_page_size)
			rebuild_abort(57);

		index->dpg_offset = space;
		index->dpg_length = temp.jrnp_length;
		p = (UCHAR *) page + space;
		do
			*p++ = *q++;
		while (--l);
	}
}


static void apply_header(HDR page,
						 JRND * record)
{
/**************************************
 *
 *	a p p l y _ h e a d e r
 *
 **************************************
 *
 * Functional description
 *	Apply changes to database header page
 *
 **************************************/
	JRNDH temp1, *clump;
	JRNDA temp2;

	if (sw_debug)
		GJRN_printf(22, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNDH *) next_clump(record, clump);) {
		if (clump->jrndh_type == JRNP_DB_HEADER)
			memcpy((SCHAR *) & temp1, (SCHAR *) clump, JRNDH_SIZE);
		else
			memcpy((SCHAR *) & temp2, (SCHAR *) clump, JRNDA_SIZE);

		if ((sw_debug) && (clump->jrndh_type == JRNP_DB_HEADER))
			GJRN_printf(203, (TEXT*) temp1.jrndh_nti,
						(TEXT*) temp1.jrndh_oat, (TEXT*) temp1.jrndh_oit,
						NULL);

		if (sw_trace)
			continue;

		switch (clump->jrndh_type) {
		case JRNP_DB_HEADER:
			page->hdr_bumped_transaction = temp1.jrndh_nti;
			page->hdr_oldest_transaction = temp1.jrndh_oit;
			page->hdr_oldest_active = temp1.jrndh_oat;
			break;

		case JRNP_DB_ATTACHMENT:
			page->hdr_attachment_id = temp2.jrnda_data;
			break;

		case JRNP_DB_HDR_PAGES:
			page->hdr_PAGES = temp2.jrnda_data;
			break;

		case JRNP_DB_HDR_FLAGS:
			page->hdr_flags = temp2.jrnda_data;
			break;

		case JRNP_DB_HDR_SDW_COUNT:
			page->hdr_shadow_count = temp2.jrnda_data;
			break;

		default:
			rebuild_abort(58);
			break;
		}
	}
}


static void apply_ids(PPG page,
					  JRND * record)
{
/**************************************
 *
 *	a p p l y _ i d s
 *
 **************************************
 *
 * Functional description
 *	Apply changes to gen ids page
 *
 **************************************/
	JRNG temp, *clump;
	SLONG *ptr;

	if (sw_debug)
		GJRN_printf(23, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNG *) next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNG_SIZE);

		if (temp.jrng_type != JRNP_GENERATOR)
			rebuild_abort(59);

		if (sw_debug)
			GJRN_printf(204, (TEXT*) temp.jrng_offset,
						(TEXT*) temp.jrng_genval, NULL, NULL);

		if (sw_trace)
			continue;

		ptr = page->ppg_page + temp.jrng_offset;
		*ptr = temp.jrng_genval;
	}
}


static void apply_index(BTR page,
						JRND * record)
{
/**************************************
 *
 *	a p p l y _ i n d e x
 *
 **************************************
 *
 * Functional description
 *	Apply changes to b-tree pages
 *
 **************************************/
	JRNB temp, *clump;
	SCHAR *p, *q;
	SLONG l;
	SLONG delta;
	BTN node, next;

	if (sw_debug)
		GJRN_printf(24, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNB *) next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNB_SIZE);

		if (sw_trace)
			continue;

		switch (clump->jrnb_type) {
		case JRNP_BTREE_NODE:
			if (sw_debug) {
				GJRN_printf(25, (TEXT*)(ULONG) temp.jrnb_offset,
							(TEXT*)(ULONG) temp.jrnb_length, NULL, NULL);
				GJRN_printf(27, (TEXT*)(SLONG) page->btr_length, NULL, NULL, NULL);
			}

			// slide down upper part by delta
			// add node and increment btr_length

			delta = temp.jrnb_delta;
			p = (SCHAR *) ((UCHAR *) page + page->btr_length);
			q = p + delta;
			if (l = page->btr_length - temp.jrnb_offset)
				do
					*--q = *--p;
				while (--l);

			// move in node , next BTN

			p = (SCHAR *) ((UCHAR *) page + temp.jrnb_offset);
			q = (SCHAR *) clump->jrnb_data;
			l = temp.jrnb_length;

			do {
				MOVE_BYTE(q, p);
			} while (--l);

			page->btr_length += delta;
			page->btr_prefix_total = temp.jrnb_prefix_total;

			if (sw_debug) {
				GJRN_printf(26, (TEXT*)(ULONG) temp.jrnb_offset,
							(TEXT*)(ULONG) temp.jrnb_length, NULL, NULL);
				GJRN_printf(27, (TEXT*)(SLONG) page->btr_length, NULL, NULL, NULL);
			}
			break;

		case JRNP_BTREE_SEGMENT:
			if (sw_debug) {
				GJRN_printf(28, (TEXT*)(SLONG) page->btr_length, NULL, NULL, NULL);
			}

			// apply change directly

			p = (SCHAR *) page;
			q = (SCHAR *) clump->jrnb_data;
			if (l = temp.jrnb_length)
				do {
					MOVE_BYTE(q, p);
				} while (--l);

			if (sw_debug) {
				GJRN_printf(29, (TEXT*)(ULONG) temp.jrnb_length, NULL, NULL, NULL);
				GJRN_printf(27, (TEXT*)(SLONG) page->btr_length, NULL, NULL, NULL);
			}
			break;

		case JRNP_BTREE_DELETE:

			// delete a node entry

			node = (BTN) ((UCHAR *) page + temp.jrnb_offset);
			next = (BTN) (node->btn_data + node->btn_length);
			QUAD_MOVE(next->btn_number, node->btn_number);
			p = (SCHAR *) node->btn_data;
			q = (SCHAR *) next->btn_data;
			l = next->btn_length;
			if (node->btn_prefix < next->btn_prefix) {
				node->btn_length = next->btn_length + next->btn_prefix
					- node->btn_prefix;
				p += next->btn_prefix - node->btn_prefix;
			}
			else {
				node->btn_length = l;
				node->btn_prefix = next->btn_prefix;
			}

			if (l)
				do
					*p++ = *q++;
				while (--l);

			// Compute length of rest of bucket and move it down.

			l = page->btr_length - ((UCHAR *) q - (UCHAR *) page);

			if (l)
				do
					*p++ = *q++;
				while (--l);

			page->btr_length = (UCHAR *) p - (UCHAR *) page;
			page->btr_prefix_total = temp.jrnb_prefix_total;

			// Error Check

			if (node->btn_prefix != temp.jrnb_delta) {
				rebuild_abort(60);
			}
			if (page->btr_length != temp.jrnb_length) {
				rebuild_abort(61);
			}

			if (sw_debug) {
				GJRN_printf(30, (TEXT*)(ULONG) temp.jrnb_offset, NULL, NULL, NULL);
				GJRN_printf(27, (TEXT*)(SLONG) page->btr_length, NULL, NULL, NULL);
			}

			break;

		default:
			rebuild_abort(72);
		}
	}
}


static void apply_log(LIP page,
					  JRND * record)
{
/**************************************
 *
 *	  a p p l y _ l o g
 *
 **************************************
 *
 * Functional description
 *	  Apply changes to database log page
 *
 **************************************/
	JRNL temp, *clump;

	if (sw_debug)

		GJRN_printf(86, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNL *) next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNL_SIZE);

		if (sw_trace)
			continue;

		page->log_flags = temp.jrnl_flags;
		page->log_mod_tid = temp.jrnl_tid;
		page->log_mod_tip = temp.jrnl_tip;
	}
}

static void apply_pip(PIP page,
					  JRND * record)
{
/**************************************
 *
 *	a p p l y _ p i p
 *
 **************************************
 *
 * Functional description
 *	Apply changes to page inventory page.
 *
 **************************************/
	JRNA temp, *clump;
	UCHAR bit;
	USHORT byte;

	if (sw_debug)

		GJRN_printf(31, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNA *) next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, sizeof(jrna));

		if (temp.jrna_type != JRNP_PIP)
			rebuild_abort(63);

		byte = temp.jrna_slot >> 3;
		bit = 1 << (temp.jrna_slot & 7);

		if (sw_debug) {
			if (temp.jrna_allocate)
				GJRN_printf(183, (TEXT*) temp.jrna_slot,
							(TEXT*) record->jrnd_page, NULL, NULL);
			else
				GJRN_printf(184, (TEXT*) temp.jrna_slot,
							(TEXT*) record->jrnd_page, NULL, NULL);
		}
		if (sw_trace)
			continue;

		if (temp.jrna_allocate) {
			page->pip_bits[byte] &= ~bit;
		}
		else {
			page->pip_bits[byte] |= bit;
			page->pip_min = MIN(page->pip_min, temp.jrna_slot);
		}
	}
}


static void apply_pointer(PPG page,
						  JRND * record)
{
/**************************************
 *
 *	a p p l y _ p o i n t e r
 *
 **************************************
 *
 * Functional description
 *	Apply incremental changes to a pointer page.
 *
 **************************************/
	JRNP temp, *clump;
	SLONG longword;

	if (sw_debug)

		GJRN_printf(32, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNP_SIZE);

		if (temp.jrnp_type != JRNP_POINTER_SLOT)
			rebuild_abort(64);

		if (sw_trace)
			continue;

		if (temp.jrnp_length) {
			memcpy(&longword, clump->jrnp_data, sizeof(SLONG));
			page->ppg_page[temp.jrnp_index] = longword;
			page->ppg_count = MAX(page->ppg_count, temp.jrnp_index + 1);
			page->ppg_min_space = MIN(page->ppg_min_space, temp.jrnp_index);
			page->ppg_max_space = MAX(page->ppg_min_space, temp.jrnp_index);
		}
		else
			page->ppg_page[temp.jrnp_index] = 0;
	}
}


static void apply_root(IRT page,
					   JRND * record)
{
/**************************************
 *
 *	a p p l y _ r o o t
 *
 **************************************
 *
 * Functional description
 *	Apply changes to index root page
 *
 **************************************/
	JRNRP temp, *clump;

	if (sw_debug)

		GJRN_printf(33, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	for (clump = NULL; clump = (JRNRP *) next_clump(record, clump);) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNRP_SIZE);

		if (temp.jrnrp_type != JRNP_ROOT_PAGE)
			rebuild_abort(65);

		if (sw_debug)
			GJRN_printf(205, (TEXT*) temp.jrnrp_id,
						(TEXT*) temp.jrnrp_page, NULL, NULL);

		if (sw_trace)
			continue;

		page->irt_rpt[temp.jrnrp_id].irt_root = temp.jrnrp_page;
	}
}

static void apply_transaction(TIP page,
							  JRND * record)
{
/**************************************
 *
 *	a p p l y _ t r a n s a c t i o n
 *
 **************************************
 *
 * Functional description
 *	Apply incremental changes to a TIP page.
 *
 **************************************/
	JRNI *clump, *clump_end;
	JRNI temp;
	JRND rec;

	if (sw_debug)

		GJRN_printf(34, (TEXT*) record->jrnd_page, NULL, NULL, NULL);

	memcpy((SCHAR *) & rec, (SCHAR *) record, JRND_SIZE);

	clump = (JRNI *) record->jrnd_data;
	clump_end = (JRNI *) (record->jrnd_data + rec.jrnd_length);

// Process clumps

	for (; clump < clump_end; clump++) {
		memcpy((SCHAR *) & temp, (SCHAR *) clump, JRNI_SIZE);

		if (sw_debug) {
			if (temp.jrni_type == JRNP_TRANSACTION)
				GJRN_printf(181,
							(TEXT*) temp.jrni_transaction, NULL, NULL, NULL);
			else if (temp.jrni_type == JRNP_NEXT_TIP)
				GJRN_printf(202,
							(TEXT*) temp.jrni_transaction, NULL, NULL, NULL);
		}
		if (sw_trace) {
			// transaction loop does not use next_clump!!

			totals[temp.jrni_type - JRN_PAGE].num_recs++;
			totals[temp.jrni_type - JRN_PAGE].num_bytes += JRNI_SIZE;
			continue;
		}

		if (temp.jrni_type == JRNP_TRANSACTION)
			page->tip_transactions[temp.jrni_position] = temp.jrni_states;
		else if (temp.jrni_type == JRNP_NEXT_TIP)
			page->tip_next = temp.jrni_transaction;
		else
			rebuild_abort(66);
	}
}

static USHORT checksum(DRB database,
					   PAG page)
{
/**************************************
 *
 *	c h e c k s u m
 *
 **************************************
 *
 * Functional description
 *	Compute the checksum of a page.
 *
 **************************************/
	ULONG chksum, *p, *end;
	USHORT old_chksum;

	end = (ULONG *) ((SCHAR *) page + database->drb_page_size);
	old_chksum = page->pag_checksum;
	page->pag_checksum = 0;
	p = (ULONG *) page;
	chksum = 0;

	do {
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
		chksum += *p++;
	} while (p < end);

	page->pag_checksum = old_chksum;

	if (chksum)
		return (USHORT) chksum;

// If the page is all zeros, return an artificial checksum

	for (p = (ULONG *) page; p < end;)
		if (*p++)
			return chksum;

// Page is all zeros -- invent a checksum

	return 12345;
}


static void close_database(DRB database)
{
/**************************************
 *
 *	c l o s e _ d a t a b a s e
 *
 **************************************
 *
 * Functional description
 *	We've reached end of processing this database
 *	Clean up database and close it.
 *
 **************************************/
	DRB *ptr;
	CACHE buffer;
	FIL fil;
	SLONG *tr2 = 0;
	SSHORT ret_val;
	ISC_STATUS_ARRAY status;

	if (sw_trace)
		return;

	fixup_header(database);

// Flush and release cache

	while (buffer = database->drb_cache) {
		database->drb_cache = buffer->cache_next;
		if ((buffer->cache_page_number != PAGE_CORRUPT) &&
			(buffer->cache_flags & PAGE_DIRTY))
			write_page(database, buffer);
		MISC_free_jrnl((int*) buffer);
	}
	ret_val = FB_SUCCESS;

	for (fil = database->drb_file; fil; fil = fil->fil_next) {
		if (LLIO_close(status, fil->fil_desc) == FB_FAILURE)
			ret_val = FB_FAILURE;
	}

	if (ret_val == FB_FAILURE)
		rebuild_abort(233);

	MISC_free_jrnl((int*) database->drb_buffers);

// Unlink from general data structures

	for (ptr = &databases; *ptr; ptr = &(*ptr)->drb_next)
		if (*ptr == database) {
			*ptr = database->drb_next;
			release_db(database);
			break;
		}

	if ((sw_partial) && (!sw_activate))
		return;

//
// Just attach to the database and detach.  This will update the
// log page to have the correct checkpoint information
//

	READY db_name AS DB_NEW
	ON_ERROR
		rebuild_abort(153);
	END_ERROR;

// cleanup rdb_$log_files for the recovered database

	if (sw_interact || sw_until || sw_partial) {
		START_TRANSACTION tr2 USING DB_NEW;

		FOR(TRANSACTION_HANDLE tr2) L IN DB_NEW.RDB$LOG_FILES
			ERASE L;
		END_FOR;

		COMMIT tr2

		ON_ERROR
			gds__print_status(gds_status);
			rebuild_abort(0);
		END_ERROR;

	}
	FINISH DB_NEW;
}


static bool commit(DRB database,
				   LTJC * record,
				   USHORT length)
{
/**************************************
 *
 *	c o m m i t
 *
 **************************************
 *
 * Functional description
 *	A commit is about to happen.  If restore is to be terminated
 *	at a given time, this is a good time to check and stop.
 *
 **************************************/

	TEXT c_time[32];
	SCHAR buf[MSG_LENGTH];
	SLONG date[2];

	if (until[0]) {
		WALR_get_blk_timestamp(WALR_handle, date);
		format_time(date, c_time);
		GJRN_printf(87, c_time, NULL, NULL, NULL);
		if (sw_interact) {
			GJRN_get_msg(88, buf, NULL, NULL, NULL);
			if (!MISC_get_line(buf, c_time, sizeof(c_time)) ||
				UPPER(c_time[0]) != 'Y') {
				close_database(database);
				return false;
			}
		}
	}
	else if (sw_verbose) {
		WALR_get_blk_timestamp(WALR_handle, date);
		format_time(date, c_time);
		GJRN_printf(87, c_time, NULL, NULL, NULL);
	}

	return true;
}


static SSHORT compress(DRB database,
					   DPG page)
{
/**************************************
 *
 *	c o m p r e s s
 *
 **************************************
 *
 * Functional description
 *	Compress a data page.  Return the high water mark.
 *
 **************************************/
	register SSHORT space, l;
	UCHAR temp[MAX_PAGE_SIZE + 4];
	register UCHAR *p, *q;

	if (database->drb_page_size > sizeof(temp))
		rebuild_abort(67);

	p = (UCHAR *) temp;
	q = (UCHAR *) page;
	l = database->drb_page_size >> 2;

	do {
		*(SLONG *) p = *(SLONG *) q;
		p += sizeof(SLONG);
		q += sizeof(SLONG);
	} while (--l);

	space = database->drb_page_size;
	dpg::dpg_repeat* end = page->dpg_rpt + page->dpg_count;

	for (dpg::dpg_repeat* index = page->dpg_rpt; index < end; index++)
		if (index->dpg_offset && index->dpg_length) {
			l = index->dpg_length;
			space -= ROUNDUP(l, ODS_ALIGNMENT);
			q = temp + index->dpg_offset;
			index->dpg_offset = space;
			p = (UCHAR *) page + space;
			if (l)
				do
					*p++ = *q++;
				while (--l);
		}

	return space;
}


static void disable(DRB database,
					LTJC * record)
{
/**************************************
 *
 *	d i s a b l e
 *
 **************************************
 *
 * Functional description
 *	We're encountered a database disable record.
 *	Clean up database and close it.
 *
 **************************************/
	TEXT ctime[32];
	SCHAR dbname[300], dir_name[300];
	SLONG date[2];

	MISC_get_wal_info(record, dbname, dir_name);

	WALR_get_blk_timestamp(WALR_handle, date);
	format_time(date, ctime);

	if (sw_verbose)
		GJRN_printf(35, (TEXT*)(ULONG) record->ltjc_length, dbname,
					(TEXT*) record->ltjc_header.jrnh_handle, ctime);

	sw_disable = 1;
	close_database(database);
}


static int error(const TEXT* filename,
				 ISC_STATUS err_num,
				 const TEXT* string,
				 ISC_STATUS operation)
{
/**************************************
 *
 *	e r r o r
 *
 **************************************
 *
 * Functional description
 *	We've had an unexpected error -- punt.
 *
 **************************************/
	ISC_STATUS_ARRAY status_vector;
	ISC_STATUS* s = status_vector;
	*s++ = isc_arg_gds;
	*s++ = isc_io_error;
	*s++ = isc_arg_string;
	*s++ = (ISC_STATUS) string;
	*s++ = isc_arg_string;
	*s++ = (ISC_STATUS) filename;
	*s++ = isc_arg_gds;
	*s++ = operation;
	*s++ = SYS_ERROR;
	*s++ = err_num;
	*s = 0;

	gds__print_status(status_vector);

	return 0;
}


static void expand_num_alloc(SCHAR *** files,
							 SSHORT * num_alloc)
{
/**************************************
 *
 *	e x p a n d _ n u m _ a l l o c
 *
 **************************************
 *
 * Functional description
 *	Expand the size of array
 *
 **************************************/
	SCHAR **temp, **temp1;
	SSHORT size;

	size = *num_alloc;
	temp1 = *files;

	temp = (SCHAR **) MISC_alloc_jrnl(2 * size * sizeof(SLONG));
	MOVE_FAST(temp1, temp, size * sizeof(SLONG));

	MISC_free_jrnl((int*) temp1);
	*num_alloc = 2 * size;
	*files = temp;
}


static void fixup_header(DRB database)
{
/**************************************
 *
 *	f i x u p _ h e a d e r
 *
 **************************************
 *
 * Functional description
 *	Fixup header page to not require recovery
 *
 **************************************/
	LIP page;
	CACHE cch;
	SCHAR *p;
	USHORT len;
	HDR header;

	if (sw_trace)
		return;
//
// After a restore, the database should be as close to the original
// database as possible.  The exceptions are when we do a recovery
// to point on time or when database name or any other information
// is changed.  In that case, the journal information and all WAL
// information must be deleted.
//

	if (sw_interact || sw_until || sw_disable || sw_activate) {
		rec_delete_hdr_entry(database, HDR_journal_server);
		rec_delete_hdr_entry(database, HDR_backup_info);
		sw_disable = 0;
	}
	if (sec_added) {
		rec_add_hdr_entry(database, HDR_file, sec_len,
						  sec_file, CLUMP_REPLACE);
	}

// Update the next transaction id from the bumped transaction id
	cch = get_page(database, (SLONG) HEADER_PAGE, true);
	MARK_HEADER(cch);
	header = (HDR) cch->cache_page;
	header->hdr_next_transaction = header->hdr_bumped_transaction;
	release_page(database, cch);

// Now update the log page
	cch = get_page(database, (SLONG) LOG_PAGE, true);
	MARK_PAGE(cch);
	page = (LIP) cch->cache_page;

	if (sw_interact || sw_until || sw_activate) {
		p = (SCHAR*) page->log_data;
		*p++ = LOG_ctrl_file1;
		*p++ = len = CTRL_FILE_LEN;
		do
			*p++ = 0;
		while (--len);

		// Set control point 2 file name

		*p++ = LOG_ctrl_file2;
		*p++ = len = CTRL_FILE_LEN;
		do
			*p++ = 0;
		while (--len);

		// Set current log file

		*p++ = LOG_logfile;
		*p++ = len = CTRL_FILE_LEN;
		do
			*p++ = 0;
		while (--len);

		*p = LOG_end;

		page->log_flags = log_no_ail;
		page->log_end = (USHORT) (p - (SCHAR *) page);

		page->log_cp_1.cp_seqno = 0;
		page->log_cp_1.cp_offset = 0;
		page->log_cp_1.cp_p_offset = 0;
		page->log_cp_1.cp_fn_length = 0;

		page->log_cp_2.cp_seqno = 0;
		page->log_cp_2.cp_offset = 0;
		page->log_cp_2.cp_p_offset = 0;
		page->log_cp_2.cp_fn_length = 0;

		page->log_file.cp_seqno = max_seqno;
		page->log_file.cp_offset = 0;
		page->log_file.cp_p_offset = 0;
		page->log_file.cp_fn_length = 0;
	}

	page->log_flags &= ~log_recover;

// At the end of a partial recovery, we will turn off log and
// set the next seqno.

	if ((sw_partial) && (!sw_activate)) {
		page->log_flags = (log_no_ail | log_partial_rebuild);
		page->log_file.cp_seqno = max_seqno;
	}

// Set the log_recovery_done flag, so that a subsequent attachment by
// the recovery process would not be denied because the shared cache
// manager is not running.
	if (!(page->log_flags & log_no_ail))
		page->log_flags |= log_recovery_done;

	release_page(database, cch);
}


static void format_time(SLONG date[2],
						TEXT* string)
{
/**************************************
 *
 *	f o r m a t _ t i m e
 *
 **************************************
 *
 * Functional description
 *	Format a date/time string.
 *
 **************************************/
	tm times;

	isc_decode_date((ISC_QUAD*) date, &times);

	sprintf(string, "%.2d:%.2d:%.2d %.2d/%.2d/%.2d",
			times.tm_hour, times.tm_min, times.tm_sec,
			times.tm_mon + 1, times.tm_mday, times.tm_year);
}


static CACHE get_free_buffer(DRB database,
							 SLONG page)
{
/**************************************
 *
 *	g e t _ f r e e _ b u f f e r
 *
 **************************************
 *
 * Functional description
 *	Find (or make) available buffer in cache.
 *
 **************************************/
	CACHE best, buffer;

	best = database->drb_cache;

	for (buffer = database->drb_cache; buffer; buffer = buffer->cache_next) {
		if (buffer->cache_use_count)
			continue;
		if (buffer->cache_age < best->cache_age)
			best = buffer;
		if (buffer->cache_page_number == PAGE_CORRUPT) {
			best = buffer;
			break;
		}
	}
	if (best->cache_use_count)
		rebuild_abort(73);

	if ((best->cache_page_number != PAGE_CORRUPT) &&
		(best->cache_flags & PAGE_DIRTY)) write_page(database, best);

	// Unlink from old slot

	MOVE_CLEAR(best->cache_page, database->drb_page_size);
	best->cache_page_number = page;

	return best;
}


static void get_log_files(SLONG db_id,
						  SSHORT use_archive,
						  SCHAR *** file_list,
						  SLONG ** p_offset,
						  SSHORT * number,
						  SLONG start_seqno)
{
/**************************************
 *
 *	g e t _ l o g _ f i l e s
 *
 **************************************
 *
 * Functional description
 *	gets a list of files for the recovery start from start_seqno
 *
 **************************************/
	SSHORT num_files;
	SSHORT num_alloc, num_alloc1;
	SCHAR **files;
	SLONG *start_p_offset;

	num_files = 0;
	num_alloc1 = num_alloc = INITIAL_ALLOC;
	files = (SCHAR **) MISC_alloc_jrnl(num_alloc * sizeof(SLONG));

	start_p_offset = (SLONG *) MISC_alloc_jrnl(num_alloc1 * sizeof(SLONG));

//
// Note that we do not allow partition offset.  Partitions are
// allowed only in raw files, and raw file have to be archived
//

	FOR(TRANSACTION_HANDLE tr1) J IN DB.JOURNAL_FILES WITH
		J.FILE_SEQUENCE >= start_seqno AND
		J.DB_ID EQ db_id SORTED BY J.FILE_SEQUENCE

		if (num_files >= num_alloc) {
			expand_num_alloc(&files, &num_alloc);
			expand_num_alloc((SCHAR***) &start_p_offset, &num_alloc1);
		}
		//
		// Stop at the first open file, but make sure there is
		// atleast one file
		//
		if ((strcmp(J.FILE_STATUS, LOG_CLOSED)) &&
			(((sw_partial) && (!sw_activate)) || (num_files)))
			break;
	
		files[num_files] = (SCHAR *) MISC_alloc_jrnl(MAX_PATH_LENGTH);
	
		if (use_archive) {
			if (strcmp(J.ARCHIVE_STATUS, ARCHIVED)) {
				if (!num_files) {
					if ((sw_partial) && (!sw_activate))
						break;
				
					// Get partition offset if single file is used
					strcpy(files[num_files], J.LOG_NAME);
					   start_p_offset[num_files] = J.PARTITION_OFFSET;
					   num_files++;
				}
				break;
			}
		
			strcpy(files[num_files], J.ARCHIVE_NAME);
		   }
		else {
			// Get partition offset if original files are used
			strcpy(files[num_files], J.LOG_NAME);
			start_p_offset[num_files] = J.PARTITION_OFFSET;
		}
	
		max_seqno = J.FILE_SEQUENCE;
	
		num_files++;
	END_FOR;

	*file_list = files;
	*p_offset = start_p_offset;
	*number = num_files;
}


static void get_old_files(SLONG db_id,
						  SCHAR *** file_list,
						  SSHORT * number,
						  SLONG * start_seqno,
						  SLONG * start_offset)
{
/**************************************
 *
 *	g e t _ o l d _ f i l e s
 *
 **************************************
 *
 * Functional description
 *	Get a list of old files for this recovery
 *	Returns number of files, start_seqno and offset.
 *
 **************************************/
	SSHORT num_files;
	SCHAR dump_num[10];
	SCHAR buf[MSG_LENGTH];
	SSHORT num_alloc;
	SCHAR **files;
	SLONG end_seqno;

	if (sw_interact) {
		sprintf(dump_num, "%d", dump_id);
		GJRN_get_msg(94, buf, NULL, NULL, NULL);
		MISC_get_new_value(buf, dump_num, sizeof(dump_num));
		dump_id = atoi(dump_num);

		// Check if this dump is for this database

			FOR(TRANSACTION_HANDLE tr1) OD IN DB.ONLINE_DUMP
				WITH OD.DUMP_ID EQ dump_id

				if (OD.DB_ID != db_id) {
					rebuild_abort(187);
				}

			END_FOR;
	}
	end_seqno = 0;

	num_files = 0;
	num_alloc = INITIAL_ALLOC;
	files = (SCHAR **) MISC_alloc_jrnl(num_alloc * sizeof(SLONG));

	FOR(TRANSACTION_HANDLE tr1) OD IN DB.ONLINE_DUMP_FILES
	   WITH OD.DUMP_ID EQ dump_id
	   SORTED BY OD.FILE_SEQNO

		if (num_files >= num_alloc) {
			expand_num_alloc(&files, &num_alloc);
		}

		files[num_files++] = (SCHAR *) MISC_alloc_jrnl(MAX_PATH_LENGTH);
		strcpy(files[num_files - 1], OD.DUMP_FILE_NAME);

	END_FOR;

	FOR(TRANSACTION_HANDLE tr1) O IN DB.ONLINE_DUMP
		WITH O.DUMP_ID EQ dump_id

		if ((O.START_SEQNO > O.END_SEQNO)
			|| ((O.START_SEQNO == O.END_SEQNO)
				&& (O.START_OFFSET > O.END_OFFSET))) {
			rebuild_abort(51);
		}

		*start_offset = O.START_OFFSET;
		*start_seqno = O.START_SEQNO;

		end_seqno = O.END_SEQNO;

	END_FOR;

// Check if online dump has been completed

	if ((sw_partial) && (!sw_activate)) {
		FOR(TRANSACTION_HANDLE tr1) J IN DB.JOURNAL_FILES
			WITH J.DB_ID EQ db_id AND J.FILE_SEQUENCE EQ end_seqno

			if (strcmp(J.FILE_STATUS, LOG_CLOSED)) {
				rebuild_abort(148);
			}

		END_FOR;
	}

	if (!num_files)
		rebuild_abort(53);

	*file_list = files;
	*number = num_files;
}


static CACHE get_page(DRB database,
					  SLONG page,
					  bool read_flag)
{
/**************************************
 *
 *	g e t _ p a g e
 *
 **************************************
 *
 * Functional description
 *	Get address of page in cache.  If read_flag is set,
 *	read the page if necessary.
 *
 **************************************/

	CACHE buffer;
	PAG pg;

	buffer = NULL;

	for (buffer = database->drb_cache; buffer; buffer = buffer->cache_next)
		if (buffer->cache_page_number == page)
			break;

	if (!buffer) {
		buffer = get_free_buffer(database, page);
		if (read_flag) {
			// stuff something in to make sure that when we read beyond
			// current end of file and not read any data, we will zero
			// things out.

			pg = (PAG) buffer->cache_page;
			pg->pag_checksum = 12345;

			read_page(database, buffer);
			if (pg->pag_checksum != checksum(database, pg))
				MOVE_CLEAR((UCHAR *) pg, database->drb_page_size);
		}
	}
	buffer->cache_age = ++database->drb_age;

	if (buffer->cache_use_count)
		rebuild_abort(68);

	buffer->cache_use_count++;

	if (page > database->drb_max_page)
		database->drb_max_page = page;

	return buffer;
}


static JRNP *next_clump(JRND * record,
						void *prior)
{
/**************************************
 *
 *	n e x t _ c l u m p
 *
 **************************************
 *
 * Functional description
 *	Given a prior clump, compute the address of the next
 *	clump on a data page.  If the prior clump is null,
 *	compute the address of the first clump.  If we run
 *	off the record, return NULL.
 *
 **************************************/
	JRNB temp1;
	USHORT offset, l;
	JRNP temp;

// Compute the offset and length of prior clump

	if (prior) {
		offset = (SCHAR *) prior - (SCHAR *) record;
		memcpy((SCHAR *) & temp, (SCHAR *) prior, JRNP_SIZE);
	}
	else {
		offset = 0;
		memcpy((SCHAR *) & temp, (SCHAR *) record->jrnd_data, JRNP_SIZE);
	}

	switch (temp.jrnp_type) {
	case JRNP_DATA_SEGMENT:
	case JRNP_POINTER_SLOT:
		l = JRNP_SIZE + temp.jrnp_length;
		break;

	case JRNP_BTREE_SEGMENT:
	case JRNP_BTREE_NODE:

		if (prior)
			memcpy((SCHAR *) & temp1, (SCHAR *) prior, JRNB_SIZE);
		else
			memcpy((SCHAR *) & temp1, (SCHAR *) record->jrnd_data, JRNB_SIZE);

		l = JRNB_SIZE + temp1.jrnb_length;
		break;

		//
		// currently DELETE node contains some debug info in jrnb_length
		// but data field is not used.
		//
	case JRNP_BTREE_DELETE:
		l = JRNB_SIZE;
		break;

	case JRNP_PIP:
		l = sizeof(jrna);
		break;

	case JRNP_DB_HEADER:
		l = JRNDH_SIZE;
		break;

	case JRNP_LOG_PAGE:
		l = JRNL_SIZE;
		break;

	case JRNP_DB_ATTACHMENT:
	case JRNP_DB_HDR_PAGES:
	case JRNP_DB_HDR_FLAGS:
	case JRNP_DB_HDR_SDW_COUNT:
		l = JRNDA_SIZE;
		break;

	case JRNP_GENERATOR:
		l = JRNG_SIZE;
		break;

	case JRNP_ROOT_PAGE:
		l = JRNRP_SIZE;
		break;

	default:
		rebuild_abort(69);
	}

	if (sw_trace) {
		totals[temp.jrnp_type - JRN_PAGE].num_recs++;
		totals[temp.jrnp_type - JRN_PAGE].num_bytes += l;
	}

// If the prior pointer is null, just return the data area

	if (!prior)
		return (JRNP *) record->jrnd_data;

	offset += l;

	if (offset & 1)
		++offset;

	if (offset >= JRND_SIZE + record->jrnd_length)
		return 0;

	return (JRNP *) ((SCHAR *) record + offset);
}


static void open_all_files(void)
{
/**************************************
 *
 *	o p e n _ a l l _ f i l e s
 *
 **************************************
 *
 * Functional description
 *	Open all secondary files of a database.
 *
 **************************************/
	cache cch;
	DRB database;
	FIL file;
	SCHAR *file_name;
	SSHORT file_length;
	UCHAR *p;
	SLONG last_page, next_page;
	SCHAR buf[MAX_PATH_LENGTH];
	PAG pg;

	if (sw_trace)
		return;

	database = databases;
	file = database->drb_file;

// allocate a temp page

	cch.cache_page = (PAG) MISC_alloc_jrnl(MAX_PAGE_SIZE);

	for (;;) {
		file_name = NULL;
		cch.cache_page_number = file->fil_min_page;
		//
		// Loop through all the header pages.  Only the primary database
		// page will have overflow header pages.
		//

		do {
			read_page(database, &cch);
			pg = (PAG) cch.cache_page;
			if (pg->pag_checksum != checksum(database, pg))
				rebuild_abort(234);

			HDR header = (HDR) cch.cache_page;

			for (p = header->hdr_data; *p != HDR_end; p += 2 + p[1])
				switch (*p) {
				case HDR_file:
					file_length = p[1];
					file_name = buf;
					MOVE_FAST(p + 2, buf, file_length);
					break;

				case HDR_last_page:
					MOVE_FAST(p + 2, &last_page, sizeof(last_page));
					break;

				default:;
				}
			next_page = header->hdr_next_page;
			cch.cache_page_number = next_page;
		} while (next_page);

		if (file->fil_min_page)
			file->fil_fudge = 1;

		if (!file_name)
			break;

		add_file(database, database->drb_file, (UCHAR*) file_name,
				 file_length, last_page + 1, false);
		file = file->fil_next;
	}
	MISC_free_jrnl((int*) cch.cache_page);
}


static DRB open_database(const TEXT* name,
						 USHORT page_size,
						 SSHORT new_file)
{
/**************************************
 *
 *	o p e n _ d a t a b a s e
 *
 **************************************
 *
 * Functional description
 *	Open database and create blocks.
 *
 **************************************/
	DRB database;
	UCHAR *buffers;
	CACHE buffer;
	USHORT n;
	SLONG file_handle;

	if (sw_trace)
		return 0;

// Allocate and partially populate database block


	database = (DRB) MISC_alloc_jrnl(sizeof(drb) + strlen(name));
	strcpy(database->drb_filename, name);

// Try to open file

	if (!open_database_file(database, name, new_file, &file_handle))
		rebuild_abort(199);


	database->drb_file = setup_file((UCHAR*) name, strlen(name), (int) file_handle);

// Link database block into world

	database->drb_next = databases;
	databases = database;

	database->drb_page_size = page_size;

// Allocate and initialize cache



	buffers = database->drb_buffers = MISC_alloc_jrnl(page_size * CACHE_SIZE);

	for (n = 0; n < CACHE_SIZE; n++) {
		buffer = (CACHE) MISC_alloc_jrnl(sizeof(cache));
		buffer->cache_next = database->drb_cache;
		database->drb_cache = buffer;
		buffer->cache_page_number = PAGE_CORRUPT;
		buffer->cache_page = (PAG) buffers;
		buffers += page_size;
	} return database;
}


static bool open_database_file(DRB database,
							   const TEXT* name,
							   bool new_file,
							   SLONG* file_handle)
{
/**************************************
 *
 *	o p e n _ d a t a b a s e _ f i l e
 *
 **************************************
 *
 * Functional description
 *	Open database and create blocks.
 *
 **************************************/
	ISC_STATUS_ARRAY status;

	if (new_file) {
		if (LLIO_open(status, name, LLIO_OPEN_NEW_RW, TRUE,
					  file_handle) == FB_FAILURE)
		return false;
	}
	else {
		if (LLIO_open(status, name, LLIO_OPEN_EXISTING_RW, TRUE,
			file_handle) == FB_FAILURE)
		return false;
	}

	return true;
}


static int open_journal(const SCHAR* dbname,
						SCHAR** files,
						SSHORT num_files,
						SLONG start_offset,
						SLONG* start_p_offset)
{
/**************************************
 *
 *	o p e n _ j o u r n a l
 *
 **************************************
 *
 * Functional description
 *	Prompt the user for a journal file name, and open the file.
 *
 **************************************/
	int i, n;
	SCHAR buf[MAX_PATH_LENGTH];
	bool stop_flag;
	SLONG *until_ptr;

	WALR_handle = 0;

	if (sw_interact)
		for (i = 0; i < num_files; i++) {
			GJRN_get_msg(154, buf, NULL, NULL, NULL);
			MISC_get_new_value(buf, files[i], MAX_PATH_LENGTH);
		}
	else if (sw_verbose)
		for (i = 0; i < num_files; i++) {
			GJRN_printf(20, files[i], NULL, NULL, NULL);
		}

	stop_flag = (sw_partial && !sw_activate);
	until_ptr = sw_until ? until : 0;

	n = WALR_open(wal_status, &WALR_handle, dbname, num_files,
				  files, start_p_offset, start_offset, until_ptr, stop_flag);

	if (n != FB_SUCCESS) {
		error(files[0], (ISC_STATUS) ERRNO, "open", isc_io_open_err);
		return 0;
	}

	return n;
}


static int process_online_dump(const SCHAR* dbname,
							   SCHAR** files,
							   SSHORT num_files)
{
/**************************************
 *
 *	p r o c e s s _ o n l i n e _ d u m p
 *
 **************************************
 *
 * Functional description
 *	Process a journal file until end of file. 
 *
 **************************************/
	OLD OLD_handle;
	SLONG ret_val;
	SSHORT len;
	SSHORT count;
	SCHAR buf[MSG_LENGTH];
	timeval first, next;

	if (sw_verbose)
		GJRN_printf(37, NULL, NULL, NULL, NULL);

	OLD_handle = 0;

//
// OLD_get () returns
//  0  	- OK
// OLD_EOD	- end of dump
//  OLD_EOF	- EOF
//  OLD_ERR	- Error
//
	count = 0;

	while (true) {
		if (sw_interact) {
			GJRN_get_msg(89, buf, NULL, NULL, NULL);
			MISC_get_new_value(buf, files[count], MAX_PATH_LENGTH);
		}
		if (sw_verbose)
			GJRN_printf(191, files[count], NULL, NULL, NULL);

		count++;

		if (OLDR_open(&OLD_handle, dbname, num_files, files) == FB_FAILURE)
			return FB_FAILURE;

		// get start time

		MISC_get_time(&first);

		while (true) {
			ret_val = OLDR_get(OLD_handle, (char*) wal_buff, &len);

			if (ret_val)
				break;

			process_old_record((JRND *) wal_buff, len);
		}
		// get finish time

		MISC_get_time(&next);

		add_time(&first, &next);

		if (ret_val == OLD_EOD) {
			ret_val = 0;
			break;
		}
		if (ret_val == OLD_ERR) {
			ret_val = -1;
			break;
		}
	}

	OLDR_close(&OLD_handle);
	OLD_handle = 0;

	if (sw_verbose)
		GJRN_printf(38, NULL, NULL, NULL, NULL);

	return ret_val;
}


static int process_journal(const SCHAR* dbname,
						   SCHAR** files,
						   SSHORT num_files,
						   SLONG start_seqno,
						   SLONG start_offset,
						   SLONG* start_p_offset)
{
/**************************************
 *
 *	p r o c e s s _ j o u r n a l
 *
 **************************************
 *
 * Functional description
 *	Process a journal file until end of file. 
 *
 **************************************/
	USHORT len;
	SLONG ret_val;
	ULONG seqno;
	ULONG offset;
	timeval first, next;
#ifdef DEV_BUILD
	JRND rec1;
	SLONG rec_checksum;
#endif

	if (sw_verbose)
		GJRN_printf(39, NULL, NULL, NULL, NULL);

	if ((open_journal(dbname, files, num_files, start_offset, start_p_offset)) < 0)
		return FB_FAILURE;

	seqno = offset = 0;

// get start time

	MISC_get_time(&first);

	max_seqno = start_seqno;

	while (true) {
		ret_val = WALR_get(wal_status, WALR_handle, wal_buff, &len,
					 (SLONG*) &seqno, (SLONG*) &offset);

		if (ret_val == -1) {
			ret_val = 0;
			break;
		}
		if (ret_val != FB_SUCCESS) {
			ret_val = -1;
			break;
		}

		// This the maximum seqno which is processed

		max_seqno = seqno;

#ifdef DEV_BUILD
		// take care of word alignment problems
		MOVE_FAST((SCHAR *) wal_buff, (SCHAR *) & rec1, JRND_SIZE);
		rec_checksum = rec1.jrnd_header.jrnh_series;
		rec1.jrnd_header.jrnh_series = 0;
		MOVE_FAST((SCHAR *) & rec1, (SCHAR *) wal_buff, JRND_SIZE);

		if (rec_checksum != MISC_checksum_log_rec(wal_buff, len, 0, 0))
			rebuild_abort(214);

		rec1.jrnd_header.jrnh_series = rec_checksum;
		MOVE_FAST((SCHAR *) & rec1, (SCHAR *) wal_buff, JRND_SIZE);
#endif
		// Will return FALSE when JRN_DISABLE is encountered

		if (!process_record((JRNH*) wal_buff, len, seqno, offset, true)){
			ret_val = 0;
			break;
		}
	}

// get end time  and increment stats

	MISC_get_time(&next);

	add_time(&first, &next);

	if ((!end_reached) && (sw_first_recover))
		if (!sw_trace)
			rebuild_abort(148);

	if (sw_verbose)
		GJRN_printf(40, NULL, NULL, NULL, NULL);

	if (!(sw_interact || sw_until))
		WALR_fixup_log_header(wal_status, WALR_handle);

	WALR_close(wal_status, &WALR_handle);

	return ret_val;
}


static SLONG process_new_file(DRB database,
							  JRNF * record)
{
/**************************************
 *
 *	p r o c e s s _ n e w _ f i l e
 *
 **************************************
 *
 * Functional description
 *	Add a file to the current database. 
 *
 **************************************/
	FIL file, next;
	HDR header;
	SLONG sequence;
	TEXT*file_name;
	SLONG start;
	CACHE cch;
	SSHORT file_len;
	TEXT new_name[MAX_PATH_LENGTH];
	SCHAR buf[MSG_LENGTH];

	start = record->jrnf_start;
	file_name = record->jrnf_filename;
	file_len = record->jrnf_length;

	memcpy(new_name, file_name, file_len);
	new_name[file_len] = 0;

	if (sw_interact) {
		GJRN_get_msg(90, buf, NULL, NULL, NULL);
		MISC_get_new_value(buf, new_name, sizeof(new_name));
		file_name = new_name;
		file_len = strlen(new_name);
	}
	if (sw_verbose)


		GJRN_printf(180, file_name,
					(TEXT*)(ULONG) record->jrnf_sequence, NULL, NULL);

	if (sw_trace)
		return 0;

// Find current last file

	for (file = database->drb_file; file->fil_next; file = file->fil_next);

// Create the file.  If the sequence number comes back zero, it didn't
// work, so punt

	if (!(sequence = add_file(database, database->drb_file, (UCHAR*) file_name,
							  file_len, start, true)))
	{
		rebuild_abort(232);
	}

	if (record->jrnf_sequence != sequence)
		rebuild_abort(71);

// Create header page for new file

	next = file->fil_next;

	cch = get_page(database, next->fil_min_page, false);
	header = (HDR) cch->cache_page;
	header->hdr_header.pag_type = pag_header;
	header->hdr_sequence = sequence;
	header->hdr_page_size = database->drb_page_size;
	header->hdr_data[0] = HDR_end;
	next->fil_sequence = sequence;

	MARK_HEADER(cch);
	release_page(database, cch);
	next->fil_fudge = 1;

// Update the previous header page to point to new file
// Header page can be updated only by setting fil_fudge to 0

	file->fil_fudge = 0;

	cch = get_page(database, file->fil_min_page, true);
	header = (HDR) cch->cache_page;

// Add clumplets for file name, starting page number

	if (!file->fil_min_page) {
		//
		// This will happen when the second file is added.  The primary
		// file header page is journalled when the file is added and that
		// record will be encountered before this new_file record is
		// added.  This code is just an error check
		//
		sec_added = true;
		sec_len = file_len;
		memcpy(sec_file, file_name, file_len);
		sec_file[sec_len] = 0;
	}
	else {
		--start;
		rec_add_clump_entry(database, header, HDR_file,
							strlen(file_name), (UCHAR*) file_name);
		rec_add_clump_entry(database, header, HDR_last_page,
							sizeof(start), (UCHAR*) & start);
	}

	MARK_HEADER(cch);
	release_page(database, cch);

// i.e if it is not the first file

	if (file->fil_min_page)
		file->fil_fudge = 1;

	return sequence;
}


static bool process_old_record(JRND * record,
							   USHORT length)
{
/**************************************
 *
 *	p r o c e s s _ o l d _ r e c o r d
 *
 **************************************
 *
 * Functional description
 *	Process a single online dump record record. 
 *	Generate a seqno/offset pair for data pages
 *
 **************************************/
	PAG page;
	ULONG seqno;
	ULONG offset;

	if (record->jrnd_header.jrnh_type == JRN_PAGE) {
		page = (PAG) record->jrnd_data;
		offset = page->pag_offset;
		seqno = page->pag_seqno;
		return process_record((JRNH*) record, length, seqno, offset, false);
	}
	offset = -1;
	seqno = -1;

	return process_record((JRNH*) record, length, seqno, offset, false);
}


static void process_page(DRB database,
						 JRND * record,
						 ULONG seqno,
						 ULONG offset,
						 bool wal_flag)
{
/**************************************
 *
 *	p r o c e s s _ p a g e
 *
 **************************************
 *
 * Functional description
 *	Update a database page.
 *	wal_flag indicates whether it is done for wal record or old record
 *
 **************************************/
	PAG page;
	JRNP *clump;
	UCHAR *p, *q;
	CACHE cch;
	JRND rec;

	clump = (JRNP *) record->jrnd_data;
	q = (UCHAR *) clump;

// wal_flag is always true for clumps

	memcpy((SCHAR *) & rec, (SCHAR *) record, JRND_SIZE);

// Print out info

	if (sw_debug)
		GJRN_printf(182, (TEXT*) rec.jrnd_page,
					(TEXT*) seqno, (TEXT*) offset, NULL);

// Handle zero length record.  This is an off shoot of PAG_release_page ()

	if (rec.jrnd_length == 0)
		return;

// Collect statistics

	bytes_processed += (rec.jrnd_length + JRND_SIZE);

	if (!sw_trace) {
		cch = get_page(database, rec.jrnd_page, wal_flag);
		page = cch->cache_page;
	}
	if ((wal_flag) && (!sw_trace)) {
		// handle freed page problem.  Ignore clumps for pages
		// which are empty.  This may have to be fixed later


		if (clump->jrnp_type > pag_max) {
			if ((page->pag_generation == 0)
				&& (page->pag_seqno == 0)
				&& (page->pag_offset == 0)) {
				release_page(database, cch);
				return;
			}
		}

		if ((page->pag_seqno > seqno) || ((page->pag_seqno == seqno)
										  && (page->pag_offset >= offset))) {
			release_page(database, cch);
			if (sw_debug) {
				if (clump->jrnp_type <= pag_max)
					GJRN_printf(41, (TEXT*) rec.jrnd_page, NULL, NULL, NULL);
				else
					GJRN_printf(43, (TEXT*) rec.jrnd_page, NULL, NULL, NULL);
			}
			return;
		}
		//
		// Error check to see if records are being skipped
		// Do not check for error if it is a full page, or clump type
		// is JRNP_BTREE_SEGMENT and the previous seqno & offset are 0
		//

		if (!((rec.jrnd_header.jrnh_prev_seqno == 0) &&
			  (rec.jrnd_header.jrnh_prev_offset == 0) &&
			  ((clump->jrnp_type <= pag_max)
			   || (clump->jrnp_type == JRNP_BTREE_SEGMENT)))) {
			if ((page->pag_seqno != rec.jrnd_header.jrnh_prev_seqno)
				|| (page->pag_offset != rec.jrnd_header.jrnh_prev_offset)) {
				release_page(database, cch);
				rebuild_abort(140);
			}
		}
	}

// Collect statistics

	bytes_applied += (rec.jrnd_length + JRND_SIZE);

	if (!sw_trace)
		MARK_PAGE(cch);

// If record is a full replacement page, handle it

	if (clump->jrnp_type <= pag_max) {
		if (sw_debug)
			GJRN_printf(42, (TEXT*) rec.jrnd_page, NULL, NULL, NULL);

		if (sw_trace) {
			totals[0].num_recs++;
			totals[0].num_bytes += rec.jrnd_length;
		}
		else {
			p = (UCHAR *) page;
			memcpy(p, q, database->drb_page_size);
			page->pag_seqno = seqno;
			page->pag_offset = offset;
			release_page(database, cch);
		}
		return;
	}

//
// If we are doing a trace, there is no page, and we will just follow
// the journal record type.  Else there is an error check.
//

	if (sw_trace) {
		switch (clump->jrnp_type) {
		case JRNP_POINTER_SLOT:
			apply_pointer((PPG) page, record);
			break;

		case JRNP_DATA_SEGMENT:
			apply_data(database, (DPG) page, record);
			break;

		case JRNP_TRANSACTION:
		case JRNP_NEXT_TIP:
			apply_transaction((TIP) page, record);
			break;

		case JRNP_PIP:
			apply_pip((PIP) page, record);
			break;
		case JRNP_DB_HEADER:
		case JRNP_DB_ATTACHMENT:
		case JRNP_DB_HDR_PAGES:
		case JRNP_DB_HDR_FLAGS:
		case JRNP_DB_HDR_SDW_COUNT:
			apply_header((HDR) page, record);
			break;
		case JRNP_ROOT_PAGE:
			apply_root((IRT) page, record);
			break;

		case JRNP_BTREE_NODE:
		case JRNP_BTREE_SEGMENT:
		case JRNP_BTREE_DELETE:
			apply_index((BTR) page, record);
			break;
		case JRNP_GENERATOR:
			apply_ids((PPG) page, record);
			break;
		case JRNP_LOG_PAGE:
			apply_log((LIP) page, record);
			break;

		default:
			if (sw_debug)
				GJRN_printf(44, (TEXT*) rec.jrnd_page, NULL, NULL, NULL);
			rebuild_abort(72);
		}
	}
	else {
		// This allows some more error checking

		switch (page->pag_type) {
		case pag_pointer:
			apply_pointer((PPG) page, record);
			break;

		case pag_data:
			apply_data(database, (DPG) page, record);
			break;

		case pag_transactions:
			apply_transaction((TIP) page, record);
			break;

		case pag_pages:
			apply_pip((PIP) page, record);
			break;
		case pag_header:
			apply_header((HDR) page, record);
			break;
		case pag_root:
			apply_root((IRT) page, record);
			break;
		case pag_index:
			apply_index((BTR) page, record);
			break;
		case pag_ids:
			apply_ids((PPG) page, record);
			break;
		case pag_log:
			apply_log((LIP) page, record);
			break;

		default:
			if (sw_debug)
				GJRN_printf(44, (TEXT*) rec.jrnd_page, NULL, NULL, NULL);
			rebuild_abort(72);
		}
	}

	if (!sw_trace) {
		page->pag_seqno = seqno;
		page->pag_offset = offset;
		page->pag_generation++;
		release_page(database, cch);
	}
}


static bool process_partial(SLONG db_id,
							SCHAR * old_db_name)
{
/**************************************
 *
 *	p r o c e s s _ p a r t i a l
 *
 **************************************
 *
 * Functional description
 *	Processes a partial recovery.  If it is the first
 *	pass, store a record and return.  Let the general
 *	recovery code handle it
 *
 **************************************/
	SCHAR buf[MSG_LENGTH];
	bool found;

	found = false;

	if (!sw_partial)
		return found;

// This is partial recovery - check if value exists in
// relation.  If not add one.

	if (!partial_db)
		rebuild_abort(195);

// Use the partial_db_name as the new db_name

	strcpy(db_name, partial_db);


	FOR(TRANSACTION_HANDLE tr1) P IN DB.PARTIAL_REBUILDS
		WITH P.DB_ID EQ db_id AND P.NEW_DB_NAME EQ partial_db

		if (P.LAST_LOG_SEQ == 0) {
			if (sw_interact) {
				GJRN_get_msg(192, P.NEW_DB_NAME, NULL, NULL, NULL);
				if ((!MISC_get_line(buf, buf, sizeof(buf)))
					|| (UPPER(buf[0]) != 'Y'))
				{
					rebuild_abort(193); // Invalid entry
				}
				ERASE P
				ON_ERROR
					rebuild_abort(194);
				END_ERROR;
				SAVE tr1
				ON_ERROR
					rebuild_abort(194);
				END_ERROR;
				break;
			}
			else
				rebuild_abort(193);
		}
		found = true;
		sw_first_recover = false;
		// This will recovery the database fully
		if (sw_verbose)
			GJRN_printf(201, partial_db, NULL, NULL, NULL);
		rebuild_partial(P.DB_ID, P.LAST_LOG_SEQ, old_db_name);
	END_FOR;
	if (found)
		return found;

	// Add entry to PARTIAL_REBUILDS
	STORE(TRANSACTION_HANDLE tr1) P IN DB.PARTIAL_REBUILDS
		P.DB_ID = db_id;
		strcpy(P.NEW_DB_NAME, partial_db);
		P.LAST_LOG_SEQ = 0;
	END_STORE
	ON_ERROR
		rebuild_abort(194);
	END_ERROR;

	if (sw_verbose)
		GJRN_printf(200, partial_db, NULL, NULL, NULL);
	return found;
}


static bool process_record(JRNH * record,
						   USHORT length,
						   ULONG seqno,
						   ULONG offset,
						   bool wal_flag)
{
/**************************************
 *
 *	p r o c e s s _ r e c o r d
 *
 **************************************
 *
 * Functional description
 *	Process a single journal record. 
 *	return FALSE when JRN_DISBALE is encountered.
 *
 **************************************/
	DRB database;
	LTJW *rec;
// Handle particular record type
	database = databases;
	switch (record->jrnh_type) {
	case JRN_PAGE:
		process_page(database, (JRND*) record, seqno, offset, wal_flag);
		break;
	case JRN_ENABLE:
		break;
	case JRN_NEW_FILE:
		process_new_file(database, (JRNF *) record);
		break;
	case JRN_DISABLE:
		disable(database, (LTJC*) record);
		return false;
	case JRN_SYNC:
		break;
	case JRN_CNTRL_PT:
		break;
	case JRN_COMMIT:
		return commit(database, (LTJC*) record, length);
		break;
	case JRN_START_ONLINE_DMP:
		if (sw_debug)
			GJRN_printf(45, (TEXT*) seqno, (TEXT*) offset, NULL, NULL);
		break;
	case JRN_END_ONLINE_DMP:

		// Check if we reached the end of online dump

		rec = (LTJW *) record;
		if (rec->ltjw_seqno == dump_id)
			end_reached = true;
		if (sw_debug)
			GJRN_printf(46, (TEXT*) seqno, (TEXT*) offset, NULL, NULL);
		break;
	default:
		GJRN_printf(45, (TEXT*)(ULONG) record->jrnh_type, NULL, NULL, NULL);
		rebuild_abort(0);
	}

	return true;
}


static void quad_move(register UCHAR * a,
					  register UCHAR * b)
{
/**************************************
 *
 *	  q u a d _ m o v e
 *
 **************************************
 *
 * Functional description
 *	  Move an unaligned longword ( 4 bytes).
 *
 **************************************/
	MOVE_BYTE(a, b);
	MOVE_BYTE(a, b);
	MOVE_BYTE(a, b);
	MOVE_BYTE(a, b);
}


static void read_page(DRB database,
					  CACHE buffer)
{
/**************************************
 *
 *	r e a d _ p a g e
 *
 **************************************
 *
 * Functional description
 *	Read a database page.
 *
 **************************************/
	FIL file;
	ISC_STATUS_ARRAY status;
	SLONG len_read, new_offset;
	file = seek_file(database, database->drb_file, buffer, &new_offset);
	if (!file)
		rebuild_abort(48);
	if (LLIO_read(status, file->fil_desc, file->fil_string, new_offset,
		 LLIO_SEEK_NONE, (UCHAR*) buffer->cache_page, database->drb_page_size,
		 &len_read) == FB_FAILURE)
	{
		gds__print_status(status);
		rebuild_abort(234);
	}

	if ((len_read != database->drb_page_size) && (len_read != 0))
		rebuild_abort(234);
}


static void rebuild_abort(SLONG err_num)
{
/**************************************
 *
 *	r e b u i l d _ a b o r t
 *
 **************************************
 *
 * Functional description
 *	Cleanup and exit
 *
 **************************************/
	DRB ptr;
	if (!(sw_partial))
		for (ptr = databases; ptr; ptr = ptr->drb_next) {
			for (FIL file = ptr->drb_file; file; file = file->fil_next)
				unlink(file->fil_string);
		}

	if (tr1)
		ROLLBACK tr1
		ON_ERROR
		END_ERROR;
	if (DB)
		FINISH DB
		ON_ERROR
		END_ERROR;
	GJRN_abort(err_num);
}


static void rebuild_partial(SLONG db_id,
							SLONG last_log_seq,
							SCHAR * old_db)
{
/**************************************
 *
 *	r e b u i l d _ p a r t i a l
 *
 **************************************
 *
 * Functional description
 *	Roll a partially rebuild database forward from a journal.
 *
 **************************************/
	SLONG *start_p_offset;
	SCHAR **files;
	SSHORT num_files;
	SSHORT use_archive;
// statistics
	bytes_processed = bytes_applied = 0;
	elapsed_time = et_sec = et_msec = 0;
	FOR(TRANSACTION_HANDLE tr1) Y IN DB.DATABASES
		WITH Y.DB_ID EQ db_id

		use_archive = (strcmp(Y.ARCHIVE_MODE, NEED_ARCH) == 0);
		open_database(db_name, Y.PAGE_SIZE, false);
		open_all_files();
		if (!test_partial_db(last_log_seq))
			rebuild_abort(198);
		//
		// Note that we do not allow partition offset.  Partitions are
		// allowed only in raw files, and raw file have to be archived
		//
		get_log_files(Y.DB_ID, use_archive,
					  &files, &start_p_offset, &num_files, last_log_seq + 1);
		if (!num_files)
			break;				  // no work to be done
		if (process_journal(Y.DB_NAME, files, num_files,
			 last_log_seq + 1, 0, start_p_offset))
		{
			rebuild_abort(54);
		}
		for (; num_files; num_files--)
			MISC_free_jrnl((int*) files[num_files - 1]);
		MISC_free_jrnl((int*) files);
		MISC_free_jrnl((int*) start_p_offset);
		break;					  // recover one database at a time
	END_FOR;

	if ((max_seqno > last_log_seq) || (sw_activate))
		update_rebuild_seqno(db_id);
}


static void rec_add_clump_entry(DRB database,
								HDR header,
								USHORT type,
								USHORT len,
								UCHAR * entry)
{
/***********************************************
 *
 *	r e c _ a d d _ c l u m p _ e n t r y
 *
 ***********************************************
 *
 * Functional description
 *	Adds an entry to header page.
 *
 **************************************/
	UCHAR *q, *p;
	int free_space;
	q = entry;
	for (p = header->hdr_data; ((*p != HDR_end)
								&& (*p != type)); p += 2 + p[1]);
// We are at HDR_end, add the entry
	free_space = database->drb_page_size - header->hdr_end;
	if (free_space > (2 + len)) {
		*p++ = type;
		*p++ = len;
		if (len)
			do
				*p++ = *q++;
			while (--len);
		*p++ = HDR_end;
		header->hdr_end = p - (UCHAR *) header;
	}
}


static bool rec_add_hdr_entry(DRB database,
							  USHORT type,
							  USHORT len,
							  UCHAR * entry,
							  USHORT mode)
{
/***********************************************
 *
 *	r e c _ a d d  _ h d r _ e n t r y
 *
 ***********************************************
 *
 * Functional description
 *	Adds an entry to header page.
 *	mode
 *		0 - add			CLUMP_ADD
 *		1 - replace		CLUMP_REPLACE
 *		2 - replace only!	CLUMP_REPLACE_ONLY
 *	returns
 *		TRUE - modified page
 *		FALSE - nothing done
 *
 *
 **************************************/
	HDR header;
	UCHAR *q, *p;
	bool found;
	CACHE cch;
	cch = get_page(database, HEADER_PAGE, true);
	header = (HDR) cch->cache_page;
	if (mode != CLUMP_ADD) {
		found = rec_find_type(database, &cch, &header, type, &q, &p);
		// If we did'nt find it and it is REPLACE_ONLY, return
		if (!found && mode == CLUMP_REPLACE_ONLY) {
			release_page(database, cch);
			return false;
		}
		if (found) {
			if (q[1] == len) {
				q += 2;
				p = entry;
				if (len) {
					do
						*q++ = *p++;
					while (--len);
					MARK_HEADER(cch);
				}

				release_page(database, cch);
				return true;
			}
			release_page(database, cch);
			rec_delete_hdr_entry(database, type);
			return rec_add_hdr_entry(database, type, len, entry, CLUMP_ADD);
		}

		// !found && REPLACE will fall through

		if (cch->cache_page_number != HEADER_PAGE) {
			release_page(database, cch);
			cch = get_page(database, HEADER_PAGE, true);
			header = (HDR) cch->cache_page;
		}
	}

// Add the entry

	rec_find_space(database, &cch, &header, type, len, entry);
	MARK_HEADER(cch);
	release_page(database, cch);
	return true;
}


static bool rec_delete_hdr_entry(DRB database,
								 USHORT type)
{
/***********************************************
 *
 *	r e c _ d e l e t e _ h d r _ e n t r y
 *
 ***********************************************
 *
 * Functional description
 *	Gets rid on the entry 'type' from header page.
 *
 **************************************/
	HDR header;
	UCHAR *q, *p, *r;
	USHORT l;
	CACHE cch;
	cch = get_page(database, HEADER_PAGE, true);
	header = (HDR) cch->cache_page;
	if (!rec_find_type(database, &cch, &header, type, &q, &p)) {
		release_page(database, cch);
		return false;
	}

	header->hdr_end -= (2 + q[1]);
	r = q + 2 + q[1];
	l = p - r + 1;
	if (l)
		do
			*q++ = *r++;
		while (--l);
	MARK_HEADER(cch);
	release_page(database, cch);
	return true;
}


static void rec_find_space(DRB database,
						   CACHE * cch_p,
						   HDR * hdr,
						   USHORT type,
						   USHORT len,
						   UCHAR * entry)
{
/***********************************************
 *
 *	r e c _ f i n d _ s p a c e
 *
 ***********************************************
 *
 * Functional description
 *		TRUE  - Found it
 *		FALSE - Not present
 *	Note: We cannot allocate space.  It should have been
 *		  allocated before.
 *
 **************************************/
	UCHAR *p, *ptr;
	SLONG next_page;
	HDR header;
	SLONG free_space;
	CACHE cch;
	header = *hdr;
	cch = *cch_p;
	ptr = entry;
	while (true) {

		p = (UCHAR *) header + header->hdr_end;
		// We are at HDR_end, add the entry
		free_space = database->drb_page_size - header->hdr_end;
		if (free_space > (2 + len)) {
			// Space available, shuffle other entries down

			*p++ = type;
			*p++ = len;
			if (len)
				do
					*p++ = *ptr++;
				while (--len);
			*p = HDR_end;
			header->hdr_end = p - (UCHAR *) header;
			*cch_p = cch;
			*hdr = header;
			return;
		}

		next_page = header->hdr_next_page;
		if (!next_page)
			break;
		release_page(database, cch);
		cch = get_page(database, next_page, true);
		header = (HDR) cch->cache_page;
	}

	rebuild_abort(50);
}


static bool rec_find_type(DRB database,
						  CACHE * cch_p,
						  HDR * hdr,
						  USHORT type,
						  UCHAR ** t_ptr,
						  UCHAR ** l_ptr)
{
/***********************************************
 *
 *	r e c _ f i n d _ t y p e
 *
 ***********************************************
 *
 * Functional description
 *		TRUE  - Found it
 *		FALSE - Not present
 *
 **************************************/
	UCHAR *q, *p;
	SLONG next_page;
	CACHE cch;
	HDR header;
	q = 0;
	cch = *cch_p;
	header = *hdr;
	while (true) {
		for (p = header->hdr_data; (*p != HDR_end); p += 2 + p[1]) {
			if (*p == type)
				q = p;
		}

		if (q) {
			*t_ptr = q;
			*l_ptr = p;
			*cch_p = cch;
			*hdr = header;
			return true;
		}

		// Follow chain of header pages

		next_page = header->hdr_next_page;
		if (next_page) {
			release_page(database, cch);
			cch = get_page(database, next_page, true);
			header = (HDR) cch->cache_page;
		}
		else {
			*hdr = header;
			*cch_p = cch;
			return false;
		}
	}
}


#ifdef NOT_USED_OR_REPLACED
static bool rec_get_hdr_entry(DRB database,
							  USHORT type,
							  USHORT * len,
							  UCHAR * entry)
{
/***********************************************
 *
 *	r e c _ g e t _ h e a d e r _ e n t r y
 *
 ***********************************************
 *
 * Functional description
 *		TRUE  - Found it
 *		FALSE - Not present
 *		***!!! NOT USED !!!***
 *
 **************************************/
	UCHAR *q, *p, *r;
	USHORT l;
	HDR header;
	CACHE cch;
	cch = get_page(database, HEADER_PAGE, true);
	header = (HDR) cch->cache_page;
	q = entry;
	if (!rec_find_type(database, &cch, &header, type, &p, &r)) {
		release_page(database, cch);
		return false;
	}

	l = *(p + 1);
	*len = l;
	p += 2;
	if (l)
		do
			*q++ = *p++;
		while (--l);
	release_page(database, cch);
	return true;
}
#endif

static void rec_restore(SCHAR * journal_name,
						SCHAR * dbn)
{
/**************************************
 *
 *	r e c _ r e s t o r e
 *
 **************************************
 *
 * Functional description
 *	Roll a database forward from a journal.
 *
 **************************************/
	SLONG start_seqno;
	SLONG start_offset;
	SLONG *start_p_offset;
	SCHAR **files;
	SSHORT num_files;
	int size;
	SCHAR buf[MSG_LENGTH];
	SSHORT use_archive;
	if (journal_name) {
		sw_journal = true;
		strcpy(journal_dir, journal_name);
	}

	if (dbn) {
		sw_db = true;
		ISC_expand_filename(dbn, 0, db_name);
		dbn = db_name;
	}

// statistics

	bytes_processed = bytes_applied = 0;
	elapsed_time = et_sec = et_msec = 0;
	max_seqno = 0;
// Scan journal looking for database to recover
	if ((sw_interact) && (!journal_name)) {
		GJRN_get_msg(91, buf, NULL, NULL, NULL);
		if (!MISC_get_line(buf, journal_dir, sizeof(journal_dir))) {
			rec_restore_manual(dbn);
			return;
		}
	}

//
// If running recovery journal server needs to be running if database
// is fully recovered
//

	if (!(sw_interact || sw_until || sw_partial)) {
		if (!(CONSOLE_test_server(journal_dir)))
			rebuild_abort(213);
	}

	size = strlen(journal_dir);
	if (!size)
		rebuild_abort(15);
	if (journal_dir[size - 1] == '/')
		size--;
#if (defined WIN_NT)
	else if (journal_dir[size - 1] == '\\')
		size--;
#endif
	strcpy(journal_dir + size, "/journal.fdb");
	READY journal_dir AS DB
	ON_ERROR
		rebuild_abort(152);
	END_ERROR;
	START_TRANSACTION tr1 USING DB;
	FOR(TRANSACTION_HANDLE tr1) X IN DB.DATABASES
		WITH X.STATUS EQ DB_ACTIVE

		if ((sw_db) && (!sw_interact)) {
			if (strcmp(db_name, X.DB_NAME))
				continue;
		}
		else {
			GJRN_get_msg(92, buf, X.DB_NAME, NULL, NULL);
			if (!MISC_get_line(buf, db_name, sizeof(db_name)))
				continue;
			if ((UPPER(db_name[0])) != 'Y')
				continue;
			strcpy(db_name, X.DB_NAME);
		}

		use_archive = (strcmp(X.ARCHIVE_MODE, NEED_ARCH) == 0);
		if (sw_partial)
			if (process_partial(X.DB_ID, X.DB_NAME))
				break;
		if ((sw_interact) && (!sw_partial)) {
			GJRN_get_msg(93, buf, NULL, NULL, NULL);
			MISC_get_new_value(buf, db_name, sizeof(db_name));
		}
	
		open_database(db_name, X.PAGE_SIZE, true);
		dump_id = X.LAST_DUMP_ID;
		get_old_files(X.DB_ID, &files, &num_files, &start_seqno, &start_offset);
		if (process_online_dump(X.DB_NAME, files, num_files) < 0)
			rebuild_abort(52);
		for (; num_files; num_files--)
			MISC_free_jrnl((int*) files[num_files - 1]);
		MISC_free_jrnl((int*) files);
		//
		// Note that we do not allow partition offset.  Partitions are
		// allowed only in raw files, and raw file have to be archived
		//
		get_log_files(X.DB_ID, use_archive,
					 &files, &start_p_offset, &num_files, start_seqno);
		if (!num_files)
			rebuild_abort(17);
		if (process_journal(X.DB_NAME, files, num_files,
			start_seqno, start_offset, start_p_offset))
		{
			rebuild_abort(54);
		}
		if ((!end_reached) && (sw_first_recover))
			if (!sw_trace)
				rebuild_abort(148);
		for (; num_files; num_files--)
			MISC_free_jrnl((int*) files[num_files - 1]);
		MISC_free_jrnl((int*) files);
		MISC_free_jrnl((int*) start_p_offset);
		update_rebuild_seqno(X.DB_ID);
		break;					  // recover one database at a time
	END_FOR;

	if ((sw_db) && (!databases) && (!sw_trace)) {
		if (!((sw_disable) || (sw_interact)))
			rebuild_abort(185);
	}

// Print out stats

	elapsed_time = et_sec + et_msec / 1000000;
	if ((sw_verbose) || (sw_trace)) {
		GJRN_printf(155, NULL, NULL, NULL, NULL);
		GJRN_printf(156, (TEXT*)
					bytes_processed, NULL, NULL, NULL);
		GJRN_printf(157, (TEXT*) bytes_applied, NULL, NULL, NULL);
		GJRN_printf(158, (TEXT*) elapsed_time, NULL, NULL, NULL);
	}

	while (databases)
		close_database(databases);
	COMMIT tr1;
	FINISH DB;
}


static void rec_restore_manual(SCHAR * dbn)
{
/**************************************
 *
 *	r e c _ r e s t o r e _ m a n u a l
 *
 **************************************
 *
 * Functional description
 *	Roll a database forward from a journal manual
 *
 **************************************/
	SLONG start_seqno;
	SLONG start_offset;
	SLONG *start_p_offset;
	SCHAR **files;
	SCHAR *msg;
	SSHORT num_files;
	SCHAR buf[MSG_LENGTH];
	SSHORT num_alloc;
	OLD OLD_handle = 0;
	SCHAR old_db_name[300];
	SSHORT page_size;
	if (!dbn) {
		GJRN_get_msg(189, old_db_name, NULL, NULL, NULL);
		if (!MISC_get_line(old_db_name, db_name, sizeof(db_name)))
			rebuild_abort(185);
	}

	strcpy(old_db_name, db_name);
	if ((sw_interact) && (!sw_trace)) {
		GJRN_get_msg(93, buf, NULL, NULL, NULL);
		MISC_get_new_value(buf, db_name, sizeof(db_name));
	}

// get online dump information

	GJRN_get_msg(4, buf, NULL, NULL, NULL);
	GJRN_output("%s\n", buf);
	num_files = 0;
	num_alloc = INITIAL_ALLOC;
	files = (SCHAR **) MISC_alloc_jrnl(num_alloc * sizeof(SLONG));
	while (true) {
		msg = (SCHAR *) MISC_alloc_jrnl(MAX_PATH_LENGTH);
		GJRN_get_msg(6, msg, NULL, NULL, NULL);
		MISC_get_line(msg, msg, MAX_PATH_LENGTH);
		if (!strlen(msg)) {
			MISC_free_jrnl((int*) msg);
			break;
		}
		files[num_files++] = msg;
		if (num_files >= num_alloc)
			expand_num_alloc(&files, &num_alloc);
	}

	start_p_offset = (SLONG *) MISC_alloc_jrnl(num_files * sizeof(SLONG));
	if (!num_files) {
		if (sw_trace) {
			if (sw_debug)
				GJRN_printf(209, NULL, NULL, NULL, NULL);
		}
		else
			rebuild_abort(53);
		page_size = MIN_PAGE_SIZE;
	}
	else {
		// Need to get page size before starting online dump

		if (OLDR_open(&OLD_handle, old_db_name, num_files, files) == FB_FAILURE)
			rebuild_abort(187);
// BRS 04/Sep/2003
// When making the first build in FB 1.5 the &start_p_offset was changed to  
// start_p_offset and the allocation moved to some lines above. 
// Claudio points that perhaps OLDR_get_info is wrongly declared and the last parameter
// should be long** instead of long*

		OLDR_get_info(OLD_handle, &page_size, &dump_id,
					  &start_seqno, &start_offset, start_p_offset);
		OLDR_close(&OLD_handle);
	}

	if (sw_verbose)
		GJRN_printf(190, db_name, (TEXT*)(SLONG) page_size, NULL, NULL);
	open_database(db_name, page_size, true);
	if (num_files) {
		if (process_online_dump(old_db_name, files, num_files) < 0)
			rebuild_abort(52);
	}

	for (; num_files; num_files--)
		MISC_free_jrnl((int*) files[num_files - 1]);
	MISC_free_jrnl((int*) files);
// Get log file information
	GJRN_get_msg(188, buf, NULL, NULL, NULL);
	GJRN_output("%s\n", buf);
	num_files = 0;
	num_alloc = INITIAL_ALLOC;
	files = (SCHAR **) MISC_alloc_jrnl(num_alloc * sizeof(SLONG));
	while (true) {
		msg = (SCHAR *) MISC_alloc_jrnl(MAX_PATH_LENGTH);
		GJRN_get_msg(6, msg, NULL, NULL, NULL);
		MISC_get_line(msg, msg, MAX_PATH_LENGTH);
		if (!strlen(msg)) {
			MISC_free_jrnl((int*) msg);
			break;
		}
		files[num_files++] = msg;
		if (num_files >= num_alloc)
			expand_num_alloc(&files, &num_alloc);
	}

	if (!num_files)
		rebuild_abort(17);
	if (process_journal(old_db_name, files, num_files, start_seqno,
						start_offset, start_p_offset))
		 rebuild_abort(54);
	if (!end_reached)
		if (!sw_trace)
			rebuild_abort(148);
	elapsed_time = et_sec + et_msec / 1000000;
	if ((sw_verbose) || (sw_trace)) {
		GJRN_printf(155, NULL, NULL, NULL, NULL);
		GJRN_printf(156, (TEXT*)
					bytes_processed, NULL, NULL, NULL);
		GJRN_printf(157, (TEXT*) bytes_applied, NULL, NULL, NULL);
		GJRN_printf(158, (TEXT*) elapsed_time, NULL, NULL, NULL);
	}

	for (; num_files; num_files--)
		MISC_free_jrnl((int*) files[num_files - 1]);
	MISC_free_jrnl((int*) files);
	MISC_free_jrnl((int*) start_p_offset);
	while (databases)
		close_database(databases);
}


static void release_db(DRB drb)
{
/**************************************
 *
 *	r e l e a s e _ d b
 *
 **************************************
 *
 * Functional description
 *	Free up the database and all file blocks.
 *
 **************************************/
	FIL fil;
	if (!drb)
		return;
	while (drb->drb_file) {
		fil = drb->drb_file;
		drb->drb_file = fil->fil_next;
		MISC_free_jrnl((int*) fil);
	}

	MISC_free_jrnl((int*) drb);
}


static void release_page(DRB database,
						 CACHE buffer)
{
/**************************************
 *
 *	r e l e a s e _ p a g e
 *
 **************************************
 *
 * Functional description
 *	Release a page.
 *
 **************************************/

	buffer->cache_use_count--;
	if (buffer->cache_use_count)
		rebuild_abort(68);
	if ((buffer->cache_page_number != HEADER_PAGE)
		&& (buffer->cache_flags & PAGE_HEADER))
	{
		write_page(database, buffer);
		buffer->cache_page_number = PAGE_CORRUPT;
		buffer->cache_flags = 0;
	}
}


static FIL seek_file(DRB database,
					 FIL fil,
					 CACHE buffer,
					 SLONG * new_offset)
{
/**************************************
 *
 *	s e e k _ f i l e
 *
 **************************************
 *
 * Functional description
 *	Given a buffer descriptor block, find the appropropriate
 *	file block and seek to the proper page in that file.
 *
 **************************************/
	ULONG page;
	ISC_STATUS_ARRAY status;
	page = buffer->cache_page_number;
	for (;; fil = fil->fil_next)
		if (!fil)
			rebuild_abort(74);
		else if (page >= fil->fil_min_page && page <= fil->fil_max_page)
			break;
	page -= (fil->fil_min_page - fil->fil_fudge);
	*new_offset = page * database->drb_page_size;
	if (LLIO_seek (status, fil->fil_desc,
		 fil->fil_string, *new_offset, LLIO_SEEK_BEGIN) == FB_FAILURE)
	{
		gds__print_status(status);
		return NULL;
	}

	return fil;
}


static FIL setup_file(const UCHAR* file_name,
					  USHORT file_length,
					  int desc)
{
/**************************************
 *
 *	s e t u p _ f i l e
 *
 **************************************
 *
 * Functional description
 *	Set up file descriptor.
 *
 **************************************/
// Allocate file block and copy file name string
	FIL file = (FIL) MISC_alloc_jrnl(sizeof(fil) + file_length + 1);
	file->fil_desc = desc;
	file->fil_length = file_length;
	file->fil_max_page = (ULONG) -1;
	MOVE_FAST(file_name, file->fil_string, file_length);
	file->fil_string[file_length] = '\0';
	return file;
}


static bool test_partial_db(SLONG expected_seqno)
{
/**************************************
 *
 *	t e s t _ p a r t i a l _ d b
 *
 **************************************
 *
 * Functional description
 *
 **************************************/
	LIP page;
	CACHE cch;
	bool ret;
	DRB database;
	database = databases;
	cch = get_page(database, LOG_PAGE, true);
	page = (LIP) cch->cache_page;
// At the end of a partial recovery, we will turn off log and
// set the next seqno.

	ret = false;
	if ((page->log_flags == (log_no_ail | log_partial_rebuild))
		&& (page->log_file.cp_seqno == expected_seqno))
	{
		ret = true;
	}
	release_page(database, cch);
	return ret;
}


static void update_rebuild_seqno(SLONG db_id)
{
/**************************************
 *
 *	u p d a t e _ r e b u i l d _ s e q n o
 *
 **************************************
 *
 * Functional description
 *	Updates the highest seqno of partial recovery.
 *	If the database is being activated, delete the entry.
 *
 **************************************/

	if (sw_partial) {
		if (sw_activate) {
			FOR(TRANSACTION_HANDLE tr1) P IN DB.PARTIAL_REBUILDS
				WITH P.DB_ID EQ db_id AND P.NEW_DB_NAME EQ partial_db
			
				ERASE P
			
				ON_ERROR
					rebuild_abort(194);
				END_ERROR;
			END_FOR;
		}
		else {
			FOR(TRANSACTION_HANDLE tr1) P IN DB.PARTIAL_REBUILDS
				WITH P.DB_ID EQ db_id AND P.NEW_DB_NAME EQ partial_db
				MODIFY P
					P.LAST_LOG_SEQ = max_seqno;
				END_MODIFY
				ON_ERROR
					rebuild_abort(194);
				END_ERROR;
			END_FOR;
		}
	}

// Do a commit retain to save changes

	SAVE tr1
	ON_ERROR
		rebuild_abort(194);
	END_ERROR;
}


static void write_page(DRB database,
					   CACHE buffer)
{
/**************************************
 *
 *	w r i t e _ p a g e
 *
 **************************************
 *
 * Functional description
 *	Write a database page.
 *
 **************************************/
	PAG page;
	FIL fil;
	SLONG len_written, new_offset;
	ISC_STATUS_ARRAY status;
	page = buffer->cache_page;
	page->pag_checksum = checksum(database, page);
	fil = seek_file(database, database->drb_file, buffer, &new_offset);
	if (!fil)
		rebuild_abort(49);
	if (LLIO_write(status, fil->fil_desc, fil->fil_string, new_offset,
				   LLIO_SEEK_NONE, (UCHAR*) buffer->cache_page,
				   database->drb_page_size, &len_written) == FB_FAILURE)
	{
		gds__print_status(status);
		rebuild_abort(49);
	}

	if (len_written != database->drb_page_size)
		rebuild_abort(49);
}