---

CdbBdbSPartitionsLayoutP.cc

Go to the documentation of this file.

// File and Version Information:
//      $Id: CdbBdbSPartitionsLayoutP.cc,v 1.10 2004/08/06 05:54:25 bartoldu Exp $

/// The implementation file for the CdbBdbSPartitionsLayoutP class
/**
  * @see CdbBdbSPartitionsLayoutP
  */

#include "BaBar/BaBar.hh"

#include "CdbBdbShared/CdbBdbSPartitionsLayoutP.hh"

#include "CdbBase/CdbEnvironment.hh"

#include <iostream>
#include <assert.h>
#include <iomanip>

#include <vector>
#include <list>
using std::cout;
using std::endl;
using std::ostream;
using std::setfill;
using std::setw;

namespace {

/////////////////////////////////
// CdbBdbSPartitionIntervalItr //
/////////////////////////////////

class CdbBdbSPartitionIntervalItr : public CdbItr< CdbBdbSPartitionInterval >::InterfaceType {

private:

    typedef CdbBdbSPartitionCell     Cell;
    typedef CdbBdbSPartitionInterval Interval;

    typedef std::vector<Cell>   Cells;
    typedef std::list<Interval> Intervals;

private:

    CdbBdbSPartitionIntervalItr( );

    CdbBdbSPartitionIntervalItr& operator=( const CdbBdbSPartitionIntervalItr& theItr );

protected:

    CdbBdbSPartitionIntervalItr( const CdbBdbSPartitionIntervalItr& theItr ) :
        _cells              (theItr._cells),
        _begin              (theItr._begin),
        _end                (theItr._end),
        _isValid            (theItr._isValid),
        _hasEverBeenAdvanced(theItr._hasEverBeenAdvanced),
        _valueIsBuilt       (theItr._valueIsBuilt),
        _values             (theItr._values),
        _currentValueItr    (theItr._currentValueItr),
        _currentValue       (theItr._currentValue)
    { }

public:

    CdbBdbSPartitionIntervalItr( const Cells&   theCells,
                                 const BdbTime& theBeginValidityTime,
                                 const BdbTime& theEndValidityTime ) :
        _cells              (theCells),
        _begin              (theBeginValidityTime),
        _end                (theEndValidityTime),
        _isValid            (false),
        _hasEverBeenAdvanced(false),
        _valueIsBuilt       (false)
    {
        assert( theBeginValidityTime < theEndValidityTime );
    }

    virtual ~CdbBdbSPartitionIntervalItr( )
    { }

    virtual CdbStatus reset( )
    {
        _isValid             = false;
        _hasEverBeenAdvanced = false;

        return CdbStatus::Success;
    }

