/*
 *	PROGRAM:	Dynamic SQL runtime support
 *	MODULE:		metd.epp
 *	DESCRIPTION:	Meta-data interface
 *
 * 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): ______________________________________.
 * 2001.11.28 Claudio Valderrama: load not only udfs but udf arguments;
 *   handle possible collisions with udf redefinitions (drop->declare).
 *   This closes SF Bug# 409769.
 * 2001.12.06 Claudio Valderrama: METD_get_charset_bpc() was added to
 *    get only the bytes per char of a field, given its charset id.
 *   This request is not cached.
 * 2001.02.23 Claudio Valderrama: Fix SF Bug #228135 with views spoiling
 *    NULLs in outer joins.
 * 2004.01.16 Vlad Horsun: make METD_get_col_default and 
 *   METD_get_domain_default return actual length of default BLR
 * 2004.01.16 Vlad Horsun: added support for default parameters 
 */

#include "firebird.h"
#include <string.h>
#include "../dsql/dsql.h"
#include "../jrd/ibase.h"
#include "../jrd/align.h"
#include "../jrd/intl.h"
#include "../jrd/thd.h"
#include "../dsql/alld_proto.h"
#include "../dsql/ddl_proto.h"
#include "../dsql/metd_proto.h"
#include "../dsql/hsh_proto.h"
#include "../dsql/make_proto.h"
#include "../dsql/errd_proto.h"
#include "../jrd/gds_proto.h"
#include "../jrd/sch_proto.h"
#include "../jrd/thread_proto.h"
#include "../jrd/why_proto.h"
#include "../common/utils_proto.h"
#include "../common/classes/init.h"

/* NOTE: The static definition of DB and gds_trans by gpre will not
         be used by the meta data routines.  Each of those routines has
         its own local definition of these variables. */

/**************************************************************
V4 Multi-threading changes.

-- direct calls to gds__ () & isc_ () entrypoints

	THREAD_EXIT();
	    gds__ () or isc_ () call.
	THREAD_ENTER();

-- calls through embedded GDML.

the following protocol will be used.  Care should be taken if
nested FOR loops are added.

    THREAD_EXIT();                // last statment before FOR loop 

    FOR ...............

	THREAD_ENTER();           // First statment in FOR loop 
	.....some C code....
	.....some C code....
	THREAD_EXIT();            // last statment in FOR loop 

    END_FOR;

    THREAD_ENTER();               // First statment after FOR loop 
***************************************************************/

DATABASE DB = STATIC "yachts.lnk";

static const UCHAR blr_bpb[] = {
	isc_bpb_version1,
	isc_bpb_source_type, 1, isc_blob_blr,
	isc_bpb_target_type, 1, isc_blob_blr
};

//static void check_array(dsql_req*, const SCHAR*, dsql_fld*);
static void convert_dtype(dsql_fld*, SSHORT);
static void	free_function (dsql_udf*);
static void free_procedure(dsql_prc*);
static void free_relation(dsql_rel*);
static void insert_symbol(dsql_dbb*, dsql_sym*);
static dsql_sym* lookup_symbol(dsql_dbb*, USHORT, const char*, SYM_TYPE, USHORT = 0);
static dsql_sym* lookup_symbol(dsql_dbb*, const dsql_str*, SYM_TYPE, USHORT = 0);

namespace {
#ifdef  SUPERSERVER
	REC_MUTX_T rec_mutex;	// Recursive metadata mutex 
	
	class RecMutexInit
	{
	public:
		static void init() {
			THD_rec_mutex_init (&rec_mutex);
		}
		static void cleanup() {
			THD_rec_mutex_destroy (&rec_mutex);
		}
	};
	Firebird::InitMutex<RecMutexInit> recMutexInited;

	class RecMutexHolder {
	public:
		RecMutexHolder() {
			recMutexInited.init();
			THREAD_EXIT();
			THD_rec_mutex_lock (&rec_mutex);
			THREAD_ENTER();
		}
		~RecMutexHolder() {
		    THD_rec_mutex_unlock (&rec_mutex);
		}
	};
#else
	class RecMutexHolder {
	public:
		RecMutexHolder() { }
		~RecMutexHolder() { }
	};
#endif
}


void METD_drop_collation(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ d r o p _ c o l l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Drop a collation from our metadata, and
 *	the next caller who wants it will
 *	look up the new version.
 *
 *	Dropping will be achieved by marking the collation
 *	as dropped.  Anyone with current access can continue
 *	accessing it.
 *
 **************************************/
	RecMutexHolder holder;

	// If the symbol wasn't defined, we've got nothing to do 
	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_intlsym_collation, 0);
    
	if (symbol)
	{
		dsql_intlsym* intlsym = (dsql_intlsym*) symbol->sym_object;
		intlsym->intlsym_flags |= INTLSYM_dropped;
	}

	// mark other potential candidates as maybe dropped 
	HSHD_set_flag(request->req_dbb,
				  reinterpret_cast<const TEXT*>(name->str_data),
				  name->str_length, SYM_intlsym_collation, INTLSYM_dropped);
}


void METD_drop_function(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ d r o p _ f u n c t i o n
 *
 **************************************
 *
 * Functional description
 *	Drop a user defined function from our metadata, and
 *	the next caller who wants it will
 *	look up the new version.
 *
 *	Dropping will be achieved by marking the function
 *	as dropped.  Anyone with current access can continue
 *	accessing it.
 *
 **************************************/
    RecMutexHolder holder;

    // If the symbol wasn't defined, we've got nothing to do 
    
    dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_udf);
    
    if (symbol) {
        dsql_udf* userFunc = (dsql_udf*) symbol->sym_object;
        userFunc->udf_flags |= UDF_dropped;
    }

    // mark other potential candidates as maybe dropped 

    HSHD_set_flag(request->req_dbb,
				  reinterpret_cast<const TEXT*>(name->str_data),
				  name->str_length, SYM_udf, UDF_dropped);

}


void METD_drop_procedure(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ d r o p _ p r o c e d u r e
 *
 **************************************
 *
 * Functional description
 *	Drop a procedure from our metadata, and
 *	the next caller who wants it will 
 *	look up the new version. 
 *	
 *	Dropping will be achieved by marking the procedure
 *	as dropped.  Anyone with current access can continue
 *	accessing it.
 *
 **************************************/
    RecMutexHolder holder;

	// If the symbol wasn't defined, we've got nothing to do 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_procedure);

	if (symbol) {
		dsql_prc* procedure = (dsql_prc*) symbol->sym_object;
		procedure->prc_flags |= PRC_dropped;
	}

	// mark other potential candidates as maybe dropped 

	HSHD_set_flag(request->req_dbb, (TEXT*)name->str_data, name->str_length,
				  SYM_procedure, PRC_dropped);

}


void METD_drop_relation(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ d r o p _ r e l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Drop a relation from our metadata, and
 *	rely on the next guy who wants it to
 *	look up the new version. 
 *
 *      Dropping will be achieved by marking the relation
 *      as dropped.  Anyone with current access can continue
 *      accessing it.
 *
 **************************************/
    RecMutexHolder holder;

	// If the symbol wasn't defined, we've got nothing to do 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_relation);

	if (symbol) {
		dsql_rel* relation = (dsql_rel*) symbol->sym_object;
		relation->rel_flags |= REL_dropped;
	}

	// mark other potential candidates as maybe dropped 

	HSHD_set_flag(request->req_dbb, (TEXT*)name->str_data, name->str_length,
				  SYM_relation, REL_dropped);

}


