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

  module      : vbd496.cpp

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

  author      : TorstenS
  responsible : TorstenS

  special area: FileHandling
  description : 


  last changed: 1999-07-23  14:00
  see also    : 

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

  copyright:    (c) 1999-2004 SAP AG



    ========== 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 "gbd496.h"

#include "gsp03.h"    // PASCAL: SP_message_constants_and_types
#include "gsp03_3.h"  // PASCAL: SP_message_constants_and_types_for_bd_layer
#include "hbd01_1.h"  // PASCAL: filesysteminterface_1
#include "hgg01_1.h"  // PASCAL: Configuration_Parameter


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


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


#define FIRST_FILE_INDEX_BD496		0
#define NUM_RESULT_FILES_BD496		62
#define NUM_OVERFLOW_FILES_BD496	2

#define MERGE_MSG_1_BD496			"Perm Leaf Pages Read    "
#define MERGE_MSG_2_BD496			"Perm Records Read       "
#define MERGE_MSG_3_BD496           "Start Final Merge Step                  "
#define MERGE_MSG_4_BD496           "Start Merge Temp Files                  "
#define MERGE_MSG_5_BD496			"Stop Final Merge Step   "
#define MERGE_MSG_6_BD496			"Stop Merge Temp Files   "
#define MERGE_MSG_7_BD496			"All Perm Leaf Pages Read"
#define MERGE_MSG_8_BD496			"All Perm Records Read   "
#define OVERFLOW_MSG_1_BD496        "Fast Merge Failed: Data Cache Too Small "
#define MEMORY_MSG_1_BD496          "Funnel: Sort Memory Too Small           "


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



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



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



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



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



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



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



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

inline tbd_current_tree
bd496_InitTempCurrent (tgg00_TransContext &Trans)
{
    tbd_current_tree	TempCurrent;

    TempCurrent              = b01nilcurrent;
    TempCurrent.curr_trans   = &Trans;
    TempCurrent.curr_tree_id = bd495InitTempFileId (NIL_PAGE_NO_GG00);

    return TempCurrent;
}

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

inline void
cbd496_FileHandling::bd496_SetEntry (
    tsp00_Int4     CurrFileIndex,
    tsp00_PageNo   PrimQueuePno,
    tsp00_PageNo   SecQueuePno)
{
    (m_pFileList + CurrFileIndex)->fliPrimPno_bd496 = PrimQueuePno;
    (m_pFileList + CurrFileIndex)->fliSecPno_bd496  = SecQueuePno;
}

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

inline void
cbd496_FileHandling::bd496_ClearEntry (
    tsp00_Int4   CurrFileIndex,
    bool         bInit)
{
    if (!bInit)
    {
        if (bd496_EntryUsed (CurrFileIndex))
        {
            if (bd496_IsQueueEntry (CurrFileIndex))
            {
                if (0 < m_NumOpenQueues) --m_NumOpenQueues;
            }
            else
                if (0 < m_NumOpenTrees) --m_NumOpenTrees;
        }
    }
    bd496_SetEntry (CurrFileIndex, NIL_PAGE_NO_GG00, NIL_PAGE_NO_GG00);
}

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

inline bool
cbd496_FileHandling::bd496_EntryUsed (
    tsp00_Int4      CurrFileIndex)
{
    return
        (NIL_PAGE_NO_GG00 != (m_pFileList + CurrFileIndex)->fliPrimPno_bd496);
}

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

inline tsp00_PageNo
cbd496_FileHandling::bd496_GetPrimQueuePno(
    tsp00_Int4  CurrFileIndex)
{
    return (m_pFileList + CurrFileIndex)->fliPrimPno_bd496;
}

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

inline tsp00_PageNo
cbd496_FileHandling::bd496_GetSecQueuePno(
    tsp00_Int4  CurrFileIndex)
{
    return (m_pFileList + CurrFileIndex)->fliSecPno_bd496;
}

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

inline tsp00_PageNo
cbd496_FileHandling::bd496_GetTempInvRoot(
    tsp00_Int4  CurrFileIndex)
{
    return (m_pFileList + CurrFileIndex)->fliPrimPno_bd496;
}

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

inline void
cbd496_FileHandling::bd496_InitFile(
    tsp00_Int4      CurrFileIndex,
    tgg00_FileId   &QueueId,
    tsp00_PageNo   &PrimQueuePno,
    tsp00_PageNo   &SecQueuePno)
{

    tbd_current_tree QueueCurrent = bd496_InitTempCurrent (m_Trans);

    cbd495_DoubleWriteQueue TempInvQueue (QueueCurrent);
    if (e_ok != m_TrError) return;

    PrimQueuePno = TempInvQueue.bd495GetPrimPno();
    SecQueuePno  = TempInvQueue.bd495GetSecPno();
    QueueId      = QueueCurrent.curr_tree_id;

    bd496_SetEntry (CurrFileIndex, PrimQueuePno, SecQueuePno);

    ++m_NumOpenQueues;
}

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

