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

    module      : Log_RecordColumnMap.cpp

    -------------------------------------------------------------------------

    author      : JuergenA
    responsible : UweH

    special area: Logging
    description : defines a class containing the index descriptions

    last changed: 2001-04-17

    -------------------------------------------------------------------------

    copyright:    (c) 2000-2004 SAP AG


    ========== licence begin  GPL
    Copyright (c) 2000-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                                                                  *
*============================================================================*/

#include <memory.h> // memcmp, memcpy, memset

#include "Logging/Log_RecordColumnMap.hpp"

#include "SAPDBCommon/SAPDB_Types.hpp"
#include "SAPDBCommon/MemoryManagement/SAPDBMem_IRawAllocator.hpp"
#include "SAPDBCommon/MemoryManagement/SAPDBMem_NewDestroy.hpp"
#include "SAPDBCommon/Tracing/SAPDBTrace_Usage.hpp" // Routine trace macros
#include "KernelCommon/Kernel_VTrace.hpp"
#include "KernelCommon/Kernel_Trace.hpp"
#include "Logging/Log_ReadWriteActionImage.hpp"

#include "gsp00.h"
#include "ggg00.h"
#include "hgg10.h"
#include "hsp30.h"

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

#define LOG_UNDEF_FILL_BYTE 0

/*===========================================================================*
*   LOCAL FUNCTIONS                                                          *
*============================================================================*/

/*---------------------------------------------------------------------------*/
const char * Log_ColumnTypeStrings [] =
{
    "FixKeyColumn",
    "VarKeyColumn",
	"FixColumn",
    "VarColumn",
    "VarLongColumn",
    "ColumnLastEnum",
    "NilColumn"
};

/*!-------------------------------------------------------------------------
declaration:  Log_ColumnType
--------------------------------------------------------------------------*/
extern const char* Log_ColumnTypeStrings[Log_NilColumn+1];

//----------------------------------------------------------------------------
/*! function:     FirstColOffset

    return value: offset of the first column
  
    arguments:    pRec [in]
*/    
inline static SAPDB_Int4 FirstColOffset ()
{
    return (RSN_RECHEAD_MXGG00);
}

//----------------------------------------------------------------------------
/*! function:     FirstFixColOffset

    return value: offset of the first fixed column
  
    arguments:    pRec [in]
*/    
inline static SAPDB_Int4 FirstFixColOffset (const tgg00_Rec *pRec)
{
    return (sizeof (tgg00_HeaderRec) + pRec->recKeyLen_gg00());
}

//----------------------------------------------------------------------------
/*! function:     FirstKeyColOffset

    return value: offset of the first key column
  
    arguments:    pRec [in]
*/    
inline static SAPDB_Int4 FirstKeyColOffset ()
{
    return (sizeof (tgg00_HeaderRec));
}

//----------------------------------------------------------------------------
/*! function:     FirstVarColOffset

    return value: offset of the first variable column
  
    arguments:    pRec [in]
*/    
inline static SAPDB_Int4 FirstVarColOffset (const tgg00_Rec *pRec)
{
    return (FirstFixColOffset(pRec) + pRec->recVarcolOffset_gg00() );
}

//--------------------------------------------------------------Log_RecordColumnMap---
/*! function:     GetColumnType

    return value: the type of the column specified by a stack entry
  
    arguments:    StackEntry [in]
*/
inline static Log_ColumnType GetColumnType (const tgg00_StackEntry &StackEntry)
{
    switch ( StackEntry.etype() )
    {
    case st_fixkey:
        return (Log_FixKeyColumn);
        
    case st_varkey:
        return (Log_VarKeyColumn);
        
    case st_fixcol:
        return (Log_FixColumn);
        
    case st_varcol:
        return (Log_VarColumn);  
        
    case st_varlongchar:
        return (Log_VarLongColumn);
        
        /* +++ UH commented because of rollback error
    case st_column:
        return (Log_Column);
        */
    default:
        return (Log_NilColumn);
    }
}

//----------------------------------------------------------------------------
/*! function:     GetInt2Length

    return value: moves an int2 length from the source specified by its pointer
  
    arguments:    pSource [in]
*/  
inline static SAPDB_Int4 GetInt2Length (const void *pSource)
{
    SAPDB_Int2 Length;
    
    memcpy (&Length, pSource, 2);
    
    return (Length);
}

//----------------------------------------------------------------------------
/*! function:     PutInt2Length

    return value: moves an int2 length to the destination specified by its pointer
  
    arguments:    pDestin [in]
*/  
inline static void PutInt2Length (void *pDestin, SAPDB_Int2 Length)
{
    memcpy (pDestin, &Length, 2);
}

//----------------------------------------------------------------------------
/*! function:    TruncateValue

    description: evaluates the truncated length of the value
  
    arguments:   pValue           [in]     the first byte of the value has to be the defined byte
                 ValueLength      [in out]
                 CreateUndefValue [out]    indicates that an undef value has to be created                      
*/  
static void TruncateValue (const SAPDB_Byte *pValue,
                           SAPDB_Int4       &ValueLength,
                           bool             &CreateUndefValue)
{    
    const SAPDB_Int4 SizeofDefinedByte = 1;

    CreateUndefValue = false;
    
    if (ValueLength < SizeofDefinedByte)
    {
        CreateUndefValue = true;
        return;
    }
    
    /*  if this part is on, ' ' will result in NULL
        this is not what we expect
    ValueLength = SizeofDefinedByte +
        s30lnr_defbyte (pValue, pValue[0], SizeofDefinedByte + 1, ValueLength - SizeofDefinedByte);
    
    if ( (SizeofDefinedByte == ValueLength) && 
         (csp_undef_byte   != pValue[0])      &&
         (csp_defined_byte != pValue[0]))   //boolean-value FALSE !!
    {
        // create a value containing an undef byte instead of any defined byte
        CreateUndefValue = true;
    }
    */
}

//----------------------------------------------------------------------------
/*! function:    Log_ColumnType operator ++
*/
void operator ++ (Log_ColumnType &ColumnType)
{
    ColumnType = Log_ColumnType (ColumnType + 1);
}


/*===========================================================================*
*   PRIVATE CLASSES                                                          *
*============================================================================*/

//===================================================================================
/*! class:       Log_ColEntryIter

    description: iterator to access a column entry
*/
class Log_ColEntryIter
{

private:
    Log_ColumnType   m_ColumnType;
    Log_ColumnEntry *m_pNext;

protected:

    Log_ColumnEntry *m_pCurr;
    Log_ColumnEntry *m_pPrev;
    bool             m_CurrIsRemoved;

public:

