---

CdbBdb2RooConversionTool.cc

Go to the documentation of this file.

// A conversion tool to convert the contents of an Objectivity/DD base
// CDB installation into the ROOT I/O based one.

#include "BaBar/BaBar.hh"
#include "BdbUtil/Bdb.hh"

#include "CdbBase/CdbPathName.hh"
#include "CdbBase/CdbTimeUtils.hh"
#include "CdbBase/CdbTransaction.hh"

#include "CdbBdbShared/CdbBdbShared.hh"
#include "CdbBdbShared/CdbBdbSRegistryP.hh"
#include "CdbBdbShared/CdbBdbSOriginCollectionP.hh"
#include "CdbBdbShared/CdbBdbSOriginP.hh"
#include "CdbBdbShared/CdbBdbSConditionCollectionP.hh"
#include "CdbBdbShared/CdbBdbSConditionP.hh"
#include "CdbBdbShared/CdbBdbSViewCollectionP.hh"
#include "CdbBdbShared/CdbBdbSViewP.hh"
#include "CdbBdbShared/CdbBdbSConfigCollectionP.hh"
#include "CdbBdbShared/CdbBdbSConfigInterval.hh"
#include "CdbBdbShared/CdbBdbSClusterCollectionP.hh"
#include "CdbBdbShared/CdbBdbSClusterP.hh"
#include "CdbBdbShared/CdbBdbSSimpleClusterP.hh"
#include "CdbBdbShared/CdbBdbSIncrement.hh"
#include "CdbBdbShared/CdbBdbSPartitionsLayoutP.hh"
#include "CdbBdbShared/CdbBdbSPartitionP.hh"
#include "CdbBdbShared/CdbBdbSMetaDataP.hh"
#include "CdbBdbShared/CdbBdbSUtils.hh"
#include "CdbBdbShared/CdbBdbSOiCollectionP.hh"
#include "CdbBdbShared/CdbBdbSRevisionP.hh"
#include "CdbBdbShared/CdbBdbSViCollectionP.hh"

#include "CdbRooReadonly/CdbRooRoFileUtils.hh"
#include "CdbRooReadonly/CdbRooRoIdR.hh"
#include "CdbRooReadonly/CdbRooRoRegistryDescriptorR.hh"
#include "CdbRooReadonly/CdbRooRoConditionCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoConditionR.hh"
#include "CdbRooReadonly/CdbRooRoViewCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoViewR.hh"
#include "CdbRooReadonly/CdbRooRoConfigCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoConfigElementR.hh"
#include "CdbRooReadonly/CdbRooRoConfigInterval.hh"
#include "CdbRooReadonly/CdbRooRoRClusterCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoRClusterR.hh"
#include "CdbRooReadonly/CdbRooRoIncrementR.hh"
#include "CdbRooReadonly/CdbRooRoOriginCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoOriginR.hh"
#include "CdbRooReadonly/CdbRooRoClusterCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoClusterR.hh"
#include "CdbRooReadonly/CdbRooRoPartitionsLayoutR.hh"
#include "CdbRooReadonly/CdbRooRoPartitionR.hh"
#include "CdbRooReadonly/CdbRooRoOiCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoRevisionR.hh"
#include "CdbRooReadonly/CdbRooRoRevCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoMetaDataR.hh"
#include "CdbRooReadonly/CdbRooRoViCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoBootRecordR.hh"

#include <TFile.h>
#include <TTree.h>
#include <TDirectory.h>

#include <iostream>
#include <fstream>
#include <string>
using std::cin;
using std::cout;
using std::cerr;
using std::endl;
#include <sstream>
using std::ostringstream;

namespace {

    bool verboseModeFlag          = false;
    bool excludeRegularFlag       = false;
    bool excludePartitionableFlag = false;
    bool doNotCompressFlag        = false;

    enum ClusteringMode {
        CLUSTERING_MODE_STANDARD,
        CLUSTERING_MODE_ALLIN3,
        CLUSTERING_MODE_1PLUS2PERORIGIN
    } clusteringMode = CLUSTERING_MODE_STANDARD;

  // Helper functions forming database names depending on a selected clustering option
  // They'll also register an entry in a special map to be saved in a boot file.
  //
  // We also have the cache and teh opening helper and open the corresponding databases.
  //
  // IMPORTANT: The same databases should be open just once! Otherwise objects may get
  //            lost! In other words, for the same file "file.root" there shouldn't be
  //            two (or more) simultaneous file opening operations:
  //
  //              TFile f("file.root")
  //
  //            within an application. Thsi problem is solved by mean of caching
  //            open files in the special cache: database2fileMap and using and polutating
  //            the cache as needed. See below.

    std::map<std::string,std::string> database2LocationMap;

    std::map<std::string, CdbCPtr<TFile> > database2fileMap;

    CdbCPtr<TFile>
    openDatabase( const std::string& theDatabaseLocation )
    {
      // Check if we have the database in the cache. If so then return the cached value.
      // We're also doing "Cd" into the top-level directory of the file.

        std::map<std::string, CdbCPtr<TFile> >::const_iterator itr = database2fileMap.find( theDatabaseLocation );
        if( itr != database2fileMap.end( )) {
            itr->second->Cd( "" );
            gDirectory->Cd( itr->second->GetPath( ));
            return itr->second;
        }

      // Try to open the database, and if we succeed - then populate the cache.

        int compressionLevel = 1;                       // "..minimal compression level but fast.."
        if( doNotCompressFlag ) compressionLevel = 0;   // "..objects written to this file will not be compressed.."

        CdbCPtr<TFile> databasePtr = new TFile( theDatabaseLocation.c_str( ),
                                                "recreate",
                                                "CDB file",
                                                compressionLevel );
        if( databasePtr->IsZombie( )) {
            cerr << "error: failed to (re-)create the database: '" << theDatabaseLocation << "'." << endl;
            return false;
        }
        database2fileMap[theDatabaseLocation] = databasePtr;

      // "Cd" into the database before returning it to a caller.

        databasePtr->Cd( "" );
        gDirectory->Cd( databasePtr->GetPath( ));

        return databasePtr;
    }

    CdbCPtr<TFile>
    openRegistryDatabase( const unsigned short theOriginId )
    {
      // Build a 'system' name of the database

        std::string systemName = "";
        {
            ostringstream s;
            s << "cdb_o" << theOriginId << "_registry.root";
            systemName = s.str( );
        }

      // Build a 'location' name according to the current clustering mode

        std::string locationName = "";
        {
            ostringstream s;

            switch( clusteringMode ) {

            case CLUSTERING_MODE_STANDARD:

                s << systemName;
                break;

            case CLUSTERING_MODE_ALLIN3:

                s << "cdb_metadata.root";
                break;

            case CLUSTERING_MODE_1PLUS2PERORIGIN:

                s << "cdb_o" << theOriginId << "_metadata.root";
                break;

            default:
                assert( 0 );
                return 0;
            }
            locationName = s.str( );
        }

      // Register a pair of names in the map

        database2LocationMap[systemName] = locationName;

      // Propagate the operation down to open/create the database (now we're going to populate/use
      // the cache). Any errors (if any) should already be reported by the called function, therefore
      // we just return the error status.

        return openDatabase( locationName );
    }

    CdbCPtr<TFile>
    openMetadataDatabase( const unsigned short theOriginId,
                          const unsigned short theClusterId,
                          const unsigned short thePartitionId,
                          const unsigned short theIncrementId,
                          const bool           isPartitionableFlag )
    {
      // Build a 'system' name of both metadata and objects databases

        std::string metadataSystemName = "";
        std::string objectsSystemName  = "";
        {
            ostringstream s;
            s << "cdb_o" << theOriginId << "_c" << theClusterId;
            if( isPartitionableFlag ) s << "_p" << thePartitionId;
            s << "_i" << theIncrementId;
            metadataSystemName =  s.str( );
            metadataSystemName += ".root";
            objectsSystemName  =  s.str( );
            objectsSystemName  += "_objects.root";
        }

      // Build 'location' names for both databases according to the current clustering mode

        std::string metadataLocation = "";
        std::string objectsLocation  = "";
        {
            ostringstream metadataStream;
            ostringstream objectsStream;

            switch( clusteringMode ) {

            case CLUSTERING_MODE_STANDARD:

                metadataStream << metadataSystemName;
                objectsStream  << objectsSystemName;

                break;

            case CLUSTERING_MODE_ALLIN3:

                metadataStream << "cdb_metadata.root";
                objectsStream  << "cdb_objects.root";

                break;

            case CLUSTERING_MODE_1PLUS2PERORIGIN:

                metadataStream << "cdb_o" << theOriginId << "_metadata.root";
                objectsStream  << "cdb_o" << theOriginId << "_objects.root";

                break;

            default:
                assert( 0 );
                return 0;
            }
            metadataLocation = metadataStream.str( );
            objectsLocation  = objectsStream.str( );
        }

      // Register both databases in the map

        database2LocationMap[metadataSystemName] = metadataLocation;
        database2LocationMap[objectsSystemName]  = objectsLocation;

      // Propagate the operation down to open/create the database (now we're going to populate/use
      // the cache). Any errors (if any) should already be reported by the called function, therefore
      // we just return the error status.

        return openDatabase( metadataLocation );
    }

  // Operations

    CdbStatus convertConfigCollection( CdbRooRoConfigCollectionR*&             theOutputConfigPtr,
                                       const BdbRef(CdbBdbSConfigCollectionP)& theInputConfigCollectionRef );

    CdbStatus convertFoldersAndConditions( CdbRooRoViewR&                     theOutView,
                                           const BdbHandle(CdbBdbSFolderP)&   theInFolderH,
                                           const CdbPathName&                 theCurrentPathName );

