/*!**************************************************************************

  module        : vbd491.cpp
  special area  : CreateIndexParticipant
  see also      : 
  author        : TorstenS
  responsible   : TorstenS
  last changed  : 1999-09-10  18:00
  copyright     : (c) 1999-2004 SAP AG
  implementation: 
  description   : 



    ========== licence begin  GPL
    Copyright (c) 1999-2005 SAP AG

    This program is free software; you can redistribute it and/or
    modify it under the terms of the GNU General Public License
    as published by the Free Software Foundation; either version 2
    of the License, or (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
    ========== licence end


*****************************************************************************/



/*===========================================================================*
 *  INCLUDES                                                                 *
 *===========================================================================*/

// Content of include files
#include "gbd491.h"
#include "gbd490.h"
#include "ggg200.h"   // CPP   : Sort Templates
#include "gsp03.h"    // PASCAL: SP_message_constants_and_types
#include "gsp03_3.h"  // PASCAL: SP_message_constants_and_types_for_bd_layer
/* */
#include "hbd03.h"    // CPP   : filesysteminterface_3
#include "hbd13.h"    // PASCAL: Nodehandling
#include "hbd30.h"    // PASCAL: Treehandling  
#include "hbd31.h"    // PASCAL: Leafhandling 
#include "hbd35.h"    // PASCAL: entryhandling 
#include "hbd50.h"    // PASCAL: Indexhandling
#include "hbd400.h"   // CPP   : Inverted Lists Add/Delete
#include "hkb33.h"    // PASCAL: KB_index_handling
#include "hgg01_1.h"  // PASCAL: Configuration_Parameter
#include "hgg01_3.h"  // PASCAL: Configuration_Parameter
#include "hgg04.h"    // PASCAL: Select_Help_Procedures
#include "SAPDB/SAPDBCommon/SAPDB_RangeCode.hpp" // Kernel_move_and_fill
#include "hgg10.h"
#include "heo56.h"    // RTE   : Vsleep
#include "heo57.h"    // RTE   : Vmalloc/Vmfree
#include "hsp30.h"    // PASCAL: Buffer handling and comparison routines

#include "Converter/Converter_IPageNoManager.hpp"
#include "KernelCommon/Kernel_Common.hpp"


#if COMPILEMODE_MEO00 >= SLOW_MEO00 
#include "hta99.h"
#endif

/*===========================================================================*
 *  DEFINES                                                                  *
 *===========================================================================*/


/*===========================================================================*
 *  MACROS                                                                   *
 *===========================================================================*/

#define INCLUSIVE_PRIM_KEY_BD491    true
#define INIT_INV_TREE_BD491         true
#define RESIZEABLE_BD491            true

#define FIRST_COLUMN_BD491          1
#define MAX_ALLOC_RETRIES_BD491     2
#define NUM_PRIM_KEY_DESC_BD491     1

#define ALLOCATE_MSG_1_BD491        "Allocate Failed: No More Memory         "
#define ALLOCATE_MSG_2_BD491        "Realloc Of Dynamic Vector Not Possible  "
#define ALLOCATE_MSG_3_BD491        "Realloc Of Dynamic Vector               "
#define OVERFLOW_MSG_1_BD491        "Quick Sort Failed: Data Cache Too Small "
#define CANCEL_MSG_1_BD491          "Create Index Sequential Canceled        "


/*===========================================================================*
 *  FORWARD DECLARATION                                                      *
 *===========================================================================*/


/*===========================================================================*
 *  LOCAL CLASSES, STRUCTURES, TYPES, UNIONS ...                             *
 *===========================================================================*/

struct tbd491_KeyDesc
{
    tsp00_BytePtr   pKey_bd491;
    tsp00_Int4      KeyLen_bd491;
#   if defined (BIT64)
    tsp00_Int4      Filler1_bd491;
#   endif
};

typedef tbd491_KeyDesc* tbd491_KeyDescPtr;

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable>
class cbd491_DynamicVector
{
public:
    cbd491_DynamicVector (tgg00_TransContext &Trans,
                          tsp00_Int4          InitialMaxItems);

    ~cbd491_DynamicVector () {if (NULL != m_pItem) vmfree (m_pItem);};


    Type*       operator [] (tsp00_Int4 Index);

    void        bd491_AppendItem (Type        &AppendItem,
                                  tsp00_Int4   NumReallocItems = 10);

    Type*       bd491_GetSpaceForItems (tsp00_Int4  NumItems,
                                        tsp00_Int4  NumReallocItems = 10);

    tsp00_Int4  bd491_MaxItems () {return m_MaxItems;};

    tsp00_Int4  bd491_NumItemsUsed () {return m_NumItemsUsed;};

    void        bd491_ReleaseSpaceForItems (tsp00_Int4  NumItems);

private:

    tgg00_BasisError    &m_TrError;
    tgg00_TransContext  &m_Trans;
    /* */
    tsp00_Int4           m_MaxItems;
    tsp00_Int4           m_NumItemsUsed;
    /* */
    Type                *m_pItem;

    void bd491_Resize (tsp00_Int4   NumReallocItems);
};

/*---------------------------------------------------------------------------*/

class cbd491_InvRecDescItem
{
public:

    cbd491_InvRecDescItem (
        tgg00_TransContext   &Trans,
        cbd497_StackDescInfo &StackDescInfo,
        tsp00_BytePtr         pUndefNullKey = NULL);

    void bd491_BuildVirtualRecord (
        tbd491_KeyDescPtr  pKeyDesc,
        tgg00_RecPtr       pRec);

    void bd491_BuildPrimKey (
        tbd491_KeyDescPtr   pKeyDesc,
        tsp00_BytePtr       pPrimKey,
        tsp00_Int4          AvailableSpace);

    void bd491_BuildSecKey (
        tbd491_KeyDescPtr   pKeyDesc,
        tsp00_BytePtr       pSecKey,
        tsp00_Int4          AvailableSpace);

    tsp00_LcompResult bd491_CompareVirtualRecords (
        bool              bInclusivePrimKey,
        tbd491_KeyDescPtr pKeyDescLeft,
        tbd491_KeyDescPtr pKeyDescRight);

    tsp00_Int4 bd491_GetPrimKeyLen (tbd491_KeyDescPtr   pKeyDesc);

    tsp00_Int4 bd491_GetSecKeyLen (tbd491_KeyDescPtr    pKeyDesc);


#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
    void                bd491_PrintVirtualRecord (tbd491_KeyDescPtr  pKeyDesc);
#   endif

protected:

    tgg00_BasisError     &m_TrError;
    tgg00_TransContext   &m_Trans;
    cbd497_StackDescInfo &m_Stack;
    const tsp00_BytePtr   m_pUndefNullKey;

protected:

    tsp00_LcompResult   bd491_ColumnCompare (
        bool          bDescending,
        bool          bLastColumn,
        bool          &bLeftColumnIsNull,  // PTS 1124759 TS 2003-10-20
        bool          &bRightColumnIsNull, // PTS 1124759 TS 2003-10-20
        tsp00_BytePtr pKeyLeft,
        tsp00_BytePtr pKeyRight,
        tsp00_Int4    KeyLenLeft,
        tsp00_Int4    KeyLenRight);
};

/*---------------------------------------------------------------------------*/

class cbd491_InvRecDescList
{
public:

    cbd491_InvRecDescList (tgg00_TransContext  &Trans,
                           tsp00_Int4           NumRec,
                           tsp00_Int4           NumPrimKeyDesc,
                           tsp00_Int4           NumSecKeyDesc);

    tbd491_KeyDescPtr   bd491_NewRecDesc();

    tsp00_Int4          bd491_NumRecDescUsed() {return m_InvRecDescList.bd491_NumItemsUsed();};

    tbd491_KeyDescPtr*  bd491_FirstRecDescAddr() {return m_InvRecDescList [0];};