inline bool
cbd496_FileHandling::bd496_IsQueueEntry(
    tsp00_Int4  CurrFileIndex)
{
    return
        (NIL_PAGE_NO_GG00 != (m_pFileList + CurrFileIndex)->fliPrimPno_bd496) &&
        (NIL_PAGE_NO_GG00 != (m_pFileList + CurrFileIndex)->fliSecPno_bd496);
}

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

void
cbd496_FileHandling::bd496_Merge (
    bool                        bLastMergeStep,
    tsp00_Int4                  LastItem,
    const cbd497_StackDescInfo  &stackDescInfo )

{
    ROUTINE_DBG_MEO00 ("bd496_Merge");


    tbd_current_tree QueueCurrent = bd496_InitTempCurrent (m_Trans);

    cbd494_Funnel   Funnel( QueueCurrent, stackDescInfo, m_pFunnel, m_NumFunnelItems );

    // Initialize Funnel Merge by creating all InvQueues within given memory
    // via placement new. Only InvQueues are created having a corresponding
    // entry within the FileList.
    // Note that the loop first creates the result files and then the ordinary sort files.
    // This warrants that the result files are always createable!

    for (tsp00_Int4 CurrFileIndex = LastItem; ((0 <= CurrFileIndex) && (e_ok == m_TrError)); --CurrFileIndex)
    {
        if (bd496_EntryUsed (CurrFileIndex))
        {
            if (bd496_IsQueueEntry (CurrFileIndex))
                Funnel.bd494Add (QUEUE_ENTRY_BD496, bd496_GetPrimQueuePno (CurrFileIndex),
                                 bd496_GetSecQueuePno (CurrFileIndex));
            else
                Funnel.bd494Add (!QUEUE_ENTRY_BD496, bd496_GetTempInvRoot (CurrFileIndex));

            if (e_ok == m_TrError)
                bd496_ClearEntry (CurrFileIndex); // Release InvQueue by ~Funnel
        }
    }

    // Because of the incredible size of a TempInvTree it is possible that bd494Add is not
    // be able to create all entries of the FileList within the given memory. Therefor another
    // Funnel Merge step with the remaining FileList entries is necessary.

    if (e_no_more_memory == m_TrError)
        m_TrError = e_ok;

    if (e_ok == m_TrError)
    {
        if (bLastMergeStep)
        {
            Funnel.bd494Merge (m_InvTree);
        }
        else
        {
            tsp00_PageNo    PrimQueuePno;
            tsp00_PageNo    SecQueuePno;

            bd496CreateFile (m_TargetFileIndex, PrimQueuePno, SecQueuePno);

            cbd494_SortableDoubleWriteQueue TempResultQueue (QueueCurrent, PrimQueuePno, SecQueuePno);

            Funnel.bd494Merge (TempResultQueue);
        }
    }

#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
    t01basis_error (bd_idx_create, "FunnelMerge ", m_TrError);
#   endif
}

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

cbd496_FileHandling::cbd496_FileHandling (
    tgg00_TransContext	    &Trans,
    cbd494_SortableInvTree  &InvTree,
    bool				    bInvIsUnique)
        :
        m_Trans			(Trans),
        m_TrError		(Trans.trError_gg00),
        m_InvTree	    (InvTree),
        m_NumOpenQueues (0),
        m_NumOpenTrees  (0),
        m_MergeCount	(0)
{
    ROUTINE_DBG_MEO00 ("cbd496_FileHandling");


    if (e_ok != m_TrError) return;

    tsp00_BytePtr	pAuxMem;

    if (!m_MemUnit.bd499GetFileListMemory (m_Trans.trTaskId_gg00,
                                           pAuxMem, m_NumFileListItems))
    {
        m_TrError = e_no_more_memory; return;
    }
    m_pFileList = REINTERPRET_CAST (tbd496_FileListItem*, pAuxMem);

    if (!m_MemUnit.bd499GetFunnelStackMemory (m_Trans.trTaskId_gg00,
            m_pFunnel, m_NumFunnelItems))
    {
        m_TrError = e_no_more_memory; return;
    }

    m_FirstFileIndex	     = FIRST_FILE_INDEX_BD496;
    m_LastFileIndex		     = m_NumFileListItems - 1;

    m_FirstSortFileIndex     = FIRST_FILE_INDEX_BD496;
    m_LastSortFileIndex      = m_LastFileIndex - NUM_RESULT_FILES_BD496 - NUM_OVERFLOW_FILES_BD496;

    m_FirstResultFileIndex   = m_LastSortFileIndex + 1;
    m_LastResultFileIndex    = m_LastFileIndex - NUM_OVERFLOW_FILES_BD496;

    m_FirstOverflowFileIndex = m_LastResultFileIndex + 1;
    m_LastOverflowFileIndex  = m_FirstOverflowFileIndex + 1;

    m_TargetFileIndex        = m_LastSortFileIndex; // Incremented before first use

    for (tsp00_Int4 iItem = m_FirstFileIndex; iItem <= m_LastFileIndex; ++iItem)
        bd496_ClearEntry (iItem, INIT_FILELIST_BD496);
}

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

