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

  module      : vbd610.cpp

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

  responsible : UweH

  special area: NodeHandling
  description : 


  last changed: 1999-02-24  11:54
  see also    : example.html ...

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

  copyright:    (c) 1998-2004 SAP AG



    ========== licence begin  GPL
    Copyright (c) 1998-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 "gsp00.h"
#include "ggg00.h"
#include "gbd00.h"
#include "gbd600.h"
#include "hsp30.h"
#include "hgg01_1.h"
#include "SAPDB/SAPDBCommon/SAPDB_RangeCode.hpp" // Kernel_move_and_fill
#include "hgg10.h"
#include "hbd06.h"


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

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

#define DUMMY_REC_OFFSET 0

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


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

struct tbd610_QuickSortElem
{
    tsp00_Int4  qseLeft;
    tsp00_Int4  qseRight;
};

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


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


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


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


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


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

void
cbd600_Node::bd610_Merge (cbd600_Node &SourceNode,
						 tsp00_Int4    Offset)
{
    ROUTINE_DBG_MEO00 ("bd610_Merge");

		  tsp00_Int4  &SourceRecCount = SourceNode.m_pNode->nd_record_cnt();
		  tsp00_Int4  &TargetRecCount = m_pNode->nd_record_cnt();
	const tsp00_Int4  ShiftLen        = SourceRecCount << DIVISOR_POINTERSIZE_BD00;
	const tsp00_Int4  OldMaxIndex     = bd600MaxRecIndex();
	
	/* move all PointerListItems from the right      */
	/* node (= SrcNode) to the left node (= m_pNode) */
	g10mv( __FILE__, 1,
		sizeof (*SourceNode.m_pNode), sizeof (*m_pNode),
		SourceNode.m_pRecIndexList-(SourceRecCount-1), POS_OFF_DIFF_BD00,
		m_pRecIndexList-(TargetRecCount-1)-SourceRecCount, POS_OFF_DIFF_BD00,
		ShiftLen, m_TrError);
	if (e_ok != m_TrError) return;
	
	m_pNode->nd_record_cnt() += SourceRecCount;
	
	bd610_UpdatePositionsRange (OldMaxIndex + 1, m_pNode->nd_record_cnt() - 1,
		DUMMY_REC_OFFSET, DUMMY_REC_OFFSET, Offset);
}

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

void 
cbd600_Node::bd610_AddPosition (tsp00_Int4 RecPos, 
                                tsp00_Int4 RecIndex)
{
    ROUTINE_DBG_MEO00 ("bd610_AddPosition");

    if (0 != m_pNode->nd_record_cnt()) // something to move = list is not empty
    {
		if (RecIndex <= bd600MaxRecIndex())
		{
			g10mv( __FILE__, 2,
				sizeof (*m_pNode), sizeof (*m_pNode),
				m_pRecIndexList-(bd600MaxRecIndex()  ), POS_OFF_DIFF_BD00,
				m_pRecIndexList-(bd600MaxRecIndex()+1), POS_OFF_DIFF_BD00,
				(bd600MaxRecIndex() - RecIndex + 1) << DIVISOR_POINTERSIZE_BD00,
				m_TrError);
		}
		else if (RecIndex > bd600MaxRecIndex() + 1)
		{
			m_TrError = e_data_page_corrupted;
		}

        if (e_ok != m_TrError)
        {
            m_TrError = e_data_page_corrupted;
            g01opmsg (sp3p_knldiag, sp3m_error, gbd610c1DataPageCorrupted_csp03,
                csp3_n_btree, "Pointerlist inconsistent", RecIndex);
            b06dump_bad_page (m_Current.curr_trans->trTaskId_gg00, 'd',
                FILE_EXT_COR_BD00, m_pNode->nd_id(), m_pNode, 1);
            b06write_filename_and_root (m_Current.curr_tree_id);
            return;
        }
    }
    
    *(m_pRecIndexList-RecIndex) = RecPos;
    m_pNode->nd_record_cnt()++;

#   if COMPILEMODE_MEO00 >= SLOW_MEO00
	t01p2int4 (bd_inv, "NewRecIndex ", RecIndex, "NewRecPos   ", RecPos);
#   endif
}

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