dsql_intlsym* METD_get_collation(dsql_req* request,
								 const dsql_str* name,
								 USHORT charset_id)
{
/**************************************
 *
 *	M E T D _ g e t _ c o l l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Look up an international text type object.
 *	If it doesn't exist, return NULL.
 *
 **************************************/
    RecMutexHolder holder;

	// Start by seeing if symbol is already defined 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_intlsym_collation, charset_id);
	if (symbol)
		return (dsql_intlsym*) symbol->sym_object;

	// Now see if it is in the database 

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	dsql_intlsym* iname = NULL;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_collation] TRANSACTION_HANDLE request->req_trans)
	X IN RDB$COLLATIONS
		CROSS Y IN RDB$CHARACTER_SETS OVER RDB$CHARACTER_SET_ID
		WITH X.RDB$COLLATION_NAME EQ name->str_data AND
			 X.RDB$CHARACTER_SET_ID EQ charset_id;

	THREAD_ENTER();

	iname = FB_NEW_RPT(*dbb->dbb_pool, name->str_length) dsql_intlsym;
	strcpy(iname->intlsym_name, (TEXT *) name->str_data);
	iname->intlsym_flags = 0;
	iname->intlsym_charset_id = X.RDB$CHARACTER_SET_ID;
	iname->intlsym_collate_id = X.RDB$COLLATION_ID;
	iname->intlsym_ttype =
		INTL_CS_COLL_TO_TTYPE(iname->intlsym_charset_id,
							  iname->intlsym_collate_id);
	iname->intlsym_bytes_per_char =
		(Y.RDB$BYTES_PER_CHARACTER.NULL) ? 1 : (Y.RDB$BYTES_PER_CHARACTER);
	THREAD_EXIT();

	END_FOR 
	ON_ERROR
		THREAD_ENTER();
		ERRD_punt(isc_status);
	END_ERROR;

	THREAD_ENTER();

	if (!iname)
		return NULL;

	// Store in the symbol table 

	symbol = iname->intlsym_symbol = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_sym;
	symbol->sym_object = (BLK) iname;
	symbol->sym_string = iname->intlsym_name;
	symbol->sym_length = name->str_length;
	symbol->sym_type = SYM_intlsym_collation;
	symbol->sym_dbb = dbb;
	insert_symbol(request->req_dbb, symbol);

	return iname;
}


USHORT METD_get_col_default(dsql_req* request,
							const char* for_rel_name,
							const char* for_col_name,
							bool* has_default,
							TEXT* buffer,
							USHORT buff_length)
{
/*************************************************************
 *
 *  M E T D _ g e t _ c o l _ d e f a u l t
 *
 **************************************************************
 *
 * Function:
 *    Gets the default value for a column of an existing table.
 *    Will check the default for the column of the table, if that is 
 *    not present, will check for the default of the relevant domain
 * 
 *    The default blr is returned in buffer. The blr is of the form
 *    blr_version4 blr_literal ..... blr_eoc
 * 
 *    Reads the system tables RDB$FIELDS and RDB$RELATION_FIELDS.
 *
 **************************************************************/
	dsql_dbb* dbb;
	ISC_STATUS_ARRAY status_vect;

	ISC_QUAD *blob_id;
	TEXT *ptr_in_buffer;

	ISC_STATUS stat;
	USHORT length, result = 0;
	isc_blob_handle blob_handle = 0;

	*has_default = false;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	// V4.x multi threading requirements 
	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_col_default] TRANSACTION_HANDLE request->req_trans)
		RFL IN RDB$RELATION_FIELDS CROSS
		FLD IN RDB$FIELDS WITH
		RFL.RDB$RELATION_NAME EQ for_rel_name AND
		RFL.RDB$FIELD_SOURCE EQ FLD.RDB$FIELD_NAME AND
		RFL.RDB$FIELD_NAME EQ for_col_name
		
	THREAD_ENTER();

	if (!RFL.RDB$DEFAULT_VALUE.NULL) {
		blob_id = &RFL.RDB$DEFAULT_VALUE;
		*has_default = true;
	}
	else if (!FLD.RDB$DEFAULT_VALUE.NULL) {
		blob_id = &FLD.RDB$DEFAULT_VALUE;
		*has_default = true;
	}
	else
		*has_default = false;

	if (*has_default) {
		// open the blob 
		THREAD_EXIT();
		stat = isc_open_blob2(status_vect, &DB, &request->req_trans,
							  &blob_handle, blob_id, sizeof(blr_bpb),
							  blr_bpb);
		THREAD_ENTER();
		if (stat)
			ERRD_punt(status_vect);

		// fetch segments. Assuming here that the buffer is big enough. 
		ptr_in_buffer = buffer;
		while (true) {
			THREAD_EXIT();
			stat = isc_get_segment(status_vect, &blob_handle,
								   &length, buff_length, ptr_in_buffer);
			THREAD_ENTER();
			ptr_in_buffer = ptr_in_buffer + length;
			buff_length = buff_length - length;
			result += length;

			if (!stat)
				continue;
			else if (stat == isc_segstr_eof) {
				// null terminate the buffer 
				*ptr_in_buffer = 0;
				break;
			}
			else
				ERRD_punt(status_vect);

		}
		THREAD_EXIT();
		isc_close_blob(status_vect, &blob_handle);
		THREAD_ENTER();

		/* the default string must be of the form:
		   blr_version4 blr_literal ..... blr_eoc  */
		fb_assert((buffer[0] == blr_version4) || (buffer[0] == blr_version5));
		fb_assert(buffer[1] == blr_literal);

	}
	else {
		if (request->req_dbb->dbb_db_SQL_dialect > SQL_DIALECT_V5)
			buffer[0] = blr_version5;
		else
			buffer[0] = blr_version4;
		buffer[1] = blr_eoc;
		result = 2;
	}

	THREAD_EXIT();

	END_FOR 
	ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
	END_ERROR 
	
	THREAD_ENTER();

	return result;
}


dsql_intlsym* METD_get_charset(dsql_req* request, USHORT length, const char* name) // UTF-8
{
/**************************************
 *
 *	M E T D _ g e t _ c h a r s e t
 *
 **************************************
 *
 * Functional description
 *	Look up an international text type object.
 *	If it doesn't exist, return NULL.
 *
 **************************************/
    RecMutexHolder holder;

	// Start by seeing if symbol is already defined 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, length, name, SYM_intlsym_charset);
	if (symbol)
		return (dsql_intlsym*) symbol->sym_object;

	// Now see if it is in the database 

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	dsql_intlsym* iname = NULL;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_charset] TRANSACTION_HANDLE request->req_trans)
	X IN RDB$COLLATIONS
		CROSS Y IN RDB$CHARACTER_SETS OVER RDB$CHARACTER_SET_ID
		CROSS Z IN RDB$TYPES
		WITH Z.RDB$TYPE EQ Y.RDB$CHARACTER_SET_ID
		AND Z.RDB$TYPE_NAME EQ name
		AND Z.RDB$FIELD_NAME EQ "RDB$CHARACTER_SET_NAME"
		AND Y.RDB$DEFAULT_COLLATE_NAME EQ X.RDB$COLLATION_NAME;

	    THREAD_ENTER();

        iname = FB_NEW_RPT(*dbb->dbb_pool, length) dsql_intlsym;
        strcpy(iname->intlsym_name, name);
        iname->intlsym_flags = 0;
        iname->intlsym_charset_id = X.RDB$CHARACTER_SET_ID;
        iname->intlsym_collate_id = X.RDB$COLLATION_ID;
        iname->intlsym_ttype =
            INTL_CS_COLL_TO_TTYPE(iname->intlsym_charset_id,
                                  iname->intlsym_collate_id);
        iname->intlsym_bytes_per_char =
            (Y.RDB$BYTES_PER_CHARACTER.NULL) ? 1 : (Y.RDB$BYTES_PER_CHARACTER);

        THREAD_EXIT();

	END_FOR 
    ON_ERROR
		THREAD_ENTER();
		ERRD_punt(isc_status);
   	END_ERROR;

	THREAD_ENTER();

	if (!iname)
		return NULL;

	// Store in the symbol table 

	symbol = iname->intlsym_symbol = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_sym;
	symbol->sym_object = (BLK) iname;
	symbol->sym_string = iname->intlsym_name;
	symbol->sym_length = length;
	symbol->sym_type = SYM_intlsym_charset;
	symbol->sym_dbb = dbb;
	insert_symbol(request->req_dbb, symbol);

	dbb->dbb_charsets_by_id.add(iname); 

	return iname;
}