    tbd491_KeyDescPtr   bd491_GetRecDesc (tsp00_Int4 iItem) {return *m_InvRecDescList [iItem];};

private:

    tgg00_BasisError    &m_TrError;
    /* */
    tsp00_Int4          m_NumPrimKeyDesc;
    tsp00_Int4          m_NumSecKeyDesc;
    /* */
    cbd491_DynamicVector <tbd491_KeyDescPtr, RESIZEABLE_BD491> m_InvRecDescList;
    cbd491_DynamicVector <tbd491_KeyDesc,   !RESIZEABLE_BD491> m_KeyDescPool;

    // +++ TODO Overflow Pool
};

/*---------------------------------------------------------------------------*/

class cbd491_CompareInvRecDescItem:
            public cbd491_InvRecDescItem,
            public cgg200ComparisonOperator<tbd491_KeyDescPtr>
{
public:

    cbd491_CompareInvRecDescItem (
        tgg00_TransContext      &Trans,
        cbd497_StackDescInfo    &StackDescInfo);

    tsp00_LcompResult_Enum gg200Compare (
        const tbd491_KeyDescPtr &Left,
        const tbd491_KeyDescPtr &Right);
};

/*---------------------------------------------------------------------------*/

class cbd491_NodeList
{
public:

    cbd491_NodeList (tbd_current_tree &Current, tsp00_Int4  NumLeavesToScan);

    ~cbd491_NodeList ();


    bool bd491_EqualLastStoredNode (tbd_node_ptrs &pNode);

    void bd491_StoreOneNode (tbd_node_ptrs &pNode);

private:

    tgg00_BasisError        &m_TrError;
    tgg00_TransContext      &m_Trans;
    tbd_current_tree        &m_Current;
    cbd491_DynamicVector <tbd_node_ptrs, RESIZEABLE_BD491> m_NodeList;
};

/*===========================================================================*
 *  EXTERNAL VARIABLES                                                       *
 *===========================================================================*/



/*===========================================================================*
 *  GLOBAL VARIABLES                                                         *
 *===========================================================================*/



/*===========================================================================*
 *  LOCAL VARIABLES                                                          *
 *===========================================================================*/



/*===========================================================================*
 *  LOCAL FUNCTIONS (PROTOTYPES)                                             *
 *===========================================================================*/



/*===========================================================================*
 *  GLOBAL FUNCTIONS (CODE)                                                  *
 *===========================================================================*/



/*===========================================================================*
 *  LOCAL FUNCTIONS (CODE)                                                   *
 *===========================================================================*/

#if COMPILEMODE_MEO00 >= SLOW_MEO00 
inline void
bd491_CheckKeyOrder (
    tsp00_LcompResult  CompResult,
    tbd491_KeyDescPtr  pPrevKeyDesc,
    tbd491_KeyDescPtr  pKeyDesc)
{
    ROUTINE_DBG_MEO00 ("bd491_CheckKeyOrder");

    if (l_greater == CompResult)
    {
        g01abort (csp3_bd_msg, csp3_n_btree, "Wrong Sec Key Order     ", 0);
    }
    if (l_equal == CompResult)
    {
        tsp00_LcompResult   PrimCompResult (tsp00_LcompResult::fromConst (l_undef));

        s30cmp (pPrevKeyDesc->pKey_bd491, POS_OFF_DIFF_BD00, pPrevKeyDesc->KeyLen_bd491,
                pKeyDesc->pKey_bd491, POS_OFF_DIFF_BD00, pKeyDesc->KeyLen_bd491,
                PrimCompResult);

        if (l_less != PrimCompResult)
        {
            t01sname (bd_idx_create, "Prim Key    ");
            t01buf   (bd_idx_create, pKeyDesc->pKey_bd491, POS_OFF_DIFF_BD00, pKeyDesc->KeyLen_bd491);
            g01abort (csp3_bd_msg, csp3_n_btree, "Wrong Prim Key Order    ", 0);
        }
    }
}
#endif

/*---------------------------------------------------------------------------*/

// PTS 1108452 TS 2000-11-24
inline tsp00_LcompResult
bd491_CompareASCIIColumn (tsp00_Byte	DefineByte,
                          tsp00_BytePtr	pKey,
                          tsp00_Int4	KeyLen,
                          tsp00_Int4	iOffset)
{
    // ASCII defined column

    while ( iOffset < KeyLen )
    {
        if ( *(pKey + iOffset) == DefineByte )
            ++iOffset;
        else
        {
            if ( *(pKey + iOffset) > DefineByte )
                return tsp00_LcompResult::fromConst (l_greater);
            else
                return tsp00_LcompResult::fromConst (l_less);
        }
    }
    return tsp00_LcompResult::fromConst (l_equal);
}

/*---------------------------------------------------------------------------*/

// PTS 1108452 TS 2000-11-24
inline tsp00_LcompResult
bd491_CompareUnicodeColumn (
    tsp00_BytePtr	pKey,
    tsp00_Int4		KeyLen,
    tsp00_Int4		iOffset)
{
    // UNICODE (UCS-2) defined column

    // PTS 1108380 TS 2000-12-27
    // No more warnings on SUN because csp_unicode_blank is used inside the
    // the function instead of char *DefineBytes in parameter call list

    while ( iOffset < KeyLen )
    {
        if (
            ( *(pKey + iOffset  ) == csp_unicode_blank[0] ) &&
            ( *(pKey + iOffset+1) == csp_unicode_blank[1] )
        )
            iOffset +=2;
        else
        {
            if ( *(pKey + iOffset) == csp_unicode_blank[0] )
            {
                if ( *(pKey + iOffset +1) > csp_unicode_blank[1] )
                    return tsp00_LcompResult::fromConst (l_greater);
                else
                    return tsp00_LcompResult::fromConst (l_less);
            }

            if ( *(pKey + iOffset) > csp_unicode_blank[0] )
                return tsp00_LcompResult::fromConst (l_greater);
            else
                return tsp00_LcompResult::fromConst (l_less);
        }
    }
    return tsp00_LcompResult::fromConst (l_equal);
}

/*---------------------------------------------------------------------------*/


inline void*
bd491_Malloc (tsp00_Int4    AllocSize,
              tsp00_TaskId  TaskId)

{
    ROUTINE_DBG_MEO00 ("bd491_Malloc");

    tsp00_Bool bOkay    = false;
    tsp00_Int4 RetryCnt = 0;
    void       *pObject = NULL;

    do
    {
        vmalloc (AllocSize, &pObject, &bOkay);

        if (bOkay) 
            return (pObject);

        ++RetryCnt;
        vsleep (TaskId, 1);
    }
    while (MAX_ALLOC_RETRIES_BD491 > RetryCnt);

    return (NULL);
}

/*===========================================================================*
 *   DEFINITION OF METHODS OF LOCAL CLASSES (CODE)                           *
 *===========================================================================*/