    /*!-------------------------------------------------------------------------------
    function:    Log_ColEntryIter ctor
    --------------------------------------------------------------------------------*/
    Log_ColEntryIter (const Log_ColumnList &ColList):
    m_ColumnType    (ColList.GetColumnType()),
    m_pNext         (NULL),
    m_pCurr         (NULL),
    m_pPrev         (NULL),
    m_CurrIsRemoved (false)
    {
        this->m_pNext = ColList.GetFirst();
    }
    /*!-------------------------------------------------------------------------------
    function:    GetLength
    returns:     the length of the column
    --------------------------------------------------------------------------------*/
    SAPDB_Int4 GetLength () const
    { 
        if (NULL == this->m_pCurr || this->m_CurrIsRemoved) return (0);
        return (this->m_pCurr->GetLength());
    }
    /*!-------------------------------------------------------------------------------
    function:    GetNumber
    returns:     the number of the variable column
    --------------------------------------------------------------------------------*/
    SAPDB_Int4 GetNumber () const
    { 
        if (NULL == this->m_pCurr || this->m_CurrIsRemoved) return (0);
        return (this->m_pCurr->GetNumber());
    } 
    /*!-------------------------------------------------------------------------------
    function:    GetOffset
    returns:     the offset of the fixed column
    --------------------------------------------------------------------------------*/
    SAPDB_Int4 GetOffset () const
    { 
        if (NULL == m_pCurr || this->m_CurrIsRemoved) return (0);
        return (this->m_pCurr->GetOffset());
    }
    /*!-------------------------------------------------------------------------------
    function:    GetValue
    returns:     pointer to the value of the column
    --------------------------------------------------------------------------------*/
    const SAPDB_Byte *GetValue () const
    { 
        if (NULL == m_pCurr || this->m_CurrIsRemoved) return (NULL);
        return (this->m_pCurr->GetValue());
    }
    /*!-------------------------------------------------------------------------------
    function:    HasMore
    --------------------------------------------------------------------------------*/
    bool HasMore ()
    {
        return (NULL != m_pNext);
    }
    /*!-------------------------------------------------------------------------------
    function:    IsEqualValue
    --------------------------------------------------------------------------------*/
    bool IsEqualValue(const SAPDB_Byte *pOtherValue,
                      SAPDB_Int4        OtherValueLength) const
    {  
        if (this->GetLength() != OtherValueLength) return (false);
        if (0 == OtherValueLength) return (true); 
        return ( 0 == memcmp (this->GetValue(), pOtherValue, OtherValueLength) );       
    }
    /*!-------------------------------------------------------------------------------
    function:    MoveColToRec
    --------------------------------------------------------------------------------*/
    void MoveColToRec (tgg00_Rec        *pRec,
                       SAPDB_Int4        RecSize,
                       tgg00_BasisError &Error) const
    {
        SAPDB_Int2 Length = this->GetLength();
        SAPDB_Int4 Offset;
        
        switch (this->m_ColumnType)
        {
        case Log_FixColumn:
            Offset = FirstFixColOffset (pRec) + this->GetOffset();
            break;
        case Log_FixKeyColumn:
        case Log_VarKeyColumn: 
            Offset = FirstKeyColOffset() + this->GetOffset();
            break;
        case Log_VarColumn:
            pRec->recBuf_gg00() [pRec->recLen_gg00()] = abs(Length);
            Offset = pRec->recLen_gg00() + 1;
            pRec->recLen_gg00() = Offset + Length;
            if (MAX_RECLEN_GG00 < pRec->recLen_gg00())
            {
                Error = e_too_long_record;
                return;
            }
            break;
        case Log_VarLongColumn:
            PutInt2Length (pRec->recBuf_gg00() + pRec->recLen_gg00(), Length);
            Offset = pRec->recLen_gg00() + 2;
            pRec->recLen_gg00() = Offset + Length;
            if (MAX_RECLEN_GG00 < pRec->recLen_gg00())
            {
                Error = e_too_long_record;
                return;
            }
            break;
        /* ++++ UH commented
        case Log_Column: 
            break;
            */
        }
            
        if (Length > 0)
            g10mv1 ("UpdMap", 1, Length, RecSize,
                GetValue(), 1, pRec, Offset + 1, Length, Error);
    }
    /*!-------------------------------------------------------------------------------
    function:    Next
    --------------------------------------------------------------------------------*/
    void Next ()
    {
        m_pPrev         = m_pCurr;
        m_pCurr         = m_pNext;
        m_CurrIsRemoved = false;
        if (NULL != m_pNext) m_pNext = m_pNext->GetNext();
    }
};
/*! endclass: Log_ColEntryIter */


//===================================================================================
/*! class:       Log_EntryIterWithRemove

    description: iterator to access a column entry
*/
class Log_EntryIterWithRemove: public Log_ColEntryIter
{

private:

    Log_ColumnList &m_ColList;

public:

    /*!-------------------------------------------------------------------------------
    function:    Log_EntryIterWithRemove
    --------------------------------------------------------------------------------*/
    inline Log_EntryIterWithRemove (Log_ColumnList &ColList)
    : Log_ColEntryIter (ColList),
      m_ColList        (ColList)
    {}
    /*!-------------------------------------------------------------------------------
    function:    Remove
    --------------------------------------------------------------------------------*/
    void Remove ()
    {
        if (NULL == m_pCurr || m_CurrIsRemoved) return;
        
        m_ColList.Remove (m_pCurr, m_pPrev);
        m_pCurr         = m_pPrev;
        m_CurrIsRemoved = true;
    }
};
/*! endclass: Log_EntryIterWithRemove */


//===================================================================================
/*! class:       Log_VarPartHandler

    description: iterator to access a column entry
*/
class Log_VarPartHandler
{
private:
    
    const tgg00_Rec *m_pOldRec;
    tgg00_Rec       *m_pNewRec;
    SAPDB_Int4       m_NewRecSize;
    SAPDB_Int4       m_SizeofLength;
    SAPDB_Int4       m_CurrSizeofLength;
    SAPDB_Int4       m_CurrOldOffset;
    SAPDB_Int4       m_CurrOldColLen;
    SAPDB_Int4       m_CurrOldColNumber;
    Log_ColumnType   m_ColumnType;
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:  ReadCurrOldColLen
    */
    void ReadCurrOldColLen ();
    
public:
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:    Log_VarPartHandler ctor
    */
    inline Log_VarPartHandler (
        Log_ColumnType   ColumnType,
        SAPDB_Int4       OldPartOffset,
        const tgg00_Rec *pOldRec,
        tgg00_Rec       *pNewRec,
        SAPDB_Int4       NewRecSize);

    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:    CopyUnchangedColumns

        description: copies columns from the old record into the new record until a column is updated
    */
    void CopyUnchangedColumns (
        SAPDB_Int4        NumberNextChangedCol,
        tgg00_BasisError &Error,
        bool              CopyRest = false);
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:    CreateMissingUndefColumns

        description: creates undef columns if there are not any in the old record. 
    */
    void CreateMissingUndefColumns (
        SAPDB_Int4        NumberNextChangedCol,
        tgg00_BasisError &Error);

    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:    FinishVarPart

        description: copies outstanding columns from the old record into the record.  
    */
    void FinishVarPart (
        SAPDB_Int4       &OffsetFollowingPart,
        tgg00_BasisError &Error);

    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:      GetOldColumn
        
        return result: pointer to the value of current column contained in the old record
    */
    inline const SAPDB_Byte *GetOldColumn() const;
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:      GetOldColumnLength
    
        return result: length of the current column contained in the old record
    */
    inline SAPDB_Int4 GetOldColumnLength() const;
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:      GetSizeLength
    
        return result: length of length-info of the var (long) column
    */
    inline SAPDB_Int4 GetSizeLength() const;
    
    //--------------------------------------------------------------Log_VarPartHandler---
    /*! function:    NextOldColumn

        description: current column becomes the following column.
    */
    void NextOldColumn();

    /*!----------------------------------------------------------------------------------
    function:    WriteToTrace
    */
    void WriteToTrace(const char * title = 0) const
    {
        Kernel_VTrace trace;
        if ( title != NULL )
            trace << title << FlushLine;
        trace << "Columntype: " << Log_ColumnTypeStrings[m_ColumnType];
        Kernel_TraceBuffer ( m_pOldRec, m_pOldRec->recLen_gg00(), "Old record" );
        trace << "New record: m_NewRecSize: " << m_NewRecSize << FlushLine;
        Kernel_TraceBuffer ( m_pNewRec, m_pNewRec->recLen_gg00(), "New record" );
        trace << "SizeofLength: " << m_SizeofLength << ", CurrSizeofLength: " << m_CurrSizeofLength;
        trace << "CurrOldColNumber: " << m_CurrOldColNumber
              << ", CurrOldOffset: " << m_CurrOldOffset
              << ", CurrOldColLen: " << m_CurrOldColLen << FlushLine;
    }
};
/*! endclass: Log_VarPartHandler */