    void buildNextNode( Cells::iterator     theInCurrent,
                        Intervals::iterator theOutCenter,
                        const BdbTime&      theBeginTime,
                        const BdbTime&      theEndTime )
    {
        const char* debugStr = "CdbBdbSCondition::CdbBdbSPartitionIntervalItr::buildNextNode()";

        for( Cells::iterator i = theInCurrent; i != _cells.end( ); ++i ) {

/**************************************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    (*i)            = " << (*i) << endl
         << "    (*theOutCenter) = " << (*theOutCenter) << endl
         << "}" << endl;
/**************************************************************/

          // Get next cell as defined by the input index and see which
          // of the following scenarious we're at:
          //
          //      theBeginTime  theEndTime
          //           :            :
          //        [--)            :       skip
          //           :            [--)    skip
          //           [--)         :       (1) split into 2, (optionally) truncate on the left, then propage down on the right
          //           [------------)       (2) split into 1, (optionally) truncate on the left and on the right, and finish
          //           :  [------)  :       (3) split into 3, then propagate down on the left and then on the right
          //           :         [--)       (4) split into 2, (optionally) truncate on the right, then propagate down on the left
          //
          // NOTE: If this is a very first call to the method then the we're modifying
          //       the collection of values is slightly different than for the non-empty
          //       vector.
          //
          // IMPLEMENTATION NOTE: The following recursive algorithm is based on
          //                      an assumption that we're using STD <list>
          //                      to implement the collection values. Otherwise
          //                      the tricks with iterators may not work since positions
          //                      of elements as defined by iterators may change as we'll
          //                      be modifying the collection.

            Cell c = (*i);

          // Check if we need to skip

            if( c.cell.endValidity   <= theBeginTime ) continue;
            if( c.cell.beginValidity >= theEndTime   ) continue;

          // Truncate if needed

            BdbTime begin = c.cell.beginValidity;
            if( begin < theBeginTime )
                begin = theBeginTime;

            BdbTime end = c.cell.endValidity;
            if( end > theEndTime )
                end = theEndTime;

/******************************************************************************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    begin        = " << setw( 10 ) << setfill( '0' ) << begin.getGmtSec( )        << endl
         << "    end          = " << setw( 10 ) << setfill( '0' ) << end.getGmtSec( )          << endl
         << "    theBeginTime = " << setw( 10 ) << setfill( '0' ) << theBeginTime.getGmtSec( ) << endl
         << "    theEndTime   = " << setw( 10 ) << setfill( '0' ) << theEndTime.getGmtSec( )   << endl
         << "}" << endl;
/******************************************************************************************************/

            if(( begin == theBeginTime ) &&
               ( end   <  theEndTime )) {

/******************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    -1-" << endl
         << "}" << endl;
/******************************************/

              // -1-

                Interval left( c.id, begin, end );
                Interval right(      end,   theEndTime );

                _values.insert( theOutCenter, left );

                Intervals::iterator outRight = theOutCenter;
                if( _values.end( ) == theOutCenter ) {
                    outRight = _values.insert( theOutCenter, right );
                } else {
                    (*theOutCenter) = right;
                }
                Cells::iterator nextCell = i;
                ++nextCell;
                buildNextNode( nextCell, outRight, end, theEndTime );   // propagate RIGHT

                return;

            } else if(( begin == theBeginTime ) &&
                      ( end   == theEndTime )) {

/******************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    -2-" << endl
         << "}" << endl;
/******************************************/

              // -2-

                Interval center( c.id, begin, end );

                if( _values.end( ) == theOutCenter )
                    _values.insert( theOutCenter, center );
                else
                    (*theOutCenter) = center;

                return;

            } else if(( begin > theBeginTime ) &&
                      ( end   < theEndTime )) {

/******************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    -3-" << endl
         << "}" << endl;
/******************************************/

              // -3-

                Interval left  (       theBeginTime, begin );
                Interval center( c.id, begin,        end );
                Interval right (       end,          theEndTime );

                Intervals::iterator outLeft = _values.insert( theOutCenter, left );
                _values.insert( theOutCenter, center );
                Intervals::iterator outRight = theOutCenter;
                if( _values.end( ) == theOutCenter ) {
                    outRight = _values.insert( theOutCenter, right );
                } else {
                    (*theOutCenter) = right;
                }
                Cells::iterator nextCell = i;
                ++nextCell;
                buildNextNode( nextCell, outLeft,  theBeginTime, begin );      // propagate LEFT
                buildNextNode( nextCell, outRight, end,          theEndTime ); // propagate RIGHT

                return;

            } else if(( begin >  theBeginTime ) &&
                      ( end   >= theEndTime )) {

/******************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    -4-" << endl
         << "}" << endl;
/******************************************/

              // -4-

                Interval left(        theBeginTime, begin );
                Interval right( c.id, begin,        end  );

                Intervals::iterator outLeft = _values.insert( theOutCenter, left );
                if( _values.end( ) == theOutCenter ) {
                    _values.insert( theOutCenter, right );
                } else {
                    (*theOutCenter) = right;
                }
                Cells::iterator nextCell = i;
                ++nextCell;
                buildNextNode( nextCell, outLeft, theBeginTime, begin );   // propagate LEFT

                return;

            } else {
                assert( 0 );
            }
        }

      // If we got in here then we need to fiil up the whole interval with an invalid
      // element unless the previous call to this function has already done it.

        if( _values.end( ) == theOutCenter ) {

/******************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "    -default-" << endl
         << "}" << endl;
/******************************************/

            _values.push_back( Interval( theBeginTime,
                                         theEndTime ));
        }
    }

    void buildVectorOfValues( )
    {
        if( !_valueIsBuilt ) {
            _valueIsBuilt = true;

            buildNextNode( _cells.begin( ),
                           _values.end( ),
                           _begin,
                           _end );
        }
        _currentValueItr = _values.begin( );
    }