    CdbStatus convertPhysicalConditions( const CdbCPtr<TFile>&              theFilePtr,
                                         const BdbHandle(CdbBdbSRegistryP)& theMasterRegistryH,
                                         const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                                         const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus convertViews( const CdbCPtr<TFile>&              theFilePtr,
                            const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                            const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus convertRegularClusters( const CdbCPtr<TFile>&              theFilePtr,
                                      const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                                      const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus convertPartitionsLayout( const CdbCPtr<TFile>&              theFilePtr,
                                       const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                                       const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus convertOrigins( const CdbCPtr<TFile>&              theFilePtr,
                              const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                              const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus convertPartitionableClusters( const CdbCPtr<TFile>&              theFilePtr,
                                            const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                                            const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus testMetaDataConversion( );

    CdbStatus convertMetaDataOfRegularCondition( const BdbHandle(CdbBdbSConditionP)& theConditionH,
                                                 const BdbHandle(CdbBdbSRegistryP)&  theMasterRegistryH,
                                                 d_UShort                            theOwnerOriginId,
                                                 const CdbRooRoRegistryDescriptorR&  theOutRegistryDescriptor );

    CdbStatus convertMetaDataOfPartitionableCondition( const BdbHandle(CdbBdbSConditionP)& theConditionH,
                                                       const BdbHandle(CdbBdbSRegistryP)&  theMasterRegistryH,
                                                       const CdbRooRoRegistryDescriptorR&  theOutRegistryDescriptor );

    CdbStatus convertMetaData( const CdbCPtr<TFile>&              theFilePtr,
                               const BdbHandle(CdbBdbSMetaDataP)& theInMetaDataH,
                               UShort_t                           theOriginId,
                               UShort_t                           theConditionId,
                               UShort_t                           theClusterId,
                               bool                               isPartitionableFlag,
                               UShort_t                           thePartitionId,
                               UShort_t                           theIncrementId,
                               const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor );

    CdbStatus translateOriginalCollection( CdbRooRoOiCollectionR&                 theOutCollection,
                                           const BdbHandle(CdbBdbSOiCollectionP)& theInCollectionH,
                                           UShort_t                               theOriginId,
                                           UShort_t                               theConditionId,
                                           UShort_t                               theClusterId,
                                           bool                                   isPartitionableFlag,
                                           UShort_t                               thePartitionId,
                                           UShort_t                               theIncrementId );

    CdbStatus translateRevision( CdbRooRoRevisionR&                 theOutRevision,
                                 const BdbHandle(CdbBdbSRevisionP)& theInRevisionH );

    CdbStatus printPersistentClasses( );

    CdbStatus printPersistentConditions( );

    void usage( )
    {
        cout << "Usage: <command> [<options>]\n"
            << "  help\n"
            << "  conversion [<options>] [-clustering { STANDARD | ALLIN3 | 1PLUS2PERORIGIN }] [-do_not_compress]\n"
            << "  testing    [<options>]\n"
            << "  classes    [<options>]\n"
            << "  conditions [<options>]\n"
            << "Options common for all commands:\n"
            << "  [-noregular] [-nopartitionable] [-verbose]" << endl;
    }
}

int
main( int argc, char* argv[] )
{
  // Get the command name

    enum CommandType {
        CMD_CONVERSION = 0,
        CMD_TESTING,
        CMD_CLASSES,
        CMD_CONDITIONS
    };
    CommandType command = CMD_CONVERSION;

    int numArgs = argc - 1;
    int nextArg = 1;

    if( 0 == numArgs ) {
        usage( );
        return 1;
    }
    if( numArgs > 0 ) {

        const std::string commandName = argv[nextArg];

        if( commandName == "help" ) {

            usage( );

            return 1;

        } else if( commandName == "conversion" ) {

            command = CMD_CONVERSION;

            numArgs--;
            nextArg++;

            if( numArgs >= 2 ) {
                const std::string option = argv[nextArg];
                if( option == "-clustering" ) {
                    const std::string value = argv[nextArg+1];
                    if     ( value == "STANDARD"        ) clusteringMode = CLUSTERING_MODE_STANDARD;
                    else if( value == "ALLIN3"          ) clusteringMode = CLUSTERING_MODE_ALLIN3;
                    else if( value == "1PLUS2PERORIGIN" ) clusteringMode = CLUSTERING_MODE_1PLUS2PERORIGIN;
                    else {
                        cerr << "error: unrecognized value '" << value << "' of the option '" << option << "'." << endl;
                        usage( );
                        return 1;
                    }
                    numArgs -= 2;
                    nextArg += 2;
                } else {
                    cerr << "error: unrecognized option '" << option << "'." << endl;
                    usage( );
                    return 1;
                }
            }
            if( numArgs >= 1 ) {
                const std::string option = argv[nextArg];
                if( option == "-do_not_compress" ) {
                    doNotCompressFlag = true;
                    numArgs -= 1;
                    nextArg += 1;
                }
            }

        } else if( commandName == "testing" ) {

            command = CMD_TESTING;

            numArgs--;
            nextArg++;

        } else if( commandName == "classes" ) {

            command = CMD_CLASSES;

            numArgs--;
            nextArg++;

        } else if( commandName == "conditions" ) {

            command = CMD_CONDITIONS;

            numArgs--;
            nextArg++;

        } else {

            cerr << "unknown command: \"" << commandName << "\"." << endl;
            usage( );
            return 1;
        }
    }
    if( numArgs > 0 ) {
        if( argv[nextArg] == std::string( "-noregular" )) {

            excludeRegularFlag = true;

            numArgs --;
            nextArg ++;
        }
    }
    if( numArgs > 0 ) {
        if( argv[nextArg] == std::string( "-nopartitionable" )) {

            excludePartitionableFlag = true;

            numArgs --;
            nextArg ++;
        }
    }
    if( numArgs > 0 ) {
        if( std::string( "-verbose" ) == argv[nextArg] ) {

            verboseModeFlag = true;

            numArgs --;
            nextArg ++;

        } else {
            cerr << "unknown command option or parameter: \"" << argv[nextArg] << "\"." << endl;
            usage( );
            return 1;
        }
    }

   // Initialize Ojectivity/DB base CDB API context

    CdbBdbShared::forceLoad( );

    CdbTransaction readOnlyTransaction( CdbBdbShared::technology( ),
                                        CdbBdbShared::implementation( ));

 // Proceed directly to the simple tests if required.

    switch( command ) {
    case CMD_TESTING   : return ( CdbStatus::Success == ::testMetaDataConversion   ( ) ? 0 : 1 );
    case CMD_CLASSES   : return ( CdbStatus::Success == ::printPersistentClasses   ( ) ? 0 : 1 );
    case CMD_CONDITIONS: return ( CdbStatus::Success == ::printPersistentConditions( ) ? 0 : 1 );
    default:
        break;
    }

  // Get the MASTER registry and the collection of origins, which we'll
  // be using for a few things below.

    BdbHandle(CdbBdbSRegistryP) masterRegistryH;
    CdbStatus result = CdbBdbSRegistryP::findMaster( masterRegistryH );
    if( CdbStatus::Success != result ) {
        cerr << "failed to find the MASTER registry." << endl;
        return 1;
    }
    assert( masterRegistryH->isMaster( ));

    BdbHandle(CdbBdbSOriginCollectionP) originCollectionH = masterRegistryH->originCollection( );
    assert( !BdbIsNull(originCollectionH));

  // Iterate over known registries and convert them one-by-one if the corresponding origins
  // are instantiated in the input database.

    CdbBdbSOriginCollectionP::IteratorOfIdentifiers itr = originCollectionH->iterator_identifiers( );
    while( itr.next( )) {

      // Get the persistent registry object

        d_UShort originId = itr.value( );

//cout << "CdbBdb2RooConversionTool::main() - DEBUG" << endl;
//if( 0 != originId ) break;

        BdbHandle(CdbBdbSOriginP) originH;
        {
            BdbRef(CdbBdbSOriginP) originRef;
            if( CdbStatus::Success != originCollectionH->find( originId,
                                                               originRef )) {

                cerr << "failed to find an origin with ID = " << originId << "." << endl
                     << "The input database may not be properly initialized." << endl;
                return 1;
            }
            originH = originRef;
        }

      // Try to get the corresponding registry. If it doesn't exist (not instantiated) then we
      // will _NOT_ create the output data structures for that registry.

        BdbHandle(CdbBdbSRegistryP) registryH;
        {
            CdbStatus result = CdbBdbSRegistryP::findByOrigin( registryH,
                                                               originId );
            if( CdbStatus::Success != result ) {
                if( CdbStatus::NotFound == result ) {

                  // The origin is not currently instantiated in the input database. This may happen
                  // when the origin's data are yet to be imported.

                    continue;

                } else {
                    cerr << "failed to find a registry for origin ID = " << originId << "." << endl;
                    return 1;
                }
            }
        }

      // Open the output file for the registry & related collections

        const CdbCPtr<TFile> databaseFilePtr = openRegistryDatabase( originId );
        if( databaseFilePtr.isNull( )) return 1;

      // Create a registry descriptor of the corresponding type

        CdbRooRoRegistryDescriptorR registryDescriptor;

        switch( originH->type( )) {

        case CdbBdbSOriginP::MASTER:

            registryDescriptor = CdbRooRoRegistryDescriptorR( originH->id( ),
                                                              originH->name( ).head( ),
                                                              registryH->description( ).head( ),
                                                              registryH->created( ),
                                                              registryH->id( ),
                                                              true,                         // hasLocalCollections
                                                              true,                         // isMaster
                                                              false,                        // isSlave
                                                              false );                      // isTest
            break;

        case CdbBdbSOriginP::SLAVE:

            registryDescriptor = CdbRooRoRegistryDescriptorR( originH->id( ),
                                                              originH->name( ).head( ),
                                                              registryH->description( ).head( ),
                                                              registryH->created( ),
                                                              registryH->id( ),
                                                              true,                         // hasLocalCollections
                                                              false,                        // isMaster
                                                              true,                         // isSlave
                                                              false );                      // isTest
            break;

        case CdbBdbSOriginP::REPLICA:

            cerr << "an obsolete and unsupprted (to the current algorithm) origin type: " << originH->type( ) << " found" << endl
                 << "for an origin with ID = " << originH->id( ) << " and NAME = \"" << originH->name( ) << "\"." << endl;
            return 1;

        case CdbBdbSOriginP::TEST:

            registryDescriptor = CdbRooRoRegistryDescriptorR( originH->id( ),
                                                              originH->name( ).head( ),
                                                              registryH->description( ).head( ),
                                                              registryH->created( ),
                                                              registryH->id( ),
                                                              true,                         // hasLocalCollections
                                                              false,                        // isMaster
                                                              false,                        // isSlave
                                                              true );                       // isTest
            break;

        default:

            cerr << "unknown (to the current algorithm) origin type: " << originH->type( ) << " found" << endl
                 << "for an origin with ID = " << originH->id( ) << " and NAME = \"" << originH->name( ) << "\"." << endl;
            return 1;
        }

      // Store the descriptor in the output file

        {
            const std::string persistentObjectName =
                CdbRooRoFileUtils::nameOfSystemContainer( registryDescriptor.originId( ),
                                                          "Registry" );

            databaseFilePtr->Cd( "" );
            gDirectory->Cd( databaseFilePtr->GetPath( ));

            Int_t bytesWritten = databaseFilePtr->WriteObject( &registryDescriptor,
                                                               persistentObjectName.c_str( ));
            if( !bytesWritten ) {
                cerr << "New object couldn't be written. Perhaps it's not" << endl
                     << "been properly instrumented for streaming." << endl;
                return 1;
            }
            cout << databaseFilePtr->GetName( ) << "::" << registryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;
        }

      // Convert and store local collections

        if( registryDescriptor.hasLocalCollections( )) {

            if( CdbStatus::Success != ( result = ::convertPhysicalConditions( databaseFilePtr,
                                                                              masterRegistryH,
                                                                              registryH,
                                                                              registryDescriptor ))) {

                cerr << "Failed to convert a collection of 'physical' conditions." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
            if( CdbStatus::Success != ( result = ::convertViews( databaseFilePtr,
                                                                 registryH,
                                                                 registryDescriptor ))) {

                cerr << "Failed to convert a collection of views." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
            if( CdbStatus::Success != ( result = ::convertRegularClusters( databaseFilePtr,
                                                                           registryH,
                                                                           registryDescriptor ))) {

                cerr << "Failed to convert a collection of 'regular' clusters." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
        }

      // MASTER & SLAVE collections only

        if( registryDescriptor.isMaster( ) || registryDescriptor.isSlave( )) {

            if( CdbStatus::Success != ( result = ::convertPartitionsLayout( databaseFilePtr,
                                                                            registryH,
                                                                            registryDescriptor ))) {

                cerr << "Failed to convert Partitions Layout." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
        }

      // MASTER collections only

        if( registryDescriptor.isMaster( )) {

            if( CdbStatus::Success != ( result = ::convertOrigins( databaseFilePtr,
                                                                   registryH,
                                                                   registryDescriptor ))) {

                cerr << "Failed to convert a collection of origins." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
            if( CdbStatus::Success != ( result = ::convertPartitionableClusters( databaseFilePtr,
                                                                                 registryH,
                                                                                 registryDescriptor ))) {

                cerr << "Failed to convert a collection of 'partitionable' clusters." << endl
                     << "for origin ID = " << registryDescriptor.originId( ) << endl;
                return 1;
            }
        }
     }

  // Create the boot file

    {
      // Get the local registry and its origin

        BdbHandle(CdbBdbSRegistryP) localRegistryH;
        CdbStatus result = CdbBdbSRegistryP::findLocal( localRegistryH );
        if( CdbStatus::Success != result ) {
            cerr << "failed to find the local registry." << endl;
            return 1;
        }

        BdbHandle(CdbBdbSOriginP) originH;
        {
            BdbRef(CdbBdbSOriginP) originRef;
            if( CdbStatus::Success != originCollectionH->find( localRegistryH->originId( ),
                                                               originRef )) {

                cerr << "failed to find the local origin with ID = " << localRegistryH->originId( ) << "." << endl
                     << "The input database may not be properly initialized." << endl;
                return 1;
            }
            originH = originRef;
        }

      // Create the boorfile record

        const std::string bootFileName = "cdb_boot.root";
        TFile boofFileDatabaseFile( bootFileName.c_str( ), "recreate" );
        if( boofFileDatabaseFile.IsZombie( )) {
            cerr << "Failed to (re-)create the file: \"" << boofFileDatabaseFile.GetName( ) << "\"." << endl;
            return 1;
        }
        const std::map<std::string,std::string> object2DatabaseMap;     // This special mapping is not implemented
                                                                        // in the current version of the conversion procedure.
        CdbRooRoBootRecordR bootRecord( originH->name( ).head( ),
                                        originH->id( ),
                                        "<local>::<recent>",
                                        database2LocationMap,
                                        object2DatabaseMap );
        const Int_t numBytes = boofFileDatabaseFile.WriteObject( &bootRecord,
                                                                 CdbRooRoBootRecordR::PersistentObjectName( ));
        if( numBytes <= 0 ) {
            cerr << "failed to save the boot record '" << CdbRooRoBootRecordR::PersistentObjectName( ) << "' in the file: " << boofFileDatabaseFile.GetName( ) << endl;
            return 1;
        }

    }

   // Done.

    return 0;
}

/////////////
// Helpers //
/////////////

namespace {

    CdbStatus
    convertConfigCollection( CdbRooRoConfigCollectionR*&             theOutputConfigPtr,
                             const BdbRef(CdbBdbSConfigCollectionP)& theInputConfigCollectionRef )
    {
        const char* errorStr = "::convertConfigCollection() - ERROR.";

        CdbStatus result = CdbStatus::Error;

        if( BdbIsNull(theInputConfigCollectionRef)) {
            theOutputConfigPtr = 0;
            return CdbStatus::Success;
        }
        theOutputConfigPtr = new CdbRooRoConfigCollectionR( );

        CdbBdbSConfigCollectionP::IteratorOfIntervals itr = theInputConfigCollectionRef->iterator( );
        while( itr.next( )) {

            CdbBdbSConfigInterval iVal = itr.value( );

            CdbRooRoConfigElementR element( iVal.value.revision,
                                            iVal.value.partition,
                                            iVal.value.useRevision );

            CdbRooRoConfigInterval interval( element,
                                             iVal.begin,
                                             iVal.end );

            if( CdbStatus::Success != ( result = theOutputConfigPtr->insert( interval ))) {
                cerr << errorStr << endl
                     << "    Failed to isert the interval into the output collection." << endl
                     << "        INTERVAL: " << interval << endl;
                return result;
            }
        }
        return CdbStatus::Success;
    }

    CdbStatus
    convertFoldersAndConditions( CdbRooRoViewR&                   theView,
                                 const BdbHandle(CdbBdbSFolderP)& theFolderH,
                                 const CdbPathName&               theCurrentPathName )
    {
        const char* errorStr = "::convertFoldersAndConditions() - ERROR.";

        assert( !BdbIsNull(theFolderH));

        CdbStatus result = CdbStatus::Error;

        // Browse conditions in the input folder and copy them

        {
            CdbItr< const char* > itr;
            if( CdbStatus::Success != ( result = theFolderH->conditionIterator( itr ))) {
                cerr << errorStr << endl
                     << "    Failed to obtain an iterator of condition names." << endl
                     << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl;
                return result;
            }
            while( itr.next( )) {

              // Find the input condition objct

                BdbHandle(CdbBdbSConditionAtFolderP) conditionH;
                {
                    BdbRef(CdbBdbSConditionAtFolderP) conditionRef;
                    if( CdbStatus::Success != ( result = theFolderH->findCondition( CdbPathName( itr.value( )),
                                                                                    conditionRef ))) {
                        cerr << errorStr << endl
                             << "    Failed to find the condition in the current folder." << endl
                             << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                             << "        CONDITION:      \"" << itr.value( ) << "\"" << endl;
                        return result;
                    }
                    conditionH = conditionRef;
                }
                assert( !BdbIsNull(conditionH));

              // Convert the configuration collection (if any)
              //
              // NOTE: The null input collection is considered a normal result since some conditions
              //       may be used through a default collection of their parent view.

                CdbRooRoConfigCollectionR* configPtr = 0;

                if( CdbStatus::Success != ( result = ::convertConfigCollection( configPtr,
                                                                                conditionH->config( )))) {
                    cerr << errorStr << endl
                         << "    Failed to convert the configuration collection of the condition in the current folder." << endl
                         << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                         << "        CONDITION:      \"" << itr.value( ) << "\"" << endl;
                    return result;
                }

              // Create a new condition by its full path in the output view

                if( CdbStatus::Success != ( result = theView.createCondition( theCurrentPathName.toString( ),
                                                                              conditionH->name( ).head( ),
                                                                              conditionH->description( ).head( ),
                                                                              conditionH->created( ),
                                                                              CdbRooRoIdR( conditionH->id( ).origin,
                                                                                           conditionH->id( ).local ),
                                                                              configPtr ))) {
                    cerr << errorStr << endl
                         << "    Failed to create the condition in the output view." << endl
                         << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                         << "        CONDITION:      \"" << itr.value( ) << "\"" << endl;
                    return result;
                }
            }
        }

        // Browse folders in the input folder and copy them

        {
            CdbItr< const char* > itr;
            if( CdbStatus::Success != ( result = theFolderH->folderIterator( itr ))) {
                cerr << errorStr << endl
                     << "    Failed to obtain an iterator of folder names." << endl
                     << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl;
                return result;
            }
            while( itr.next( )) {

              // Find the input folder objct

                BdbHandle(CdbBdbSFolderP) folderH;
                {
                    BdbRef(CdbBdbSFolderP) folderRef;
                    if( CdbStatus::Success != ( result = theFolderH->findFolder( CdbPathName( itr.value( )),
                                                                                 folderRef ))) {
                        cerr << errorStr << endl
                             << "    Failed to find the folder in the current folder." << endl
                             << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                             << "        FOLDER:         \"" << itr.value( ) << "\"" << endl;
                        return result;
                    }
                    folderH = folderRef;
                }
                assert( !BdbIsNull(folderH));

              // Create a new folder by its full path in the output view

                CdbPathName folderFullPathName = theCurrentPathName + itr.value( );

                if( CdbStatus::Success != ( result = theView.createFolder( folderFullPathName.toString( ),
                                                                           folderH->description( ).head( ),
                                                                           folderH->created( )))) {
                    cerr << errorStr << endl
                         << "    Failed to create the folder in the output view." << endl
                         << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                         << "        FOLDER:         \"" << itr.value( ) << "\"" << endl;
                    return result;
                }

              // Proceed down to the found folder

                if( CdbStatus::Success != ( result = ::convertFoldersAndConditions( theView,
                                                                                    folderH,
                                                                                    folderFullPathName ))) {
                    cerr << errorStr << endl
                         << "    Failed to convert folders starting from the found one." << endl
                         << "        CURRENT FOLDER: \"" << theCurrentPathName << "\"" << endl
                         << "        FOLDER:         \"" << itr.value( ) << "\"" << endl;
                    return result;
                }
            }
        }
        return CdbStatus::Success;
    }

    CdbStatus
    convertPhysicalConditions( const CdbCPtr<TFile>&              theFilePtr,
                               const BdbHandle(CdbBdbSRegistryP)& theMasterRegistryH,
                               const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                               const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertPhysicalConditions() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        CdbRooRoConditionCollectionR conditionCollection;

        BdbHandle(CdbBdbSConditionCollectionP) conditionCollectionH = theInRegistryH->conditionCollection( );

        CdbBdbSConditionCollectionP::IteratorOfIdentifiers itr = conditionCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the input condition

            d_UShort conditionId = itr.value( );

            BdbHandle(CdbBdbSConditionP) conditionH;
            {
                BdbRef(CdbBdbSConditionP) conditionRef;
                if( CdbStatus::Success != ( result = conditionCollectionH->find( conditionId,
                                                                                 conditionRef ))) {
                    cerr << errorStr << endl
                         << "    A condition with ID = " << conditionId << " was not found." << endl;
                    return result;
                }
                conditionH = conditionRef;
            }

          // See if we need to exclude it

            if( conditionH->isPartitionable( )) {
                if( excludePartitionableFlag ) continue;
            } else {
                if( excludeRegularFlag ) continue;
            }

          // Create the output one and put into the collection

            CdbRooRoConditionR condition( conditionH->name( ).head( ),
                                          conditionH->id( ),
                                          conditionH->isPartitionable( ),
                                          conditionH->cluster( )->id( ),
                                          conditionH->created( ),
                                          conditionH->description( ).head( ));

            if( CdbStatus::Success != ( result = conditionCollection.add( condition ))) {
                cerr << errorStr << endl
                     << "    Failed to add a condition with ID = " << conditionId << " to a collection." << endl;
                return result;
            }
            if( verboseModeFlag ) {
                condition.dump( cout );
                cout << endl;
            }

          // Find all relevant MetaData objects and convert them too

            if( !conditionH->isPartitionable( )) {

                if( CdbStatus::Success != ( result = ::convertMetaDataOfRegularCondition( conditionH,
                                                                                          theMasterRegistryH,
                                                                                          theInRegistryH->originId( ),
                                                                                          theOutRegistryDescriptor ))) {
                    cerr << errorStr << endl
                         << "    Failed to convert MetaDaa of a 'regular' condition with ID = " << conditionId << "." << endl;
                    return result;
                }

            } else {

                if( CdbStatus::Success != ( result = ::convertMetaDataOfPartitionableCondition( conditionH,
                                                                                                theMasterRegistryH,
                                                                                                theOutRegistryDescriptor ))) {
                    cerr << errorStr << endl
                         << "    Failed to convert MetaDaa of a 'partitionable' condition with ID = " << conditionId << "." << endl;
                    return result;
                }
            }
        }

      // Save the output collection of 'physical' conditions

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "ConditionCollection" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != conditionCollection.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                           theOutRegistryDescriptor.originId( )),
                                                               theFilePtr,
                                                               bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written." << endl
                 << "    Perhaps it's not been properly instrumented for the streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

    CdbStatus
    convertViews( const CdbCPtr<TFile>&              theFilePtr,
                  const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                  const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertViews() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        CdbRooRoViewCollectionR viewCollection;

        BdbHandle(CdbBdbSViewCollectionP) viewCollectionH = theInRegistryH->viewCollection( );
        assert( !BdbIsNull(viewCollectionH));

        CdbBdbSViewCollectionP::IteratorOfIdentifiers itr = viewCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the input condition

            d_UShort viewId = itr.value( );

            BdbHandle(CdbBdbSViewP) viewH;
            {
                BdbRef(CdbBdbSViewP) viewRef;
                if( CdbStatus::Success != ( result = viewCollectionH->find( viewId,
                                                                            viewRef ))) {
                    cerr << errorStr << endl
                         << "    A view with ID = " << viewId << " was not found." << endl;
                    return result;
                }
                viewH = viewRef;
            }

          // Convert the configuration collection (if any)
          //
          // NOTE: The null input collection is considered a normal result since conditions
          //       normally have their own conditions.

            CdbRooRoConfigCollectionR* configPtr = 0;

            if( CdbStatus::Success != ( result = ::convertConfigCollection( configPtr,
                                                                            viewH->defaultConfig( )))) {
                cerr << errorStr << endl
                     << "    Failed to convert the default configuration collection of a view with ID = " << viewId << endl;
                return result;
            }

          // Create the output view

            CdbRooRoViewR view( viewH->name( ).head( ),
                                viewH->id( ),
                                viewH->description( ).head( ),
                                viewH->created( ),
                                viewH->minValidity( ),
                                viewH->maxValidity( ),
                                configPtr );

          // Proceed down to folders
          //
          // NOTES:
          //
          //   (1) The "/" folder must already exist in the output collection as it's created
          //       automatically at the same time the view gets created.
          //
          //   (2) The path name would be automatically extended as the recursive copying
          //       procedure of folders and conditions will keep going.

            BdbHandle(CdbBdbSFolderP) rootFolderH = viewH->rootFolder( );
            assert( !BdbIsNull(rootFolderH));

            if( CdbStatus::Success != ( result = ::convertFoldersAndConditions( view,
                                                                                rootFolderH,
                                                                                CdbPathName( rootFolderH->name( ).head( ))))) {
                cerr << errorStr << endl
                     << "    Failed to convert folders starting from \"/\"" << endl
                     << "    when copying view \"" << view.name( ) << "\"" << endl;
                return result;
            }
                                                        
          // Finally, store the whole graph in the collection

            if( CdbStatus::Success != ( result = viewCollection.add( view ))) {
                cerr << errorStr << endl
                     << "    Failed to add a view to a collection" << endl;
                return result;
            }
            if( verboseModeFlag ) {
                view.dump( cout );
                cout << endl;
            }
        }

      // Save the output collection

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "ViewCollection" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != viewCollection.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                      theOutRegistryDescriptor.originId( )),
                                                          theFilePtr,
                                                          bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written. Perhaps it's not" << endl
                 << "    been properly instrumented for streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

    CdbStatus
    convertRegularClusters( const CdbCPtr<TFile>&              theFilePtr,
                            const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                            const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertRegularClusters() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        CdbRooRoRClusterCollectionR clusterCollection;

        BdbHandle(CdbBdbSClusterCollectionP) clusterCollectionH = theInRegistryH->clusterCollection( );
        assert( !BdbIsNull(clusterCollectionH));

        CdbBdbSClusterCollectionP::IteratorOfIdentifiers itr = clusterCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the input condition

            d_UShort clusterId = itr.value( );

            BdbHandle(CdbBdbSSimpleClusterP) clusterH;
            {
                BdbRef(CdbBdbSSimpleClusterP) clusterRef;
                if( CdbStatus::Success != ( result = clusterCollectionH->find( clusterId,
                                                                               clusterRef ))) {
                    cerr << errorStr << endl
                         << "    A 'regular' cluster with ID = " << clusterId << " was not found." << endl;
                    return result;
                }
                clusterH = clusterRef;
            }

          // Prepare a vector with increments converted from the input cluster

            std::vector<CdbRooRoIncrementR> vectorOfIncrements;
            {
                d_UShort numIncrements = clusterH->numIncrements( );
                for( d_UShort i = 0; i < numIncrements; ++i ) {

                    CdbBdbSIncrement inputIncrement;

                    if( CdbStatus::Success != ( result = clusterH->increment( i,
                                                                              inputIncrement ))) {
                        cerr << errorStr << endl
                             << "    An increment with ID = " << i << " was not found in a cluster with ID = " << clusterId << endl;
                        return result;
                    }

                    CdbRooRoIncrementR outputIncrement( inputIncrement.begin,
                                                        inputIncrement.end );

                    vectorOfIncrements.push_back( outputIncrement );
                }
            }

          // Create the output cluster and store it in the collection

            CdbRooRoRClusterR cluster( clusterH->id( ),
                                       clusterH->name( ).head( ),
                                       clusterH->created( ),
                                       clusterH->description( ).head( ),
                                       vectorOfIncrements );

            if( CdbStatus::Success != ( result = clusterCollection.add( cluster ))) {
                cerr << errorStr << endl
                     << "    Failed to add a 'regular' cluster to a collection" << endl;
                return result;
            }
            if( verboseModeFlag ) {
                cluster.dump( cout );
                cout << endl;
            }
        }

      // Save the output collection

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "RegularClusterCollection" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != clusterCollection.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                         theOutRegistryDescriptor.originId( )),
                                                             theFilePtr,
                                                             bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written. Perhaps it's not" << endl
                 << "    been properly instrumented for streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

    CdbStatus
    convertPartitionsLayout( const CdbCPtr<TFile>&              theFilePtr,
                             const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                             const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertPartitionsLayout() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        std::vector<CdbRooRoPartitionR> vectorOfPartitions;

        BdbHandle(CdbBdbSPartitionsLayoutP) partitionsLayoutH = theInRegistryH->partitionsLayout( );
        assert( !BdbIsNull(partitionsLayoutH));

        CdbItr<d_UShort> itr = partitionsLayoutH->iterator( );
        while( itr.next( )) {

            d_UShort partitionId = itr.value( );

          // Get the input partition

            BdbHandle(CdbBdbSPartitionP) partitionH;
            {
                BdbRef(CdbBdbSPartitionP) partitionRef;
                if( CdbStatus::Success != ( result = partitionsLayoutH->find( partitionId,
                                                                              partitionRef ))) {
                    cerr << errorStr << endl
                         << "    A partition with ID = " << partitionId << " was not found." << endl;
                    return result;
                }
                partitionH = partitionRef;
            }

          // Prepare a vector with increments converted from the input cluster

            std::vector<CdbRooRoIncrementR> vectorOfIncrements;
            {
                d_UShort numIncrements = partitionH->numIncrements( );
                for( d_UShort i = 0; i < numIncrements; ++i ) {

                    CdbBdbSIncrement inputIncrement;

                    if( CdbStatus::Success != ( result = partitionH->increment( i,
                                                                                inputIncrement ))) {
                        cerr << errorStr << endl
                             << "    An increment with ID = " << i << " was not found in a partition with ID = " << partitionId << endl;
                        return result;
                    }

                    CdbRooRoIncrementR outputIncrement( inputIncrement.begin,
                                                        inputIncrement.end );

                    vectorOfIncrements.push_back( outputIncrement );
                }
            }

          // Create the output cluster and store it in the collection

            CdbRooRoPartitionR partition( vectorOfIncrements,
                                          partitionH->id( ),
                                          partitionH->originId( ),
                                          partitionH->isInstantiated( ),
                                          partitionH->isClosed( ),
                                          partitionH->created( ),
                                          partitionH->modified( ),
                                          partitionH->cell( ).beginValidity,
                                          partitionH->cell( ).endValidity,
                                          partitionH->cell( ).beginInsertion,
                                          partitionH->cell( ).endInsertion,
                                          partitionH->description( ).head( ));

            vectorOfPartitions.push_back( partition );

            if( verboseModeFlag ) {
                partition.dump( cout );
                cout << endl;
            }
        }

      // Finally, create the output layout using the prepared vector of partitions

        CdbRooRoPartitionsLayoutR partitionsLayout( vectorOfPartitions );

      // Save the output collection

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "PartitionsLayout" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != partitionsLayout.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                        theOutRegistryDescriptor.originId( )),
                                                            theFilePtr,
                                                            bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written. Perhaps it's not" << endl
                 << "    been properly instrumented for streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

    CdbStatus
    convertOrigins( const CdbCPtr<TFile>&              theFilePtr,
                    const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                    const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertOrigins() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        CdbRooRoOriginCollectionR originCollection;

        BdbHandle(CdbBdbSOriginCollectionP) originCollectionH = theInRegistryH->originCollection( );
        assert( !BdbIsNull(originCollectionH));

        CdbBdbSOriginCollectionP::IteratorOfIdentifiers itr = originCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the input condition

            d_UShort originId = itr.value( );

            BdbHandle(CdbBdbSOriginP) originH;
            {
                BdbRef(CdbBdbSOriginP) originRef;
                if( CdbStatus::Success != ( result = originCollectionH->find( originId,
                                                                              originRef ))) {
                    cerr << errorStr << endl
                         << "    An origin with ID = " << originId << " was not found." << endl;
                    return result;
                }
                originH = originRef;
            }

          // Remap the origin type

            CdbRooRoOriginR::OriginType originType = CdbRooRoOriginR::FIRST;

            switch( originH->type( )) {

            case CdbBdbSOriginP::MASTER  : originType = CdbRooRoOriginR::MASTER;  break;
            case CdbBdbSOriginP::SLAVE   : originType = CdbRooRoOriginR::SLAVE;   break;
            case CdbBdbSOriginP::REPLICA : originType = CdbRooRoOriginR::REPLICA; break;
            case CdbBdbSOriginP::TEST    : originType = CdbRooRoOriginR::TEST;    break;

            default:

                cerr << errorStr << endl
                     << "    An unknown (to the current algorithm) origin type: " << originH->type( ) << endl;
                return CdbStatus::Error;
            }

          // Create the output one and put into the collection

            CdbRooRoOriginR origin( originH->name( ).head( ),
                                    originH->id( ),
                                    originType,
                                    originH->created( ),
                                    originH->description( ).head( ));

            if( CdbStatus::Success != ( result = originCollection.add( origin ))) {
                cerr << errorStr << endl
                     << "    Failed to add an origin to a collection" << endl;
                return result;
            }
            if( verboseModeFlag ) {
                origin.dump( cout );
                cout << endl;
            }
        }

      // Save the output collection

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "OriginCollection" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != originCollection.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                        theOutRegistryDescriptor.originId( )),
                                                            theFilePtr,
                                                            bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written. Perhaps it's not" << endl
                 << "    been properly instrumented for streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << "  ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

    CdbStatus
    convertPartitionableClusters( const CdbCPtr<TFile>&              theFilePtr,
                                  const BdbHandle(CdbBdbSRegistryP)& theInRegistryH,
                                  const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertPartitionableClusters() - ERROR.";

        assert( !BdbIsNull(theInRegistryH));

        CdbStatus result = CdbStatus::Error;

        CdbRooRoClusterCollectionR clusterCollection;

        BdbHandle(CdbBdbSClusterCollectionP) clusterCollectionH = theInRegistryH->pClusterCollection( );
        assert( !BdbIsNull(clusterCollectionH));

        CdbBdbSClusterCollectionP::IteratorOfIdentifiers itr = clusterCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the input condition

            d_UShort clusterId = itr.value( );

            BdbHandle(CdbBdbSClusterP) clusterH;
            {
                BdbRef(CdbBdbSClusterP) clusterRef;
                if( CdbStatus::Success != ( result = clusterCollectionH->find( clusterId,
                                                                               clusterRef ))) {
                    cerr << errorStr << endl
                         << "    A 'partitionable' cluster with ID = " << clusterId << " was not found." << endl;
                    return result;
                }
                clusterH = clusterRef;
            }

          // Create the output cluster and store it in the collection

            CdbRooRoClusterR cluster( clusterH->id( ),
                                      clusterH->name( ).head( ),
                                      clusterH->created( ),
                                      clusterH->description( ).head( ));

            if( CdbStatus::Success != ( result = clusterCollection.add( cluster ))) {
                cerr << errorStr << endl
                     << "    Failed to add a 'partitionable' cluster to a collection" << endl;
                return result;
            }
            if( verboseModeFlag ) {
                cluster.dump( cout );
                cout << endl;
            }
        }

      // Save the output collection

        const std::string persistentObjectName =
            CdbRooRoFileUtils::nameOfSystemContainer( theOutRegistryDescriptor.originId( ),
                                                      "PartitionableClusterCollection" );

        Int_t bytesWritten = 0;

        theFilePtr->Cd( "" );
        gDirectory->Cd( theFilePtr->GetPath( ));

        if( CdbStatus::Success != clusterCollection.storeAt( CdbRooRoCollectionAddressR::createRegistry( persistentObjectName,
                                                                                                         theOutRegistryDescriptor.originId( )),
                                                             theFilePtr,
                                                             bytesWritten )) {
            cerr << errorStr << endl
                 << "    New object couldn't be written. Perhaps it's not" << endl
                 << "    been properly instrumented for streaming." << endl;
            return CdbStatus::Error;
        }
        cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;

        return CdbStatus::Success;
    }

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

        CdbStatus result = CdbStatus::Error;

      // Ask a user which collection to convert

        d_UShort originId = 0;

        cout << endl
             << "Please, enter the parameters of a condition." << endl
             << endl;

        d_UShort conditionId = 0xFFFF;
        {
            cout << "  Condition identifier : "; cin >> conditionId;
        }
        bool isPartitionableFlag = false;
        {
            int yesOrNo;
            cout << "Is partitionable [0/1] : "; cin >> yesOrNo;
            isPartitionableFlag = ( 0 != yesOrNo );
        }
        d_UShort partitionId = 0xFFFF;
        {
            cout << "  Partition identifier : "; cin >> partitionId;
        }
        d_UShort clusterId = 0xFFFF;
        {
            cout << "    Cluster identifier : "; cin >> clusterId;
        }
        d_UShort incrementNumber = 0xFFFF;
        {
            cout << "      Increment number : "; cin >> incrementNumber;
        }
        std::string dbIdRange = "";
        {
            cout << "       DBID Range Name : "; cin >> dbIdRange;
        }
        cout << endl
             << "Just for the record, you've entered the following parameters:" << endl
             << endl
             << "  Condition identifier : "   << conditionId << endl
             << "      Is partitionable : "   << ( isPartitionableFlag ? "Yes" : "No" ) << endl
             << "  Partition identifier : "   << partitionId << endl
             << "    Cluster identifier : "   << clusterId << endl
             << "      Increment number : "   << incrementNumber << endl
             << "       DBID Range Name : \"" << dbIdRange << "\"" << endl
             << endl;

      // Find the input metadata collection 

        BdbHandle(CdbBdbSMetaDataP) mdH;
        {
            BdbRef(CdbBdbSMetaDataP) mdRef;
            if( CdbStatus::Success != ( result = CdbBdbSUtils::findMetaData( mdRef,
                                                                             conditionId,
                                                                             isPartitionableFlag,
                                                                             partitionId,
                                                                             clusterId,
                                                                             incrementNumber,
                                                                             dbIdRange ))) {
                cerr << errorStr << endl
                     << "    Failed to locate the input MetaData object." << endl;
                return result;
            }
            mdH = mdRef;
        }

      // Create the output file

        const CdbCPtr<TFile> databaseFilePtr = new TFile( "cdb_test.root",
                                                          "recreate" );
        if( databaseFilePtr->IsZombie( )) {
            cerr << errorStr << endl
                 << "    Failed to (re-)create the file: '" << databaseFilePtr->GetName( ) << "'." << endl;
            return CdbStatus::Error;
        }

      // Create a fake descriptor

        CdbRooRoRegistryDescriptorR registryDescriptor;

      // Proceed to the converion of the found MetaData object

        return ::convertMetaData( databaseFilePtr,
                                  mdH,
                                  originId,
                                  conditionId,
                                  clusterId,
                                  isPartitionableFlag,
                                  partitionId,
                                  incrementNumber,
                                  registryDescriptor );
    }

    CdbStatus
    convertMetaDataOfRegularCondition( const BdbHandle(CdbBdbSConditionP)& theConditionH,
                                       const BdbHandle(CdbBdbSRegistryP)&  theMasterRegistryH,
                                       d_UShort                            theOwnerOriginId,
                                       const CdbRooRoRegistryDescriptorR&  theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertMetaDataOfRegularCondition() - ERROR.";

        assert( !BdbIsNull(theConditionH));
        assert( !theConditionH->isPartitionable( ));
        assert( !BdbIsNull(theMasterRegistryH));

        CdbStatus result = CdbStatus::Error;

      // Get the DBID range of the condition's origin.

        std::string dbidRangeName;
        {
            BdbRef(CdbBdbSOriginP) originRef;
            if( CdbStatus::Success != ( result = theMasterRegistryH->originCollection( )->find( theOwnerOriginId,
                                                                                                originRef ))) {
                cerr << errorStr << endl
                     << "    An origin with ID=" << theOwnerOriginId << " was not found." << endl;
                return result;
            }
            dbidRangeName = originRef->dbidrange( ).head( );
        }

      // Browse through increments of a cluster the condition is a member of. Select only
      // those increments the condition is participating in.
      //
      // NOTE: The condition might be creaed after certain increments were already
      //       closed. That's why we shoul skip those ones closed before.

        BdbHandle(CdbBdbSSimpleClusterP) clusterH = (const BdbRef(CdbBdbSSimpleClusterP)&)theConditionH->cluster( );

        d_ULong numIncrements = clusterH->numIncrements( );
        for( d_ULong i = 0; i < numIncrements; ++i ) {

            if( verboseModeFlag ) {
                cout << "increment=" << i << endl;
            }

            CdbBdbSIncrement increment;
            if( CdbStatus::Success != ( result = clusterH->increment( i,
                                                                      increment ))) {
                cerr << errorStr << endl
                     << "    Failed to get increment " << i << " from cluster " << clusterH->id( ) << endl
                     << "    of condition \"" << theConditionH->name( ) << "\"" << endl;
                return result;
            }

          // Skip increments closed at or before the condition creation time.
          // These increments are not supposed to know about this condition.

            if( increment.end <= theConditionH->created( )) {
                if( verboseModeFlag ) {
                    cout << "skipping the increment=" << i << " since it was" << endl
                         << "    ended before the condition was created." << endl
                         << "    CONDITION CREATED: " << theConditionH->created( ) << endl
                         << "    PARTITION CLOSED : " << increment.end << endl;
                }
                continue;
            }

          // Find out the input "MetaData" object

            BdbHandle(CdbBdbSMetaDataP) mdH;
            {
                BdbRef(CdbBdbSMetaDataP) mdRef;
                if( CdbStatus::Success != ( result = CdbBdbSUtils::findMetaData( mdRef,
                                                                                 theConditionH->id( ),
                                                                                 theConditionH->isPartitionable( ),
                                                                                 0,
                                                                                 clusterH->id( ),
                                                                                 i,
                                                                                 dbidRangeName.c_str( )))) {
                    cerr << errorStr << endl
                         << "    Failed to find the metadata object for condition \"" << theConditionH->name( ) << "\"." << endl
                         << "    See details above." << endl;
                    return result;
                }
                mdH = mdRef;
            }

          // Open the output database name for that <condition_owner_origin>-<cluster>-<increment>.
          //
          // NOTE: The database file is open in the "update" mode, not "recreate". Otherwise
          //       we would destroy its previous contents.

            const CdbCPtr<TFile> databaseFilePtr = openMetadataDatabase( theOwnerOriginId,
                                                                         clusterH->id( ),
                                                                         0,
                                                                         i,
                                                                         theConditionH->isPartitionable( ));
            if( databaseFilePtr.isNull( )) return CdbStatus::Error;

          // Convert the object

            if( CdbStatus::Success != ( result = ::convertMetaData( databaseFilePtr,
                                                                    mdH,
                                                                    theOwnerOriginId,
                                                                    theConditionH->id( ),
                                                                    clusterH->id( ),
                                                                    theConditionH->isPartitionable( ),
                                                                    0,
                                                                    i,
                                                                    theOutRegistryDescriptor ))) {
                cerr << errorStr << endl
                     << "    Failed to translate the metadata object for condition \"" << theConditionH->name( ) << "\"." << endl
                     << "    See details above." << endl;
                return result;
            }
        }
        return CdbStatus::Success;
    }

    CdbStatus
    convertMetaDataOfPartitionableCondition( const BdbHandle(CdbBdbSConditionP)& theConditionH,
                                             const BdbHandle(CdbBdbSRegistryP)&  theMasterRegistryH,
                                             const CdbRooRoRegistryDescriptorR&  theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertMetaDataOfPartitionableCondition() : ";

        assert( !BdbIsNull(theConditionH));
        assert( theConditionH->isPartitionable( ));
        assert( !BdbIsNull(theMasterRegistryH));

        CdbStatus result = CdbStatus::Error;

      // Get to the cluster.

        BdbHandle(CdbBdbSSimpleClusterP) clusterH = (const BdbRef(CdbBdbSSimpleClusterP)&)theConditionH->cluster( );

      // Get to the PartitinLayout at MASTER's Registry and iterate over
      // _instantiated_ partitions only.

        BdbHandle(CdbBdbSPartitionsLayoutP) pLayoutH = theMasterRegistryH->partitionsLayout( );

        CdbItr< d_UShort > pItr = pLayoutH->iterator( );
        while( pItr.next( )) {

            d_UShort partitionId = pItr.value( );

            if( verboseModeFlag ) {
                cout << "partition ID=" << partitionId << endl;
            }

            BdbHandle( CdbBdbSPartitionP ) partitionH;
            {
                BdbRef( CdbBdbSPartitionP ) partitionRef;
                if( CdbStatus::Success != ( result = pLayoutH->find( partitionId,
                                                                     partitionRef ))) {
                    cerr << errorStr << endl
                         << "   Failed to find the partition corresponding to ID=" << partitionId << endl;
                    return result;
                }
                partitionH = partitionRef;
            }

          // Skip this partition if it's not instantiated yet or if it's closed at the time
          // of before the current condition was created.

            if( !partitionH->isInstantiated( )) {
                if( verboseModeFlag ) {
                    cout << "skipping non-instantiated partition ID=" << partitionId << endl;
                }
                continue;
            }
            if( partitionH->isClosed( ) && ( partitionH->cell( ).endInsertion <= theConditionH->created( ))) {
                if( verboseModeFlag ) {
                    cout << "skipping closed partition ID=" << partitionId << " since it was" << endl
                         << "    closed before the condition was created." << endl
                         << "    CONDITION CREATED: " << theConditionH->created( ) << endl
                         << "    PARTITION CLOSED : " << partitionH->cell( ).endInsertion << endl;
                }
                continue;
            }

          // Get the DBID range of the partition's orgin.

            std::string dbidRangeName;
            {
                d_UShort originId = partitionH->originId( );

                BdbRef(CdbBdbSOriginP) originRef;
                if( CdbStatus::Success != ( result = theMasterRegistryH->originCollection( )->find( originId,
                                                                                                    originRef ))) {
                    cerr << errorStr << endl
                         << "    An origin with ID=" << originId << " was not found." << endl;
                    return result;
                }
                dbidRangeName = originRef->dbidrange( ).head( );
            }

          //  Browse through increments

            d_ULong numIncrements = partitionH->numIncrements( );
            for( d_ULong i = 0; i < numIncrements; ++i ) {

                if( verboseModeFlag ) {
                    cout << "increment=" << i << endl;
                }

                CdbBdbSIncrement increment;
                if( CdbStatus::Success != ( result = partitionH->increment( i,
                                                                            increment ))) {
                    cerr << errorStr << endl
                         << "    Failed to get increment " << i << " from partition " << partitionId << endl
                         << "    of condition \"" << theConditionH->name( ) << "\"" << endl;
                    return result;
                }

              // Skip increments closed at or before the condition creation time.
              // These increments are not supposed to know about this condition.

                if( increment.end <= theConditionH->created( )) {
                    if( verboseModeFlag ) {
                        cout << "skipping the increment=" << i << " since it was" << endl
                         << "    ended before the condition was created." << endl
                         << "    CONDITION CREATED: " << theConditionH->created( ) << endl
                         << "    PARTITION CLOSED : " << increment.end << endl;
                    }
                    continue;
                }

              // Find out the input "MetaData" object

                BdbHandle(CdbBdbSMetaDataP) mdH;
                {
                    BdbRef(CdbBdbSMetaDataP) mdRef;
                    if( CdbStatus::Success != ( result = CdbBdbSUtils::findMetaData( mdRef,
                                                                                     theConditionH->id( ),
                                                                                     theConditionH->isPartitionable( ),
                                                                                     partitionId,
                                                                                     clusterH->id( ),
                                                                                     i,
                                                                                     dbidRangeName ))) {
                        cerr << errorStr << endl
                             << "    Failed to find the metadata object for condition \"" << theConditionH->name( ) << "\"." << endl
                             << "    See details above." << endl;
                        return result;
                    }
                    mdH = mdRef;
                }

              // Open the output database name for that <partition_owner_origin>-<cluster>-<partition>-<increment>.
              //
              // NOTE: The database file is open in the "update" mode, not "recreate". Otherwise
              //       we would destroy its previous contents.

                const CdbCPtr<TFile> databaseFilePtr = openMetadataDatabase( partitionH->originId( ),
                                                                             clusterH->id( ),
                                                                             partitionH->id( ),
                                                                             i,
                                                                             theConditionH->isPartitionable( ));
                if( databaseFilePtr.isNull( )) return CdbStatus::Error;

              // Convert the object

                if( CdbStatus::Success != ( result = ::convertMetaData( databaseFilePtr,
                                                                        mdH,
                                                                        partitionH->originId( ),
                                                                        theConditionH->id( ),
                                                                        clusterH->id( ),
                                                                        theConditionH->isPartitionable( ),
                                                                        partitionId,
                                                                        i,
                                                                        theOutRegistryDescriptor ))) {
                    cerr << errorStr << endl
                         << "    Failed to translate the metadata object for condition \"" << theConditionH->name( ) << "\"." << endl
                         << "    See details above." << endl;
                    return result;
                }
            }
        }
        return CdbStatus::Success;
    }

    CdbStatus
    convertMetaData( const CdbCPtr<TFile>&              theFilePtr,
                     const BdbHandle(CdbBdbSMetaDataP)& theInMetaDataH,
                     UShort_t                           theOriginId,
                     UShort_t                           theConditionId,
                     UShort_t                           theClusterId,
                     bool                               isPartitionableFlag,
                     UShort_t                           thePartitionId,
                     UShort_t                           theIncrementId,
                     const CdbRooRoRegistryDescriptorR& theOutRegistryDescriptor )
    {
        const char* errorStr = "::convertMetaData() - ERROR.";

        assert( !BdbIsNull(theInMetaDataH));

        CdbStatus result = CdbStatus::Error;

      // Translate the collection of 'original intervals

        CdbRooRoOiCollectionR originalCollection;
        if( CdbStatus::Success != ( result = ::translateOriginalCollection( originalCollection,
                                                                            theInMetaDataH->originaIntervalsCollection( ),
                                                                            theOriginId,
                                                                            theConditionId,
                                                                            theClusterId,
                                                                            isPartitionableFlag,
                                                                            thePartitionId,
                                                                            theIncrementId ))) {
            cerr << errorStr << endl
                 << "    Failed to translate a collection of the 'original' intervals." << endl;
            return result;
        }

      // Iterate over known revisions and translate them into a collection of revisions

        CdbRooRoRevCollectionR revisionCollection;
        {
            CdbItr<BdbTime> itr;
            if( CdbStatus::Success != ( result = theInMetaDataH->revisionIdIterator( itr ))) {
                cerr << errorStr << endl
                     << "    Failed to set up an iterator of revision identifiers." << endl;
                return result;
            }
            while( itr.next( )) {

                BdbTime revisionId = itr.value( );

              // Find the input revision

                BdbHandle(CdbBdbSRevisionP) theInRevisionH;
                {
                    BdbRef(CdbBdbSRevisionP) revisionRef;
                    if( CdbStatus::Success != ( result = theInMetaDataH->findRevision( revisionId,
                                                                                       revisionRef ))) {
                        cerr << errorStr << endl
                            << "    Failed to find a revision with ID = " << CdbTimeUtils::time2string( revisionId ) << endl;
                        return result;
                    }
                    theInRevisionH = revisionRef;
                }

              // Translate the revision

                CdbRooRoRevisionR revision;
                if( CdbStatus::Success != ( result = ::translateRevision( revision,
                                                                          theInRevisionH ))) {
                    cerr << errorStr << endl
                         << "    Failed to translate a revision with ID = " << CdbTimeUtils::time2string( revisionId ) << endl;
                    return result;
                }

              // Put it into the output collection of revisions

                if( CdbStatus::Success != ( result = revisionCollection.insert( revision ))) {
                    cerr << errorStr << endl
                         << "    Failed to insert a revision with ID = " << CdbTimeUtils::time2string( revisionId ) << endl
                         << "    into the output collection." << endl;
                    return result;
                }
            }
        }

      // Create the output MetaData object

        CdbRooRoMetaDataR metadata( theInMetaDataH->minValidity( ),
                                    theInMetaDataH->maxValidity( ),
                                    theInMetaDataH->minInsertion( ),
                                    theInMetaDataH->maxInsertion( ),
                                    theInMetaDataH->modified( ),
                                    originalCollection,
                                    revisionCollection );

        if( verboseModeFlag ) {
            cout << endl;
            metadata.dump( cout );
            cout << endl;
        }

      // Build the name of a persistent object for the "MetaData" and store it

        {
            const std::string persistentObjectName =
                CdbRooRoFileUtils::nameOfMetaDataContainer( theConditionId,
                                                            isPartitionableFlag,
                                                            thePartitionId,
                                                            theClusterId,
                                                            theIncrementId,
                                                            theOriginId );

            Int_t bytesWritten = 0;

            theFilePtr->Cd( "" );
            gDirectory->Cd( theFilePtr->GetPath( ));

            CdbRooRoCollectionAddressR collectionAddress;
            {
                if( isPartitionableFlag ) {
                    collectionAddress = CdbRooRoCollectionAddressR::createPartitionable( persistentObjectName,
                                                                                         theOriginId,
                                                                                         theClusterId,
                                                                                         thePartitionId,
                                                                                         theIncrementId );
                } else {
                    collectionAddress = CdbRooRoCollectionAddressR::createRegular( persistentObjectName,
                                                                                   theOriginId,
                                                                                   theClusterId,
                                                                                   theIncrementId );
                }
            }
            if( CdbStatus::Success != metadata.storeAt( collectionAddress,
                                                        theFilePtr,
                                                        bytesWritten )) {
                cerr << errorStr << endl
                     << "    New object couldn't be written." << endl
                     << "    Perhaps it's not been properly instrumented for the streaming." << endl;
                return CdbStatus::Error;
            }
            cout << theFilePtr->GetName( ) << "::" << theOutRegistryDescriptor.originName( ) << "::" << persistentObjectName << " ** " << bytesWritten << " bytes written **" << endl;
        }
        return CdbStatus::Success;
    }

  // A generator of locations of ID-s for user-defined "payload".
  //
  // The corresponding objects will be produced during the second path of the conversion
  // run by users.
  //
  // GENERATOR DESCRIPTION:
  //
  //   1. Containers may only have objects of the same (final) persistent type.
  //
  //      1.1 It's assumed that there is one-to-one correspondance between Objectivity/DB
  //          and ROOT I/O persistent classes.
  //
  //      1.2 The object index varies from 0 up to 0xFFFE.
  //
  //   2. When a container is full then a new one gets allocated.
  //
  //   3. There is a variable with an ID of the "next available (for allocation) container"
  //      It gets incremented every time a new container is allocated.
  //
  //   4. There is a map with the current mapping from an input class name onto a pair
  //      of the currently allocated (for the class) container and the last index used
  //      of an output "payload" object "stored" (the one which would be actually stored
  //      during the second phase of the conversion).
  //
  //      4.1 The object index varies from 0 up to 0xFFFE.
  //
  //   5. When no more containers is available then the current function will
  //      exit with the error status.
  //
  // The class defined below assists in producing the location information for
  // objects given their class names as keys. The class is implemented as an iterator
  // with an internal state to be used in the right context.
  //
  // Here is how it should be used:
  //
  //    LocationGenerator generator;
  //
  //    if( !generator.tryNextFor( )) {
  //        cerr << "error: an overflow in the current conditions. Not able to store" << endl
  //             << "       more than 2^32 objects." << endl;
  //        ...
  //    }
  //    UShort_t currentContainer   = generator.container( );
  //    UShort_t currentObjectIndex = generator.index    ( );
  //    ..
  //
  // USAGE NOTES:
  //
  //    1. If no single call to the "LocationGenerator::tryNextFor()" method is made
  //       then the current context would be invalid.
  //
  //    2. Each successfull call to the "LocationGenerator::tryNextFor()" would setup
  //       proper context for the specified value of the key. Obviously, this value
  //       can't be used for other purposes.
  //
  //    3. If the "LocationGenerator::tryNextFor()" returns "false" then the current
  //       context of the generator object gets invalid.
  //

    struct Location {

        Location( UShort_t theCurrentContainerId = 0xFFFF,
                  UShort_t theCurrentObjectIndex = 0xFFFF) :
            container(theCurrentContainerId),
            index    (theCurrentObjectIndex)
        { }

        UShort_t container;
        UShort_t index;
    };

    class LocationGenerator {

    public:

        LocationGenerator( ) :
            _nextContainerId (0),
            _currentContainer(0xFFFF),
            _currentIndex    (0xFFFF)
        { }

        bool tryNextFor( const std::string& theKey )
        {
            std::map< std::string, ::Location >::iterator itr = _locations.find( theKey );
            if( itr == _locations.end( )) {

              // New key passed

                _currentContainer = _nextContainerId;
                _currentIndex     = 0;

                if( _currentContainer == 0xFFFF ) {
                    _currentIndex = 0xFFFF;
                    return false;
                }
                _locations[theKey] = ::Location( _currentContainer,
                                                 _currentIndex );

              // Give up one more container from the pool of spare ones. We're going to check
              // it on the next invocation of the current method if we've run out of containers.

                ++_nextContainerId;

            } else {

              // This key is already known

                ::Location location = (*itr).second;    // reuse the found value to save one lookup operation with map

                _currentContainer = location.container;
                _currentIndex     = ++location.index;

                if( location.index == 0xFFFF ) {

                  // Remove the key from the map and do all over again as if the key
                  // never existed (which would be absolutelly correct since a particular
                  // value of the key is associated with a container.

                    _locations.erase( theKey );

                    return tryNextFor( theKey );
                }

              // Update the map

                _locations[theKey] = location;

            }
            return true;
        }

        UShort_t container( ) const { return _currentContainer; }
        UShort_t index    ( ) const { return _currentIndex;     }

    private:

        UShort_t _nextContainerId;      // next available container identifier

        std::map< std::string, ::Location > _locations;     // the current mapping for recently used keys

        UShort_t _currentContainer;     // a cached value for the recently used key
        UShort_t _currentIndex;         // a cached value for the recently used key
    };

    CdbStatus
    translateOriginalCollection( CdbRooRoOiCollectionR&                 theOutCollection,
                                 const BdbHandle(CdbBdbSOiCollectionP)& theInCollectionH,
                                 UShort_t                               theOriginId,
                                 UShort_t                               theConditionId,
                                 UShort_t                               theClusterId,
                                 bool                                   isPartitionableFlag,
                                 UShort_t                               thePartitionId,
                                 UShort_t                               theIncrementId )
    {
        const char* errorStr = "::translateOriginalCollection() - ERROR.";

        assert( !BdbIsNull(theInCollectionH));

        CdbStatus result = CdbStatus::Error;

      // Prepare a generator of output object ID-s for user-defined "payload".

        ::LocationGenerator objectLocationGenerator;

      // Scan the input collection and fill a vector of intervals to be used
      // to initialize the output collection.

        std::vector< CdbRooRoOiR > vectorOfIntervals;
        {
            d_ULong numOriginalElements = theInCollectionH->size( );

            for( d_ULong idx = 1; idx < numOriginalElements; ++ idx ) {

                CdbBdbSOi oi;

                BdbTime beginDuration;  // will be filled by the call, but we're not going to use it
                BdbTime endDuration;    // -//-

                if( CdbStatus::Success != ( result = theInCollectionH->find( idx,
                                                                             oi,
                                                                             beginDuration,
                                                                             endDuration ))) {
                    cerr << errorStr << endl
                         << "    Failed to obtain an 'original' interval with IDX = " << idx << " from the collection." << endl;
                    return result;
                }

              // Create an output "original" object initialized with a preallocated ID of
              // a "payload" object. The corresponding object will be produced and stored
              // in the suggested (by ID) location during the second phase of the conversion.

                if( !objectLocationGenerator.tryNextFor( oi.object.typeName( ))) {

                    cerr << errorStr << endl
                         << "   An overflow in the current condition. The current implementation of" << endl
                         << "   the conversion procedure doesn't allo storing more than 2^32 'original'" << endl
                         << "   objects per MetaData." << endl
                         << "       ORIGIN ID        : " << theOriginId << endl
                         << "       CONDITION ID     : " << theConditionId << endl
                         << "       CLUSTER ID       : " << theClusterId << endl
                         << "       IS PARTITIONABLE : " << ( isPartitionableFlag ? "Yes" : "No" ) << endl
                         << "       PARTITION        : " << thePartitionId << endl
                         << "       INCREMENT        : " << theIncrementId << endl;

                    return CdbStatus::Error;
                }

                CdbRooRoObjectIdR outputPersistentObjectId;
                if( isPartitionableFlag ) {
                    outputPersistentObjectId = CdbRooRoObjectIdR::createPartitionable( theOriginId,
                                                                                       theConditionId,
                                                                                       theClusterId,
                                                                                       thePartitionId,
                                                                                       theIncrementId,
                                                                                       objectLocationGenerator.container( ),
                                                                                       objectLocationGenerator.index    ( ));
                } else {
                    outputPersistentObjectId = CdbRooRoObjectIdR::createRegular( theOriginId,
                                                                                 theConditionId,
                                                                                 theClusterId,
                                                                                 theIncrementId,
                                                                                 objectLocationGenerator.container( ),
                                                                                 objectLocationGenerator.index    ( ));
                }
                vectorOfIntervals.push_back( CdbRooRoOiR( oi.begin,
                                                          oi.end,
                                                          oi.inserted,
                                                          outputPersistentObjectId,
                                                          oi.object.sprint( )));
            }
        }

      // Create the output collection object

        theOutCollection = CdbRooRoOiCollectionR( vectorOfIntervals );

        if( verboseModeFlag ) {
            cout << endl;
            theOutCollection.dump( cout );
            cout << endl;
        }
        return CdbStatus::Success;
    }

    CdbStatus
    translateRevision( CdbRooRoRevisionR&                 theOutRevision,
                       const BdbHandle(CdbBdbSRevisionP)& theInRevisionH )
    {
        const char* errorStr = "::translateRevision() - ERROR.";

        assert( !BdbIsNull(theInRevisionH));

        CdbStatus result = CdbStatus::Error;

      // Get the collection of 'visible' intervals

        BdbHandle(CdbBdbSViCollectionP) viCollectionH = theInRevisionH->collection( );
        assert( !BdbIsNull(viCollectionH));

      // Scan the input collection and fill an output collection of 'visible' intervals
      // to be used to initialize the output revision.

        CdbRooRoViCollectionR viCollection;
        {
            CdbBdbSViCollectionP::IteratorOfIntervals itr = viCollectionH->iterator( );
            while( itr.next( )) {

                CdbBdbSVi iVal = itr.value( );

                if( CdbStatus::Success != ( result = viCollection.insert( CdbRooRoViR( iVal.value,
                                                                                       iVal.begin,
                                                                                       iVal.end )))) {
                    cerr << errorStr << endl
                         << "    Failed to insert an interval into the collection." << endl;
                    return result;
                }
            }
        }

      // Create the output revision object

        theOutRevision = CdbRooRoRevisionR( theInRevisionH->id( ),
                                            theInRevisionH->name( ).head( ),
                                            viCollection,
                                            theInRevisionH->created( ),
                                            theInRevisionH->description( ).head( ));
        if( verboseModeFlag ) {
            cout << endl;
            theOutRevision.dump( cout );
            cout << endl;
        }
        return CdbStatus::Success;
    }

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

        CdbStatus result = CdbStatus::Error;

      // Get the MASTER registry.

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != ( result = CdbBdbSRegistryP::findMaster( masterRegistryH ))) {
            cerr << errorStr << endl
                 << "    Failed to find the MASTER registry." << endl;
            return result;
        }

      // Iterate over known registries and analyses them one-by-one if the corresponding origins
      // are instantiated in the input database.

        BdbHandle(CdbBdbSOriginCollectionP) originCollectionH = masterRegistryH->originCollection( );

        CdbBdbSOriginCollectionP::IteratorOfIdentifiers itr = originCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the persistent registry object

            d_UShort conditionOwnerOriginId = itr.value( );

            BdbHandle(CdbBdbSOriginP) originH;
            {
                BdbRef(CdbBdbSOriginP) originRef;
                if( CdbStatus::Success != ( result = originCollectionH->find( conditionOwnerOriginId,
                                                                              originRef ))) {

                    cerr << errorStr << endl
                         << "    Failed to find an origin with ID = " << conditionOwnerOriginId << "." << endl
                         << "    The input database may not be properly initialized." << endl;
                    return result;
                }
                originH = originRef;
            }

          // Try to get the corresponding registry. If it doesn't exist (not instantiated)
          // then we will skip that registry.

            BdbHandle(CdbBdbSRegistryP) registryH;
            {
                if( CdbStatus::Success != ( result = CdbBdbSRegistryP::findByOrigin( registryH,
                                                                                     conditionOwnerOriginId ))) {
                    if( CdbStatus::NotFound == result ) {

                      // The origin is not currently instantiated in the input database. This may happen
                      // when the origin's data are yet to be imported.

                        continue;

                    } else {
                        cerr << errorStr << endl
                             << "    Failed to find a registry for origin ID = " << conditionOwnerOriginId << "." << endl;
                        return result;
                    }
                }
            }

          // Proceed to 'physical' conditions of the collection

            if( registryH->hasLocalCollections( )) {

                BdbHandle(CdbBdbSConditionCollectionP) conditionCollectionH = registryH->conditionCollection( );

                CdbBdbSConditionCollectionP::IteratorOfIdentifiers itr = conditionCollectionH->iterator_identifiers( );
                while( itr.next( )) {

                  // Get the input condition

                    d_UShort conditionId = itr.value( );

                    BdbHandle(CdbBdbSConditionP) conditionH;
                    {
                        BdbRef(CdbBdbSConditionP) conditionRef;
                        if( CdbStatus::Success != ( result = conditionCollectionH->find( conditionId,
                                                                                         conditionRef ))) {
                            cerr << errorStr << endl
                                 << "    A condition with ID = " << conditionId << " was not found." << endl;
                            return result;
                        }
                        conditionH = conditionRef;
                    }

                  // Find all relevant MetaData objects

                    if( conditionH->isPartitionable( )) {

                      // ** PARTITIONABLE **

                        if( excludePartitionableFlag ) continue;

                        BdbHandle(CdbBdbSSimpleClusterP) clusterH = (const BdbRef(CdbBdbSSimpleClusterP)&)conditionH->cluster( );

                      // Get to the PartitinLayout at MASTER's Registry and iterate over
                      // _instantiated_ partitions only.

                        BdbHandle(CdbBdbSPartitionsLayoutP) pLayoutH = masterRegistryH->partitionsLayout( );

                        CdbItr< d_UShort > pItr = pLayoutH->iterator( );
                        while( pItr.next( )) {

                            d_UShort partitionId = pItr.value( );

                            BdbHandle( CdbBdbSPartitionP ) partitionH;
                            {
                                BdbRef( CdbBdbSPartitionP ) partitionRef;
                                if( CdbStatus::Success != ( result = pLayoutH->find( partitionId,
                                                                                     partitionRef ))) {
                                    cerr << errorStr << endl
                                         << "   Failed to find the partition corresponding to ID=" << partitionId << endl;
                                    return result;
                                }
                                partitionH = partitionRef;
                            }

                          // Skip this partition if it's not instantiated yet or if it's closed at the time
                          // of before the current condition was created.

                            if( !partitionH->isInstantiated( )) continue;
                            if( partitionH->isClosed( ) && ( partitionH->cell( ).endInsertion <= conditionH->created( ))) continue;

                          // Get the DBID range of the partition's orgin.

                            std::string dbidRangeName;
                            {
                                BdbRef(CdbBdbSOriginP) originRef;
                                if( CdbStatus::Success != ( result = masterRegistryH->originCollection( )->find( partitionH->originId( ),
                                                                                                                 originRef ))) {
                                    cerr << errorStr << endl
                                         << "    An origin with ID=" << partitionH->originId( ) << " was not found." << endl;
                                    return result;
                                }
                                dbidRangeName = originRef->dbidrange( ).head( );
                            }

                          //  Browse through increments

                            d_ULong numIncrements = partitionH->numIncrements( );
                            for( d_ULong i = 0; i < numIncrements; ++i ) {

                                CdbBdbSIncrement increment;
                                if( CdbStatus::Success != ( result = partitionH->increment( i,
                                                                                            increment ))) {
                                    cerr << errorStr << endl
                                         << "    Failed to get increment " << i << " from partition " << partitionId << endl
                                         << "    of condition \"" << conditionH->name( ) << "\"" << endl;
                                    return result;
                                }

                              // Skip increments closed at or before the condition creation time.
                              // These increments are not supposed to know about this condition.

                                if( increment.end <= conditionH->created( )) continue;

                              // Find out the input "MetaData" object

                                BdbHandle(CdbBdbSMetaDataP) mdH;
                                {
                                    BdbRef(CdbBdbSMetaDataP) mdRef;
                                    if( CdbStatus::Success != ( result = CdbBdbSUtils::findMetaData( mdRef,
                                                                                                     conditionH->id( ),
                                                                                                     conditionH->isPartitionable( ),
                                                                                                     partitionId,
                                                                                                     clusterH->id( ),
                                                                                                     i,
                                                                                                     dbidRangeName ))) {
                                        cerr << errorStr << endl
                                             << "    Failed to find the metadata object for condition \"" << conditionH->name( ) << "\"." << endl
                                             << "    See details above." << endl;
                                        return result;
                                    }
                                    mdH = mdRef;
                                }

                              // Browse the 'original' collection

                                std::map<std::string,unsigned int> counters;

                                const BdbHandle(CdbBdbSOiCollectionP) collectionH = mdH->originaIntervalsCollection( );

                                d_ULong numOriginalElements = collectionH->size( );

                                for( d_ULong idx = 1; idx < numOriginalElements; ++idx ) {

                                    CdbBdbSOi oi;

                                    BdbTime beginDuration;  // will be filled by the call, but we're not going to use it
                                    BdbTime endDuration;    // -//-

                                    if( CdbStatus::Success != ( result = collectionH->find( idx,
                                                                                            oi,
                                                                                            beginDuration,
                                                                                            endDuration ))) {
                                        cerr << errorStr << endl
                                             << "    Failed to obtain an 'original' interval with IDX = " << idx << " from the collection." << endl;
                                        return result;
                                    }

                                    std::string typeName = oi.object.typeName( );

                                    unsigned int nObjectsOfThisClass = 0;
                                    if( 0 != counters.count( typeName )) nObjectsOfThisClass = counters[ typeName ];
                                    counters[ typeName ] = ++nObjectsOfThisClass;
                                }

                              // Print results

                                for( std::map<std::string,unsigned int>::const_iterator itr = counters.begin( );
                                                                                        itr != counters.end( );
                                                                                      ++itr ) {
                                    std::string typeName = (*itr).first;
                                    unsigned int nObjectsOfThisClass = (*itr).second;
                                    cout << conditionOwnerOriginId << "::" << conditionH->id( ) << " [" << partitionH->id( ) << "," << clusterH->id( ) << "," << i << "] \"" << typeName << "\" " << nObjectsOfThisClass << endl;
                                }
                            }
                        }

                    } else {

                      // ** REGULAR **

                        if( excludeRegularFlag ) continue;

                      // Get the DBID range of the condition's origin.

                        std::string dbidRangeName;
                        {
                            BdbRef(CdbBdbSOriginP) originRef;
                            if( CdbStatus::Success != ( result = masterRegistryH->originCollection( )->find( conditionOwnerOriginId,
                                                                                                             originRef ))) {
                                cerr << errorStr << endl
                                     << "    An origin with ID=" << conditionOwnerOriginId << " was not found." << endl;
                                return result;
                            }
                            dbidRangeName = originRef->dbidrange( ).head( );
                        }

                      // Browse through increments of a cluster the condition is a member of. Select only
                      // those increments the condition is participating in.
                      //
                      // NOTE: The condition might be creaed after certain increments were already
                      //       closed. That's why we shoul skip those ones closed before.

                        BdbHandle(CdbBdbSSimpleClusterP) clusterH = (const BdbRef(CdbBdbSSimpleClusterP)&)conditionH->cluster( );

                        d_ULong numIncrements = clusterH->numIncrements( );
                        for( d_ULong i = 0; i < numIncrements; ++i ) {

                            CdbBdbSIncrement increment;
                            if( CdbStatus::Success != ( result = clusterH->increment( i,
                                                                                      increment ))) {
                                cerr << errorStr << endl
                                     << "    Failed to get increment " << i << " from cluster " << clusterH->id( ) << endl
                                     << "    of condition \"" << conditionH->name( ) << "\"" << endl;
                                return result;
                            }

                          // Skip increments closed at or before the condition creation time.
                          // These increments are not supposed to know about this condition.

                            if( increment.end <= conditionH->created( )) continue;

                          // Find out the input "MetaData" object

                            BdbHandle(CdbBdbSMetaDataP) mdH;
                            {
                                BdbRef(CdbBdbSMetaDataP) mdRef;
                                if( CdbStatus::Success != ( result = CdbBdbSUtils::findMetaData( mdRef,
                                                                                                 conditionH->id( ),
                                                                                                 conditionH->isPartitionable( ),
                                                                                                 0,
                                                                                                 clusterH->id( ),
                                                                                                 i,
                                                                                                 dbidRangeName.c_str( )))) {
                                    cerr << errorStr << endl
                                         << "    Failed to find the metadata object for condition \"" << conditionH->name( ) << "\"." << endl
                                         << "    See details above." << endl;
                                    return result;
                                }
                                mdH = mdRef;
                            }

                          // Browse the 'original' collection

                            std::map<std::string,unsigned int> counters;

                            const BdbHandle(CdbBdbSOiCollectionP) collectionH = mdH->originaIntervalsCollection( );

                            d_ULong numOriginalElements = collectionH->size( );

                            for( d_ULong idx = 1; idx < numOriginalElements; ++idx ) {

                                CdbBdbSOi oi;

                                BdbTime beginDuration;  // will be filled by the call, but we're not going to use it
                                BdbTime endDuration;    // -//-

                                if( CdbStatus::Success != ( result = collectionH->find( idx,
                                                                                        oi,
                                                                                        beginDuration,
                                                                                        endDuration ))) {
                                    cerr << errorStr << endl
                                         << "    Failed to obtain an 'original' interval with IDX = " << idx << " from the collection." << endl;
                                    return result;
                                }

                                std::string typeName = oi.object.typeName( );

                                unsigned int nObjectsOfThisClass = 0;
                                if( 0 != counters.count( typeName )) nObjectsOfThisClass = counters[ typeName ];
                                counters[ typeName ] = ++nObjectsOfThisClass;
                            }

                          // Print results

                            for( std::map<std::string,unsigned int>::const_iterator itr = counters.begin( );
                                                                                    itr != counters.end( );
                                                                                  ++itr ) {
                                std::string typeName = (*itr).first;
                                unsigned int nObjectsOfThisClass = (*itr).second;
                                cout << conditionOwnerOriginId << "::" << conditionH->id( ) << " [" << 0 << "," << clusterH->id( ) << "," << i << "] \"" << typeName << "\" " << nObjectsOfThisClass << endl;
                            }
                        }
                    }
                }
            }
        }
        return CdbStatus::Success;
    }

    CdbStatus
    browseConditionsInFolder( d_UShort                      theOriginId,
                              d_UShort                      theViewId,
                              const CdbPathName&            theParentFolderPath,
                              const BdbRef(CdbBdbSFolderP)& theParentFolderRef  )
    {
        const char* errorStr = "::browseConditionsInFolder() - ERROR.";

        assert( theParentFolderPath.isAbsolute( ));
        assert( !BdbIsNull(theParentFolderRef));

        const BdbHandle(CdbBdbSFolderP) parentFolderH = theParentFolderRef;

        CdbStatus result = CdbStatus::Error;

      // Find all conditions directly attached to the parent folder

        {
            CdbItr< const char* > itr;
            if( CdbStatus::Success != ( result = parentFolderH->conditionIterator( itr ))) {
                cerr << errorStr << endl
                     << "    Failed to set up an iterator of conditions for folder \"" << theParentFolderPath.toString( ) << "\"." << endl;
                return result;
            }
            while( itr.next( )) {

                CdbPathName conditionPath( itr.value( ));

                BdbRef(CdbBdbSConditionAtFolderP) conditionRef;
                if( CdbStatus::Success != ( result = parentFolderH->findCondition( conditionPath,
                                                                                   conditionRef ))) {
                    cerr << errorStr << endl
                         << "    Failed to find the condition \"" <<  conditionPath.toString( ) << "\"" << endl
                         << "    in folder \"" << theParentFolderPath.toString( ) << "\"." << endl;
                    return result;
                }

                CdbBdbSId   id                = conditionRef->id( );
                CdbPathName conditionFullPath = theParentFolderPath + conditionPath;

                cout << theOriginId << "::" << theViewId << " " << id.origin << "::" << id.local << " \"" << conditionFullPath.toString( ) << "\"" << endl;
            }
        }

      // Find all sub-folders directly attached to the parent folder and proceed to them

        {
            CdbItr< const char* > itr;
            if( CdbStatus::Success != ( result = parentFolderH->folderIterator( itr ))) {
                cerr << errorStr << endl
                     << "    Failed to set up an iterator of sub-folders for folder \"" << theParentFolderPath.toString( ) << "\"." << endl;
                return result;
            }
            while( itr.next( )) {

                CdbPathName folderPath( itr.value( ));

                BdbRef(CdbBdbSFolderP) folderRef;
                if( CdbStatus::Success != ( result = parentFolderH->findFolder( folderPath,
                                                                                folderRef ))) {
                    cerr << errorStr << endl
                         << "    Failed to find the sub-folder \"" <<  folderPath.toString( ) << "\"" << endl
                         << "    in folder \"" << theParentFolderPath.toString( ) << "\"." << endl;
                    return result;
                }

                CdbPathName folderFullPath = theParentFolderPath + folderPath;

                if( CdbStatus::Success != ( result = ::browseConditionsInFolder( theOriginId,
                                                                                 theViewId,
                                                                                 folderFullPath,
                                                                                 folderRef ))) {
                    cerr << errorStr << endl
                         << "    Failed to browse conditions in the sub-folder \"" <<  folderPath.toString( ) << "\"" << endl
                         << "    of folder \"" << theParentFolderPath.toString( ) << "\"." << endl;
                    return result;
                }
            }
        }
        return CdbStatus::Success;
    }

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

        CdbStatus result = CdbStatus::Error;

      // Get the MASTER registry.

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != ( result = CdbBdbSRegistryP::findMaster( masterRegistryH ))) {
            cerr << errorStr << endl
                 << "    Failed to find the MASTER registry." << endl;
            return result;
        }

      // Iterate over known registries and analyses them one-by-one if the corresponding origins
      // are instantiated in the input database.

        BdbHandle(CdbBdbSOriginCollectionP) originCollectionH = masterRegistryH->originCollection( );

        CdbBdbSOriginCollectionP::IteratorOfIdentifiers itr = originCollectionH->iterator_identifiers( );
        while( itr.next( )) {

          // Get the persistent registry object

            d_UShort viewOwnerOriginId = itr.value( );

            BdbHandle(CdbBdbSOriginP) originH;
            {
                BdbRef(CdbBdbSOriginP) originRef;
                if( CdbStatus::Success != ( result = originCollectionH->find( viewOwnerOriginId,
                                                                              originRef ))) {

                    cerr << errorStr << endl
                         << "    Failed to find an origin with ID = " << viewOwnerOriginId << "." << endl
                         << "    The input database may not be properly initialized." << endl;
                    return result;
                }
                originH = originRef;
            }

          // Try to get the corresponding registry. If it doesn't exist (not instantiated)
          // then we will skip that registry.

            BdbHandle(CdbBdbSRegistryP) registryH;
            {
                if( CdbStatus::Success != ( result = CdbBdbSRegistryP::findByOrigin( registryH,
                                                                                     viewOwnerOriginId ))) {
                    if( CdbStatus::NotFound == result ) {

                      // The origin is not currently instantiated in the input database. This may happen
                      // when the origin's data are yet to be imported.

                        continue;

                    } else {
                        cerr << errorStr << endl
                             << "    Failed to find a registry for origin ID = " << viewOwnerOriginId << "." << endl;
                        return result;
                    }
                }
            }

          // Proceed to 'views' of this registry

            if( registryH->hasLocalCollections( )) {

                BdbHandle(CdbBdbSViewCollectionP) viewCollectionH = registryH->viewCollection( );

                CdbBdbSViewCollectionP::IteratorOfIdentifiers itr = viewCollectionH->iterator_identifiers( );
                while( itr.next( )) {

                  // Get the view

                    d_UShort viewId = itr.value( );

                    BdbHandle(CdbBdbSViewP) viewH;
                    {
                        BdbRef(CdbBdbSViewP) viewRef;
                        if( CdbStatus::Success != ( result = viewCollectionH->find( viewId,
                                                                                    viewRef ))) {
                            cerr << errorStr << endl
                                 << "    A view with ID = " << viewId << " was not found." << endl;
                            return result;
                        }
                        viewH = viewRef;
                    }

                  // Skip the "main" view as it's not supposed to have any valueable information

                    if( std::string( "main" ) == viewH->name( )) continue;

                  // Browse the view

                    if( CdbStatus::Success != ( result = ::browseConditionsInFolder( viewOwnerOriginId,
                                                                                     viewId,
                                                                                     CdbPathName( viewH->rootFolder( )->name( ).head( )),
                                                                                     viewH->rootFolder( )))) {
                        cerr << errorStr << endl
                             << "    Failed to browse conditions in the 'root' folder of a view with ID = " << viewId << "." << endl;
                        return result;
                    }
                }
            }
        }
        return CdbStatus::Success;
    }
}

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

---