/*===========================================================================*
*  METHODS of Log_VarPartHandler                                             *
*============================================================================*/

inline Log_VarPartHandler::Log_VarPartHandler (Log_ColumnType   ColumnType,
                                               SAPDB_Int4       OldPartOffset,
                                               const tgg00_Rec *pOldRec,
                                               tgg00_Rec       *pNewRec,
                                               SAPDB_Int4       NewRecSize):
m_pOldRec          (pOldRec),
m_pNewRec          (pNewRec),
m_NewRecSize       (NewRecSize),
m_SizeofLength     (1),
m_CurrSizeofLength (0),
m_CurrOldOffset    (OldPartOffset),
m_CurrOldColLen    (0),
m_CurrOldColNumber (1),
m_ColumnType       (ColumnType)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::Log_VarPartHandler", LogAction_Trace, 5);
    if (Log_VarColumn == ColumnType)
    {
        this->m_pNewRec->recLen_gg00()       = FirstVarColOffset (pNewRec);
        this->m_pNewRec->recVarcolCnt_gg00() = 0;
    }
    else
    {
        //Log_VarLongColumn
        this->m_SizeofLength = 2;
    }
    
    this->ReadCurrOldColLen ();
}

//----------------------------------------------------------------------------

void Log_VarPartHandler::CopyUnchangedColumns (SAPDB_Int4        NumberNextChangedCol,
                                               tgg00_BasisError &Error,
                                               bool              CopyRest)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::CopyUnchangedColumns", LogAction_Trace, 5);

    SAPDB_Int4 StartOffset = this->m_CurrOldOffset;
    
    while ( (this->m_CurrOldColLen > 0)
        && (CopyRest || (this->m_CurrOldColNumber < NumberNextChangedCol)) )        
    {
        this->NextOldColumn();
    }
    
    if (this->m_CurrOldOffset > StartOffset) 
    {
        if ( m_NewRecSize
             <
             m_pNewRec->recLen_gg00() + m_CurrOldOffset - StartOffset )
        {
            // PTS 1115833 UH 2002-08-28 NULL columns were added
            m_NewRecSize = m_pNewRec->recLen_gg00() + m_CurrOldOffset - StartOffset;
        }
        
        g10mv1 ("UpdMap", 2, m_pOldRec->recLen_gg00(), m_NewRecSize,
            m_pOldRec, StartOffset+1,
            m_pNewRec, m_pNewRec->recLen_gg00()+1,
            m_CurrOldOffset - StartOffset, Error);
        
        if (e_ok != Error)
        {
            #ifdef SAPDB_SLOW
            Kernel_VTrace() << "StartOffset: " << StartOffset
                            << ", m_NewRecSize: " << m_NewRecSize
                            << ", m_CurrOldOffset: " << m_CurrOldOffset
                            << ", m_pOldRec->recLen_gg00(): " << m_pOldRec->recLen_gg00();
            WriteToTrace("CopyUnchangedColumns");
            #endif
            return;
        }
        
        m_pNewRec->recLen_gg00() += m_CurrOldOffset - StartOffset;
    }
}

//----------------------------------------------------------------------------

void Log_VarPartHandler::CreateMissingUndefColumns (SAPDB_Int4        NumberNextChangedCol,
                                                    tgg00_BasisError &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::CreateMissingUndefColumns", LogAction_Trace, 5);

    const SAPDB_UInt1 SizeofUndefByte = 1;
    
    while (this->m_CurrOldColNumber < NumberNextChangedCol)
    {        
        if (this->m_pNewRec->recLen_gg00() + this->m_SizeofLength + SizeofUndefByte > this->m_NewRecSize)
        {
            Error = e_column_trunc; // new record too small
            return;
        }
        
        if (Log_VarLongColumn == this->m_ColumnType)
        {
            PutInt2Length 
                (this->m_pNewRec->recBuf_gg00() + this->m_pNewRec->recLen_gg00(), SizeofUndefByte);                
        }
        else
        {
            this->m_pNewRec->recBuf_gg00() [this->m_pNewRec->recLen_gg00()] = SizeofUndefByte;
        }
        
        this->m_pNewRec->recLen_gg00() += this->m_SizeofLength;
        
        this->m_pNewRec->recBuf_gg00() [this->m_pNewRec->recLen_gg00()]  = csp_undef_byte;
        
        this->m_pNewRec->recLen_gg00() += SizeofUndefByte;
        
        this->NextOldColumn();
    }
}

//----------------------------------------------------------------------------

void Log_VarPartHandler::FinishVarPart (SAPDB_Int4       &OffsetFollowingPart,
                                        tgg00_BasisError &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::FinishVarPart", LogAction_Trace, 5);

    const bool CopyRest = true;
    
    this->CopyUnchangedColumns (0, Error, CopyRest);
    
    if (e_ok != Error) return;
    
    if (Log_VarColumn == this->m_ColumnType)
    {
        this->m_pNewRec->recVarcolCnt_gg00() = this->m_CurrOldColNumber - 1;
    }
    
    OffsetFollowingPart = this->m_CurrOldOffset;
}

//----------------------------------------------------------------------------

inline const SAPDB_Byte *Log_VarPartHandler::GetOldColumn() const
{
    if (this->m_CurrOldColLen > 0)
    {
        return ( this->m_pOldRec->recBuf_gg00() + this->m_CurrOldOffset + this->m_CurrSizeofLength );
    }
    
    return (NULL);
}

//----------------------------------------------------------------------------

inline SAPDB_Int4 Log_VarPartHandler::GetOldColumnLength() const
{
    return ( this->m_CurrOldColLen );
}

//----------------------------------------------------------------------------

inline SAPDB_Int4 Log_VarPartHandler::GetSizeLength() const
{
    return ( this->m_SizeofLength );
}

//----------------------------------------------------------------------------

void Log_VarPartHandler::ReadCurrOldColLen ()
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::ReadCurrOldColLen", LogAction_Trace, 5);

    this->m_CurrOldColLen    = 0;
    this->m_CurrSizeofLength = 0;
    
    if (Log_VarLongColumn == this->m_ColumnType)
    {
        if ( this->m_CurrOldOffset + this->m_SizeofLength > this->m_pOldRec->recLen_gg00() ) return;
        
        this->m_CurrOldColLen = GetInt2Length (this->m_pOldRec->recBuf_gg00() + this->m_CurrOldOffset);
    }
    else
    {
        if ( this->m_CurrOldColNumber > this->m_pOldRec->recVarcolCnt_gg00() ) return;
        
        this->m_CurrOldColLen = this->m_pOldRec->recBuf_gg00() [this->m_CurrOldOffset];
    }
    
    this->m_CurrSizeofLength = this->m_SizeofLength;
}

//----------------------------------------------------------------------------

void Log_VarPartHandler::NextOldColumn()
{
    SAPDBTRACE_METHOD_DEBUG ("Log_VarPartHandler::NextOldColumn", LogAction_Trace, 5);

    ++this->m_CurrOldColNumber;
    
    this->m_CurrOldOffset += this->m_CurrSizeofLength + this->m_CurrOldColLen;
    
    this->ReadCurrOldColLen();
}


/*===========================================================================*
*  METHODS of Log_ColumnEntry                                                *
*============================================================================*/