USHORT METD_get_charset_bpc(dsql_req* request, SSHORT charset_id)
{
/**************************************
 *
 *	M E T D _ g e t _ c h a r s e t _ b p c
 *
 **************************************
 *
 * Functional description
 *	Look up an international text type object.
 *	If it doesn't exist, return NULL.
 *	Go directly to system tables & return only the
 *	number of bytes per character. Lookup by
 *	charset' id, not by name.
 *
 **************************************/
	dsql_dbb* dbb;
	USHORT bpc = 0;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	if (charset_id == CS_dynamic)
		charset_id = dbb->dbb_att_charset;

	dsql_intlsym* cs_sym = 0;
	size_t pos = 0;
	if (dbb->dbb_charsets_by_id.find(charset_id, pos)) {
		cs_sym = dbb->dbb_charsets_by_id[pos];
	}
	else {
		Firebird::MetaName cs_name = METD_get_charset_name(request, charset_id);
		cs_sym = METD_get_charset(request, cs_name.length(), cs_name.c_str());
	}
	return cs_sym->intlsym_bytes_per_char;
}


Firebird::MetaName METD_get_charset_name(dsql_req* request, SSHORT charset_id)
{
/**************************************
 *
 *	M E T D _ g e t _ c h a r s e t _ n a m e
 *
 **************************************
 *
 * Functional description
 *	Look up an international text type object.
 *	If it doesn't exist, return empty string.
 *	Go directly to system tables & return only the
 *	name.
 *
 **************************************/
    Firebird::MetaName name;
    
	ISC_STATUS_ARRAY isc_status = {0};
    dsql_dbb* dbb = request->req_dbb;
    isc_db_handle DB = dbb->dbb_database_handle;
    
    if (charset_id == CS_dynamic)
		charset_id = dbb->dbb_att_charset;
    
	size_t pos = 0;
	if (dbb->dbb_charsets_by_id.find(charset_id, pos))
	{
		return dbb->dbb_charsets_by_id[pos]->intlsym_name;
	}

    THREAD_EXIT();
    
    FOR (REQUEST_HANDLE dbb->dbb_requests[irq_cs_name] TRANSACTION_HANDLE request->req_trans)
        Y IN RDB$CHARACTER_SETS
        WITH Y.RDB$CHARACTER_SET_ID EQ charset_id

        THREAD_ENTER();

		name = Y.RDB$CHARACTER_SET_NAME;

        THREAD_EXIT();
    END_FOR
    ON_ERROR
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

    THREAD_ENTER();

	// put new charset into hash table if needed
	METD_get_charset(request, name.length(), name.c_str());

    return name;
}


dsql_str* METD_get_default_charset(dsql_req* request)
{
/**************************************
 *
 *	M E T D _ g e t _ d e f a u l t _ c h a r s e t
 *
 **************************************
 *
 * Functional description
 *	Find the default character set for a database
 *
 **************************************/
	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	if (dbb->dbb_flags & DBB_no_charset)
		return NULL;

	if (dbb->dbb_dfl_charset)
		return dbb->dbb_dfl_charset;

	// Now see if it is in the database 

	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_default_cs] TRANSACTION_HANDLE request->req_trans)
		FIRST 1 DBB IN RDB$DATABASE
		WITH DBB.RDB$CHARACTER_SET_NAME NOT MISSING;

	    THREAD_ENTER();

        // Terminate ASCIIZ string on first trailing blank 
        fb_utils::exact_name(DBB.RDB$CHARACTER_SET_NAME);
        USHORT length = strlen(DBB.RDB$CHARACTER_SET_NAME);
        dbb->dbb_dfl_charset = FB_NEW_RPT(*dbb->dbb_pool, length) dsql_str;
        dbb->dbb_dfl_charset->str_length = length;
        dbb->dbb_dfl_charset->str_charset = NULL;
		const UCHAR* str_charset = (UCHAR*) DBB.RDB$CHARACTER_SET_NAME;
        for (char* p = dbb->dbb_dfl_charset->str_data; length; --length) {
            *p++ = *str_charset++;
		}

        THREAD_EXIT();
	END_FOR 
    ON_ERROR
		THREAD_ENTER();
		ERRD_punt(isc_status);
   	END_ERROR;

	THREAD_ENTER();

	if (dbb->dbb_dfl_charset == NULL)
		dbb->dbb_flags |= DBB_no_charset;

	return dbb->dbb_dfl_charset;
}


USHORT METD_get_domain(dsql_req* request, dsql_fld* field, const char* name) // UTF-8
{
/**************************************
 *
 *	M E T D _ g e t _ d o m a i n
 *
 **************************************
 *
 * Functional description
 *	Fetch domain information for field defined as 'name'
 *
 **************************************/

	bool found = false;

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_domain] TRANSACTION_HANDLE request->req_trans)
        FLX IN RDB$FIELDS WITH FLX.RDB$FIELD_NAME EQ name 

        THREAD_ENTER();

	    found = true;
        field->fld_length = FLX.RDB$FIELD_LENGTH;
        field->fld_scale = FLX.RDB$FIELD_SCALE;
        field->fld_sub_type = FLX.RDB$FIELD_SUB_TYPE;

        field->fld_character_set_id = 0;
        if (!FLX.RDB$CHARACTER_SET_ID.NULL)
            field->fld_character_set_id = FLX.RDB$CHARACTER_SET_ID;
        field->fld_collation_id = 0;
        if (!FLX.RDB$COLLATION_ID.NULL)
            field->fld_collation_id = FLX.RDB$COLLATION_ID;
        field->fld_character_length = 0;
        if (!FLX.RDB$CHARACTER_LENGTH.NULL)
            field->fld_character_length = FLX.RDB$CHARACTER_LENGTH;
        
        if (!FLX.RDB$COMPUTED_BLR.NULL)
            field->fld_flags |= FLD_computed;

		if (FLX.RDB$SYSTEM_FLAG == 1)
			field->fld_flags |= FLD_system;

        convert_dtype(field, FLX.RDB$FIELD_TYPE);
    
        if (FLX.RDB$FIELD_TYPE == blr_blob) {
            field->fld_seg_length = FLX.RDB$SEGMENT_LENGTH;
        }

        THREAD_EXIT();

	END_FOR 
    ON_ERROR
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();
    
	return found ? TRUE : FALSE;
}