template <class Type, bool Resizeable> inline
cbd491_DynamicVector<Type, Resizeable>::cbd491_DynamicVector (
    tgg00_TransContext &Trans,
    tsp00_Int4          InitialMaxItems
)
        :
        m_TrError      (Trans.trError_gg00),
        m_Trans        (Trans),
        m_NumItemsUsed (0),
        m_pItem        (NULL)
{
    ROUTINE_DBG_MEO00 ("cbd491_DynamicVector");


    if (e_ok != m_TrError) return;

    const tsp00_Int4 AllocSize = InitialMaxItems * sizeof (Type);

    m_pItem = REINTERPRET_CAST (Type*,
                                (bd491_Malloc (AllocSize, m_Trans.trTaskId_gg00)));

    if (NULL == m_pItem)
    {
        m_MaxItems = 0;
        m_TrError  = e_no_more_memory;
        g01optextmsg (sp3p_knldiag, sp3m_warning, BD491_ALLOCATE_1_SP03,
                      csp3_n_index, ALLOCATE_MSG_1_BD491);
        return;
    }
    m_MaxItems = InitialMaxItems;
}

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable> inline Type *
cbd491_DynamicVector<Type, Resizeable>::operator [] (tsp00_Int4 Index)
{
    if ((NULL != m_pItem) && (0 <= Index) && (Index <=  m_NumItemsUsed))
    {
        return  m_pItem + Index;
    }
    return NULL;
}

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable> inline void
cbd491_DynamicVector<Type, Resizeable>::bd491_AppendItem (
    Type        &AppendItem,
    tsp00_Int4   NumReallocItems)
{
    ROUTINE_DBG_MEO00 ("bd491_AppendItem");


    if ((m_NumItemsUsed + 1) > m_MaxItems)
    {
        if (Resizeable)
        {
            bd491_Resize (NumReallocItems);
        }
        else
        {
            m_TrError = e_no_more_memory;
            g01optextmsg (sp3p_knldiag, sp3m_warning, BD491_ALLOCATE_3_SP03,
                          csp3_n_index, ALLOCATE_MSG_2_BD491);
        }
    }

    if (e_ok != m_TrError) return;

    *(m_pItem + m_NumItemsUsed) = AppendItem;
    ++m_NumItemsUsed;

    return;
}

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable> inline Type *
cbd491_DynamicVector<Type, Resizeable>::bd491_GetSpaceForItems (tsp00_Int4  NumItems,
        tsp00_Int4  NumReallocItems)
{
    ROUTINE_DBG_MEO00 ("bd491_GetSpaceForItems");


    if ((m_NumItemsUsed + NumItems) > m_MaxItems)
    {
        if (Resizeable)
        {
            const tsp00_Int4 Delta = m_NumItemsUsed + NumItems - m_MaxItems;

            bd491_Resize ((Delta > NumReallocItems) ? Delta : NumReallocItems);
        }
        else
        {
            m_TrError = e_no_more_memory;
            g01optextmsg (sp3p_knldiag, sp3m_warning, BD491_ALLOCATE_3_SP03,
                          csp3_n_index, ALLOCATE_MSG_2_BD491);
        }
    }

    if (e_ok != m_TrError) return NULL;

    tsp00_Int4 OldNumItemsUsed = m_NumItemsUsed;
    m_NumItemsUsed += NumItems;

    return m_pItem + OldNumItemsUsed;
}

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable> inline void
cbd491_DynamicVector<Type, Resizeable>::bd491_ReleaseSpaceForItems (tsp00_Int4  NumItems)
{
    ROUTINE_DBG_MEO00 ("bd491_ReleaseSpaceForItems");

    if ((0 < NumItems) && (m_NumItemsUsed >= NumItems)) m_NumItemsUsed-= NumItems;
}

/*---------------------------------------------------------------------------*/

template <class Type, bool Resizeable> inline void
cbd491_DynamicVector<Type, Resizeable>::bd491_Resize (tsp00_Int4   NumReallocItems)
{
    ROUTINE_DBG_MEO00 ("bd491_Resize");


    const tsp00_Int4    NewMaxItems  = m_MaxItems + NumReallocItems;
    const tsp00_Int4    NewAllocSize = NewMaxItems * sizeof (Type);

    Type *pAuxItem = NULL;

    g01optextmsg (sp3p_console, sp3m_info, BD491_ALLOCATE_2_SP03,
                  csp3_n_index, ALLOCATE_MSG_3_BD491);

    pAuxItem = REINTERPRET_CAST (Type*,
                                 (bd491_Malloc (NewAllocSize, m_Trans.trTaskId_gg00)));

    if (NULL == pAuxItem)
    {
        m_TrError = e_no_more_memory;
        g01optextmsg (sp3p_knldiag, sp3m_warning, BD491_ALLOCATE_3_SP03,
                      csp3_n_index, ALLOCATE_MSG_1_BD491);
        return;
    }
    // Copy items of old dynamic vector into new dynamic vector

    for (tsp00_Int4 Item = 0; Item < m_MaxItems; ++Item)
        *(pAuxItem + Item) = *(m_pItem + Item);

    // Free old dynamic vector

    vmfree (m_pItem);
    m_pItem    = pAuxItem;
    m_MaxItems = NewMaxItems;
}

/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

cbd491_InvRecDescItem::cbd491_InvRecDescItem (tgg00_TransContext    &Trans,
        cbd497_StackDescInfo  &StackDescInfo,
        tsp00_BytePtr          pUndefNullKey)
        :
        m_TrError       (Trans.trError_gg00),
        m_Trans         (Trans),
        m_Stack         (StackDescInfo),
        m_pUndefNullKey (pUndefNullKey)
{
    ROUTINE_DBG_MEO00 ("cbd491_InvRecDescItem");
}

/*---------------------------------------------------------------------------*/

inline void
cbd491_InvRecDescItem::bd491_BuildVirtualRecord (
    tbd491_KeyDescPtr  pKeyDesc,
    tgg00_RecPtr       pRec)
{
    ROUTINE_DBG_MEO00 ("bd491_BuildVirtualRecord");


    bool                ColumnIsEmpty;
    tsp00_Int4          CurrColumn = m_Stack.bd497FirstColumn ();
    tsp00_Int4          CurrColumnLen;
    tsp00_Int4          CurrColumnPos;
    tgg00_VarColPosList VarColPos;
    /* */
    tgg00_StEntryAddr   pStackEntry = NULL;


    VarColPos.vpl_last = -1;

    if (NULL == m_pUndefNullKey)
    {
        m_TrError = e_not_implemented; // caused by a wrong constructor call
        return;
    }

    do
    {
        pStackEntry = m_Stack.bd497GetStackEntryPtr (CurrColumn);
        if (NULL == pStackEntry) return; // e_stack_type_illegal

        g04locate_col (*pStackEntry, pRec, VarColPos, CurrColumnPos, CurrColumnLen);

        if ((1 < CurrColumnLen) && (CurrColumn == m_Stack.bd497LastColumn()))
        {
            // Cut trailing define bytes of the last (!) indexed column by
            // reduceing the column length

            CurrColumnLen = 1 + // Define byte
                            s30lnr_defbyte (pRec,
                                            pRec->recBuf_gg00() [CurrColumnPos - POS_OFF_DIFF_BD00],
                                            CurrColumnPos + 1, CurrColumnLen - 1);
        }

        ColumnIsEmpty = ((0 == CurrColumnLen) ||
                         (csp_undef_byte == pRec->recBuf_gg00() [CurrColumnPos - POS_OFF_DIFF_BD00]));

        if (FIRST_COLUMN_BD491 == CurrColumn)
        {
            // Set pointer on the primary key

            pKeyDesc->KeyLen_bd491 = pRec->recKeyLen_gg00();
            pKeyDesc->pKey_bd491   = &pRec->recBody_gg00()[0];
        }

        // Set pointer on the columns representing the secondary key

        if (ColumnIsEmpty)
        {
            // Simulate "null-value" column for multiple indexes

            if  (
                (CurrColumn == m_Stack.bd497LastColumn ()     ) &&
                (!m_Stack.bd497IsColumnDescending (CurrColumn))
            )
                (pKeyDesc + CurrColumn)->KeyLen_bd491 = 1;
            else
                (pKeyDesc + CurrColumn)->KeyLen_bd491 = m_Stack.bd497GetColumnLength (CurrColumn);
            (pKeyDesc + CurrColumn)->pKey_bd491 = m_pUndefNullKey;
        }
        else
        {
            (pKeyDesc + CurrColumn)->KeyLen_bd491 = CurrColumnLen;
            (pKeyDesc + CurrColumn)->pKey_bd491   = &pRec->recBuf_gg00()[CurrColumnPos - POS_OFF_DIFF_BD00];
        }
        ++CurrColumn;
    }
    while ((e_ok == m_TrError) && (CurrColumn <= m_Stack.bd497LastColumn()));
}