void Log_ColumnEntry::AllocateAndInitFixCol (SAPDBMem_IRawAllocator &RawAllocator,
                                             SAPDB_Int4              ColumnOffset,
                                             SAPDB_Int4              FixedColumnLen,
                                             SAPDB_Int4              ValueLength,
                                             const SAPDB_Byte       *pValue,
                                             tgg00_BasisError       &Error)
{
    if (e_ok != Error) return;
    
    this->m_pAllocatedValue = REINTERPRET_CAST (SAPDB_Byte*, RawAllocator.Allocate(FixedColumnLen));
    if (NULL == this->m_pAllocatedValue)
    {
        Error = e_no_more_memory;
        return;
    }
    
    m_pAllocator = &RawAllocator;

    this->m_ValueLength    = FixedColumnLen;
    this->m_OffsetOrNumber = ColumnOffset;
    this->m_pValue         = this->m_pAllocatedValue;
    
    memcpy (this->m_pAllocatedValue, pValue, ValueLength);
    
    if ( csp_unicode_def_byte == pValue[0] )
    {
        // fill trailing unicode defined bytes
        
        g10filuni ("UpdMap", 1, this->m_ValueLength,
            this->m_pAllocatedValue, ValueLength + 1, FixedColumnLen - ValueLength,
            csp_unicode_blank, Error);
        
        if (e_ok != Error) return;
    }
    else
    {
        if ( csp_undef_byte == pValue[0] )
        {
            // fill trailing bytes of undef value
            memset (this->m_pAllocatedValue + ValueLength,
                LOG_UNDEF_FILL_BYTE,
                FixedColumnLen - ValueLength);
        }
        else
        {
            // fill trailing defined bytes
            memset (                                       
                this->m_pAllocatedValue + ValueLength,
                pValue[0],
                FixedColumnLen - ValueLength);
        }
    }
}

//----------------------------------------------------------------------------

void Log_ColumnEntry::AllocateAndInitUndefCol (SAPDBMem_IRawAllocator &RawAllocator,
                                               SAPDB_Int4              OffsetOrNumber,
                                               SAPDB_Int4              FixedColumnLen,
                                               tgg00_BasisError       &Error)
{
    if (e_ok != Error) return;
    
    const SAPDB_Int4 SizeofDefinedByte = 1;
    
    this->m_ValueLength = (FixedColumnLen > 0) ? FixedColumnLen : SizeofDefinedByte;
    
    this->m_pAllocatedValue = REINTERPRET_CAST (SAPDB_Byte*, RawAllocator.Allocate(this->m_ValueLength));
    
    if (NULL == this->m_pAllocatedValue)
    {
        this->m_ValueLength = 0;
        Error = e_no_more_memory;
        return;
    }
    
    m_pAllocator = &RawAllocator;

    this->m_OffsetOrNumber = OffsetOrNumber;
    this->m_pValue         = this->m_pAllocatedValue;
    
    this->m_pAllocatedValue[0] = csp_undef_byte;
    
    if (this->m_ValueLength > SizeofDefinedByte)
    {
        memset (this->m_pAllocatedValue+1, LOG_UNDEF_FILL_BYTE, this->m_ValueLength - SizeofDefinedByte);
    }
}

//----------------------------------------------------------------------------

void Log_ColumnEntry::Insert (Log_ColumnEntry  *pPrevEntry,
                              Log_ColumnEntry *&pFirstListEntry)
{
    if (NULL == pPrevEntry)
    {
        this->m_pNext   = pFirstListEntry;
        pFirstListEntry = this;
        return;
    }
    
    this->m_pNext       = pPrevEntry->m_pNext;
    pPrevEntry->m_pNext = this;
}

//----------------------------------------------------------------------------

void Log_ColumnEntry::Remove (Log_ColumnEntry  *pPrevEntry,
                              Log_ColumnEntry *&pFirstListEntry)
{
    if (NULL == pPrevEntry)
    {
        pFirstListEntry = this->m_pNext;
        return;
    }
    
    pPrevEntry->m_pNext = this->m_pNext;
    this->m_pNext       = NULL;
}

//----------------------------------------------------------------------------

void Log_ColumnEntry::WriteToTrace(const char *title) const
{
    Kernel_VTrace trace;
    if ( title != NULL )
        trace << title << FlushLine;
    trace << "Offset/No: " << m_OffsetOrNumber
          << ", length: "  << m_ValueLength << FlushLine;
    Kernel_TraceBuffer ( m_pValue, m_ValueLength, "New record" );
}

/*===========================================================================*
*  METHODS of Log_ColumnList                                                 *
*============================================================================*/

void Log_ColumnList::InsertSorted (Log_ColumnEntry *pEntry)
{
    Log_ColumnEntry *pCurr = this->m_pFirst;
    Log_ColumnEntry *pPrev = NULL;
    
    while (NULL != pCurr)
    {
        if (*pCurr > *pEntry) break;
        
        pPrev = pCurr;
        pCurr = pCurr->GetNext();
    }
    
    pEntry->Insert (pPrev, this->m_pFirst);
    
    this->m_SumValueLength += pEntry->GetLength();
    
    ++this->m_EntryCount;
}


/*===========================================================================*
*  METHODS of Log_RecordColumnMap                                            *
*============================================================================*/

Log_RecordColumnMap::Log_RecordColumnMap (Log_ActionType ActionType):

Log_ReadWriteActionImage (ActionType),
m_EntryHeader            (),
m_pRawAllocator          (NULL),
m_pAllocatedMap          (NULL),
m_ColList                (),
m_AllocatedEntries       (0),
m_UsedEntries            (0),
m_pAllValuesForRead      (NULL),
m_pAllValuesAllocator    (NULL),
m_PseudoNullValLen       (0)
{ 
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::Log_RecordColumnMap", LogAction_Trace, 5);
    for (Log_ColumnType Curr = Log_FixKeyColumn; Curr < Log_ColumnLastEnum; ++Curr)
        m_ColList[Curr].Init (Curr);
}

//----------------------------------------------------------------------------