USHORT METD_get_domain_default(dsql_req* request,
							   const TEXT* domain_name,
							   bool* has_default,
							   TEXT* buffer,
							   USHORT buff_length)
{
/*************************************************************
 *
 *  M E T D _ g e t _ d o m a i n _ d e f a u l t
 *
 **************************************************************
 *
 * Function:
 *    Gets the default value for a domain of an existing table.
 *
 **************************************************************/
	ISC_STATUS_ARRAY status_vect;

	*has_default = false;

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	USHORT result = 0;

	// V4.x multi threading requirements 
	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_domain_2] TRANSACTION_HANDLE request->req_trans)
        FLD IN RDB$FIELDS WITH FLD.RDB$FIELD_NAME EQ domain_name
        
		THREAD_ENTER();

        ISC_QUAD* blob_id;
	    if (!FLD.RDB$DEFAULT_VALUE.NULL) {
            blob_id = &FLD.RDB$DEFAULT_VALUE;
            *has_default = true;
        }
        else
            *has_default = false;

        if (*has_default) {
            // open the blob 
            THREAD_EXIT();
            isc_blob_handle blob_handle = 0;
            ISC_STATUS stat = isc_open_blob2(status_vect, &DB, &request->req_trans,
                                  &blob_handle, blob_id, sizeof(blr_bpb),
                                  blr_bpb);
            THREAD_ENTER();
            if (stat)
                ERRD_punt(status_vect);

            // fetch segments. Assume buffer is big enough.  
            TEXT* ptr_in_buffer = buffer;
            while (true) {
                THREAD_EXIT();
                USHORT length;
                stat = isc_get_segment(status_vect,
                                       &blob_handle,
                                       &length, buff_length, ptr_in_buffer);
                THREAD_ENTER();
                ptr_in_buffer = ptr_in_buffer + length;
                buff_length = buff_length - length;
				result += length;
                
                if (!stat)
                    continue;
                else if (stat == isc_segstr_eof) {
                    // null terminate the buffer 
                    *ptr_in_buffer = 0;
                    break;
                }
                else
                    ERRD_punt(status_vect);

            }
            THREAD_EXIT();
            isc_close_blob(status_vect, &blob_handle);
            THREAD_ENTER();

            /* the default string must be of the form:
               blr_version4 blr_literal ..... blr_eoc  */
            fb_assert((buffer[0] == blr_version4) || (buffer[0] == blr_version5));
            fb_assert(buffer[1] == blr_literal);

        }
        else {
            if (request->req_dbb->dbb_db_SQL_dialect > SQL_DIALECT_V5)
                buffer[0] = blr_version5;
            else
                buffer[0] = blr_version4;
            buffer[1] = blr_eoc;
			result = 2;
        }
        
        THREAD_EXIT();
        
    END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR 

    THREAD_ENTER();

	return result;
}


bool METD_get_exception(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ g e t _ e x c e p t i o n 
 *
 **************************************
 *
 * Functional description
 *	Look up an exception.
 *
 **************************************/
	dsql_dbb* dbb;
	bool found = false;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_exception] TRANSACTION_HANDLE request->req_trans)
		X IN RDB$EXCEPTIONS WITH
		X.RDB$EXCEPTION_NAME EQ name->str_data;

		THREAD_ENTER();

		found = true;

		THREAD_EXIT();
        
	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	return found;
}


dsql_udf* METD_get_function(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ g e t _ f u n c t i o n
 *
 **************************************
 *
 * Functional description
 *	Look up a user defined function.  If it doesn't exist,
 *	return NULL.
 *
 **************************************/
	dsql_nod* udf_param_node;
	dsql_nod** ptr;

    RecMutexHolder holder;

	// Start by seeing if symbol is already defined 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_udf);
	if (symbol)
        return (dsql_udf*) symbol->sym_object;

	// Now see if it is in the database 

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	dsql_udf* userFunc = NULL;
    DsqlNodStack stack;
    USHORT return_arg;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_function] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$FUNCTIONS WITH
		X.RDB$FUNCTION_NAME EQ name->str_data 

        THREAD_ENTER();

	    userFunc = FB_NEW_RPT(*dbb->dbb_pool, name->str_length) dsql_udf;
        /* Moved below as still can't say for sure it will be stored.
           Following the same logic for MET_get_procedure and MET_get_relation
        userFunc->udf_next = dbb->dbb_functions;
        dbb->dbb_functions = userFunc;
        */

        strcpy(userFunc->udf_name, (char*) name->str_data);
        return_arg = X.RDB$RETURN_ARGUMENT;
        userFunc->udf_arguments = 0;

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	if (!userFunc) {
		return NULL;
    }
    
	// Note: The following two requests differ in the fields which are
	// new since ODS7 (DBB_v3 flag).  One of the two requests
	// here will be cached in dbb_requests[ird_func_return],
	// the one that is appropriate for the database we are
	// working against.

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_func_return] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$FUNCTION_ARGUMENTS WITH
        X.RDB$FUNCTION_NAME EQ name->str_data
        SORTED BY X.RDB$ARGUMENT_POSITION

        THREAD_ENTER();

        if (X.RDB$ARGUMENT_POSITION == return_arg) {
            userFunc->udf_dtype = (X.RDB$FIELD_TYPE != blr_blob) ?
                gds_cvt_blr_dtype[X.RDB$FIELD_TYPE] : dtype_blob;
            userFunc->udf_scale = X.RDB$FIELD_SCALE;
            
            if (!X.RDB$FIELD_SUB_TYPE.NULL) {
                userFunc->udf_sub_type = X.RDB$FIELD_SUB_TYPE;
            }
            else {
                userFunc->udf_sub_type = 0;
            }
            /* CVC: We are overcoming a bug in ddl.c:put_field()
               when any field is defined: the length is not given for blobs. */
            if (X.RDB$FIELD_TYPE == blr_blob)
                userFunc->udf_length = sizeof (ISC_QUAD);
            else
                userFunc->udf_length = X.RDB$FIELD_LENGTH;
            
            if (!X.RDB$CHARACTER_SET_ID.NULL) {
                userFunc->udf_character_set_id = X.RDB$CHARACTER_SET_ID;
            }
        }

        else {
            DSC* d;
            /* udf_param_node = MAKE_node (type_nod, 0); */
		    udf_param_node = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_nod;
            d = &udf_param_node->nod_desc;
            d->dsc_dtype = (X.RDB$FIELD_TYPE != blr_blob) ?
                gds_cvt_blr_dtype[X.RDB$FIELD_TYPE] : dtype_blob;
            d->dsc_scale = X.RDB$FIELD_SCALE;
            if (!X.RDB$FIELD_SUB_TYPE.NULL) {
                d->dsc_sub_type = X.RDB$FIELD_SUB_TYPE;
            }
            else {
                d->dsc_sub_type = 0;
            }
            d->dsc_length = X.RDB$FIELD_LENGTH;
            if (d->dsc_dtype == dtype_varying) {
                d->dsc_length += sizeof (USHORT);
            }
            d->dsc_address = NULL;
            
            if (!X.RDB$CHARACTER_SET_ID.NULL) {
                if (d->dsc_dtype != dtype_blob) {
                    d->dsc_ttype() = X.RDB$CHARACTER_SET_ID;
                }
                else {
                    d->dsc_scale = X.RDB$CHARACTER_SET_ID;
                }
            }
            
            if (X.RDB$MECHANISM != Jrd::FUN_value && X.RDB$MECHANISM != Jrd::FUN_reference)
			{
                d->dsc_flags = DSC_nullable;
            }
            
            stack.push(udf_param_node);
        }

        THREAD_EXIT();

    END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();
    
	// Adjust the return type & length of the UDF to account for 
	// cstring & varying.  While a UDF can return CSTRING, we convert it
	// to VARCHAR for manipulation as CSTRING is not a SQL type.

	if (userFunc->udf_dtype == dtype_cstring) {
		userFunc->udf_dtype = dtype_varying;
		userFunc->udf_length += sizeof(USHORT);
		if (userFunc->udf_length > MAX_SSHORT)
			userFunc->udf_length = MAX_SSHORT;
	}
	else if (userFunc->udf_dtype == dtype_varying)
		userFunc->udf_length += sizeof(USHORT);

    /* Can't call MAKE_list because it allocates an initial list node
       in the default pool. */
	USHORT arg_count = stack.getCount();

    /* userFunc->udf_arguments = MAKE_node (nod_list, count); */
    userFunc->udf_arguments = FB_NEW_RPT(*dbb->dbb_pool, arg_count) dsql_nod;
    userFunc->udf_arguments->nod_type = nod_list;
    userFunc->udf_arguments->nod_count = arg_count;
    
    ptr = userFunc->udf_arguments->nod_arg + arg_count;
    while (stack.hasData())
	{
        *--ptr = stack.pop();
		// dimitr: adjust the UDF arguments for CSTRING
		if ((*ptr)->nod_desc.dsc_dtype == dtype_cstring)
			(*ptr)->nod_desc.dsc_dtype = dtype_text;
    }

    /* Since we could give up control due to the THREAD_EXIT and THEAD_ENTER
     * calls, another thread may have added the same udf in the mean time
     */

	if ((symbol = lookup_symbol(request->req_dbb, name, SYM_udf)))
	{
		// Get rid of all the stuff we just read in. Use existing one 
		free_function (userFunc);
		return (dsql_udf*) symbol->sym_object;
    }

    // Add udf in the front of the list. 
    userFunc->udf_next = dbb->dbb_functions;
    dbb->dbb_functions = userFunc;

	// Store in the symbol table 
	// The UDF_new_udf flag is not used, so nothing extra to check. 

	symbol = userFunc->udf_symbol = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_sym;
	symbol->sym_object = (BLK) userFunc;
	symbol->sym_string = userFunc->udf_name;
	symbol->sym_length = name->str_length;
	symbol->sym_type = SYM_udf;
	symbol->sym_dbb = dbb;
	insert_symbol(request->req_dbb, symbol);

	return userFunc;
}