void
cbd496_FileHandling::bd496ConvertFile (
    tsp00_Int4   CurrFileIndex,
    tsp00_PageNo TempInvRoot)
{
    ROUTINE_DBG_MEO00 ("bd496ConvertFile");


    // Drop all pages of the DoubleWriteQueue and replace the anchor
    // entry with the pno of the TempInvTree

    if (bd496_EntryUsed (CurrFileIndex))
    {
        tbd_current_tree QueueCurrent = bd496_InitTempCurrent (m_Trans);

        cbd495_DoubleWriteQueue TempInvQueue (QueueCurrent,
                                              bd496_GetPrimQueuePno (CurrFileIndex),
                                              bd496_GetSecQueuePno  (CurrFileIndex));
        if (e_ok != m_TrError) return;

        TempInvQueue.bd495Free();
        --m_NumOpenQueues;
    }
    bd496_SetEntry (CurrFileIndex, TempInvRoot);
    ++m_NumOpenTrees;
}

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

void
cbd496_FileHandling::bd496CreateFile (
    tsp00_Int4       NewFileIndex,
    tsp00_PageNo    &PrimQueuePno,
    tsp00_PageNo    &SecQueuePno)
{
    ROUTINE_DBG_MEO00 ("bd496CreateFile");


    tgg00_FileId    DummyId;

    bd496_InitFile (NewFileIndex, DummyId, PrimQueuePno, SecQueuePno);
}

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

void
cbd496_FileHandling::bd496CreateFile (
    tsp00_Int4      &NewFileIndex,
    tgg00_FileId    &QueueId,
    tsp00_PageNo    &PrimQueuePno,
    tsp00_PageNo    &SecQueuePno)
{
    ROUTINE_DBG_MEO00 ("bd496CreateFile");


    bool		bFound = false;
    /* */
    tsp00_Int4	CurrFileIndex = m_FirstSortFileIndex;
    tsp00_Int4	LastFileIndex = m_LastSortFileIndex;

    while ((e_ok == m_TrError) && (CurrFileIndex <= LastFileIndex) && (!bFound))
    {
        if (bd496_EntryUsed (CurrFileIndex))
        {
            ++CurrFileIndex;
        }
        else
        {
            bd496_InitFile (CurrFileIndex, QueueId, PrimQueuePno, SecQueuePno);
            NewFileIndex = CurrFileIndex;
            bFound       = true;
        }
    }
    if (!bFound)
        m_TrError = e_file_limit; // All files in use -> Merge files and try again

#   if COMPILEMODE_MEO00 >= SLOW_MEO00 
    t01basis_error (bd_idx_create, "CreateFile  ", m_TrError);
#   endif
}

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

void
cbd496_FileHandling::bd496DeleteAllFiles ()
{
    ROUTINE_DBG_MEO00 ("bd496DeleteAllFiles");


    for (tsp00_Int4 CurrFileIndex = m_FirstFileIndex;
            CurrFileIndex <= m_LastFileIndex;
            ++CurrFileIndex)
        bd496DeleteFile (CurrFileIndex);
}

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