/*---------------------------------------------------------------------------*/

inline void
cbd491_InvRecDescItem::bd491_BuildPrimKey (tbd491_KeyDescPtr    pKeyDesc,
        tsp00_BytePtr        pPrimKey,
        tsp00_Int4           AvailableSpace)
{
    ROUTINE_DBG_MEO00 ("bd491_BuildPrimKey");

    if (e_ok != m_TrError) return;

    g10mv( __FILE__, 1,
           pKeyDesc->KeyLen_bd491, AvailableSpace,
           pKeyDesc->pKey_bd491, POS_OFF_DIFF_BD00, pPrimKey, POS_OFF_DIFF_BD00,
           pKeyDesc->KeyLen_bd491, m_TrError);
}

/*---------------------------------------------------------------------------*/

inline void
cbd491_InvRecDescItem::bd491_BuildSecKey (tbd491_KeyDescPtr pKeyDesc,
        tsp00_BytePtr     pSecKey,
        tsp00_Int4        AvailableSpace)
{
    ROUTINE_DBG_MEO00 ("bd491_BuildSecKey");

    if (e_ok != m_TrError) return;

    tsp00_Int4  CurrColumn = m_Stack.bd497FirstColumn ();
    tsp00_Int4  FixedColumnLength;
    tsp00_Int4  Offset;
    tsp00_Int4  SecKeyLen = 0;


    tsp00_Int4 CurrColumnLength;
    SecKeyLen = 0;

    do
    {
        Offset           = SecKeyLen;
        CurrColumnLength = (pKeyDesc + CurrColumn)->KeyLen_bd491;

        g10mv( __FILE__, 1,
               CurrColumnLength, AvailableSpace,
               (pKeyDesc + CurrColumn)->pKey_bd491, POS_OFF_DIFF_BD00,
               pSecKey, Offset + POS_OFF_DIFF_BD00,
               CurrColumnLength, m_TrError);

        if (e_ok != m_TrError) 
            return;

        if (
            (CurrColumn == m_Stack.bd497LastColumn ()     ) &&
            (!m_Stack.bd497IsColumnDescending (CurrColumn))
        )
        {
            SecKeyLen += CurrColumnLength;
        }
        else
        {
            FixedColumnLength = m_Stack.bd497GetColumnLength (CurrColumn);
            if (CurrColumnLength < FixedColumnLength)
            {
                if (csp_unicode_def_byte == *(pSecKey + Offset))
                    g10filuni ("VBD491", 2,
                               AvailableSpace, pSecKey,
                               Offset + CurrColumnLength + POS_OFF_DIFF_BD00,
                               FixedColumnLength - CurrColumnLength,
                               csp_unicode_blank, m_TrError);
                else
                    SAPDB_RangeFill( __FILE__, 3,
                            AvailableSpace, pSecKey,
                            Offset + CurrColumnLength + POS_OFF_DIFF_BD00,
                            FixedColumnLength - CurrColumnLength,
                            *(pSecKey + Offset), m_TrError);
            }
            if (e_ok != m_TrError) return;
            SecKeyLen += FixedColumnLength;
        }
        if (m_Stack.bd497IsColumnDescending (CurrColumn))
        {
            for (tsp00_Int4 i = Offset; i < SecKeyLen; ++i)
                *(pSecKey + i) = ~*(pSecKey + i);
        }
        ++CurrColumn;
    }
    while ((e_ok == m_TrError) && (CurrColumn <= m_Stack.bd497LastColumn()));
}

/*--------------------------------------------------------------------------*/

inline tsp00_LcompResult
cbd491_InvRecDescItem::bd491_ColumnCompare (
    bool          bDescending,
    bool          bLastColumn,
    bool          &bLeftColumnIsNull,  // PTS 1124759 TS 2003-10-20
    bool          &bRightColumnIsNull, // PTS 1124759 TS 2003-10-20
    tsp00_BytePtr pKeyLeft,
    tsp00_BytePtr pKeyRight,
    tsp00_Int4    KeyLenLeft,
    tsp00_Int4    KeyLenRight)
{
    // Is one of both columns empty

    if ( 0 >= KeyLenLeft )
    {
        if ( 0 >= KeyLenRight )
            return tsp00_LcompResult::fromConst (l_equal);
        else
            return tsp00_LcompResult::fromConst (l_less);
    }
    else if ( 0 >= KeyLenRight )
        return tsp00_LcompResult::fromConst (l_greater);

    // KeyLenLeft > 0 and KeyLenRight > 0

    if( csp_undef_byte != *pKeyLeft ) { // PTS 1124759 TS 2003-10-20
        bLeftColumnIsNull = false;
    }
    if( csp_undef_byte != *pKeyRight ){ // PTS 1124759 TS 2003-10-20
        bRightColumnIsNull = false;
    }

    const tsp00_Int4    PrefixLen = (KeyLenLeft > KeyLenRight) ? KeyLenRight : KeyLenLeft;
    tsp00_Int4          iOffset = 0;

    while ( iOffset < PrefixLen )
    {
        if ( *(pKeyLeft + iOffset) == *(pKeyRight + iOffset) ) ++iOffset;
        else
        {
            if ( *(pKeyLeft + iOffset) > *(pKeyRight + iOffset) )
                return tsp00_LcompResult::fromConst (l_greater);
            else
                return tsp00_LcompResult::fromConst (l_less);
        }
    }

    // iOffset >= PrefixLen and CompResult == l_equal

    if (KeyLenLeft == KeyLenRight){
        return tsp00_LcompResult::fromConst (l_equal);
    }

    if ((bLastColumn) && (!bDescending))
    {
        if (KeyLenLeft > KeyLenRight)
            return tsp00_LcompResult::fromConst (l_greater);
        else
            return tsp00_LcompResult::fromConst (l_less);
    }
    else
    {
        // Handle 'NON-LAST' columns of a multiple index and
        // the LAST column of multiple or single index if it
        // is descending. Pay attention to bytes smaller than
        // the define byte!
        // Note that the longer column could be smaller than the
        // shorter column, namely if the longer one is filled with
        // bytes whose value is smaller than the define byte.
        // For example left column contains "<define byte>,A" and
        // right column contains "<define byte>,A,<BEL>. The right
        // column is smaller although is longer because the value of
        // the <BEL> byte is smaller than the define byte left column
        // is filled up.

        // KeyLenLeft <> KeyLenRight and iOffset 'points' to the
        // first different byte

        if (KeyLenLeft > KeyLenRight)
        {
            const tsp00_Byte	DefineByte = *pKeyRight;

            if (csp_unicode_def_byte == DefineByte)
                return bd491_CompareUnicodeColumn (pKeyLeft, KeyLenLeft, iOffset);
            else
                return bd491_CompareASCIIColumn (DefineByte, pKeyLeft, KeyLenLeft, iOffset);
        }
        else
        {
            const tsp00_Byte	DefineByte = *pKeyLeft;
            tsp00_LcompResult	CompResult (tsp00_LcompResult::fromConst (l_equal));

            if (csp_unicode_def_byte == DefineByte)
            {
                CompResult = bd491_CompareUnicodeColumn (pKeyRight, KeyLenRight, iOffset);

                if (l_less == CompResult)
                    return tsp00_LcompResult::fromConst (l_greater);
                else if (l_greater == CompResult)
                    return tsp00_LcompResult::fromConst (l_less);
                else
                    return CompResult;
            }
            else
            {
                CompResult = bd491_CompareASCIIColumn (DefineByte, pKeyRight, KeyLenRight, iOffset);

                if (l_less == CompResult)
                    return tsp00_LcompResult::fromConst (l_greater);
                else if (l_greater == CompResult)
                    return tsp00_LcompResult::fromConst (l_less);
                else
                    return CompResult;
            }
        }

    }
}