dsql_nod* METD_get_primary_key(dsql_req* request, const dsql_str* relation_name)
{
/**************************************
 *
 *	M E T D _ g e t _ p r i m a r y _ k e y
 *
 **************************************
 *
 * Functional description
 *	Lookup the fields for the primary key
 *	index on a relation, returning a list 
 *	node of the fields.
 *
 **************************************/

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	dsql_nod* list = NULL;
	USHORT count = 0;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_primary_key] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$INDICES CROSS
		Y IN RDB$INDEX_SEGMENTS
		OVER RDB$INDEX_NAME CROSS
		Z IN RDB$RELATION_CONSTRAINTS
		OVER RDB$INDEX_NAME
		WITH Z.RDB$RELATION_NAME EQ relation_name->str_data
		AND Z.RDB$CONSTRAINT_TYPE EQ "PRIMARY KEY"
		SORTED BY Y.RDB$FIELD_POSITION 

        THREAD_ENTER();

	    if (!list)
            list = MAKE_node(nod_list, (int) X.RDB$SEGMENT_COUNT);
		dsql_str* field_name = MAKE_cstring(Y.RDB$FIELD_NAME);
        dsql_nod* field_node = MAKE_node(nod_field_name, (int) e_fln_count);
        field_node->nod_arg[e_fln_name] = (dsql_nod*) field_name;
        list->nod_arg[count] = field_node;
        count++;

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();
    
	return list;
}


dsql_prc* METD_get_procedure(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ g e t _ p r o c e d u r e
 *
 **************************************
 *
 * Functional description
 *	Look up a procedure.  If it doesn't exist, return NULL.
 *	If it does, fetch field information as well.
 *	If it is marked dropped, try to read from system tables
 *
 **************************************/
	dsql_dbb* dbb;
	dsql_fld* parameter;
	dsql_fld** ptr;
	dsql_prc* temp;
	SSHORT type, count, defaults;

    RecMutexHolder holder;

	// Start by seeing if symbol is already defined 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_procedure);
	if (symbol)
		return (dsql_prc*) symbol->sym_object;

	// see if the procedure is the one currently being defined in this request 

	if (((temp = request->req_procedure) != NULL) &&
		!strcmp((char*) temp->prc_name, (char*) name->str_data)) {
		return temp;
	}

	// now see if it is in the database 

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	dsql_prc* procedure = NULL;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_procedure] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$PROCEDURES WITH
		X.RDB$PROCEDURE_NAME EQ name->str_data 

        THREAD_ENTER();

	    fb_utils::exact_name(X.RDB$OWNER_NAME);

        procedure = FB_NEW_RPT(*dbb->dbb_pool,
                                 name->str_length + strlen(X.RDB$OWNER_NAME)) dsql_prc;
        procedure->prc_id = X.RDB$PROCEDURE_ID;
        
        procedure->prc_name = procedure->prc_data;
        procedure->prc_owner = procedure->prc_data + name->str_length + 1;
        strcpy(procedure->prc_name, (TEXT *) name->str_data);
        strcpy(procedure->prc_owner, X.RDB$OWNER_NAME);

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	if (!procedure) {
		return NULL;
	}

	// Lookup parameter stuff 

	for (type = 0; type < 2; type++) {
		if (type)
			ptr = &procedure->prc_outputs;
		else
			ptr = &procedure->prc_inputs;
		count = defaults = 0; 

		THREAD_EXIT();

		FOR(REQUEST_HANDLE dbb->dbb_requests[irq_parameters] TRANSACTION_HANDLE request->req_trans)
		    PR IN RDB$PROCEDURE_PARAMETERS CROSS
			FLD IN RDB$FIELDS
			WITH FLD.RDB$FIELD_NAME EQ PR.RDB$FIELD_SOURCE
			AND PR.RDB$PROCEDURE_NAME EQ name->str_data
			AND PR.RDB$PARAMETER_TYPE = type
			SORTED BY DESCENDING PR.RDB$PARAMETER_NUMBER 

			SSHORT pr_collation_id_null = TRUE;
			SSHORT pr_collation_id;
			SSHORT pr_default_value_null = TRUE;
			///SSHORT pr_null_flag_null = TRUE;
			///SSHORT pr_null_flag;

			if (ENCODE_ODS(dbb->dbb_ods_version, dbb->dbb_minor_version) >= ODS_11_1)
			{
				FOR(REQUEST_HANDLE dbb->dbb_requests[irq_parameters2] TRANSACTION_HANDLE request->req_trans)
					PR2 IN RDB$PROCEDURE_PARAMETERS
					WITH PR2.RDB$PROCEDURE_NAME EQ PR.RDB$PROCEDURE_NAME AND
						 PR2.RDB$PARAMETER_NAME EQ PR.RDB$PARAMETER_NAME

					pr_collation_id_null = PR2.RDB$COLLATION_ID.NULL;
					pr_collation_id = PR2.RDB$COLLATION_ID;

					pr_default_value_null = PR2.RDB$DEFAULT_VALUE.NULL;

					///pr_null_flag_null = PR2.RDB$NULL_FLAG.NULL;
					///pr_null_flag = PR2.RDB$NULL_FLAG;

				END_FOR
			}
 
			THREAD_ENTER();

			count++;
			// allocate the field block 

			fb_utils::exact_name(PR.RDB$PARAMETER_NAME);
			parameter = FB_NEW_RPT(*dbb->dbb_pool,
								   strlen(PR.RDB$PARAMETER_NAME)) dsql_fld;
			parameter->fld_next = *ptr;
			*ptr = parameter;

			// get parameter information 

			strcpy(parameter->fld_name, PR.RDB$PARAMETER_NAME);
			parameter->fld_id = PR.RDB$PARAMETER_NUMBER;
			parameter->fld_length = FLD.RDB$FIELD_LENGTH;
			parameter->fld_scale = FLD.RDB$FIELD_SCALE;
			parameter->fld_sub_type = FLD.RDB$FIELD_SUB_TYPE;
			parameter->fld_procedure = procedure;

			if (!FLD.RDB$CHARACTER_SET_ID.NULL)
				parameter->fld_character_set_id = FLD.RDB$CHARACTER_SET_ID;

			if (!pr_collation_id_null)
				parameter->fld_collation_id = pr_collation_id;
			else if (!FLD.RDB$COLLATION_ID.NULL)
				parameter->fld_collation_id = FLD.RDB$COLLATION_ID;

			convert_dtype(parameter, FLD.RDB$FIELD_TYPE);

			/***            
			if (!pr_null_flag_null)
			{
				if (!pr_null_flag)
					parameter->fld_flags |= FLD_nullable;
			}
			else if (!FLD.RDB$NULL_FLAG)
				parameter->fld_flags |= FLD_nullable;
			***/
			parameter->fld_flags |= FLD_nullable;

			if (FLD.RDB$FIELD_TYPE == blr_blob)
				parameter->fld_seg_length = FLD.RDB$SEGMENT_LENGTH;

			if (type == 0 &&
				(!pr_default_value_null ||
				 (fb_utils::implicit_domain(FLD.RDB$FIELD_NAME) &&
				  !FLD.RDB$DEFAULT_VALUE.NULL)))
			{
				defaults++;
			}

			THREAD_EXIT();

		END_FOR 
        ON_ERROR 
			THREAD_ENTER();
			ERRD_punt(isc_status);
        END_ERROR;

		THREAD_ENTER();

		if (type)
			procedure->prc_out_count = count;
		else {
			procedure->prc_in_count = count;
			procedure->prc_def_count = defaults;
		}
	}

	// Since we could give up control due to the THREAD_EXIT and THEAD_ENTER
	// calls, another thread may have added the same procedure in the mean time

	if ((symbol = lookup_symbol(request->req_dbb, name, SYM_procedure)))
	{
		// Get rid of all the stuff we just read in. Use existing one 
		free_procedure(procedure);
		return (dsql_prc*) symbol->sym_object;
	}

	// store in the symbol table unless the procedure is not yet committed 
	// CVC: This is strange, because PRC_new_procedure is never set. 

	if (!(procedure->prc_flags & PRC_new_procedure)) {
		// add procedure in the front of the list 

		procedure->prc_next = dbb->dbb_procedures;
		dbb->dbb_procedures = procedure;

		symbol = procedure->prc_symbol = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_sym;
		symbol->sym_object = (BLK) procedure;
		symbol->sym_string = procedure->prc_name;
		symbol->sym_length = name->str_length;
		symbol->sym_type = SYM_procedure;
		symbol->sym_dbb = dbb;
		insert_symbol(request->req_dbb, symbol);
	}

	return procedure;
}