    bool tryNext( )
    {
        bool result = _currentValueItr != _values.end( );
        if( result ) {
            _currentValue = *(_currentValueItr++);
        }
        return result;
    }

    virtual bool next( )
    {
        if( _hasEverBeenAdvanced ) {

            if( _isValid ) {
                _isValid = tryNext( );
            } else {

              // stay in the BEYOND_LAST_ELEMENT state
                ;
            }

        } else {

            buildVectorOfValues( );

            _hasEverBeenAdvanced = true;
            _isValid             = tryNext( );
        }
        return _isValid;
    }

    virtual ValueType value( )
    {
        if( !_isValid ) {
            assert( 0 );
        }
        return _currentValue;
    }

    virtual bool isValid( )
    {
        return _isValid;
    }

    virtual InterfaceType* clone( ) const
    {
        return new CdbBdbSPartitionIntervalItr( *this );
    }

private:

    Cells _cells;

    BdbTime _begin;
    BdbTime _end;

    bool _isValid;
    bool _hasEverBeenAdvanced;

  // The output collection is built only once during the lifetime
  // of the iterator.

    bool _valueIsBuilt;

    Intervals           _values;
    Intervals::iterator _currentValueItr;
    Interval            _currentValue;
};

};

//////////////////////////
// CdbBdbSPartitionCell //
//////////////////////////

CdbBdbSPartitionCell::CdbBdbSPartitionCell( ) :
    cell(CdbBdbSCell( )),
    id(0),
    partition(0)
{ }

CdbBdbSPartitionCell::CdbBdbSPartitionCell( const CdbBdbSCell&                     theCell,
                                            d_UShort                               theId,
                                            const BdbRef( CdbBdbSPartitionBaseP )& theRef ) :
    cell(theCell),
    id(theId),
    partition(theRef)
{ }

CdbBdbSPartitionCell::CdbBdbSPartitionCell( const CdbBdbSPartitionCell& theOther ) :
    cell(theOther.cell),
    id(theOther.id),
    partition(theOther.partition)
{ }

CdbBdbSPartitionCell::~CdbBdbSPartitionCell( )
{ }

CdbBdbSPartitionCell&
CdbBdbSPartitionCell::operator=( const CdbBdbSPartitionCell& theOther )
{
    const_cast< CdbBdbSCell& >                    ( cell      ) = theOther.cell;
    const_cast< d_UShort& >                       ( id        ) = theOther.id;
    const_cast< BdbRef( CdbBdbSPartitionBaseP )& >( partition ) = theOther.partition;
    return *this;
}

void
CdbBdbSPartitionCell::dump( ostream& o ) const
{
    o << cell << endl
      << "ID        : " << id << endl
      << "PARTITION : " << partition.sprint( ) << endl;
}

//////////////////////////////
// CdbBdbSPartitionInterval //
//////////////////////////////

CdbBdbSPartitionInterval::CdbBdbSPartitionInterval( const BdbTime& theBeginTime,
                                                    const BdbTime& theEndTime ) :
    isEmpty(true),
    begin  (theBeginTime),
    end    (theEndTime)
{ }

CdbBdbSPartitionInterval::CdbBdbSPartitionInterval( d_UShort       theId,
                                                    const BdbTime& theBeginTime,
                                                    const BdbTime& theEndTime ) :
    isEmpty(false),
    id     (theId),
    begin  (theBeginTime),
    end    (theEndTime)
{ }

CdbBdbSPartitionInterval::CdbBdbSPartitionInterval( const CdbBdbSPartitionInterval& theOther ) :
    isEmpty(theOther.isEmpty),
    id     (theOther.id),
    begin  (theOther.begin),
    end    (theOther.end)
{ }

CdbBdbSPartitionInterval::~CdbBdbSPartitionInterval( )
{ }

CdbBdbSPartitionInterval&
CdbBdbSPartitionInterval::operator=( const CdbBdbSPartitionInterval& theOther )
{
    if( this != &theOther ) {
        isEmpty = theOther.isEmpty;
        id      = theOther.id;
        begin   = theOther.begin;
        end     = theOther.end;
    }
    return *this;
}