void
cbd600_Node::bd610_DelPosition (tsp00_Int4 RecIndex)
{
    ROUTINE_DBG_MEO00 ("bd610_DelPosition");

   /* This function deletes the pointerlist entry */
   /* with index RecIndex from the pointerlist    */

    if (0 == m_pNode->nd_record_cnt())
	{
		// No Record in page to delete

        m_TrError = e_data_page_corrupted;
        g01opmsg (sp3p_knldiag, sp3m_error, gbd610c2DataPageCorrupted_csp03,
            csp3_n_btree, "Data page is empty      ", RecIndex);
        b06dump_bad_page (m_Current.curr_trans->trTaskId_gg00, 'd',
            FILE_EXT_COR_BD00, m_pNode->nd_id(), m_pNode, 1);
        b06write_filename_and_root (m_Current.curr_tree_id);
	}
    else
    {
		SAPDB_RangeOverlappingMove ("VBD610", 3,
			sizeof (*m_pNode), sizeof (*m_pNode),
			m_pRecIndexList-bd600MaxRecIndex()    , POS_OFF_DIFF_BD00,
			m_pRecIndexList-(bd600MaxRecIndex()-1), POS_OFF_DIFF_BD00,
			(bd600MaxRecIndex() - RecIndex) << DIVISOR_POINTERSIZE_BD00,
			m_TrError);
    
		if (e_move_error == m_TrError)
		{
			m_TrError = e_data_page_corrupted;
            g01opmsg (sp3p_knldiag, sp3m_error, gbd610c3DataPageCorrupted_csp03,
            csp3_n_btree, "Pointerlist is corrupted", RecIndex);
			b06dump_bad_page (m_Current.curr_trans->trTaskId_gg00, 'd',
				FILE_EXT_COR_BD00, m_pNode->nd_id(), m_pNode, 1);
            b06write_filename_and_root (m_Current.curr_tree_id);
#           if COMPILEMODE_MEO00 >= SLOW_MEO00 
			bd600PrintPointerList();
#           endif
			return;
		}
		m_pNode->nd_record_cnt() --;
	}
#   if COMPILEMODE_MEO00 >= SLOW_MEO00
	t01int4 (bd_inv, "OldRecIndex ", RecIndex);
#   endif
}
    
/*---------------------------------------------------------------------------*/

void 
cbd600_Node::bd610_CopyPointerList (const cbd600_Node &SrcNode)
{
    ROUTINE_DBG_MEO00 ("bd610_CopyPointerList");

    m_pNode->nd_id()         = NIL_PAGE_NO_GG00;
    m_pNode->nd_record_cnt() = SrcNode.m_pNode->nd_record_cnt();
    m_pNode->nd_bottom  ()   = SrcNode.m_pNode->nd_bottom  ();  // only for check
    g10mv( __FILE__, 4,    
        sizeof (*SrcNode.m_pNode), sizeof (*m_pNode),
        SrcNode.m_pRecIndexList - (m_pNode->nd_record_cnt()-1), POS_OFF_DIFF_BD00,
                m_pRecIndexList - (m_pNode->nd_record_cnt()-1), POS_OFF_DIFF_BD00,
        m_pNode->nd_record_cnt() << DIVISOR_POINTERSIZE_BD00,
        m_TrError);
    
}


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

void 
cbd600_Node::bd610_SortForDistribution (tsp00_Int4 LastDistribIndex)
{
    ROUTINE_DBG_MEO00 ("bd610_SortForDistribution");
    
	const tsp00_Int4 NumDistribIndexes = bd600MaxRecIndex() - LastDistribIndex + 1;

    if (NumDistribIndexes > 1)
        bd610_QuickSort (NumDistribIndexes, m_pRecIndexList-bd600MaxRecIndex());

    if (LastDistribIndex > 1)
        bd610_QuickSort (LastDistribIndex, m_pRecIndexList-(LastDistribIndex-1));
}

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

void
cbd600_Node::bd610_UpdatePositions (tsp00_Int4 TargetPos,
								   tsp00_Int4 ShiftLen)
{
    ROUTINE_DBG_MEO00 ("bd610_UpdatePositions");
	
	for (tsp00_Int4 RecIndex = FIRST_REC_INDEX_BD00; RecIndex <= bd600MaxRecIndex(); RecIndex ++)
	{
		if (bd600RecOffset(RecIndex) + POS_OFF_DIFF_BD00 >= TargetPos) 
			*(m_pRecIndexList-RecIndex) += ShiftLen;
		
		if (e_ok != m_TrError) return;
	}
}

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

void
cbd600_Node::bd610_UpdatePositionsRange (tsp00_Int4 FirstIndex,
										tsp00_Int4 LastIndex,
										tsp00_Int4 MinRecOffset,
										tsp00_Int4 MaxRecOffset,
										tsp00_Int4 ShiftLen)
{
    ROUTINE_DBG_MEO00 ("bd610_UpdatePositionsRange");

	const bool        bIgnoreOffsets = DUMMY_REC_OFFSET == MinRecOffset && DUMMY_REC_OFFSET == MaxRecOffset;
	tsp00_Int4        CurrIndex      = FirstIndex;
	tsp00_Int4        CurrRecOffset;
	
	while (CurrIndex <= LastIndex)
	{
		CurrRecOffset = bd600RecOffset(CurrIndex);
		if (e_ok != m_TrError) return;
		
		if (bIgnoreOffsets || (CurrRecOffset >= MinRecOffset && CurrRecOffset < MaxRecOffset)) 
		{
			*(m_pRecIndexList-CurrIndex) += ShiftLen;
		}
		CurrIndex++;
	}
}

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