dsql_rel* METD_get_relation(dsql_req* request, const dsql_str* name)
{
/**************************************
 *
 *	M E T D _ g e t _ r e l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Look up a relation.  If it doesn't exist, return NULL.
 *	If it does, fetch field information as well.
 *
 **************************************/
	dsql_dbb* dbb;
	dsql_fld* field;
	dsql_fld** ptr;
	dsql_rel* temp;
	tsql* tdsql;

	tdsql = DSQL_get_thread_data();

    RecMutexHolder holder;

	// Start by seeing if symbol is already defined 

	dsql_sym* symbol = lookup_symbol(request->req_dbb, name, SYM_relation);
	if (symbol)
		return (dsql_rel*) symbol->sym_object;

	// See if the relation is the one currently being defined in this request 

	if (((temp = request->req_relation) != NULL) &&
		!strcmp((char*) temp->rel_name, (char*) name->str_data)) {
		return temp;
	}

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	// If the relation id or any of the field ids have not yet been assigned,
	// and this is a type of request which does not use ids, prepare a
	// temporary relation block to provide information without caching it

	bool permanent = true;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_rel_ids] TRANSACTION_HANDLE request->req_trans)
	    REL IN RDB$RELATIONS
	    CROSS RFR IN RDB$RELATION_FIELDS OVER RDB$RELATION_NAME
	    WITH REL.RDB$RELATION_NAME EQ name->str_data
		AND (REL.RDB$RELATION_ID MISSING OR RFR.RDB$FIELD_ID MISSING)

        THREAD_ENTER();

		permanent = false;

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	// Now see if it is in the database 

	DsqlMemoryPool& pool =
		permanent ? *dbb->dbb_pool : *tdsql->getDefaultPool();

	dsql_rel* relation = NULL;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_relation] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$RELATIONS WITH X.RDB$RELATION_NAME EQ name->str_data

        THREAD_ENTER();

	    fb_utils::exact_name(X.RDB$OWNER_NAME);
	    const int length = name->str_length + strlen(X.RDB$OWNER_NAME);

        // Allocate from default or permanent pool as appropriate 

        if (!X.RDB$RELATION_ID.NULL) {
            relation = FB_NEW_RPT(pool, length) dsql_rel;
            relation->rel_id = X.RDB$RELATION_ID;
        }
        else if (!DDL_ids(request)) {
            relation = FB_NEW_RPT(pool, length) dsql_rel;
        }

        // fill out the relation information 

        if (relation) {
            relation->rel_name = relation->rel_data;
            relation->rel_owner = relation->rel_data + name->str_length + 1;
            strcpy(relation->rel_name, (TEXT *) name->str_data);
            strcpy(relation->rel_owner, X.RDB$OWNER_NAME);
            if (!(relation->rel_dbkey_length = X.RDB$DBKEY_LENGTH))
                relation->rel_dbkey_length = 8;
            // CVC: let's see if this is a table or a view. 
            if (!X.RDB$VIEW_BLR.NULL) {
                relation->rel_flags |= REL_view;
            }
			if (!X.RDB$EXTERNAL_FILE.NULL) {
				relation->rel_flags |= REL_external;
			}
        }

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	if (!relation) {
		return NULL;
	}

	// Lookup field stuff 

	ptr = &relation->rel_fields;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_fields] TRANSACTION_HANDLE request->req_trans)
	    FLX IN RDB$FIELDS CROSS
		RFR IN RDB$RELATION_FIELDS
		WITH FLX.RDB$FIELD_NAME EQ RFR.RDB$FIELD_SOURCE
		AND RFR.RDB$RELATION_NAME EQ name->str_data
		SORTED BY RFR.RDB$FIELD_POSITION

		THREAD_ENTER();

	    // allocate the field block 

	    fb_utils::exact_name(RFR.RDB$FIELD_NAME);
	    const int length = strlen(RFR.RDB$FIELD_NAME);

        // Allocate from default or permanent pool as appropriate 

		field = NULL;

        if (!RFR.RDB$FIELD_ID.NULL) {
            field = FB_NEW_RPT(pool, length) dsql_fld;
			field->fld_id = RFR.RDB$FIELD_ID;
        }
        else if (!DDL_ids(request)) {
            field = FB_NEW_RPT(pool, length) dsql_fld;
        }

		if (field) {
			*ptr = field;
			ptr = &field->fld_next;

			// get field information 

			strcpy(field->fld_name, RFR.RDB$FIELD_NAME);
			field->fld_length = FLX.RDB$FIELD_LENGTH;
			field->fld_scale = FLX.RDB$FIELD_SCALE;
			field->fld_sub_type = FLX.RDB$FIELD_SUB_TYPE;
			field->fld_relation = relation;

			if (!FLX.RDB$COMPUTED_BLR.NULL)
				field->fld_flags |= FLD_computed;

			convert_dtype(field, FLX.RDB$FIELD_TYPE);

			if (FLX.RDB$FIELD_TYPE == blr_blob) {
				field->fld_seg_length = FLX.RDB$SEGMENT_LENGTH;
			}

			if (!FLX.RDB$DIMENSIONS.NULL && FLX.RDB$DIMENSIONS) {
				field->fld_element_dtype = field->fld_dtype;
				field->fld_element_length = field->fld_length;
				field->fld_dtype = dtype_array;
				field->fld_length = sizeof(ISC_QUAD);
				field->fld_dimensions = FLX.RDB$DIMENSIONS;
			}

			if (!FLX.RDB$CHARACTER_SET_ID.NULL)
				field->fld_character_set_id = FLX.RDB$CHARACTER_SET_ID;

			if (!RFR.RDB$COLLATION_ID.NULL)
				field->fld_collation_id = RFR.RDB$COLLATION_ID;
			else if (!FLX.RDB$COLLATION_ID.NULL)
				field->fld_collation_id = FLX.RDB$COLLATION_ID;

			if (!(RFR.RDB$NULL_FLAG || FLX.RDB$NULL_FLAG)
				|| (relation->rel_flags & REL_view))
			{
				field->fld_flags |= FLD_nullable;
			}

            if (RFR.RDB$SYSTEM_FLAG == 1 || FLX.RDB$SYSTEM_FLAG == 1)
                field->fld_flags |= FLD_system;
		}

        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	// Since we could give up control due to the THREAD_EXIT and THREAD_ENTER,
	// another thread may have added the same table in the mean time

	if ((symbol = lookup_symbol(request->req_dbb, name, SYM_relation)))
	{
		free_relation(relation);
		return (dsql_rel*) symbol->sym_object;
	}

	// Add relation to front of list 

	if (permanent) {
		relation->rel_next = dbb->dbb_relations;
		dbb->dbb_relations = relation;

		// store in the symbol table unless the relation is not yet committed 

		symbol = relation->rel_symbol = FB_NEW_RPT(*dbb->dbb_pool, 0) dsql_sym;
		symbol->sym_object = (BLK) relation;
		symbol->sym_string = relation->rel_name;
		symbol->sym_length = name->str_length;
		symbol->sym_type = SYM_relation;
		symbol->sym_dbb = dbb;
		insert_symbol(request->req_dbb, symbol);
	}
	else {
		relation->rel_flags |= REL_new_relation;
	}

	return relation;
}