/*---------------------------------------------------------------------------*/

inline tsp00_LcompResult
cbd491_InvRecDescItem::bd491_CompareVirtualRecords (
    bool              bInclusivePrimKey,
    tbd491_KeyDescPtr pKeyDescLeft,
    tbd491_KeyDescPtr pKeyDescRight)
{
    ROUTINE_DBG_MEO00 ("bd491_CompareVirtualRecords");


    bool                bLeftColumnIsNull  = true; // PTS 1124759 TS 2003-10-20
    bool                bRightColumnIsNull = true; // PTS 1124759 TS 2003-10-20
    tsp00_LcompResult   CompResult;
    tsp00_Int4          CurrColumn = m_Stack.bd497FirstColumn ();

    if (e_ok != m_TrError){
        return tsp00_LcompResult::fromConst (l_undef);
    }

    do
    {
        CompResult = bd491_ColumnCompare (
                         m_Stack.bd497IsColumnDescending( CurrColumn ),
                         CurrColumn == m_Stack.bd497LastColumn(),
                         bLeftColumnIsNull, bRightColumnIsNull,
                         (pKeyDescLeft + CurrColumn)->pKey_bd491,  (pKeyDescRight + CurrColumn)->pKey_bd491,
                         (pKeyDescLeft + CurrColumn)->KeyLen_bd491,(pKeyDescRight + CurrColumn)->KeyLen_bd491);

        if (m_Stack.bd497IsColumnDescending(CurrColumn))
        {
            // Note that in the case of descending order the corresponding column
            // is temporary 'build' as an ascending column, because the complement of
            // the bytes is quit difficult to build with column pointers pointing
            // direct into the primary records. (Don't change primary data!)
            // Therefore descending columns handled as ascending columns and final
            // the compare result will be reversed.

            if (l_less == CompResult)
                CompResult = (tsp00_LcompResult::fromConst (l_greater));
            else if (l_greater == CompResult)
                CompResult = (tsp00_LcompResult::fromConst (l_less));
        }
        ++CurrColumn;
    }
    while
    (
        (l_equal == CompResult) &&
        (e_ok == m_TrError    ) &&
        (CurrColumn <= m_Stack.bd497LastColumn())
    );

    if (
        (e_ok == m_TrError    ) &&
        (l_equal == CompResult) &&
        (bInclusivePrimKey    )
    )
    {
        // secondary keys are equal

        s30cmp (pKeyDescLeft->pKey_bd491, POS_OFF_DIFF_BD00, pKeyDescLeft->KeyLen_bd491,
                pKeyDescRight->pKey_bd491, POS_OFF_DIFF_BD00, pKeyDescRight->KeyLen_bd491,
                CompResult);

        const bool bBothColumnsAreNull = bLeftColumnIsNull && bRightColumnIsNull;

        if (
            (m_Stack.bd497IsInvUnique()) &&
            (l_equal != CompResult     ) && // primary keys are different
            (! bBothColumnsAreNull     )    // PTS 1124759 TS 2003-10-20
        )
        {
            m_TrError = e_duplicate_invkey;
        }
    }

    return CompResult;

}

/*---------------------------------------------------------------------------*/

inline tsp00_Int4
cbd491_InvRecDescItem::bd491_GetPrimKeyLen (tbd491_KeyDescPtr   pKeyDesc)
{
    ROUTINE_DBG_MEO00 ("bd491_GetPrimKeyLen");

    return (pKeyDesc->KeyLen_bd491);
}

/*---------------------------------------------------------------------------*/

inline tsp00_Int4
cbd491_InvRecDescItem::bd491_GetSecKeyLen (tbd491_KeyDescPtr    pKeyDesc)
{
    ROUTINE_DBG_MEO00 ("bd491_GetSecKeyLen");


    tsp00_Int4  CurrColumn = m_Stack.bd497FirstColumn();
    tsp00_Int4  SecKeyLen  = 0;

    do
    {
        if (
            (CurrColumn == m_Stack.bd497LastColumn()      ) &&
            (!m_Stack.bd497IsColumnDescending (CurrColumn))
        )
        {
            SecKeyLen += (pKeyDesc + CurrColumn)->KeyLen_bd491;
        }
        else
        {
            SecKeyLen += m_Stack.bd497GetColumnLength (CurrColumn);
        }
        ++CurrColumn;
    }
    while ((e_ok == m_TrError) && (CurrColumn <= m_Stack.bd497LastColumn()));

    return SecKeyLen;
}

/*---------------------------------------------------------------------------*/

#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
inline void
cbd491_InvRecDescItem::bd491_PrintVirtualRecord (tbd491_KeyDescPtr  pKeyDesc)
{
    ROUTINE_DBG_MEO00 ("bd491_PrintVirtualRecord");


    tsp00_Int4  CurrColumn = m_Stack.bd497FirstColumn();

    t01sname (bd_idx_create, "Prim Key    ");
    t01buf   (bd_idx_create, pKeyDesc->pKey_bd491, POS_OFF_DIFF_BD00, pKeyDesc->KeyLen_bd491);

    do
    {
        t01int4 (bd_idx_create, "Sec Key Col ", CurrColumn);
        t01buf  (bd_idx_create, (pKeyDesc + CurrColumn)->pKey_bd491, POS_OFF_DIFF_BD00,
                 (pKeyDesc + CurrColumn)->KeyLen_bd491);

        ++CurrColumn;
    }
    while (CurrColumn <= m_Stack.bd497LastColumn());
}
#   endif

/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

cbd491_InvRecDescList::cbd491_InvRecDescList (tgg00_TransContext &Trans,
        tsp00_Int4          NumRec,
        tsp00_Int4          NumPrimKeyDesc,
        tsp00_Int4          NumSecKeyDesc)
        :
        m_TrError        (Trans.trError_gg00),
        m_NumPrimKeyDesc (NumPrimKeyDesc),
        m_NumSecKeyDesc  (NumSecKeyDesc),
        m_InvRecDescList (Trans, NumRec),
        m_KeyDescPool    (Trans, NumRec * (NumPrimKeyDesc + NumSecKeyDesc))
{
    ROUTINE_DBG_MEO00 ("cbd491_InvRecDescList");

}

/*---------------------------------------------------------------------------*/

inline tbd491_KeyDescPtr
cbd491_InvRecDescList::bd491_NewRecDesc ()
{
    ROUTINE_DBG_MEO00 ("bd491_NewRecDesc");


    tbd491_KeyDescPtr  pKeyDesc;

    pKeyDesc = m_KeyDescPool.bd491_GetSpaceForItems (m_NumPrimKeyDesc + m_NumSecKeyDesc);

    if (e_ok != m_TrError) return NULL;

    m_InvRecDescList.bd491_AppendItem (pKeyDesc);

    if (e_ok != m_TrError) return NULL;

    return pKeyDesc;
}

/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

cbd491_CompareInvRecDescItem::cbd491_CompareInvRecDescItem (tgg00_TransContext   &Trans,
        cbd497_StackDescInfo &StackDescInfo)
        :
        cbd491_InvRecDescItem (Trans, StackDescInfo)
{
    ROUTINE_DBG_MEO00 ("cbd491_CompareInvRecDescItem");
}

/*---------------------------------------------------------------------------*/

inline tsp00_LcompResult_Enum
cbd491_CompareInvRecDescItem::gg200Compare (
    const tbd491_KeyDescPtr   &Left,
    const tbd491_KeyDescPtr   &Right)
{
    return bd491_CompareVirtualRecords (INCLUSIVE_PRIM_KEY_BD491, Left, Right);
}

/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