void
CdbBdbSPartitionInterval::dump( ostream& o ) const
{
    o << "IS_EMPTY=" << ( isEmpty ? "Y" : "N" )
      << "  "
      << "BEGIN="
      << setw( 10 ) << setfill( '0' ) << begin.getGmtSec( )
      << "."
      << setw( 10 ) << setfill( '0' ) << begin.getGmtNsec( ) 
      << "  "
      << "END="
      << setw( 10 ) << setfill( '0' ) << end.getGmtSec( )
      << "."
      << setw( 10 ) << setfill( '0' ) << end.getGmtNsec( );

    if( !isEmpty ) {
      cout << "  "
           << "ID=" << id;
    }
}

//////////////////////////////
// CdbBdbSPartitionsLayoutP //
//////////////////////////////

CdbBdbSPartitionsLayoutP::CdbBdbSPartitionsLayoutP( ) :
    _collectionP(0),
    _trivialIndexP(0),
    _complexIndexP(0)
{
  // Note, we're not going to use names of partition objects in
  // the collection.

    bool useIdFlag   = true;
    bool useNameFlag = false;

    _collectionP = new( ooThis( )) CdbBdbSCollectionP( useIdFlag,
                                                       useNameFlag );

  // This is a trivial index for the persistent partition object. It's meant in here
  // to speed up lookup operations to avoid bringing the whole persistent partition
  // objects into the memory.
  //
  // Another reason to use this index is that actual partition objects may reside
  // in different containers. Therefore by using an extra inidex we may avoid locking
  // unnessesary containers.
  //
  // The index does not contain any elements by default.

    _trivialIndexP = new( ooThis( )) CdbBdbSPagedVarrayP< CdbBdbSPartitionCell >( );
}

CdbBdbSPartitionsLayoutP::~CdbBdbSPartitionsLayoutP( ) 
{
  // Delete all the partition objects first
  // Then delete the collection itself.

    d_ULong num = _trivialIndexP->size( );
    for( d_ULong i = 0; i < num; ++i ) {

        CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );
        BdbDelete(pCell.partition);
    }
    BdbDelete(_collectionP);

  // ...and the trivial index

    BdbDelete(_trivialIndexP);

  // And never attempt to delete the optional index. This has to be done
  // by an application knowing about the actual persistent type (and it virtual
  // tables) of the corresponding index.

    if( ! BdbIsNull(_complexIndexP)) {
        cout << "CdbBdbSPartitionsLayoutP::~CdbBdbSPartitionsLayoutP( ) - WARNING!" << endl
             << "    The destructor has detected non-zero optional COMPLEX index. This index" << endl
             << "    can't be deleted by the current destructor. Use more recent implementation" << endl
             << "    for this kind of operation." << endl
             << "        INDEX OID: " << _complexIndexP.sprint( ) << endl;
    }
}

d_ULong
CdbBdbSPartitionsLayoutP::numPartitions( ) const
{
    return _collectionP->elements( );
}