dsql_str* METD_get_trigger_relation(dsql_req* request, const dsql_str* name, USHORT* trig_type)
{
/**************************************
 *
 *	M E T D _ g e t _ t r i g g e r _ r e l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Look up a trigger's relation and return it's current type
 *
 **************************************/
	dsql_dbb* dbb;
	dsql_str* relation;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	relation = NULL;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_trigger] TRANSACTION_HANDLE request->req_trans)
		X IN RDB$TRIGGERS WITH X.RDB$TRIGGER_NAME EQ name->str_data

		THREAD_ENTER();

		fb_utils::exact_name(X.RDB$RELATION_NAME);

		relation = MAKE_string(X.RDB$RELATION_NAME, strlen(X.RDB$RELATION_NAME));
		*trig_type = X.RDB$TRIGGER_TYPE;

		THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	return relation;
}


bool METD_get_type(dsql_req* request, const dsql_str* name, char* field, SSHORT* value)
{
/**************************************
 *
 *	M E T D _ g e t _ t y p e 
 *
 **************************************
 *
 * Functional description
 *	Look up a symbolic name in RDB$TYPES
 *
 **************************************/
	dsql_dbb* dbb;
	bool found = false;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_type] TRANSACTION_HANDLE request->req_trans)
		X IN RDB$TYPES WITH
		X.RDB$FIELD_NAME EQ field AND X.RDB$TYPE_NAME EQ name->str_data;

		THREAD_ENTER();

		found = true;
		*value = X.RDB$TYPE;

		THREAD_EXIT();
        
	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	return found;
}


dsql_rel* METD_get_view_base(dsql_req* request,
							 const char* view_name,	// UTF-8
							 MetaNamePairMap& fields)
{
/**************************************
 *
 *	M E T D _ g e t _ v i e w _ b a s e
 *
 **************************************
 *
 * Functional description
 *	Return the base table of a view or NULL if there
 *  is more than one.
 *  If there is only one base, put in fields a map of
 *  top view field name / bottom base field name.
 *
 **************************************/

	dsql_rel* relation = NULL;

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;
	bool first = true;
	bool cont = true;
	MetaNamePairMap aux;

	fields.clear();

	while (cont)
	{
		THREAD_EXIT();

		FOR(REQUEST_HANDLE dbb->dbb_requests[irq_view_base]
			TRANSACTION_HANDLE request->req_trans)
			X IN RDB$VIEW_RELATIONS
			WITH X.RDB$VIEW_NAME EQ view_name

			// return NULL if there is more than one context
			if (X.RDB$VIEW_CONTEXT != 1)
			{
				relation = NULL;
				cont = false;
				break;
			}

			THREAD_ENTER();

			fb_utils::exact_name(X.RDB$CONTEXT_NAME);
			fb_utils::exact_name(X.RDB$RELATION_NAME);

			dsql_str* relation_name =
				MAKE_string(X.RDB$RELATION_NAME, strlen(X.RDB$RELATION_NAME));
			relation = METD_get_relation(request, relation_name);
			delete relation_name;

			THREAD_EXIT();

			if (!relation)
			{
				cont = false;
				break;
			}

			FOR(REQUEST_HANDLE dbb->dbb_requests[irq_view_base_flds]
				TRANSACTION_HANDLE request->req_trans)
				RFL IN RDB$RELATION_FIELDS
				WITH RFL.RDB$RELATION_NAME EQ X.RDB$VIEW_NAME

				THREAD_ENTER();

				if (first)
				{
					fields.put(
						(RFL.RDB$FIELD_NAME.NULL ? "" : RFL.RDB$FIELD_NAME),
						(RFL.RDB$BASE_FIELD.NULL ? "" : RFL.RDB$BASE_FIELD));

					aux.put(
						(RFL.RDB$BASE_FIELD.NULL ? "" : RFL.RDB$BASE_FIELD),
						(RFL.RDB$FIELD_NAME.NULL ? "" : RFL.RDB$FIELD_NAME));
				}
				else
				{
					Firebird::MetaName field;

					if (aux.get(RFL.RDB$FIELD_NAME, field))
					{
						fields.put(field,
							(RFL.RDB$BASE_FIELD.NULL ? "" : RFL.RDB$BASE_FIELD));

						aux.remove(RFL.RDB$FIELD_NAME);
						aux.put((RFL.RDB$BASE_FIELD.NULL ? "" : RFL.RDB$BASE_FIELD), field);
					}
				}

				THREAD_EXIT();
			END_FOR

			if (relation->rel_flags & REL_view)
				view_name = X.RDB$RELATION_NAME;
			else
			{
				cont = false;
				break;
			}

			first = false;

		END_FOR 
		ON_ERROR 
			THREAD_ENTER();
			ERRD_punt(isc_status);
		END_ERROR;

		THREAD_ENTER();
	}

	if (!relation)
		fields.clear();

	return relation;
}


dsql_rel* METD_get_view_relation(dsql_req* request,
								const char* view_name,         // UTF-8
								const char* relation_or_alias, // UTF-8
								USHORT level)
{
/**************************************
 *
 *	M E T D _ g e t _ v i e w _ r e l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Return TRUE if the passed view_name represents a 
 *	view with the passed relation as a base table 
 *	(the relation could be an alias).
 *
 **************************************/

	dsql_rel* relation = NULL;

	ISC_STATUS_ARRAY isc_status = {0};
	dsql_dbb* dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_view] TRANSACTION_HANDLE request->req_trans, LEVEL level)
		X IN RDB$VIEW_RELATIONS WITH X.RDB$VIEW_NAME EQ view_name

		THREAD_ENTER();

		fb_utils::exact_name(X.RDB$CONTEXT_NAME);
		fb_utils::exact_name(X.RDB$RELATION_NAME);

		if (!strcmp(X.RDB$RELATION_NAME, relation_or_alias) ||
			!strcmp(X.RDB$CONTEXT_NAME, relation_or_alias))
		{
			dsql_str* relation_name = MAKE_string(X.RDB$RELATION_NAME,
										strlen(X.RDB$RELATION_NAME));
			relation = METD_get_relation(request, relation_name);
			delete relation_name;
			return relation;
		}

		relation =
			METD_get_view_relation(request, X.RDB$RELATION_NAME,
								   relation_or_alias, (USHORT)level + 1);
		if (relation) {
			return relation;
		}

		THREAD_EXIT();

	END_FOR 
    ON_ERROR 
		THREAD_ENTER();
		ERRD_punt(isc_status);
    END_ERROR;

	THREAD_ENTER();

	return NULL;
}