cbd491_NodeList::cbd491_NodeList (
    tbd_current_tree &Current,
    tsp00_Int4        NumLeavesToScan
)
        :
        m_TrError  (Current.curr_trans->trError_gg00),
        m_Trans    (*Current.curr_trans),
        m_Current  (Current),
        m_NodeList (*Current.curr_trans, NumLeavesToScan)
{
    ROUTINE_DBG_MEO00 ("cbd491_NodeList");
}

/*---------------------------------------------------------------------------*/

cbd491_NodeList::~cbd491_NodeList ()
{
    ROUTINE_DBG_MEO00 ("~cbd491_NodeList");


    if (0 >=  m_NodeList.bd491_NumItemsUsed()) return;

    for (tsp00_Int4 Item = 0; Item < m_NodeList.bd491_NumItemsUsed(); Item++)
    {
        if (NULL != m_NodeList [Item]->np_ptr())
            b13r_release_node (*m_NodeList [Item], m_Current, lru_last);
    }
}

/*---------------------------------------------------------------------------*/

inline bool
cbd491_NodeList::bd491_EqualLastStoredNode (tbd_node_ptrs &pNode)
{
    ROUTINE_DBG_MEO00 ("bd491_EqualLastStoredNode");


    if (0 >=  m_NodeList.bd491_NumItemsUsed()) return (false);

    return (m_NodeList [m_NodeList.bd491_NumItemsUsed() - 1]->np_ptr() == pNode.np_ptr());
}

/*---------------------------------------------------------------------------*/

inline void
cbd491_NodeList::bd491_StoreOneNode (tbd_node_ptrs &pNode)
{
    ROUTINE_DBG_MEO00 ("bd491_StoreOneNode");

    m_NodeList.bd491_AppendItem (pNode);
}

/*===========================================================================*
 *  DEFINITION OF METHODS DECLARED IN gbd491.h (CODE)                        * 
 *===========================================================================*/

cbd491_CreateIndexParticipant::cbd491_CreateIndexParticipant(
    tbd_current_tree   &PrimCurrent,
    tgg00_StackDesc    &StackDesc,
    tgg00_IndexCounter &IndexCounter,
    bool                bUseMemorySort
)
        :
        m_TrError       (PrimCurrent.curr_trans->trError_gg00),
        m_Trans         (*PrimCurrent.curr_trans),
        m_PrimCurrent   (PrimCurrent),
        m_IndexCounter  (IndexCounter),
        m_Stack         (PrimCurrent.curr_trans->trError_gg00, StackDesc),
        m_pUndefNullKey (NULL)
{
    ROUTINE_DBG_MEO00 ("cbd491_CreateIndexParticipant");


    bd490InitIndexCounter (m_IndexCounter); // Reset Counter

    if ((e_ok != m_TrError) || (!bUseMemorySort)) return;

    SAPDB_RangeFill( __FILE__, 8,
            sizeof (m_UndefNullKey), m_UndefNullKey,
            POS_OFF_DIFF_BD00, sizeof (m_UndefNullKey), '\00', m_TrError);

    m_UndefNullKey [0] = csp_undef_byte;
    m_pUndefNullKey    = &m_UndefNullKey[0];
}

/*---------------------------------------------------------------------------*/