void
cbd496_FileHandling::bd496DeleteFile (tsp00_Int4   CurrFileIndex)
{
    ROUTINE_DBG_MEO00 ("bd496DeleteFile");


    tgg00_BasisError    AuxError = m_TrError;

    m_TrError = e_ok;

    if (bd496_EntryUsed (CurrFileIndex))
    {
        if (bd496_IsQueueEntry (CurrFileIndex))
        {
            // Drop DoubleWriteQueue
            tbd_current_tree QueueCurrent = bd496_InitTempCurrent (m_Trans);

            cbd495_DoubleWriteQueue TempInvQueue (QueueCurrent,
                                                  bd496_GetPrimQueuePno (CurrFileIndex),
                                                  bd496_GetSecQueuePno  (CurrFileIndex));
            if (e_ok != m_TrError) return;

            TempInvQueue.bd495Free();
        }
        else
        {
            // Drop TempInvTree
            tgg00_FileId    TempFileId = bd495InitTempFileId (bd496_GetTempInvRoot (CurrFileIndex));

            cbd495_TempInvTree TempInvTree (m_Trans, TempFileId, DROP_TREE_BD495);
            if (e_ok != m_TrError) return;

            TempInvTree.bd495Free();
        }
        bd496_ClearEntry (CurrFileIndex);
    }
    m_TrError = AuxError;
}

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

void
cbd496_FileHandling::bd496MergeFiles (
    tsp00_Bool			        bNoMorePrimData,
    const cbd497_StackDescInfo  &stackDescInfo,
    tgg00_IndexCounter	        &IndexCounter)
{
    ROUTINE_DBG_MEO00 ("bd496MergeFiles");


    bool        bLastMergeStep = false;
    tsp00_Int4  LastFileIndex;

    m_TrError = e_ok; // m_TrError could be only e_file_limit

    if (bNoMorePrimData)
    {
        g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03, csp3_n_index,
                  MERGE_MSG_7_BD496, IndexCounter.idc_prim_leafnodes);

        g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03, csp3_n_index,
                  MERGE_MSG_8_BD496, IndexCounter.idc_prim_keycount);
    }
    else
    {
        g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03, csp3_n_index,
                  MERGE_MSG_1_BD496, IndexCounter.idc_prim_leafnodes);

        g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03, csp3_n_index,
                  MERGE_MSG_2_BD496, IndexCounter.idc_prim_keycount);
    }

    do
    {
        ++m_MergeCount;

        if (bNoMorePrimData)
        {
            // The decision if the last Funnel Merge step could be started depends
            // on two factors. First: all primary data is scaned. Second: all object
            // handler for the FileList could be instantiated in the given memory.

            bLastMergeStep = cbd494_Funnel::bd494MemoryAvailable (m_NumFunnelItems,
                             m_NumOpenQueues, m_NumOpenTrees);
        }

        if (bLastMergeStep)
            g01optextmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03,
                          csp3_n_index, MERGE_MSG_3_BD496);
        else
            g01optextmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03,
                          csp3_n_index, MERGE_MSG_4_BD496);

        if (bLastMergeStep)
        {
            LastFileIndex = m_LastFileIndex;
        }
        else
        {
            // Determine TargetFileIndex

            // FileList has an overflow therefore a funnel merge step is
            // necessary to obtain new free FileList entries.

            if (m_TargetFileIndex < m_LastResultFileIndex)
            {
                ++m_TargetFileIndex;
                LastFileIndex = m_LastSortFileIndex;
            }
            else
            {
                // All result files are full therefore the next funnel
                // step has to merge both ordinary sort files and result
                // files into an overflow result file.

                LastFileIndex = m_LastFileIndex;

                if (bd496_EntryUsed (m_FirstOverflowFileIndex))
                    m_TargetFileIndex = m_LastOverflowFileIndex;
                else
                    m_TargetFileIndex = m_FirstOverflowFileIndex;
            }
        }
        if (e_ok == m_TrError)
            bd496_Merge (bLastMergeStep, LastFileIndex, stackDescInfo);

        if (
            (e_ok == m_TrError) &&
            (
                (m_TargetFileIndex == m_FirstOverflowFileIndex) ||
                (m_TargetFileIndex == m_LastOverflowFileIndex )
            )
        )
            m_TargetFileIndex = m_LastSortFileIndex; // incremented before first use

        if (bLastMergeStep)
            g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03,
                      csp3_n_index, MERGE_MSG_5_BD496, m_MergeCount);
        else
            g01opmsg (sp3p_console, sp3m_info, BD496_MERGE_SP03,
                      csp3_n_index, MERGE_MSG_6_BD496, m_MergeCount);
    }
    while ((bNoMorePrimData) && (!bLastMergeStep) && (e_ok == m_TrError));

    if (e_sysbuffer_overflow == m_TrError)
        g01optextmsg (sp3p_knldiag, sp3m_warning, BD496_OVERFLOW_1_SP03,
                      csp3_n_index, OVERFLOW_MSG_1_BD496);

    if (e_no_more_memory == m_TrError)
        g01optextmsg (sp3p_knldiag, sp3m_warning, BD496_MEMORY_1_SP03,
                      csp3_n_index, MEMORY_MSG_1_BD496);
}

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