#ifdef NOT_USED_OR_REPLACED
/* - to make gpre happy too
static void check_array(dsql_req* request, const TEXT* field_name, dsql_fld* field)
{
 **************************************
 *
 *	c h e c k _ a r r a y
 *
 **************************************
 *
 * Functional description
 *	Try to get field dimensions (RDB$FIELD_DIMENSIONS may not be
 *	defined in the meta-data).  If we're not successful, indicate
 *	so in the database block to avoid wasting further effort.
 *
 **************************************
	dsql_dbb* dbb;

	ISC_STATUS_ARRAY isc_status = {0};
	dbb = request->req_dbb;
	isc_db_handle DB = dbb->dbb_database_handle;

	THREAD_EXIT();

	FOR(REQUEST_HANDLE dbb->dbb_requests[irq_dimensions] TRANSACTION_HANDLE request->req_trans)
	    X IN RDB$FIELDS WITH X.RDB$FIELD_NAME EQ field_name 

        THREAD_ENTER();

	    if (!X.RDB$DIMENSIONS.NULL && X.RDB$DIMENSIONS) {
            field->fld_element_dtype = field->fld_dtype;
            field->fld_element_length = field->fld_length;
            field->fld_dtype = dtype_array;
            field->fld_length = sizeof(ISC_QUAD);
            field->fld_dimensions = X.RDB$DIMENSIONS;
        }
        
        THREAD_EXIT();

	END_FOR 
    ON_ERROR 
        dbb->dbb_flags |= DBB_no_arrays;
	END_ERROR;

	THREAD_ENTER();
}
*/
#endif

static void convert_dtype(dsql_fld* field, SSHORT field_type)
{
/**************************************
 *
 *	c o n v e r t _ d t y p e
 *
 **************************************
 *
 * Functional description
 *	Convert from the blr_<type> stored in system metadata
 *	to the internal dtype_* descriptor.  Also set field
 *	length.
 *
 **************************************/

// fill out the type descriptor 

	if (field_type == blr_text) {
		field->fld_dtype = dtype_text;
	}
	else if (field_type == blr_varying) {
		field->fld_dtype = dtype_varying;
		field->fld_length += sizeof(USHORT);
	}
	else if (field_type == blr_blob) {
		field->fld_dtype = dtype_blob;
		field->fld_length = type_lengths[field->fld_dtype];
	}
	else {
		field->fld_dtype = gds_cvt_blr_dtype[field_type];
		field->fld_length = type_lengths[field->fld_dtype];

		fb_assert(field->fld_dtype != dtype_unknown);
	}
}


static void free_function(dsql_udf* userFunc)
{
/**************************************
 *
 *	f r e e _ f u n c t i o n
 *
 **************************************
 *
 * Functional description
 *	Free memory allocated for a function block and args
 *
 **************************************/

    // release the arguments blocks 

    if (userFunc->udf_arguments) {
        dsql_nod* args = userFunc->udf_arguments;
        for (USHORT arg_count = 0; arg_count < args->nod_count; ++arg_count) {
            dsql_nod* udf_argument = args->nod_arg[arg_count];
            delete udf_argument;
		}
	}

    /* release the udf & symbol blocks */

    delete userFunc;
}

static void free_procedure(dsql_prc* procedure)
{
/**************************************
 *
 *	f r e e _ p r o c e d u r e
 *
 **************************************
 *
 * Functional description
 *	Free memory allocated for a procedure block and params
 *
 **************************************/
	dsql_fld* param;
	dsql_fld* temp;

/* release the input & output parameter blocks */

	for (param = procedure->prc_inputs; param;) {
		temp = param;
		param = param->fld_next;
		delete temp;
	}

	for (param = procedure->prc_outputs; param;) {
		temp = param;
		param = param->fld_next;
		delete temp;
	}

/* release the procedure & symbol blocks */

	delete procedure;
}


static void free_relation(dsql_rel* relation)
{
/**************************************
 *
 *	f r e e _ r e l a t i o n
 *
 **************************************
 *
 * Functional description
 *	Free memory allocated for a relation block and fields
 *
 **************************************/
	dsql_fld* field;
	dsql_fld* temp;

// release the field blocks 

	for (field = relation->rel_fields; field;) {
		temp = field;
		field = field->fld_next;
		delete temp;
	}

/* release the relation & symbol blocks */

	delete relation;
}


static void insert_symbol(dsql_dbb* dbb, dsql_sym* symbol)
{
/**************************************
 *
 *	i n s e r t _ s y m b o l
 *
 **************************************
 *
 * Functional description
 *	Insert a symbol in the hash table and
 *	inform the engine that we're using it.
 *
 **************************************/
	HSHD_insert(symbol);

#ifndef SUPERSERVER
	ISC_STATUS_ARRAY status;

	THREAD_EXIT();
	const ISC_STATUS s = gds__dsql_cache(status, &dbb->dbb_database_handle,
		DSQL_CACHE_USE, symbol->sym_type, symbol->sym_string, NULL);
	THREAD_ENTER();

	if (s)
		ERRD_punt(status);
#endif	// SUPERSERVER
}


static dsql_sym* lookup_symbol(dsql_dbb* dbb, USHORT length, const char* name, SYM_TYPE type, USHORT charset_id)
{
/**************************************
 *
 *	l o o k u p _ s y m b o l
 *
 **************************************
 *
 * Functional description
 *	Lookup a symbol in the hash table and
 *	inform the engine that we're using it.
 *  The engine may inform that the symbol
 *	is obsolete, in this case mark it as
 *	dropped and return NULL.
 *
 **************************************/
	dsql_intlsym* intlSym;
	dsql_prc* procedure;
	dsql_rel* relation;
	dsql_udf* userFunc;

    dsql_sym* symbol = HSHD_lookup(dbb, name, length, type, 0);

    for (; symbol; symbol = symbol->sym_homonym)
    {
        if (symbol->sym_type == type)
		{
			if (type == SYM_intlsym_charset)
				break;
			else if (type == SYM_intlsym_collation &&
					 (intlSym = (dsql_intlsym*) symbol->sym_object) &&
					 (!(intlSym->intlsym_flags & INTLSYM_dropped)) &&
					 (charset_id == 0 || intlSym->intlsym_charset_id == charset_id))
			{
				break;
			}
			else if (type == SYM_procedure &&
					 (procedure = (dsql_prc*) symbol->sym_object) &&
					 (!(procedure->prc_flags & PRC_dropped)))
			{
				break;
			}
			else if (type == SYM_relation &&
					 (relation = (dsql_rel*) symbol->sym_object) &&
					 (!(relation->rel_flags & REL_dropped)))
			{
				break;
			}
			else if (type == SYM_udf &&
					 (userFunc = (dsql_udf*) symbol->sym_object) &&
					 (!(userFunc->udf_flags & UDF_dropped)))
			{
				break;
			}
		}
    }

#ifndef SUPERSERVER
	if (symbol)
	{
		ISC_STATUS_ARRAY status;
		bool obsolete;

		THREAD_EXIT();
		const ISC_STATUS s = gds__dsql_cache(status, &dbb->dbb_database_handle,
			DSQL_CACHE_USE, type, name, &obsolete);
		THREAD_ENTER();

		if (s)
			ERRD_punt(status);

		if (obsolete)
		{
			switch (type)
			{
				case SYM_intlsym_collation:
					intlSym->intlsym_flags |= INTLSYM_dropped;
					break;

				case SYM_procedure:
					procedure->prc_flags |= PRC_dropped;
					break;

				case SYM_relation:
					relation->rel_flags |= REL_dropped;
					break;

				case SYM_udf:
					userFunc->udf_flags |= UDF_dropped;
					break;

				default:
					return symbol;
			}

			symbol = NULL;
		}
	}
#endif	// SUPERSERVER

	return symbol;    
}


static dsql_sym* lookup_symbol(dsql_dbb* dbb, const dsql_str* name, SYM_TYPE type, USHORT charset_id)
{
/**************************************
 *
 *	l o o k u p _ s y m b o l
 *
 **************************************
 *
 * Functional description
 *	Wrapper for the above function.
 *
 **************************************/
	return lookup_symbol(dbb, name->str_length,
		reinterpret_cast<const TEXT*>(name->str_data), type, charset_id);
}