cbd491_CreateIndexParticipant::~cbd491_CreateIndexParticipant()
{
}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491InsertionSort(
    cbd300_InvCurrent &InvCurrent,
    tgg00_IndexMode    IndexMode,
    tsp00_BytePtr      pStartPrimKey,
    tsp00_BytePtr      pStopPrimKey,
    tsp00_Int4         StartPrimKeyLen,
    tsp00_Int4         StopPrimKeyLen,
    tsp00_Bool        &bEntriesExist,
    tsp00_Bool        &bErrorOnBaseTable )
{
    ROUTINE_DBG_MEO00 ("bd491InsertionSort");


    if (e_ok != m_TrError) 
        return;

    tbd_searchresult        PrimKeySearchResult;
    tsp00_LcompResult       CompResult;
    /* */
    tsp00_Int4              PrimRecIndex;
    tsp00_PageNo            OldLeafPno = NIL_PAGE_NO_GG00;
    /* */
    tgg00_RecPtr            pRec = NULL;
    tbd_node_ptrs           pIndexNptr;
    tbd_node_ptrs           pPrimNptr;
    /* */
    tbd_neighbors           Neighbors;

    pIndexNptr.np_ptr()   = NULL;
    pIndexNptr.np_cbptr() = NULL;

    pPrimNptr.np_ptr()    = NULL;
    pPrimNptr.np_cbptr()  = NULL;

    do
    {
        if (NULL == pPrimNptr.np_ptr())
        {
            bd30SearchRecord (m_PrimCurrent, pStartPrimKey, StartPrimKeyLen, ! LOCK_LEAF_EXCL_BD00,
                              pPrimNptr, PrimRecIndex, Neighbors, PrimKeySearchResult);

            if (e_ok != m_TrError) 
            {
                bErrorOnBaseTable = true;
                break;
            }

            if (nonefound == PrimKeySearchResult) 
                break; // e_no_next_record;
        }
        else
        {
            OldLeafPno = pPrimNptr.np_ptr()->nd_id();

            b31next_search( pPrimNptr, PrimRecIndex, m_PrimCurrent );

            if (e_ok != m_TrError)
            {
                bErrorOnBaseTable = true;
                break;
            }
        }

        pRec = bd50GetRecPtr (m_PrimCurrent, pPrimNptr, PrimRecIndex);

        if (NULL == pRec) break; // e_illegal_entrypos

        s30cmp (pRec->recBody_gg00(), POS_OFF_DIFF_BD00, pRec->recKeyLen_gg00(),
                pStopPrimKey, POS_OFF_DIFF_BD00, StopPrimKeyLen, CompResult);

        if (l_greater == CompResult) break; // e_no_next_record;

        if (OldLeafPno != pPrimNptr.np_ptr()->nd_id())
        {
            bd490Inc (m_IndexCounter.idc_prim_leafnodes, 1);
        }

        if( m_Trans.trRteCommPtr_gg00->to_cancel )
        {
            m_TrError = e_cancelled;
            g01optextmsg (sp3p_knldiag, sp3m_info, csp3_bd_msg, csp3_n_index, CANCEL_MSG_1_BD491);
            break;
        }
        
        // PTS 1124334 TS 2003-09-23
        if(( e_ok == m_TrError ) && ( 0 == m_IndexCounter.idc_prim_leafnodes % 10 ))
        {
           Converter_IPageNoManager::Instance().HandleDBFull( m_Trans, 10 );
        }

        bd491_AddInvRecord (InvCurrent, IndexMode, pRec);

    }
    while (e_ok == m_TrError);

    if( NULL != pPrimNptr.np_ptr()){
        b13r_release_node( pPrimNptr, m_PrimCurrent, lru_last );
    }

    if( NULL != pIndexNptr.np_ptr()){
        b13r_release_node( pIndexNptr, m_PrimCurrent, lru_normal );
    }

    bd30ReleaseSubTree( m_PrimCurrent );

    if (e_no_next_record == m_TrError)
        m_TrError = e_ok;

    if (e_ok == m_TrError)
    {
        bEntriesExist = (0 < m_IndexCounter.idc_prim_keycount);
        bd490Inc (m_IndexCounter.idc_prim_keycount, m_IndexCounter.idc_nullcount);
    }

#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
    t01basis_error( bi, "base error  ", m_TrError );
#   endif

}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491MemorySort(
    cbd495_DoubleWriteQueue  &InvQueue,
    tsp00_BytePtr             pStartPrimKey,
    tsp00_BytePtr             pStopPrimKey,
    tsp00_Int4                StartPrimKeyLen,
    tsp00_Int4                StopPrimKeyLen,
    tsp00_Int4                NumLeavesToScan,
    tsp00_Int4                AvgRecordsPerLeaf,
    tsp00_Bool               &bEntriesExist,
    tsp00_Bool               &bErrorOnBaseTable )
{
    ROUTINE_DBG_MEO00 ("bd491MemorySort");


    if (e_ok != m_TrError)
        return;

    cbd491_NodeList  NodeList (m_PrimCurrent, NumLeavesToScan);

    if (e_ok != m_TrError)
        return;

    cbd491_InvRecDescList InvRecDescList (m_Trans,
                                          (NumLeavesToScan * AvgRecordsPerLeaf),
                                          NUM_PRIM_KEY_DESC_BD491, m_Stack.bd497NumColumns());
    if (e_ok != m_TrError)
        return;

    bd491_ProcessPrimTreeRange (NodeList, InvRecDescList,
                                pStartPrimKey, pStopPrimKey, 
                                StartPrimKeyLen, StopPrimKeyLen,
                                bErrorOnBaseTable);

    if( e_ok == m_TrError )
    {
        bEntriesExist = (0 < InvRecDescList.bd491_NumRecDescUsed());

        if( bEntriesExist )
        {
            cbd491_CompareInvRecDescItem CompareOperator (m_Trans, m_Stack);

#           if COMPILEMODE_MEO00 >= SLOW_MEO00 
            t01int4 (bd_idx_create, "Num Sort Rec", InvRecDescList.bd491_NumRecDescUsed());
#           endif

            const bool bSuccess = gg200QuickSort( InvRecDescList.bd491_FirstRecDescAddr(),
                                                  InvRecDescList.bd491_NumRecDescUsed(),
                                                  CompareOperator );
            if( e_ok == m_TrError )
            {
                if(  bSuccess )
                    bd491_AppendInvRecords( InvQueue, InvRecDescList );
                else
                    m_TrError = e_no_more_memory;  // Use sequential create index
            }
        }
    }

    bd30ReleaseSubTree( m_PrimCurrent );

#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
    t01basis_error( bi, "base error  ", m_TrError );
#   endif
}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491_AddInvRecord (
    cbd300_InvCurrent    &InvCurrent,
    tgg00_IndexMode       IndexMode,
    tgg00_RecPtr          pRec)
{
    ROUTINE_DBG_MEO00 ("bd491_AddInvRecord");


    tgg00_FileId        &InvId     = InvCurrent.curr_tree_id;
    const tsp00_Int4    PrimKeyLen = pRec->recKeyLen_gg00();
    const tsp00_BytePtr pPrimKey   = REINTERPRET_CAST (tsp00_BytePtr, &pRec->recBody_gg00());

    bool                bSecKeyFound;
    tsp00_Int4          SecKeyLen = 0;
    tgg00_Lkey          InvKey;         // substitute Lkey by pKey and KeyLen
    tsp00_BytePtr       pSecKey  = REINTERPRET_CAST (tsp00_BytePtr, &InvKey.keyVal_gg00());
    tsp00_Bool          bAllColumnsUndef = SAPDB_FALSE; // PTS 1121337 UH 2003-03-27


    k33mult_inv_key (m_Trans, InvId, InvId, !INIT_INV_TREE_BD491,
                     m_Stack.bd497GetStackListPtr(), pRec,
                     m_Stack.bd497FirstColumn(), m_Stack.bd497LastColumn(),
                     InvKey, bAllColumnsUndef, m_TrError); // PTS 1121337 UH 2003-03-27

    if (e_ok != m_TrError)
        return;

    SecKeyLen = InvKey.keyLen_gg00();

    if ( ! bAllColumnsUndef // PTS 1121337 UH 2003-03-27 do not check if is null
            &&
            m_Stack.bd497IsInvUnique() )
    {
        bd400TestUniqueInv (InvCurrent, pSecKey, SecKeyLen, pPrimKey, PrimKeyLen);
        if (e_ok != m_TrError)
            return;
    }

#if COMPILEMODE_MEO00 >= SLOW_MEO00 
    t01sname (bd_idx_create, "Prim Key    ");
    t01buf   (bd_idx_create, pPrimKey, POS_OFF_DIFF_BD00, PrimKeyLen);

    t01sname (bd_idx_create, "Sec Key     ");
    t01buf   (bd_idx_create, pSecKey, POS_OFF_DIFF_BD00, SecKeyLen);
#endif

    bd400AddToInvTree (InvCurrent, pSecKey, SecKeyLen, pPrimKey, PrimKeyLen, bSecKeyFound);

    switch (IndexMode)
    {
    case create_index:
        {
            if ((tfnTempInv_egg00 != InvId.fileTfn_gg00()) && (!bSecKeyFound))
                bd490Inc (m_IndexCounter.idc_sec_keycount, 1);
            bd490Inc (m_IndexCounter.idc_prim_keycount, 1);
            return;
        }
    case test_index:
        {
            if (e_duplicate_key == m_TrError)
            {
                m_TrError = e_ok; // Normal case, because the primary key must exists in the inv tree
            }
            else if (e_ok == m_TrError)
            {
                // Error, because the primary key doesn't exist in the inv tree

                bd490Inc (m_IndexCounter.idc_prim_keycount, 1); // count missing primary keys

                if (g01vtrace.vtrAll_gg00 || g01vtrace.vtrBdIndex_gg00)
                {
                    bd03KeyVtrace (m_Trans, pSecKey, SecKeyLen, pPrimKey, PrimKeyLen, b03test);
                }
            }
            return;
        }
    default:
        m_TrError = e_not_implemented;
    }
}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491_AppendInvRecords(
    cbd495_DoubleWriteQueue   &InvQueue,
    cbd491_InvRecDescList     &InvRecDescList)
{
    ROUTINE_DBG_MEO00 ("bd491_AppendInvRecords");


    cbd491_InvRecDescItem   InvRecDescItem (m_Trans, m_Stack);
    /* */
    tbd491_KeyDescPtr       pKeyDesc      = InvRecDescList.bd491_GetRecDesc (0);
    tbd491_KeyDescPtr       pStartKeyDesc = pKeyDesc;
    tsp00_BytePtr           pPrimKey      = NULL;
    tsp00_BytePtr           pSecKey       = NULL;
    /* */
    tsp00_Int4              MinRequiredPrimKeySpace = 0;
    tsp00_Int4              NumPrimKeys = 0;
    tsp00_Int4              NumRecs     = InvRecDescList.bd491_NumRecDescUsed();
    tsp00_Int4              PrimKeyLen;
    tsp00_Int4              RecIndex = 0;
    tsp00_Int4              SecKeyLen;
    /* */
    tsp00_LcompResult       CompResult (tsp00_LcompResult::fromConst (l_undef));

    do
    {
        // Insert primary key into inv queue and count number of primary keys

        PrimKeyLen = InvRecDescItem.bd491_GetPrimKeyLen (pKeyDesc);
        InvQueue.bd495ReservePrimKeySpace (pPrimKey, PrimKeyLen);
        InvRecDescItem.bd491_BuildPrimKey (pKeyDesc, pPrimKey, PrimKeyLen);
        ++NumPrimKeys;
        MinRequiredPrimKeySpace += PrimKeyLen;

        ++RecIndex;

        if (RecIndex < NumRecs)
        {
            tbd491_KeyDescPtr   pPrevKeyDesc = pKeyDesc;

            pKeyDesc = InvRecDescList.bd491_GetRecDesc (RecIndex);

            CompResult = InvRecDescItem.bd491_CompareVirtualRecords (
                             !INCLUSIVE_PRIM_KEY_BD491, pStartKeyDesc, pKeyDesc);

#           if COMPILEMODE_MEO00 >= SLOW_MEO00 
            bd491_CheckKeyOrder (CompResult, pPrevKeyDesc, pKeyDesc);
#           endif
        }

        if ((l_less == CompResult) || (RecIndex >= NumRecs))
        {
            // Insert number of primary keys, min amount of storage used by
            // primary keys and secondary key into inv queue

#           if COMPILEMODE_MEO00 >= SLOW_MEO00 
            t01int4 (bd_idx_create, "Num Prim Key", NumPrimKeys);
#           endif

            InvQueue.bd495PushNumPrimKeys (NumPrimKeys);
            InvQueue.bd495PushMinRequiredPrimKeySpace (MinRequiredPrimKeySpace);

            SecKeyLen = InvRecDescItem.bd491_GetSecKeyLen (pStartKeyDesc);
            InvQueue.bd495ReserveSecKeySpace (pSecKey, SecKeyLen);
            InvRecDescItem.bd491_BuildSecKey (pStartKeyDesc, pSecKey, SecKeyLen);

            pStartKeyDesc           = pKeyDesc;
            NumPrimKeys             = 0;
            MinRequiredPrimKeySpace = 0;
        }
    }
    while  ((e_ok == m_TrError) && (RecIndex < NumRecs));
}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491_ProcessPrimTreeRange(
    cbd491_NodeList         &NodeList,
    cbd491_InvRecDescList   &InvRecDescList,
    tsp00_BytePtr            pStartPrimKey,
    tsp00_BytePtr            pStopPrimKey,
    tsp00_Int4               StartPrimKeyLen,
    tsp00_Int4               StopPrimKeyLen,
    tsp00_Bool              &bErrorOnBaseTable)
{
    // Temp solution as long as no primtree class is available

    ROUTINE_DBG_MEO00 ("bd491_ProcessPrimTreeRange");

    tbd_searchresult        PrimKeySearchResult;
    tsp00_LcompResult       CompResult;
    /* */
    tsp00_Int4              PrimRecIndex;
    tsp00_PageNo            OldLeafPno = NIL_PAGE_NO_GG00;
    /* */
    tgg00_RecPtr            pRec     = NULL;
    tbd491_KeyDescPtr       pKeyDesc = NULL;
    tbd_node_ptrs           pIndexNptr;
    tbd_node_ptrs           pPrimNptr;
    /* */
    tbd_neighbors           Neighbors;
    /* */
    cbd491_InvRecDescItem   InvRecDescItem (m_Trans, m_Stack, m_pUndefNullKey);

    pIndexNptr.np_ptr()   = NULL;
    pIndexNptr.np_cbptr() = NULL;

    pPrimNptr.np_ptr()   = NULL;
    pPrimNptr.np_cbptr() = NULL;
    
    do
    {
        if (NULL == pPrimNptr.np_ptr())
        {
            bd30SearchRecord (m_PrimCurrent, pStartPrimKey, StartPrimKeyLen, ! LOCK_LEAF_EXCL_BD00,
                              pPrimNptr, PrimRecIndex, Neighbors, PrimKeySearchResult);

            if (e_ok != m_TrError)
            {
                bErrorOnBaseTable = true;
                break;
            }

            if (nonefound == PrimKeySearchResult) break; // e_no_next_record;
        }
        else
        {
            OldLeafPno = pPrimNptr.np_ptr()->nd_id();

            bd491_NextRecord( OldLeafPno, pIndexNptr, pPrimNptr, PrimRecIndex );

            if (e_ok != m_TrError)
            {
                bErrorOnBaseTable = true;
                break;
            }
        }
        pRec = bd50GetRecPtr (m_PrimCurrent, pPrimNptr, PrimRecIndex);

        if (NULL == pRec) break; // e_illegal_entrypos

        s30cmp (pRec->recBody_gg00(), POS_OFF_DIFF_BD00, pRec->recKeyLen_gg00(),
                pStopPrimKey, POS_OFF_DIFF_BD00, StopPrimKeyLen, CompResult);

        if (l_greater == CompResult) break; // e_no_next_record;

        if (OldLeafPno != pPrimNptr.np_ptr()->nd_id())
        {
            bd490Inc (m_IndexCounter.idc_prim_leafnodes, 1);
            NodeList.bd491_StoreOneNode (pPrimNptr);
        }
        pKeyDesc = InvRecDescList.bd491_NewRecDesc ();
        if (NULL == pKeyDesc) break; // e_no_more_memory

        InvRecDescItem.bd491_BuildVirtualRecord (pKeyDesc, pRec);

        if (e_ok == m_TrError)
        {
            bd490Inc (m_IndexCounter.idc_prim_keycount, 1);
        }
    }
    while (e_ok == m_TrError);

    if (NULL != pPrimNptr.np_ptr())
    {
        if (!NodeList.bd491_EqualLastStoredNode (pPrimNptr))
            b13r_release_node (pPrimNptr, m_PrimCurrent, lru_last);
    }

    if( NULL != pIndexNptr.np_ptr()){
        b13r_release_node( pIndexNptr, m_PrimCurrent, lru_normal );
    }

    if (e_no_next_record == m_TrError) m_TrError = e_ok;

    if (e_ok == m_TrError)
        bd490Inc (m_IndexCounter.idc_prim_keycount, m_IndexCounter.idc_nullcount);

    if (e_sysbuffer_overflow == m_TrError)
    {
        g01optextmsg (sp3p_knldiag, sp3m_warning, BD491_OVERFLOW_1_SP03,
                      csp3_n_index, OVERFLOW_MSG_1_BD491);
    }
}

