---

CdbBdbSPartitionP.cc

Go to the documentation of this file.

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

/// Implementation file for the CdbBdbSPartitionP class
/**
  * @see CdbBdbSPartitionP
  */

#include "BaBar/BaBar.hh"

#include "CdbBdbShared/CdbBdbSPartitionP.hh"
#include "CdbBdbShared/CdbBdbSRegistryP.hh"
#include "CdbBdbShared/CdbBdbSClusterP.hh"
#include "CdbBdbShared/CdbBdbSConditionP.hh"
#include "CdbBdbShared/CdbBdbSUtils.hh"
#include "CdbBdbShared/CdbBdbSMetaDataP.hh"
#include "BdbClustering/BdbCondIndexClusteringHint.hh"

extern "C" {
#include <stdio.h>
using std::cout;
using std::endl;
using std::ostream;
};

CdbBdbSPartitionP::CdbBdbSPartitionP( d_UShort                                 theId,
                                      const char*                              theDescription,
                                      const BdbTime&                           theCreationTime,
                                      const CdbBdbSCell&                       theCell,
                                      d_UShort                                 theOriginId,
                                      const BdbRef(CdbBdbSPartitionListenerP)& theListenerRef ) :
    CdbBdbSPartitionBaseP( theId,
                           theDescription,
                           theCreationTime,
                           theCell ),
    _listenerRef(theListenerRef),
    _modified(theCreationTime),
    _isInstantiated(d_False),
    _originId(theOriginId)
{ }

CdbBdbSPartitionP::~CdbBdbSPartitionP( ) 
{ }

d_UShort
CdbBdbSPartitionP::originId( ) const
{
    return _originId;
}

bool
CdbBdbSPartitionP::isInstantiated( ) const
{
    return _isInstantiated;
}

BdbTime
CdbBdbSPartitionP::modified( ) const
{
    return _modified;
}