Log_RecordColumnMap::~Log_RecordColumnMap()
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::~Log_RecordColumnMap", LogAction_Trace, 5);

    if (NULL != this->m_pAllValuesAllocator)
    {
        this->m_pAllValuesAllocator->Deallocate (this->m_pAllValuesForRead);
        this->m_pAllValuesForRead = NULL;
    }
    
    if ((NULL != this->m_pAllocatedMap) && (this->m_AllocatedEntries > 0))
    {
        destroyarray (this->m_pAllocatedMap, this->m_AllocatedEntries, *(this->m_pRawAllocator));
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::AllocateColumnMap (SAPDBMem_IRawAllocator &RawAllocator,
                                             SAPDB_Int4              MaxEntries,
                                             bool                   &IsOk)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::AllocateColumnMap", LogAction_Trace, 5);
    
    SAPDBTRACE_WRITELN (LogAction_Trace, 6, "MaxEntries: " << MaxEntries);

    IsOk = (NULL == this->m_pAllocatedMap);
    
    if ( ! IsOk ) return;
    
    this->m_pRawAllocator = &RawAllocator;
    
    newarray (this->m_pAllocatedMap, MaxEntries, RawAllocator);
    
    IsOk = (NULL != this->m_pAllocatedMap);
    
    if ( ! IsOk ) return;
    
    this->m_AllocatedEntries = MaxEntries;
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CalculatePersistentImage()
{    
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CalculatePersistentImage", LogAction_Trace, 5);
    this->m_EntryHeader.ImageLen = 
        Log_AlignedImageLen   (sizeof (this->m_EntryHeader))
        + Log_AlignedImageLen (this->m_UsedEntries * sizeof (PersistentColEntry));
    
    this->m_EntryHeader.LengthAllValues = 0; 
    
    for (Log_ColumnType Curr = Log_FixKeyColumn; Curr < Log_ColumnLastEnum; ++Curr)
    {        
        this->m_EntryHeader.LengthAllValues += this->m_ColList[Curr].GetSumValueLength();
    }
    this->m_EntryHeader.ImageEntries     = this->m_UsedEntries;
    this->m_EntryHeader.FixColCnt        = this->m_ColList [Log_FixColumn    ].GetEntryCount();
    this->m_EntryHeader.VarColCnt        = this->m_ColList [Log_VarColumn    ].GetEntryCount();
    this->m_EntryHeader.VarLongColCnt    = this->m_ColList [Log_VarLongColumn].GetEntryCount();

    this->m_EntryHeader.ImageLen += Log_AlignedImageLen (this->m_EntryHeader.LengthAllValues);
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateAnyVarColPart (const Log_ColumnList &ColList,
                                               SAPDB_Int4           &OffsetFollowingOldPart,
                                               const tgg00_Rec      *pOldRec,
                                               tgg00_Rec            *pNewRec,
                                               SAPDB_Int4            NewRecSize,
                                               tgg00_BasisError     &Error)     const
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateAnyVarColPart", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    if (e_ok != Error) return;
    
    Log_ColEntryIter Iter (ColList);
    
    Log_VarPartHandler VarPartHandler (ColList.GetColumnType(),
        OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize);
    
    while ( Iter.HasMore() )
    {
        Iter.Next();
        VarPartHandler.CopyUnchangedColumns (Iter.GetNumber(), Error);
        
        if (e_ok != Error) return;
        
        VarPartHandler.CreateMissingUndefColumns (Iter.GetNumber(), Error);
        
        if (e_ok != Error) return;
        
        Iter.MoveColToRec (pNewRec, NewRecSize, Error);
        
        if (e_ok != Error) return;
        
        VarPartHandler.NextOldColumn();
    } 
    
    VarPartHandler.FinishVarPart (OffsetFollowingOldPart, Error);
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateAnyVarColPartAndBeforeImage (Log_ColumnList      &ColList,
                                                             SAPDB_Int4          &OffsetFollowingOldPart,
                                                             const tgg00_Rec     *pOldRec,
                                                             tgg00_Rec           *pNewRec,
                                                             SAPDB_Int4           NewRecSize,
                                                             Log_RecordColumnMap &BeforeImage,
                                                             tgg00_BasisError    &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateAnyVarColPartAndBeforeImage", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);

    const SAPDB_Int4 SizeofDefinedByte = 1;

    if (e_ok != Error) return;
    
    Log_EntryIterWithRemove Iter (ColList);
    
    Log_VarPartHandler VarPartHandler (ColList.GetColumnType(),
        OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize);
    
    SAPDBTRACE_IF (LogAction_Trace,6,VarPartHandler.WriteToTrace("Log_RecordColumnMap::CreateAnyVarColPartAndBeforeImage::VarPartHandler begin"));

    SAPDB_UInt2 OldRecLen;

    while ( Iter.HasMore() )
    {
        Iter.Next();
        
        VarPartHandler.CopyUnchangedColumns (Iter.GetNumber(), Error);
        
        if (e_ok != Error) return;
        
        OldRecLen = pNewRec->recLen_gg00();
        VarPartHandler.CreateMissingUndefColumns (Iter.GetNumber(), Error);
        BeforeImage.AddNullValLen (pNewRec->recLen_gg00() - OldRecLen);
        
        if (e_ok != Error) return;
        
        if ( Iter.IsEqualValue (VarPartHandler.GetOldColumn(), VarPartHandler.GetOldColumnLength()) )
        {
            -- this->m_UsedEntries;
            Iter.Remove ();
        }
        else
        {           
            if (VarPartHandler.GetOldColumnLength() > 0)
            {
                BeforeImage.InsertValue (ColList.GetColumnType(), Iter.GetNumber(), 
                    VarPartHandler.GetOldColumnLength(), VarPartHandler.GetOldColumn(), Error);
            }
            else
            {
                BeforeImage.InsertVarUndefValue (ColList.GetColumnType(), Iter.GetNumber(), Error);
                BeforeImage.AddNullValLen (VarPartHandler.GetSizeLength() + SizeofDefinedByte);
            }
            
            if (e_ok != Error) return;
            
            Iter.MoveColToRec (pNewRec, NewRecSize, Error);
            
            if (e_ok != Error) return;
            
            VarPartHandler.NextOldColumn();
        }
    } 
    
    VarPartHandler.FinishVarPart (OffsetFollowingOldPart, Error);

    SAPDBTRACE_IF (LogAction_Trace,6,VarPartHandler.WriteToTrace("Log_RecordColumnMap::CreateAnyVarColPartAndBeforeImage::VarPartHandler end"));
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateColDescForExistingNewRec (SAPDBMem_IRawAllocator &RawAllocator,
                                                          const tgg00_Rec        *pOldRecord,
                                                          const tgg00_Rec        *pNewRecord,
                                                          Log_RecordColumnMap    &BeforeImage,
                                                          tgg00_BasisError       &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateColDescForExistingNewRec", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    SAPDBTRACE_IF         (LogAction_Trace,5,Kernel_TraceBuffer(pOldRecord,pOldRecord->recLen_gg00(),"OldRecord"));
    SAPDBTRACE_IF         (LogAction_Trace,5,Kernel_TraceBuffer(pNewRecord,pNewRecord->recLen_gg00(),"NewRecord"));
    
    if (e_ok != Error) return;
    
    if ( pOldRecord->recKeyLen_gg00() != pNewRecord->recKeyLen_gg00() )
    {
        Error = e_key_update;
        return;
    }
    
    if (0 != memcmp
        (pOldRecord->recBody_gg00()+0, pNewRecord->recBody_gg00()+0, pOldRecord->recKeyLen_gg00()))
    {
        // key of OldRecord != key of NewRecord
        Error = e_key_update;
        return;
    }
    
    {
        const SAPDB_Byte *pOldRest        = pOldRecord->recBuf_gg00() + FirstFixColOffset (pOldRecord);
        const SAPDB_Byte *pNewRest        = pNewRecord->recBuf_gg00() + FirstFixColOffset (pNewRecord);
        const SAPDB_Int   OldLen          = pOldRecord->recLen_gg00() - FirstFixColOffset (pOldRecord);
        const SAPDB_Int   NewLen          = pNewRecord->recLen_gg00() - FirstFixColOffset (pNewRecord);
        const SAPDB_Int   MinLen          = (OldLen > NewLen) ? NewLen : OldLen;
        const SAPDB_Int   ComparePartSize = 32;
        const SAPDB_Int   CompareEntries  = ((MinLen+ComparePartSize-1) / ComparePartSize);
        
        bool      IsOk       = true;
        SAPDB_Int CurrOffset = 0;
        SAPDB_Int CompareLen;
        
        if (OldLen == NewLen)
        {
            if (0 == OldLen)
            {
                Error = e_skip_upd;
                return;
            }
            
            if ( 0 == memcmp (pOldRest, pNewRest, OldLen) )
            {
                Error = e_skip_upd;
                return;
            }
        }
        
        this->AllocateColumnMap (RawAllocator, CompareEntries+1, IsOk);
        
        if (IsOk)
        {
            BeforeImage.AllocateColumnMap (RawAllocator, CompareEntries+1, IsOk);
        }
        
        if ( ! IsOk)
        {
            Error = e_no_more_memory;
            return;
        }
        
        while (CurrOffset < MinLen)
        {
            CompareLen = (CurrOffset + ComparePartSize <= MinLen) ? ComparePartSize : MinLen - CurrOffset;
            
            if ( 0 != memcmp (pOldRest+CurrOffset, pNewRest+CurrOffset, CompareLen) )
            {
                BeforeImage.InsertValue (Log_FixColumn, CurrOffset, CompareLen, pOldRest+CurrOffset, Error);
                
                if (e_ok != Error) return;
                
                this->InsertValue (Log_FixColumn, CurrOffset, CompareLen, pNewRest+CurrOffset, Error);
                
                if (e_ok != Error) return;
            }
            
            CurrOffset += CompareLen;
        }
        
        if (OldLen > MinLen)
        {
            BeforeImage.InsertValue (Log_FixColumn, CurrOffset, OldLen-MinLen, pOldRest+MinLen, Error);
            
            if (e_ok != Error) return;
        }
        
        if (NewLen > MinLen)
        {
            this->InsertValue (Log_FixColumn, CurrOffset, NewLen-MinLen, pNewRest+MinLen, Error);
            
            if (e_ok != Error) return;
        }
    }
    
    BeforeImage.CalculatePersistentImage();
    this->CalculatePersistentImage();
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateFixColPartAndBeforeImage (const tgg00_Rec     *pOldRec,
                                                          tgg00_Rec           *pNewRec,
                                                          SAPDB_Int4           NewRecSize,
                                                          Log_RecordColumnMap &BeforeImage,
                                                          tgg00_BasisError    &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateFixColPartAndBeforeImage", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    if (e_ok != Error) return;
    
    SAPDB_Int4 MoveLen = FirstVarColOffset(pOldRec) - FirstFixColOffset(pOldRec);
    
    if (MoveLen > 0)
    {
        // copy all fixed columns of the old record to the new record
        g10mv1 ("UpdMap", 3, pOldRec->recLen_gg00(), NewRecSize,
            pOldRec, FirstFixColOffset(pOldRec) + 1,
            pNewRec, FirstFixColOffset(pNewRec) + 1,
            MoveLen, Error);
        
        if (e_ok != Error) return;
    }
    
    if ( this->m_ColList[Log_FixColumn].IsEmpty() ) return;
    
    {
        Log_EntryIterWithRemove  Iter (this->m_ColList[Log_FixColumn]);
        const SAPDB_Byte        *pOldColumn;
        
        while ( Iter.HasMore() )
        {
            Iter.Next();
            
            pOldColumn = pOldRec->recBuf_gg00() + (FirstFixColOffset(pOldRec) + Iter.GetOffset());
            
            if ( Iter.IsEqualValue (pOldColumn, Iter.GetLength()) )
            {
                // old fix column == new value
                -- this->m_UsedEntries;
                Iter.Remove (); 
            }
            else
            {
                BeforeImage.InsertValue
                    (Log_FixColumn, Iter.GetOffset(), Iter.GetLength(), pOldColumn, Error);
                
                if (e_ok != Error) return;
                
                // replace the old fixed column
                Iter.MoveColToRec (pNewRec, NewRecSize, Error);
                
                if (e_ok != Error) return;
            }
        }
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateKeyAndFixColPart (const tgg00_Rec  *pOldRec,
                                                  tgg00_Rec        *pNewRec,
                                                  bool              WithoutVarParts,
                                                  tgg00_BasisError &Error)         const
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateKeyAndFixColPart", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    if (e_ok != Error) return;
    
    SAPDB_Int4 MoveLen = FirstVarColOffset (pOldRec);
    
    if (WithoutVarParts)
    {
        MoveLen = (pOldRec->recLen_gg00() < pNewRec->recLen_gg00()) ?
            pOldRec->recLen_gg00() : pNewRec->recLen_gg00();
    }
    
    MoveLen -= sizeof (pOldRec->recLen_gg00());
    
    // copy header without len, all key columns and all fixed columns or the rest of the record
    g10mv1 ("UpdMap", 4, pOldRec->recLen_gg00(), pNewRec->recLen_gg00(),
        pOldRec, sizeof (pOldRec->recLen_gg00()) + 1,
        pNewRec, sizeof (pNewRec->recLen_gg00()) + 1, MoveLen, Error);
    
    if (e_ok != Error) return;
    
    if ( ! this->m_ColList[Log_FixColumn].IsEmpty() )    
    {
        Log_ColEntryIter Iter (this->m_ColList[Log_FixColumn]);
        
        while ( Iter.HasMore() )
        {
            Iter.Next();
            
            // replace the old fixed column   
            Iter.MoveColToRec (pNewRec, pNewRec->recLen_gg00(), Error);
            
            if (e_ok != Error) return;
        }
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateKeyPart (const tgg00_Rec  *pOldRec,
                                         tgg00_Rec        *pNewRec,
                                         SAPDB_Int4        NewRecSize,
                                         bool             &KeyUpdated,
                                         tgg00_BasisError &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateKeyPart", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    SAPDB_Int4        OldColumnLen;
    const SAPDB_Byte *pOldColumn;
    
    if (e_ok != Error) return;
    
    KeyUpdated = false;
    
    // copy header and all key columns of the old record to the new record
    g10mv1 ("UpdMap", 5, pOldRec->recLen_gg00(), NewRecSize,
        pOldRec, 1, pNewRec, 1, FirstFixColOffset (pOldRec), Error);
    
    if (e_ok != Error) return;
    
    pNewRec->recLen_gg00()       = FirstVarColOffset (pNewRec);
    pNewRec->recVarcolCnt_gg00() = 0;
    
    for (Log_ColumnType CurrList = Log_FixKeyColumn; CurrList <= Log_VarKeyColumn; ++CurrList)
    {
        if ( ! this->m_ColList[CurrList].IsEmpty() )
        {
            // replace changed key columns
            
            Log_EntryIterWithRemove Iter (this->m_ColList[CurrList]);
            
            while ( Iter.HasMore() )
            {
                Iter.Next();
                
                OldColumnLen = (Log_FixKeyColumn == CurrList) ?
                    Iter.GetLength() : pOldRec->recKeyLen_gg00() - Iter.GetOffset();
                
                pOldColumn = pOldRec->recBuf_gg00() + FirstKeyColOffset() + Iter.GetOffset();
                
                if ( Iter.IsEqualValue (pOldColumn, OldColumnLen) )
                {
                    // old fix key == new value
                    -- this->m_UsedEntries;
                    Iter.Remove (); 
                }
                else
                {
                    KeyUpdated = true;
                    
                    Iter.MoveColToRec (pNewRec, NewRecSize, Error);
                    
                    if (e_ok != Error) return;
                    
                    if (Log_VarKeyColumn == CurrList)
                    { 
                        pNewRec->recKeyLen_gg00() = Iter.GetOffset() + Iter.GetLength();
                        pNewRec->recLen_gg00()    = FirstVarColOffset (pNewRec);
                    }
                }
            } 
        }
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateNewRecord (const tgg00_Rec  *pOldRec,
                                           tgg00_Rec        *pNewRec,
                                           tgg00_BasisError &Error)     const
{   
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateNewRecord", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    SAPDBTRACE_IF (LogAction_Trace, 6, Kernel_TraceBuffer(pOldRec,pOldRec->recLen_gg00(),"OldRecord"));
    
    if (e_ok != Error) return;
    
    const SAPDB_Int4 NewRecSize = pNewRec->recLen_gg00();
    
    const bool WithoutVarParts = 
        (this->m_ColList[Log_VarColumn].IsEmpty()
        &&
        this->m_ColList[Log_VarLongColumn].IsEmpty() );
    
    this->CreateKeyAndFixColPart (pOldRec, pNewRec, WithoutVarParts, Error);
    
    if (e_ok != Error) return;
    
    
    if ( ! WithoutVarParts )
    {
        SAPDB_Int4 OffsetFollowingOldPart = FirstVarColOffset (pOldRec);
        
        pNewRec->recLen_gg00()       = FirstVarColOffset (pNewRec);
        pNewRec->recVarcolCnt_gg00() = 0;
        
        this->CreateAnyVarColPart (this->m_ColList [Log_VarColumn],
            OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize, Error);
        
        if (e_ok != Error) return;

        this->CreateAnyVarColPart (this->m_ColList [Log_VarLongColumn],
            OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize, Error);
    }

    SAPDBTRACE_IF(LogAction_Trace,5,Kernel_TraceBuffer(pNewRec,pNewRec->recLen_gg00(),"NewRecord"));
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::CreateNewRecordAndBeforeImage (const tgg00_Rec     *pOldRec,
                                                         tgg00_Rec           *pNewRec,
                                                         SAPDB_Int4           NewRecSize,
                                                         Log_RecordColumnMap &BeforeImage,
                                                         tgg00_BasisError    &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::CreateNewRecordAndBeforeImage", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    SAPDBTRACE_IF(LogAction_Trace,5,Kernel_TraceBuffer(pOldRec,pOldRec->recLen_gg00(),"OldRecord"));
    
    if (e_ok != Error) return;
    
    bool KeyUpdated = false;
    
    SAPDBTRACE_IF (LogAction_Trace,6,WriteToTrace("Log_RecordColumnMap::CreateNewRecordAndBeforeImage 1"));
    
    this->CreateKeyPart (pOldRec, pNewRec, NewRecSize, KeyUpdated, Error);
    
    SAPDBTRACE_IF (LogAction_Trace,6,WriteToTrace("Log_RecordColumnMap::CreateNewRecordAndBeforeImage 2"));
    
    if (e_ok != Error) return;
    
    this->CreateFixColPartAndBeforeImage (pOldRec, pNewRec, NewRecSize, BeforeImage, Error);
    
    SAPDBTRACE_IF (LogAction_Trace,6,WriteToTrace("Log_RecordColumnMap::CreateNewRecordAndBeforeImage 3"));
    
    if (e_ok != Error) return;
    
    SAPDB_Int4 OffsetFollowingOldPart = FirstVarColOffset (pOldRec);
    
    this->CreateAnyVarColPartAndBeforeImage (this->m_ColList [Log_VarColumn],
        OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize, BeforeImage, Error);
    
    SAPDBTRACE_IF (LogAction_Trace,6,WriteToTrace("Log_RecordColumnMap::CreateNewRecordAndBeforeImage 4"));

    if (e_ok != Error) return;
    
    this->CreateAnyVarColPartAndBeforeImage (this->m_ColList [Log_VarLongColumn],
        OffsetFollowingOldPart, pOldRec, pNewRec, NewRecSize, BeforeImage, Error);
    
    SAPDBTRACE_IF (LogAction_Trace,6,WriteToTrace("Log_RecordColumnMap::CreateNewRecordAndBeforeImage 5"));

    if (e_ok != Error) return;
    
    if (KeyUpdated)
    {
        Error = e_key_update;
        return;
    }
    
    BeforeImage.CalculatePersistentImage();
    this->CalculatePersistentImage();
    
    if (this->m_EntryHeader.ImageEntries < 1) Error = e_skip_upd;
    
    SAPDBTRACE_IF(LogAction_Trace,5,Kernel_TraceBuffer(pNewRec,pNewRec->recLen_gg00(),"NewRecord"));
}

//----------------------------------------------------------------------------

bool Log_RecordColumnMap::ExistsEntry (const tgg00_StackEntry &StackEntry) const
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::ExistsEntry", LogAction_Trace, 5);
    const Log_ColumnType ColumnType = GetColumnType (StackEntry);
    
    if ( this->m_ColList [ColumnType].IsEmpty() ) return (false);
    {
        const Log_ColumnEntry *CurrEntry = this->m_ColList [ColumnType].GetFirst();
        
        while (NULL != CurrEntry)
        {
            if ((Log_FixKeyColumn == ColumnType) ||
                (Log_VarKeyColumn == ColumnType) ||
                (Log_FixColumn    == ColumnType) )
            {
                if ( CurrEntry->GetOffset()+1 == StackEntry.epos() ) return (true);
            }
            else
            {
                if ( CurrEntry->GetNumber() == StackEntry.ecolno() ) return (true);
            }
            
            CurrEntry = CurrEntry->GetNext();
        }
    }
    return (false);
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::InsertColumn (const tgg00_StackEntry &StackEntry,
                                        const SAPDB_Byte       *pValue,
                                        SAPDB_Int4              ValueLength,
                                        tgg00_BasisError       &Error)
{   
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::InsertColumn", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    if (e_ok != Error) return;
    
    const Log_ColumnType ColumnType = GetColumnType(StackEntry);
    
    const SAPDB_Int4 ColOffsetOrNumber =
        ((Log_VarColumn == ColumnType) || (Log_VarLongColumn == ColumnType)) ?
        StackEntry.ecolno() : (StackEntry.epos()-1);
    
    const SAPDB_Int4 FixedColumnLen =
        ((Log_FixColumn == ColumnType) || (Log_FixKeyColumn == ColumnType)) ? StackEntry.elen_var() : 0;
    
    SAPDB_Int4       TruncValueLen    = ValueLength;
    bool             CreateUndefValue = false;
    Log_ColumnEntry *pNewEntry        = this->GetFreeEntry();
    
    if ((NULL == pNewEntry) || (Log_NilColumn == ColumnType))
    {
        Error = e_stack_type_illegal;
        return;
    }
    
    TruncateValue (pValue, TruncValueLen, CreateUndefValue);
    
    if (CreateUndefValue)
    {
        pNewEntry->AllocateAndInitUndefCol
            ( *(this->m_pRawAllocator), ColOffsetOrNumber, FixedColumnLen, Error);
        
        if (e_ok != Error) return;
    }
    else
    {
        if ((Log_FixColumn == ColumnType) || (Log_FixKeyColumn == ColumnType))
        {
            if (TruncValueLen > FixedColumnLen)
            {
                Error = e_column_trunc; // value too large
                return;
            }
            
            if (ValueLength < FixedColumnLen)
            {
                // enlarge value with trailing defined bytes
                pNewEntry->AllocateAndInitFixCol ( *(this->m_pRawAllocator),
                    ColOffsetOrNumber, FixedColumnLen, ValueLength, pValue, Error);
                
                if (e_ok != Error) return;
            }
            else  
            {
                pNewEntry->Init (ColOffsetOrNumber, FixedColumnLen, pValue);
            }
        }
        else
        {
            pNewEntry->Init (ColOffsetOrNumber, TruncValueLen, pValue);
        }
    }
    
    switch (ColumnType)
    {
    case Log_FixKeyColumn:
    case Log_VarKeyColumn:
        {
            if (ColOffsetOrNumber + pNewEntry->GetLength() > SAPDB_MAX_UINT2 )
            {
                Error = e_column_trunc; // key too large
                return;
            }
            
            if (csp_undef_byte == pNewEntry->GetValue()[0])
            {
                Error = e_null_value_illegal;
                return;
            }
            
            this->m_ColList[ColumnType].InsertOnTop (pNewEntry);
            break;
        }
    case Log_FixColumn:
        {
            this->m_ColList[Log_FixColumn].InsertOnTop (pNewEntry);
            break;
        }
    case Log_VarColumn:
        {
            if (TruncValueLen > SAPDB_MAX_UINT1)
            {
                Error = e_column_trunc; // value too large
                return;
            }
            this->m_ColList[Log_VarColumn].InsertSorted (pNewEntry);
            break;
        }
    case Log_VarLongColumn:
        {
            if (TruncValueLen > SAPDB_MAX_UINT2)
            {
                Error = e_column_trunc; // value too large
                return;
            }
            this->m_ColList[Log_VarLongColumn].InsertSorted (pNewEntry);
            break;
        }
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::InsertValue (Log_ColumnType    ColumnType,
                                       SAPDB_Int4        OffsetOrNumber,
                                       SAPDB_Int4        ValueLength,
                                       const SAPDB_Byte *pValue,
                                       tgg00_BasisError &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::InsertValue", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);

    if (e_ok != Error) return;
    
    Log_ColumnEntry *pNewEntry = this->GetFreeEntry();
    
    if ((ValueLength < 1) || (NULL == pNewEntry))
    {
        Error = e_stack_type_illegal;
        return;
    }
    
    pNewEntry->Init (OffsetOrNumber, ValueLength, pValue);
    
    switch (ColumnType)
    {
    case Log_FixColumn:
        this->m_ColList[ColumnType].InsertOnTop (pNewEntry);
        break;
        
    case Log_VarColumn:
    case Log_VarLongColumn:
        this->m_ColList[ColumnType].InsertSorted (pNewEntry);
        break;
        
    default:
        Error = e_stack_type_illegal;
        return;
    }
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::InsertVarUndefValue (Log_ColumnType    ColumnType,
                                               SAPDB_Int4        OffsetOrNumber,
                                               tgg00_BasisError &Error)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::InsertVarUndefValue", LogAction_Trace, 5);
    
    KERNEL_TRACE_BASIS_ERROR (Error, LogAction_Trace, 5);
    
    if (e_ok != Error) return;
    
    Log_ColumnEntry *pNewEntry = this->GetFreeEntry();
    
    if (NULL == pNewEntry)
    {
        Error = e_stack_type_illegal;
        return;
    }
    
    const SAPDB_Int4 UndefColumnLen = 0;
    
    pNewEntry->AllocateAndInitUndefCol
        ( *(this->m_pRawAllocator), OffsetOrNumber, UndefColumnLen, Error);
    
    if (e_ok != Error) return;
    
    if ((Log_VarColumn     == ColumnType) ||
        (Log_VarLongColumn == ColumnType) )
    {
        this->m_ColList[ColumnType].InsertSorted (pNewEntry);
        return;
    }
    
    Error = e_stack_type_illegal;
    return;
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::ReadColumnMapPersistent (Log_ActionImageReader &ImageReader,
                                                   SAPDB_Int4             PersistentLen,
                                                   bool                  &IsOk)
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::ReadColumnMapPersistent", LogAction_Trace, 5);
    
    KERNEL_TRACE_IS_OK (IsOk, LogAction_Trace, 5);
    
    PersistentColEntry     *pPersistEntries          = NULL;
    SAPDBMem_IRawAllocator *pPersistEntriesAllocator = NULL;
    
    ImageReader.CopyImage (
        REINTERPRET_CAST (SAPDB_Byte*, &(this->m_EntryHeader)),
        sizeof(this->m_EntryHeader),
        IsOk);
    
    if ( ! IsOk ) return;
    
    if (this->m_EntryHeader.ImageEntries > 0)
    {
        this->AllocateColumnMap (ImageReader.GetAllocator(), this->m_EntryHeader.ImageEntries, IsOk);
        
        if ( ! IsOk ) return;
        
        ImageReader.PersistentRead (REINTERPRET_CAST (SAPDB_Byte*&, pPersistEntries),
            this->m_EntryHeader.ImageEntries * sizeof (PersistentColEntry), pPersistEntriesAllocator, IsOk);
        
        if (IsOk && (this->m_EntryHeader.LengthAllValues > 0))
        {
            ImageReader.PersistentRead (this->m_pAllValuesForRead,
                this->m_EntryHeader.LengthAllValues, this->m_pAllValuesAllocator, IsOk);
        }
        
        if (IsOk)
        {
            SAPDB_Int4       EntryCntPerList;
            SAPDB_Int4       CurrPersist     = 0;
            SAPDB_Int4       CurrValueOffset = 0;
            Log_ColumnType   CurrList        = Log_FixColumn; // update image does not contain any altered key
            tgg00_BasisError Error           = e_ok;
            SAPDB_UInt2      EntryCnt [Log_ColumnLastEnum];
            
            EntryCnt [Log_FixColumn    ] = this->m_EntryHeader.FixColCnt;
            EntryCnt [Log_VarColumn    ] = this->m_EntryHeader.VarColCnt;
            EntryCnt [Log_VarLongColumn] = this->m_EntryHeader.VarLongColCnt;
            
            do
            {
                EntryCntPerList = 0;
                
                while (IsOk && (EntryCntPerList < EntryCnt[CurrList]))
                { 
                    this->InsertValue (
                        CurrList,
                        (pPersistEntries + CurrPersist)->OffsetOrNumber,
                        (pPersistEntries + CurrPersist)->ValueLength,
                        this->m_pAllValuesForRead + CurrValueOffset,
                        Error);
                    
                    IsOk = (e_ok == Error);
                    
                    CurrValueOffset += (pPersistEntries + CurrPersist)->ValueLength;
                    ++EntryCntPerList;
                    ++CurrPersist;
                    
                    if (CurrValueOffset > this->m_EntryHeader.LengthAllValues) IsOk = false;
                }          
                ++CurrList; 
            }
            while (IsOk && (CurrList < Log_ColumnLastEnum));
        }
        
        if (NULL != pPersistEntriesAllocator)
        {
            pPersistEntriesAllocator->Deallocate (pPersistEntries);
        }
    }
    SAPDBTRACE_IF (LogAction_Trace, 6, WriteToTrace("ReadColumnMapPersistent"));
}

//----------------------------------------------------------------------------

void Log_RecordColumnMap::WriteColumnMapPersistent (Log_ActionImageWriter &ImageWriter,
                                                    bool                  &IsOk)       const
{
    SAPDBTRACE_METHOD_DEBUG ("Log_RecordColumnMap::WriteColumnMapPersistent", LogAction_Trace, 5);
    
    KERNEL_TRACE_IS_OK(IsOk, LogAction_Trace, 5);
    
    IsOk = true;
    
    SAPDBTRACE_IF (LogAction_Trace, 6, WriteToTrace("WriteColumnMapPersistent"));

    ImageWriter.PersistentWrite (&(this->m_EntryHeader), sizeof (this->m_EntryHeader), IsOk);
    
    if ( ! IsOk ) return;
    
    if (this->m_EntryHeader.ImageEntries > 0)
    {
        const Log_ColumnEntry *pCurrEntry;
        Log_ColumnType         CurrColList;
        PersistentColEntry    *pPersistEntries = NULL;
        SAPDB_Int4             CurrPersist     = 0;
        
        newarray (pPersistEntries, this->m_EntryHeader.ImageEntries, *(this->m_pRawAllocator));
        
        if (NULL == pPersistEntries)
        {
            IsOk = false;
            return;
        }
        
        // update image does not contain any altered key, therefore FixColumn .. VarLongColumn
        
        for (CurrColList = Log_FixColumn; CurrColList < Log_ColumnLastEnum; ++CurrColList)
        {
            pCurrEntry = this->m_ColList[CurrColList].GetFirst();
            
            while (NULL != pCurrEntry)
            {
                (pPersistEntries + CurrPersist)->OffsetOrNumber = pCurrEntry->GetOffset();
                (pPersistEntries + CurrPersist)->ValueLength    = pCurrEntry->GetLength();
                pCurrEntry = pCurrEntry->GetNext();
                ++CurrPersist;
            }
        }
        
        ImageWriter.PersistentWrite (pPersistEntries,
            this->m_EntryHeader.ImageEntries * sizeof (PersistentColEntry), IsOk);
        
        destroyarray (pPersistEntries, this->m_EntryHeader.ImageEntries, *(this->m_pRawAllocator));
        
        if ( ! IsOk ) return;
        
        if (this->m_EntryHeader.LengthAllValues > 0)
        {
            for (CurrColList = Log_FixKeyColumn; CurrColList < Log_ColumnLastEnum; ++CurrColList)
            {
                pCurrEntry = this->m_ColList[CurrColList].GetFirst();
                
                while (NULL != pCurrEntry)
                {
                    if ( NULL != pCurrEntry->GetValue() )
                    {
                        ImageWriter.UnalignedPersistentWrite
                            (pCurrEntry->GetValue(), pCurrEntry->GetLength(), IsOk);
                        
                        if ( ! IsOk ) return;
                    }
                    
                    pCurrEntry = pCurrEntry->GetNext();
                }
            }
        }
        
        ImageWriter.AlignImage (this->m_EntryHeader.LengthAllValues, IsOk);
    }
}

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