void 
cbd600_Node::bd610_QuickSort (tsp00_Int4      NumRecPosElem,
                             tgg00_PagePos  *pMinRecPosElem)
{
    
    ROUTINE_DBG_MEO00 ("bd610_QuickSort");

    const tsp00_Int4 MaxQuickSortStack = 16;
    tsp00_Int4       i;
    tsp00_Int4       j;
    tsp00_Int4       Left;
    tsp00_Int4       Right;
    tsp00_Int4       PivotIndex;
    tsp00_Int4       StackIndex = 0;
    tgg00_PagePos    AuxElem;
    tgg00_PagePos    PivotElem;
    ARRAY_VAR_MSP00 (Stack, tbd610_QuickSortElem, MaxQuickSortStack);
    
    
    Stack [StackIndex].qseLeft  = 0;
    Stack [StackIndex].qseRight = NumRecPosElem - 1;
    do
    {
        Left  = Stack [StackIndex].qseLeft;
        Right = Stack [StackIndex].qseRight;
        StackIndex --;
        do
        {
            i          = Left;
            j          = Right;
            PivotIndex = (Left + Right) >> 1;
            PivotElem  = *(pMinRecPosElem + PivotIndex);
            do
            {
                while (*(pMinRecPosElem + i)  < PivotElem) i++;
                while (PivotElem < *(pMinRecPosElem + j)) j--;
                if (i <= j)
                {
                    AuxElem = *(pMinRecPosElem + i);
                    *(pMinRecPosElem + i) = *(pMinRecPosElem + j); 
                    *(pMinRecPosElem + j) = AuxElem;    
                    i++;
                    j--;
                }
            }
            while (i <= j);
            
            if ((j - Left) < (Right - i))
            {
                if (i < Right)
                {
                    StackIndex++;
                    Stack [StackIndex].qseLeft  = i;
                    Stack [StackIndex].qseRight = Right;
                }
                Right = j;
            }
            else
            {
                if (Left < j)
                {
                    StackIndex++;
                    Stack [StackIndex].qseLeft  = Left;
                    Stack [StackIndex].qseRight = j;
                }
                Left = i;
            }
        }
        while (Left < Right);
    }
    while (StackIndex >= 0);
}

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

void  
cbd600_Node::bd610_DelHighIndexPositions (tsp00_Int4 NumDeletedRecords)
{
    ROUTINE_DBG_MEO00 ("bd610_DelHighIndexPositions");

	/* This function deletes the highest NumOfDeletedRecords */ 
	/* pointerlist entries from the pointerlist              */

    if (0 == m_pNode->nd_record_cnt())
	{
		// No Record in page to delete

        m_TrError = e_data_page_corrupted;
        g01opmsg (sp3p_knldiag, sp3m_error, gbd610c4DataPageCorrupted_csp03,
            csp3_n_btree, "Data page is empty  ", NumDeletedRecords);
        b06dump_bad_page (m_Current.curr_trans->trTaskId_gg00, 'd',
            FILE_EXT_COR_BD00, m_pNode->nd_id(), m_pNode, 1);
        b06write_filename_and_root (m_Current.curr_tree_id);
	}
    else
    {
		const tsp00_Int4     MoveLen      = (bd600NumRecords() - NumDeletedRecords) << DIVISOR_POINTERSIZE_BD00;
			  tgg00_PagePos *pPointerList = m_pRecIndexList - (m_pNode->nd_record_cnt()-1);

		SAPDB_RangeOverlappingMove ("VBD610", 6,
			sizeof (*m_pNode),                sizeof (*m_pNode),
			pPointerList,                     POS_OFF_DIFF_BD00,
			pPointerList + NumDeletedRecords, POS_OFF_DIFF_BD00,
			MoveLen,
			m_TrError);
    
		if (e_move_error == m_TrError)
		{
            m_TrError = e_data_page_corrupted;
            g01opmsg (sp3p_knldiag, sp3m_error, gbd610c5DataPageCorrupted_csp03,
                csp3_n_btree, "Pointerlist is corrupted", MoveLen);
            b06dump_bad_page (m_Current.curr_trans->trTaskId_gg00, 'd',
                FILE_EXT_COR_BD00, m_pNode->nd_id(), m_pNode, 1);
            b06write_filename_and_root (m_Current.curr_tree_id);
#           if COMPILEMODE_MEO00 >= SLOW_MEO00 
            bd600PrintPointerList();
#			endif
            return;
		}

		m_pNode->nd_record_cnt() -= NumDeletedRecords;
	}
#   if COMPILEMODE_MEO00 >= SLOW_MEO00
	t01int4 (bd_inv, "#deleted rec", NumDeletedRecords);
#   endif
}
	
/*===========================================================================*
 *  END OF CODE                                                              *
 *============================0===============================================*/