/*---------------------------------------------------------------------------*/

void
cbd491_CreateIndexParticipant::bd491_NextRecord(
    const tsp00_PageNo  leafPageNo,
    tbd_node_ptrs       &pIndexNptr,
    tbd_node_ptrs       &pPrimNptr,
    tsp00_Int4          &primRecIndex )
{
    ROUTINE_DBG_MEO00 ("bd491_NextRecord");

    tsp00_Bool  bIsLastRecordOnPage;

    b35next_entry( pPrimNptr.np_ptr(), primRecIndex, bIsLastRecordOnPage );

    if( ! bIsLastRecordOnPage )
    {
        m_TrError = e_ok;
        return;
    }

    const tsp00_PageNo  nextLeafPageNo = pPrimNptr.np_ptr()->nd_right();

    // original nptrs are stored within separate list, which is
    // used to free this nptrs later, after sorting all records

    pPrimNptr.np_ptr()   = NULL;
    pPrimNptr.np_cbptr() = NULL;

    if( NIL_PAGE_NO_GG00 == nextLeafPageNo )
    {
        m_TrError = e_no_next_record;
        return;
    }
    const tsp00_PageNo nextIndexPageNo =
        m_PrimCurrent.currIndexNptrs_bd00.np_ptr()->nd_right();

    SAPDBERR_ASSERT_STATE( m_PrimCurrent.curr_tree_id.fileType_gg00().includes( ftsPerm_egg00 ));

    if(
        ( leafPageNo == m_PrimCurrent.currRightBound_bd00   ) &&
        ( NULL != m_PrimCurrent.currIndexNptrs_bd00.np_ptr()) &&
        ( NIL_PAGE_NO_GG00 != nextIndexPageNo               )
    )
    {
        SAPDBERR_ASSERT_STATE( NULL == pIndexNptr.np_ptr() );

        bd13GetNode( m_PrimCurrent, nextIndexPageNo, plmLock_ebd00, nr_for_read, pIndexNptr );

        if( e_ok != m_TrError )
        {
            if( NULL != pIndexNptr.np_ptr()){
                b13r_release_node( pIndexNptr, m_PrimCurrent, lru_normal );
            }
            return;
        }
    }

    bd13GetNode( m_PrimCurrent, nextLeafPageNo, plmLock_ebd00, nr_for_read, pPrimNptr );

    primRecIndex = FIRST_REC_INDEX_BD00;
}

/*===========================================================================*
 *  END OF CODE                                                              *
 *===========================================================================*/