CdbStatus
CdbBdbSPartitionsLayoutP::insert( const BdbRef( CdbBdbSPartitionBaseP )& theRef )
{
    const char* errorStr = "CdbBdbSPartitionsLayoutP::insert() - ERROR.";

    ooUpdate( );

    CdbStatus result = CdbStatus::Error;
    do {

      // Verify parameters.

        if( BdbIsNull(theRef)) {
            cout << errorStr << endl
                 << "    0-0-0-0 persistent reference passed where a pointer onto" << endl
                 << "    a partition object was expected." << endl;
            break;
        }

      // Check if there is no such partition in the layout.

        d_ULong id = theRef->id( );

        BdbRef( CdbBdbSPartitionBaseP ) ref;

        if( CdbStatus::Success == _collectionP->find( id,
                                                      ref )) {
            cout << errorStr << endl
                 << "    The Layout already has a registered partition with the same identifier." << endl
                 << "        ID: " << id << endl
                 << "        PASSED     PARTITION OID: " << theRef.sprint( ) << endl
                 << "        REGISTERED PARTITION OID: " << ref.sprint( ) << endl;
            break;
        }

      // Check if the cell of passed partition does not overlap with none of
      // the known partitions.

        CdbBdbSCell cell = theRef->cell( );

        bool overlap = false;
        {
            d_ULong num = _trivialIndexP->size( );
            for( d_ULong i = 0; i < num; ++i ) {

                CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

                if( pCell.cell.overlap( cell )) {
                    overlap = true;
                    break;
                }
            }
        }
        if( overlap ) {
            cout << errorStr << endl
                 << "    One of the registered with the Layout partitions overlaps with" << endl
                 << "    with the one passed at the 2-dimension space of the VALIDITY" << endl
                 << "    and INSERTION timelines." << endl;
            break;
        }

      // Add new partition to the collection and then - to the trivial index.
      //
      // NOTE: A failure to add into a trivial index will be followed by
      //       a proper "removal" operation to leave the Layout in the self-consistent
      //       state.

        if( CdbStatus::Success != _collectionP->insert( theRef )) {
            break;
        }
        d_ULong num = _trivialIndexP->size( );
        if( BdbcSuccess != _trivialIndexP->resize( num + 1 )) {

            cout << errorStr << endl
                 << "    Failed to resize the trivial index to make a room for a new partition." << endl
                 << "        CURRENT SIZE: " << num << endl;

            _collectionP->remove( id );

            break;
        }
        _trivialIndexP->setElementAt( num,
                                      CdbBdbSPartitionCell( cell,
                                                            id,
                                                            theRef ));

      // Done.

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSPartitionsLayoutP::update( const BdbRef( CdbBdbSPartitionBaseP )& theNewRef )
{
    const char* errorStr = "CdbBdbSPartitionsLayoutP::update() - ERROR.";

    ooUpdate( );

    CdbStatus result = CdbStatus::NotFound;
    do {

      // Verify parameters.

        if( BdbIsNull(theNewRef)) {
            cout << errorStr << endl
                 << "    0-0-0-0 persistent reference passed where a pointer onto" << endl
                 << "    a partition object was expected." << endl;
            break;
        }

      // Use trivial index to find the partition cell corresponding to
      // the old partition

        bool    updateIndexOnly = false;    // If the passed objecxt is known
        d_ULong oldCellIdx      = 0;        // Will need to replace with updated cell
        {
            const d_UShort pId = theNewRef->id( );

            CdbBdbSPartitionCell pCell;

            bool found = false;

            d_ULong num = _trivialIndexP->size( );
            for( d_ULong i = 0; i < num; ++i ) {

                pCell = _trivialIndexP->elementAt( i );
                if( pCell.id == pId ) {

                    oldCellIdx = i;

                    found = true;
                    break;
                }
            }
            if( !found) break;

          // Check if the cell of passed partition is the same (except its upper border) 
          // than the one known to the local layout. The upper border of the passed
          // partition is allowed to be "closed".

            CdbBdbSCell newCell = theNewRef->cell( );

            if(( newCell.beginValidity  != pCell.cell.beginValidity )  ||
               ( newCell.endValidity    != pCell.cell.endValidity )    ||
               ( newCell.beginInsertion != pCell.cell.beginInsertion )) {

                cout << errorStr << endl
                     << "    The validity/insertion cell occupied by the new partition passed" << endl
                     << "    to the procedure does not match the old one." << endl
                     << "        NEW ->" << endl
                     << newCell << endl
                     << "        OLD ->" << endl
                     << pCell.cell << endl;

                result = CdbStatus::Error;

                break;
            }

          // Also check if the found partition objects has the same OID.
          // If this is the case the we only need to update indexes.

            if( theNewRef == pCell.partition ) {
                updateIndexOnly = true;
            }
        }

      // Replace old cell in the trivial index.

        _trivialIndexP->setElementAt( oldCellIdx,
                                      CdbBdbSPartitionCell( theNewRef->cell( ),
                                                            theNewRef->id( ),
                                                            theNewRef ));

      // Replace the partition object in the current Layout if needed.

        if( ! updateIndexOnly ) {

          // The following method will also physically destroy the partition object
          // corresponding to the passed identifier.

            if( CdbStatus::Success != _collectionP->remove( theNewRef->id( ))) {

                cout << errorStr << endl
                     << "    Failed to remove old parition object out of the collection." << endl
                     << "        PARTITION ID: " << theNewRef->id( ) << endl;

                result = CdbStatus::Error;

                break;
            }

          // And insert

            if( CdbStatus::Success != _collectionP->insert( theNewRef )) {

                cout << errorStr << endl
                     << "    Failed to insert new parition object into the collection." << endl
                     << "        PARTITION ID: " << theNewRef->id( ) << endl;

                result = CdbStatus::Error;

                break;
            }
        }

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSPartitionsLayoutP::find( d_UShort                         theId,
                                BdbRef( CdbBdbSPartitionBaseP )& theRef ) const
{
    return _collectionP->find( theId,
                               theRef );
}

CdbStatus
CdbBdbSPartitionsLayoutP::find( const BdbTime&                   theValidityTime,
                                const BdbTime&                   theInsertionTime,
                                BdbRef( CdbBdbSPartitionBaseP )& theRef ) const
{
    CdbStatus result = CdbStatus::NotFound;
    {
      // Use trivial index

        d_ULong num = _trivialIndexP->size( );
        for( d_ULong i = 0; i < num; ++i ) {

            CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

            if( pCell.cell.in( theValidityTime,
                               theInsertionTime )) {

                theRef = pCell.partition;

                result = CdbStatus::Success;
                break;
            }
        }
    }
    return result;
}

CdbStatus
CdbBdbSPartitionsLayoutP::find( const BdbTime&                   theValidityTime,
                                BdbRef( CdbBdbSPartitionBaseP )& theRef ) const
{
    CdbStatus result = CdbStatus::NotFound;
    {
      // Use trivial index to find an "open" partition

        d_ULong num = _trivialIndexP->size( );
        if( num > 0 ) {

          // NOTE: That we're going from the most recently registered partition
          //       stored at the end of the list down to the one at a position 0.
          //       This is done since we expect that "open" partitions are located
          //       at the end of the index.

            int first = num - 1;
            assert( first >= 0 );       // We're are not expecting more than 10^31-1
                                        // partitions.

            for( int i = first; i >= 0; --i ) {

                CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

                if( pCell.cell.inValidityRange( theValidityTime ) &&
                    ( BdbTime::plusInfinity == pCell.cell.endInsertion )) {

                    theRef = pCell.partition;

                    result = CdbStatus::Success;
                    break;
                }
            }
        }
    }
    return result;
}
CdbStatus
CdbBdbSPartitionsLayoutP::topmost( const BdbTime&                   theValidityTime,
                                   BdbRef( CdbBdbSPartitionBaseP )& theRef ) const
{
    CdbStatus result = CdbStatus::NotFound;
    {
      // Use trivial index to find a "topmost" partition

        d_ULong num = _trivialIndexP->size( );
        if( num > 0 ) {

          // NOTE: That we're going from the most recently registered partition
          //       stored at the end of the list down to the one at a position 0.
          //       This is done since we expect that "topmost" partitions are located
          //       at the end of the index.

            int first = num - 1;
            assert( first >= 0 );       // We're are not expecting more than 10^31-1
                                        // partitions.

            for( int i = first; i >= 0; --i ) {

                CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

                if( pCell.cell.inValidityRange( theValidityTime )) {

                    theRef = pCell.partition;

                    result = CdbStatus::Success;
                    break;
                }
            }
        }
    }
    return result;
}

CdbStatus
CdbBdbSPartitionsLayoutP::next( const BdbTime&                   theValidityTime,
                                BdbRef( CdbBdbSPartitionBaseP )& theRef ) const
{
    assert( !BdbIsNull(theRef));

    CdbStatus result = CdbStatus::NotFound;
    {
      // Use trivial index to find a "next" partition

      // NOTE: That we're going from the most recently registered partition
      //       stored at the end of the list down to the one at a position 0.
      //       This is done since we expect that "next" partitions are located
      //       at the end of the index.

        d_ULong num = _trivialIndexP->size( );
        if( num > 0 ) {

          // Find the specified partition first. The actual search will
          // go down from that partition.

            bool    idxIsFound = false;
            d_ULong idx = 0;
            {
                int first = num - 1;
                assert( first >=0 );      // We're are not expecting more than 10^31-1
                                            // partitions.

                for( int i = first; i >= 0; --i ) {

                    CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

                    if( theRef == pCell.partition ) {
                        idxIsFound = true;
                        idx        = i;
                        break;
                    }
                }
            }

          // Proceed down the index till a partition containing the
          // the specified validity time is met.

            if( idxIsFound && ( 0 != idx )) {

                int first = idx - 1;
                assert( first >= 0 );       // We're are not expecting more than 10^31-1
                                            // partitions.

                for( int i = first; i >= 0; --i ) {

                    CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

                    if( pCell.cell.inValidityRange( theValidityTime )) {

                        theRef = pCell.partition;

                        result = CdbStatus::Success;
                        break;
                    }
                }
            }
        }
    }
    return result;
}

CdbStatus
CdbBdbSPartitionsLayoutP::close( d_UShort theId )
{
    const char* fatalStr = "CdbBdbSPartitionsLayoutP::close() - FATAL ERROR.";

    ooUpdate( );

  // Find this partition in a collection first

    BdbRef( CdbBdbSPartitionBaseP ) ref;

    CdbStatus result = find( theId,
                             ref );
    if( CdbStatus::Success != result ) {
        return result;
    }

  // Find this partition in the trivial index then

    d_ULong foundIndex = 0;
    bool    indexWasFound = false;
    {
        d_ULong num = _trivialIndexP->size( );
        for( d_ULong i = 0; i < num; ++i ) {

            CdbBdbSPartitionCell pCell = _trivialIndexP->elementAt( i );

            if( theId == pCell.id ) {

                if( ref != pCell.partition ) {
                    cout << fatalStr << endl
                         << "    The trivial index is not compatible with the collection of partitions." << endl;
                    assert( 0 );

                } else {
                    foundIndex = i;
                    indexWasFound = true;
                }
                break;
            }
        }
    }
    if( ! indexWasFound ) {

        cout << fatalStr << endl
             << "    The trivial index is not compatible with the collection of partitions." << endl;

        assert( 0 );

        return CdbStatus::Error;
    }

  // And finally update the index

     _trivialIndexP->setElementAt( foundIndex,
                                   CdbBdbSPartitionCell( ref->cell( ),
                                                         theId,
                                                         ref ));
     return CdbStatus::Success;
}

CdbItr< d_UShort >
CdbBdbSPartitionsLayoutP::iterator( ) const
{
    return _collectionP->iterator_identifiers( );
}

CdbItr< CdbBdbSPartitionInterval >
CdbBdbSPartitionsLayoutP::iterator( const BdbTime& theInsertionTime,
                                    const BdbTime& theBeginValidityTime,
                                    const BdbTime& theEndValidityTime ) const
{
    const char* debugStr = "CdbBdbSPartitionsLayoutP::iterator()";

    assert( !BdbIsNull(_trivialIndexP));

  // Build a transient vector of cells and pass it down to the iterator
  //
  // NOTE: This vector will only contains a subset of
  //       elements satisfying the selection criteria
  //       explained in the description of this method
  //
  // NOTE: The vector will be sorted at reverse order
  //
  // @see CdbBdbSPartitionsLayoutP::iterator()

    const CdbBdbSCell iteratorCell( theBeginValidityTime,
                                    theEndValidityTime,
                                    BdbTime::minusInfinity, // The bottom of the iterator's limit
                                    theInsertionTime );

/*****************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "iteratorCell ->" << endl
         << iteratorCell << endl
         << "}" << endl;
/*****************************************/

    std::vector<CdbBdbSPartitionCell> cells;

    d_ULong num = _trivialIndexP->size( );
    for( d_ULong i = 0; i < num; ++i ) {
        CdbBdbSPartitionCell pc = _trivialIndexP->elementAt( i );

/*****************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "pc.cell ->" << endl
         << pc.cell << endl
         << "}" << endl;
/*****************************************/

        if( iteratorCell.overlap( pc.cell )) {
/*****************************************/
if( CdbEnvironment::getDebugMode( ) != 0 )
    cout << debugStr << endl
         << "{" << endl
         << "'pc.cell' overlaps with 'iteratorCell'" << endl
         << "}" << endl;
/*****************************************/
            cells.insert( cells.begin( ),
                          pc );
        }
    }

  // ATTENTION: This is not a memory leak - the created object
  //            will be destroyed by the iterator.

    return CdbItr< CdbBdbSPartitionInterval >( new CdbBdbSPartitionIntervalItr( cells,
                                                                                theBeginValidityTime,
                                                                                theEndValidityTime ));
}

void
CdbBdbSPartitionsLayoutP::dump( ostream& o ) const
{
    o << "TRIVIAL INDEX: " << _trivialIndexP.sprint( ) << endl
      << "COMPLEX INDEX: " << _complexIndexP.sprint( ) << endl
      << endl;

    _collectionP->dump( o );
}

/////////////////
// End Of File //
/////////////////

---