CdbStatus
CdbBdbSPartitionP::update( const BdbTime& theModificationTime )
{
    const char* errorStr = "CdbBdbSPartitionP::update() - ERROR.";

    ooUpdate( );

    CdbStatus result = CdbStatus::Error;
    do {

      // Verify parameters

        if(( BdbTime::minusInfinity == theModificationTime ) || ( BdbTime::plusInfinity == theModificationTime )) {
            cout << errorStr << endl
                 << "    Inappropriate value of the modification time passed. This value" << endl
                 << "    should not be neither -Infinity nor +Infinity." << endl;
            break;
        }

      // Check if the local origin matches this partition's origin.

        if( CdbStatus::Success != matchOrigin( )) break;

      // Check if this partition is still open for modifications

        if( isClosed( )) {
            cout << errorStr << endl
                 << "    The parition is closed. Any modifications are not allowed." << endl;
            break;
        }

      // Modify only if the passed time is newer.

        if( theModificationTime > _modified ) _modified = theModificationTime;

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSPartitionP::instantiate( bool reinstantiateFlag )
{
    const char* warningStr = "CdbBdbSPartitionP::instantiate() - WARNING.";
    const char* errorStr   = "CdbBdbSPartitionP::instantiate() - ERROR.";
    const char* fatalStr   = "CdbBdbSPartitionP::instantiate() - FATAL ERROR.";

    ooUpdate( );

    CdbStatus result = CdbStatus::Error;
    do {

      // Check if the local origin matches this partition's origin.

        if( CdbStatus::Success != matchOrigin( )) break;

      // Check if this partition is still open for modifications

        if( isClosed( )) {
            cout << errorStr << endl
                 << "    The parition is already closed. The requested operation" << endl
                 << "    is not possible." << endl;
            break;
        }

      // Check if this partition has not been instantiated yet unless
      // we're not in the "reinstantiate" mode.

        if( isInstantiated( ) && !reinstantiateFlag ) {
            cout << warningStr << endl
                 << "    The parition is already instantiated. Attempting to do it again" << endl
                 << "    may be an indication of problems in the application's logic." << endl;
            result = CdbStatus::Warning;
            break;
        }

      // Get to the clusters of partitionable conditions at the MASTER registry

        BdbHandle(CdbBdbSRegistryP) rH;
        if( CdbStatus::Success != CdbBdbSRegistryP::findMaster( rH )) {
            cout << errorStr << endl
                 << "    Failed to find master registry." << endl;
            break;
        }
        BdbRef(CdbBdbSClusterCollectionP) clusterCollRef = rH->pClusterCollection( );
        if( BdbIsNull(clusterCollRef)) {
            cout << fatalStr << endl
                 << "    Failed to get to the collection of clusters at master." << endl;
            assert( 0 );
            break;
        }

      // Find all the "partitionable" conditions at the MASTER and do databases and
      // containers instantion for them.
      //
      // NOTE: If we're in the "reinstantiate" mode then we're destroying each
      //       existing container before to recreate it.

        bool failed = false;
        {
            CdbBdbSClusterCollectionP::IteratorOfIdentifiers itr = clusterCollRef->iterator_identifiers( );
            while( itr.next( )) {

                d_UShort clusterId = itr.value( );

                BdbRef(CdbBdbSClusterP) clusterRef;
                if( CdbStatus::Success != clusterCollRef->find( clusterId,
                                                                clusterRef )) {
                    cout << fatalStr << endl
                         << "    A cluster with ID = " << clusterId << " was not found." << endl;
                    assert( 0 );

                    failed = true;
                    break;
                }

                CdbItr< BdbRef(CdbBdbSConditionP)> conditionItr;
                if( CdbStatus::Success != clusterRef->iterator( conditionItr )) {

                    cout << errorStr << endl
                         << "    Failed to set up an iterator of conditions - members of a cluster." << endl;

                    failed = true;
                    break;
                }
                while( conditionItr.next( )) {

                    BdbRef(CdbBdbSConditionP) condRef = conditionItr.value( );
                    d_UShort condId = condRef->id( );

                    char condIdStr[12];
                    sprintf( condIdStr, "%u", condId );

                  // Build database file name suffix according to specified mode of data
                  // placement for the condition.

                    std::string databaseSuffix;
                    {

                      // -------------------------------
                      // Step I: Placement specific part
                      // -------------------------------

                      // The current partition identifier is also a part of the database name

                        char partitionIdStr[12];
                        sprintf( partitionIdStr, "%u", ooThis( )->id( ));

                        databaseSuffix = databaseSuffix +
                                         std::string( "p" ) + std::string( partitionIdStr );

                      // The new condition will be placed at the most recent increment of the cluster.

                        d_UShort num = numIncrements( );
                        assert( num > 0 );
                        d_UShort clusterIncrement = num - 1;

                      // Build the database file suffix.

                        char clusterIdStr[12];
                        char clusterIncrementStr[12];

                        sprintf( clusterIdStr, "%u", clusterId );
                        sprintf( clusterIncrementStr, "%u", clusterIncrement );

                        databaseSuffix = databaseSuffix +
                                         std::string( "_c" ) + std::string( clusterIdStr ) +
                                         std::string( "_i" ) + std::string( clusterIncrementStr );

                      // ---------------------------------------
                      // Step II: DBID range name specific part.
                      // ---------------------------------------

                        std::string localRangeName;

                        CdbStatus result = CdbBdbSUtils::getLocalDbIdRange( localRangeName );
                        if( CdbStatus::Success != result ) {
                            cout << errorStr << endl
                                 << "Failed to obtain local DBID range name." << endl;
                            failed = true;
                            break;
                        }
                        databaseSuffix = databaseSuffix +
                                         std::string( "_" ) +
                                         localRangeName;
                    }

                  // Build containers names

                    std::string mdContainerName = std::string( "MetaData." ) + std::string( condIdStr );
                    std::string obContainerName = std::string( "Objects."  ) + std::string( condIdStr );

                  // Create containers for metadata and condition objects at specified
                  // database.

                    BdbCondIndexClusteringHint mdHint( "cdb",
                                                       mdContainerName.c_str( ),
                                                       databaseSuffix.c_str( ));

                    BdbHandle(BdbContObj) mdContH;
                    mdContH = mdHint.updatedHint( );
                    assert( ! BdbIsNull(mdContH));

                    if( !reinstantiateFlag ) {
                        assert( mdHint.newContainerCreated( ));
                    } else {

                      // Destroy the existing container

                        BdbDelete( mdContH );

                      // Create a new one
                      //
                      // NOTE: We're using a fresh instance of the Clustering Hint
                      //       to avoid problems with the context in the old one.

                        BdbCondIndexClusteringHint mdHint( "cdb",
                                                           mdContainerName.c_str( ),
                                                           databaseSuffix.c_str( ));
                        mdContH = mdHint.updatedHint( );

                        assert( ! BdbIsNull(mdContH));
                        assert( mdHint.newContainerCreated( ));
                    }

                    BdbCondIndexClusteringHint obHint( "cdb",
                                                       obContainerName.c_str( ),
                                                       databaseSuffix.c_str( ));

                    BdbHandle(BdbContObj) obContH;
                    obContH = obHint.updatedHint( );
                    assert( ! BdbIsNull(obContH));

                    if( !reinstantiateFlag ) {
                        assert( obHint.newContainerCreated( ));
                    } else {

                      // Destroy the existing container

                        BdbDelete( obContH );

                      // Create a new one
                      //
                      // NOTE: We're using a fresh instance of the Clustering Hint
                      //       to avoid problems with the context in the old one.

                        BdbCondIndexClusteringHint newObHint( "cdb",
                                                              obContainerName.c_str( ),
                                                              databaseSuffix.c_str( ));
                        obContH = newObHint.updatedHint( );

                        assert( ! BdbIsNull(obContH));
                        assert( newObHint.newContainerCreated( ));
                    }

                  // Create initial infrastructure in the newely created
                  // container for metadata.
                  //
                  // NOTE: The "size" of this meta-data object in the 2-dimension space
                  //       is exactly the same as the current partition has.

                    CdbBdbSCell myCell = cell( );

                    BdbHandle(CdbBdbSMetaDataP) mdH;
                    mdH = new( mdContH ) CdbBdbSMetaDataP( myCell.beginValidity,
                                                           myCell.endValidity,
                                                           myCell.beginInsertion );
                    assert( ! BdbIsNull(mdH));

                  //  Name the metadata structure to be able to discover it later.

                    BdbStatus status;
                    status = mdH.nameObj( mdContH,
                                          "MetaData" );
                    assert( BdbcSuccess == status );
                }
                if( failed ) break;
            }
        }
        if( failed ) break;

        _isInstantiated = true;

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSPartitionP::close( const BdbTime& theTime )
{
    const char* warningStr = "CdbBdbSPartitionP::close() -- WARNING";
    const char* errorStr   = "CdbBdbSPartitionP::close() -- ERROR";
    const char* fatalStr   = "CdbBdbSPartitionP::close() -- FATAL";

    ooUpdate( );

    CdbStatus result = CdbStatus::Error;
    do {

      // Check if the local origin matches this partition's origin.

        if( CdbStatus::Success != matchOrigin( )) break;

      // Check if this partition has not been instantiated

        if( ! isInstantiated( )) {
            cout << errorStr << endl
                 << "    The parition is not instantiated yet. The requested operation" << endl
                 << "    only applies to instantiated partitions." << endl;
            break;
        }

      // Check if this partition has not been closed yet

        if( isClosed( )) {
            cout << warningStr << endl
                 << "    The parition is already closed. Attempting to close it again" << endl
                 << "    may be an indication of problems in the application's logic." << endl;
            result = CdbStatus::Warning;
            break;
        }

      // Locate the highest increment number

        d_UShort incrementNumber = 0;
        {
            d_UShort num = _increments.numIncrements( );

            assert( num >= 1 );      // FATAL ERROR: There must be at least 1 increment
                                     //              in the cluster.
            incrementNumber = num - 1;
        }

      // Need this name to locate the metadata objects at the local
      // database files of the partition.

        std::string localRangeName;
        {
            CdbStatus returnStatus = CdbBdbSUtils::getLocalDbIdRange( localRangeName );
            assert( CdbStatus::Success == returnStatus );
        }

      // For each cluster of conditions
      //
      // NOTE: We're using MASTER's collection of "partitionable" clusters because:
      // 
      //         (1) partitionable condition's increments are managed by each partition
      //             unlike regular conditions, whose increments are managed dircetly by
      //             a "simple cluster", which is a subclass of the basic cluster.
      //
      //         (2) it just does not make any sense to have local collection at the SLAVE
      //             federation although the current design of the "SLAVE" registry
      //             allows so.

        BdbHandle(CdbBdbSRegistryP) rH;
        if( CdbStatus::Success != CdbBdbSRegistryP::findMaster( rH )) {

            cout << fatalStr << endl
                 << "    Failed to find Master registry." << endl;

            assert( 0 );
            break;
        }
        BdbRef(CdbBdbSClusterCollectionP) clusterCollRef = rH->pClusterCollection( );
        assert( !BdbIsNull(clusterCollRef));

        bool failed = false;

        CdbBdbSClusterCollectionP::IteratorOfIdentifiers clusterItr =
            clusterCollRef->iterator_identifiers( );

        while( clusterItr.next( )) {

            d_UShort clusterId = clusterItr.value( );

            BdbRef(CdbBdbSClusterP) clusterRef;
            if( CdbStatus::Success != clusterCollRef->find( clusterId,
                                                            clusterRef )) {
                cout << errorStr << endl
                     << "    Failed to find a cluster with ID #" << clusterId << endl;

                failed = true;
                break;
            }

          // For each condition in this cluster

            CdbItr< BdbRef(CdbBdbSConditionP) > itr;
            if( CdbStatus::Success != clusterRef->iterator( itr )) {
                cout << errorStr << endl
                     << "    Failed to obtain the conditions - members of the cluster ID #" << clusterId << endl;

                failed = true;
                break;
            }
            while( itr.next( )) {

              // Locate condition

                BdbRef(CdbBdbSConditionP) conditionRef = itr.value( );
                assert( !BdbIsNull(conditionRef));

                d_UShort conditionId = conditionRef->id( );

              // Locate topmost metadata of the old increment

                BdbRef(CdbBdbSMetaDataP) mdRef;
                if( CdbStatus::Success != CdbBdbSUtils::findMetaData( mdRef,
                                                                      conditionId,
                                                                      true,         // partitionable
                                                                      id( ),        // partition identifier
                                                                      clusterId,    // current cluster identifier
                                                                      incrementNumber,
                                                                      localRangeName )) {

                    cout << errorStr << endl
                         << "    Failed to find MetaData object for increment #" << incrementNumber << endl;

                    failed = true;
                    break;
                }

              // Close  the found metadata object.
              //
              // NOTE: The closure procedure will also create a final revision in
              //       in the metadata container at the most recent modification
              //       time of that metadata container.

                BdbTime modificationTime = mdRef->modified( );
                char finalRevisionNameStr[32];

                sprintf( finalRevisionNameStr,
                         "final_%u.%u",
                         modificationTime.getGmtSec( ),
                         modificationTime.getGmtNsec( ));

                if( CdbStatus::Success != mdRef->close( theTime,
                                                        finalRevisionNameStr,
                                                        theTime,
                                                        "automatically created by the partition closure procedure" )) {
                    cout << errorStr << endl
                         << "    Failed to close the MetaData object of the latest increment." << endl;

                    failed = true;
                    break;
                }
            }
            if( failed ) break;
        }
        if( failed ) break;

      // Proceed and inform the listener if any by passing it the current
      // partition's identifier

        if( CdbStatus::Success != closeCell( theTime )) {
            break;
        }
        if( ! BdbIsNull(_listenerRef)) {
            result = _listenerRef->close( id( ));
        } else {
            result = CdbStatus::Success;
        }

    } while( false );

    return result;
}

d_UShort
CdbBdbSPartitionP::numIncrements( ) const
{
    return _increments.numIncrements( );
}

CdbStatus
CdbBdbSPartitionP::increment( d_UShort          theNumber,
                              CdbBdbSIncrement& theIncrement ) const
{
    return _increments.increment( theNumber,
                                  theIncrement );
}

CdbStatus
CdbBdbSPartitionP::createIncrement( const BdbTime& theSplitTime )
{
    const char* errorStr = "CdbBdbSPartitionP::createIncrement() -- ERROR";
    const char* fatalStr = "CdbBdbSPartitionP::createIncrement() -- FATAL";

    ooUpdate( );

    CdbStatus result = CdbStatus::Error;
    do {

      // Check if the local origin matches this partition's origin.

        if( CdbStatus::Success != matchOrigin( )) break;

      // Check if this partition has not been instantiated

        if( ! isInstantiated( )) {
            cout << errorStr << endl
                 << "    The parition is not instantiated yet. The requested operation" << endl
                 << "    only applies to instantiated partitions." << endl;
            break;
        }

      // Check if this partition has not been closed yet

        if( isClosed( )) {
            cout << errorStr << endl
                 << "    The partition is already closed. The requested operation is" << endl
                 << "    only allowed for open partitions." << endl;
            break;
        }

      // Locate the highest old increment number

        d_UShort oldIncrementNumber = 0;
        {
            d_UShort num = _increments.numIncrements( );

            assert( num >= 1 );      // FATAL ERROR: There must be at least 1 increment
                                     //              in the cluster.
            oldIncrementNumber = num - 1;
        }

      // Create new increment number

        if( CdbStatus::Success != _increments.createIncrement( theSplitTime )) {
            cout << errorStr << endl
                 << "    Failed to create the new partition increment." << endl;
            break;
        }

      // Locate the highest new increment number

        d_UShort newIncrementNumber = 0;
        {
            d_UShort num = _increments.numIncrements( );

            assert( num >= 2 );      // FATAL ERROR: There must be at least 1+1 increment
                                     //              in the cluster (the last one we've just
                                     //              created)
            newIncrementNumber = num - 1;
        }

      // Prepare invariant parts of database names.

        std::string localRangeName;
        {
            CdbStatus returnStatus = CdbBdbSUtils::getLocalDbIdRange( localRangeName );
            assert( CdbStatus::Success == returnStatus );
        }
        char partitionIdStr[12];
        char newIncrementStr[12];

        sprintf( partitionIdStr, "%u", id( ));
        sprintf( newIncrementStr, "%u", newIncrementNumber );

      // For each cluster of conditions
      //
      // NOTE: For the reasons already explained at the ::close() method above,
      //       we should always use the MASTER's collection of clusters. SLAVE-s
      //       are not allowed to have partitionable clusters anymore.

        BdbHandle(CdbBdbSRegistryP) rH;
        if( CdbStatus::Success != CdbBdbSRegistryP::findMaster( rH )) {

            cout << fatalStr << endl
                 << "    Failed to find Master registry." << endl;

            assert( 0 );
            break;
        }
        BdbRef(CdbBdbSClusterCollectionP) clusterCollRef = rH->pClusterCollection( );
        assert( !BdbIsNull(clusterCollRef));

        bool failed = false;

        CdbBdbSClusterCollectionP::IteratorOfIdentifiers clusterItr =
            clusterCollRef->iterator_identifiers( );

        while( clusterItr.next( )) {

            d_UShort clusterId = clusterItr.value( );

            BdbRef(CdbBdbSClusterP) clusterRef;
            if( CdbStatus::Success != clusterCollRef->find( clusterId,
                                                            clusterRef )) {
                cout << errorStr << endl
                     << "    Failed to find a cluster with ID #" << clusterId << endl;

                failed = true;
                break;
            }

          // Build the name of the new database for the new increment

            std::string newDatabaseSuffix;
            {
                char clusterIdStr[12];
                sprintf( clusterIdStr, "%u", clusterId );

                newDatabaseSuffix = std::string( "p" )  + std::string( partitionIdStr ) +
                                    std::string( "_c" ) + std::string( clusterIdStr ) +
                                    std::string( "_i" ) + std::string( newIncrementStr );

                newDatabaseSuffix = newDatabaseSuffix +
                                    std::string( "_" ) +
                                    localRangeName;
            }

          // Create the new database by asking to create the "System" container
          // in it.

            BdbCondIndexClusteringHint systemHint( "cdb",
                                                   "System",
                                                   newDatabaseSuffix.c_str( ));

            BdbHandle(BdbContObj) systemContH;
            systemContH = systemHint.updatedHint( );

            assert( ! BdbIsNull(systemContH));
            assert( systemHint.newContainerCreated( ));

          // For each condition in this cluster

            CdbItr< BdbRef(CdbBdbSConditionP) > itr;
            if( CdbStatus::Success != clusterRef->iterator( itr )) {
                cout << errorStr << endl
                     << "    Failed to obtain the conditions - members of the cluster ID #" << clusterId << endl;

                failed = true;
                break;
            }

            while( itr.next( )) {

              // Locate condition

                BdbRef(CdbBdbSConditionP) conditionRef = itr.value( );
                assert( !BdbIsNull(conditionRef));

                d_UShort conditionId = conditionRef->id( );

              // Locate topmost metadata of the old increment

                BdbRef(CdbBdbSMetaDataP) oldMetaDataRef;
                if( CdbStatus::Success != CdbBdbSUtils::findMetaData( oldMetaDataRef,
                                                                      conditionId,
                                                                      true,         // partitionable
                                                                      id( ),        // partition identifier
                                                                      clusterId,    // current cluster identifier
                                                                      oldIncrementNumber,
                                                                      localRangeName )) {

                    cout << errorStr << endl
                         << "    Failed to find MetaData object for increment #" << oldIncrementNumber << endl;

                    failed = true;
                    break;
                }

              // Close  the found metadata object.
              //
              // NOTE: The closure procedure will also create a final revision in
              //       in the metadata container at the most recent modification
              //       time of that metadata container.

                BdbTime modificationTime = oldMetaDataRef->modified( );
                char finalRevisionNameStr[32];

                sprintf( finalRevisionNameStr,
                         "final_%u.%u",
                         modificationTime.getGmtSec( ),
                         modificationTime.getGmtNsec( ));

                if( CdbStatus::Success != oldMetaDataRef->close( theSplitTime,
                                                                 finalRevisionNameStr,
                                                                 theSplitTime,
                                                                 "automatically created by the increment creation procedure" )) {
                    cout << errorStr << endl
                         << "    Failed to close the old MetaData object." << endl;

                    failed = true;
                    break;
                }

              // Create new metadata object for this condition in the new database
              //
              //   NOTE: We need to extend the API of MetaData object to pass it a reference onto
              //         the previous metadata object to be initialized from.

                char conditionIdStr[12];
                sprintf( conditionIdStr, "%u", conditionId );

                std::string mdContainerName = std::string( "MetaData." ) + std::string( conditionIdStr );
                std::string obContainerName = std::string( "Objects."  ) + std::string( conditionIdStr );

              // Create containers for metadata and condition objects at specified
              // database.

                BdbCondIndexClusteringHint mdHint( "cdb",
                                                   mdContainerName.c_str( ),
                                                   newDatabaseSuffix.c_str( ));

                BdbHandle(BdbContObj) mdContH;
                mdContH = mdHint.updatedHint( );

                assert( ! BdbIsNull(mdContH));
                assert( mdHint.newContainerCreated( ));

                BdbCondIndexClusteringHint obHint( "cdb",
                                                   obContainerName.c_str( ),
                                                   newDatabaseSuffix.c_str( ));

                BdbHandle(BdbContObj) obContH;
                obContH = obHint.updatedHint( );

                assert( ! BdbIsNull(obContH));
                assert( obHint.newContainerCreated( ));

              // Create initial infrastructure in the newely created
              // container for metadata. The new medatada will have the same validity interval
              // as the old, and its begin insertion time will be where the old one ends.
              //
              // NOTE: Remember, that intervals collections of the new metadata object
              //       will be incomplete since they will not include intervals from previous
              //       increments. It's up to the intervals locator algorithm running on the top
              //       of metadata objects to resolve this issue.

                BdbHandle(CdbBdbSMetaDataP) newMetaDataH;
                newMetaDataH = new( mdContH ) CdbBdbSMetaDataP( oldMetaDataRef->minValidity( ), // begin of the validity
                                                                oldMetaDataRef->maxValidity( ), // end   of the validity
                                                                theSplitTime );
                assert( ! BdbIsNull(newMetaDataH));

              //  Name the metadata structure to be able to discover it later.

                BdbStatus status;
                status = newMetaDataH.nameObj( mdContH,
                                              "MetaData" );
                assert( BdbcSuccess == status );
            }
            if( failed ) break;
        }
        if( failed ) break;

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

void
CdbBdbSPartitionP::dump( ostream& o ) const
{
    CdbBdbSPartitionBaseP::dump( o );

    o << endl
      << "         ORIGIN ID: " << _originId << endl
      << "   IS INSTANTIATED: " << ( _isInstantiated ? "Yes" : "No" ) << endl
      << "LAST TIME MODIFIED: " << _modified << endl
      << endl
      << "INCREMENTS: " << endl;

    d_UShort num = numIncrements( );
    for( d_UShort i = 0; i < num; ++i ) {
        CdbBdbSIncrement inc;
        CdbStatus result = increment( i, inc );
        assert( CdbStatus::Success == result );
        o << " -> ";
        inc.dump( o );
    }
}

CdbStatus
CdbBdbSPartitionP::matchOrigin( ) const
{
    const char* errorStr   = "CdbBdbSPartitionP::matchOrigin() - ERROR.";

    CdbStatus result = CdbStatus::Error;
    do {

      // Check if the local origin matches this partition's origin.

        BdbHandle(CdbBdbSRegistryP) rH;
        if( CdbStatus::Success != CdbBdbSRegistryP::findLocal( rH )) {
            cout << errorStr << endl
                 << "    Failed to find local registry." << endl;
            break;
        }
        if( rH->originId( ) != _originId ) {
            cout << errorStr << endl
                 << "    The origin of the current partition does not match the origin of" << endl
                 << "    the local database." << endl
                 << "        THE ORIGIN OF THIS PARTITION'S OWNER: " << _originId << endl
                 << "        THE ORIGIN OF THE LOCAL DATABASE:     " << rH->originId( ) << endl;
            break;
        }

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

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

---