---

CdbBdb2RooPayloadConversionFwk.cc

Go to the documentation of this file.

// File and Version Information:
//      $Id: CdbBdb2RooPayloadConversionFwk.cc,v 1.12 2005/11/11 02:07:41 gapon Exp $

/// The implementation of the CdbBdb2RooPayloadConversionFwk class.
/**
  * @see CdbBdb2RooPayloadConversionFwk
  */
#include "BaBar/BaBar.hh"

#include "CdbRooConversionFwk/CdbBdb2RooPayloadConversionFwk.hh"

#include "CdbBase/CdbId.hh"
#include "CdbBase/CdbDatabase.hh"
#include "CdbBase/CdbOrigin.hh"
#include "CdbBase/CdbPartition.hh"
#include "CdbBase/CdbCondition.hh"
#include "CdbBase/CdbTransaction.hh"
#include "CdbBase/CdbStringUtils.hh"

#include "BdbCond/BdbObject.hh"

#include "CdbBdb/CdbBdbSchemaUtils.hh"

#include "CdbBdbShared/CdbBdbShared.hh"

#include "CdbRoo/CdbRooObjectR.hh"
#include "CdbRoo/CdbRooObjectHasMovedR.hh"

#include "CdbRooReadonly/CdbRooReadonly.hh"
#include "CdbRooReadonly/CdbRooRoFileUtils.hh"
#include "CdbRooReadonly/CdbRooRoMetaDataR.hh"
#include "CdbRooReadonly/CdbRooRoOiCollectionR.hh"
#include "CdbRooReadonly/CdbRooRoObjectIdR.hh"

#include "CdbRooConversionFwk/CdbRooConverterBase.hh"
#include "CdbRooConversionFwk/CdbBdb2RooPayloadConvertersDict.hh"
#include "CdbRooConversionFwk/CdbRooConverterDict.hh"

#include <stdio.h>  // sscanf(), sprintf()

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

#include <string>
#include <vector>
#include <map>
#include <iostream>
using std::cin;
using std::cout;
using std::cerr;
using std::endl;

#include <fstream>
using std::ifstream;

#include <sstream>
using std::ostringstream;

#include "BbrStdUtils/Tokenize.hh"
#include "BbrStdUtils/String.hh"
using namespace babar::String;

namespace {

  /// Convert a vector into a string
  /**
    * USE & PARAMETERS:
    *
    *     1. The converted values will be separated by a single space in an output string
    *
    *     2. The "<<" operator is expected to be available for the elements of the vector.
    */
    template< class T >
    std::string
    vector2string( const std::vector<T>& theVector )
    {
        ostringstream result;
        for( typename std::vector<T>::const_iterator itr =  theVector.begin( );
                                                     itr != theVector.end( );
                                                   ++itr ) {
            result << (*itr) << " ";
        }
        return result.str( );
    }

  // A helper function

    bool translate_OID( ooRef(ooObj)&      theRef,
                        const std::string& theString )
    {
        assert( !theString.empty( ));

        ooId id;
        {
            int db;
            int cont;
            int page;
            int slot;

            if( 4 != sscanf( theString.c_str( ), "#%d-%d-%d-%d", &db, &cont, &page, &slot )) {
                if( 4 != sscanf( theString.c_str( ), "%d-%d-%d-%d", &db, &cont, &page, &slot )) {
                    return false;
                }
            }

            id.set_DB  ( db );
            id.set_OC  ( cont );
            id.set_page( page );
            id.set_slot( slot );
        }
        theRef = id;

        return true;
    }

  /// A supplementary helper class storing parameters of a MetaData

    struct CdbBdb2RooPayloadMetaDataDescriptor {

      /// Constructor for regular conversion

        CdbBdb2RooPayloadMetaDataDescriptor( UShort_t theOriginId,
                                             UShort_t theConditionId,
                                             bool     isPartitionableFlag,
                                             UShort_t thePartitionId,
                                             UShort_t theClusterId,
                                             UShort_t theIncrementId ) :
            originId       (theOriginId),
            conditionId    (1, theConditionId),
            isPartitionable(isPartitionableFlag),
            partitionId    (thePartitionId),
            clusterId      (theClusterId),
            incrementId    (theIncrementId)
        { }

      /// Constructor for merging conversion

        CdbBdb2RooPayloadMetaDataDescriptor( UShort_t                       theOriginId,
                                             const std::vector< UShort_t >& theConditionId,
                                             bool                           isPartitionableFlag,
                                             UShort_t                       thePartitionId,
                                             UShort_t                       theClusterId,
                                             UShort_t                       theIncrementId ) :
            originId       (theOriginId),
            conditionId    (theConditionId),
            isPartitionable(isPartitionableFlag),
            partitionId    (thePartitionId),
            clusterId      (theClusterId),
            incrementId    (theIncrementId)
        { }

        std::string toString( ) const
        {
            ostringstream result;
            result << originId << "::" << conditionId[0] << " [" << partitionId << "," << clusterId << "," << incrementId << "] ";
            return result.str( );
        }

        UShort_t                originId;
        std::vector< UShort_t > conditionId;
        bool                    isPartitionable;
        UShort_t                partitionId;
        UShort_t                clusterId;
        UShort_t                incrementId;
    };

  // A helper function

    void printMetDataParameters( UShort_t theOriginId,
                                 UShort_t theConditionId,
                                 bool     isPartitionableFlag,
                                 UShort_t thePartitionId,
                                 UShort_t theClusterId,
                                 UShort_t theIncrementId )
    {
        cout << "         Origin identifier : "   << theOriginId << endl
             << "      Condition identifier : "   << theConditionId << endl
             << "          Is partitionable : "   << ( isPartitionableFlag ? "Yes" : "No" ) << endl
             << "      Partition identifier : "   << thePartitionId << endl
             << "        Cluster identifier : "   << theClusterId << endl
             << "          Increment number : "   << theIncrementId << endl;
    }

  // Supplementary helper classes storing the context of a "payload" container (TTree/TBranch)

    struct CdbBdb2RooPayloadContainerContext {

        CdbBdb2RooPayloadContainerContext( const std::string&                             theTreeName,
                                           const std::string&                             theTreeDescription,
                                           const CdbCPtr<CdbRooConverterBase>& theConverterPtr ) :
            tree( theTreeName.c_str( ),
                  theTreeDescription.c_str( )),
            converterPtr(theConverterPtr)
        { }

        void add( const BdbRef(BdbObject)& theInputLegacyObjectRef,
                  const CdbRooRoObjectIdR& theOutputObjectId )
        {
             inObjects.push_back( theInputLegacyObjectRef );
            outObjects.push_back( theOutputObjectId );
        }

      // Data members

        TTree tree;

        CdbCPtr<CdbRooConverterBase> converterPtr;

        std::vector< BdbRef(BdbObject) >  inObjects;
        std::vector< CdbRooRoObjectIdR > outObjects;
    };

    struct CdbBdb2RooPayloadMainContainerContext {

        CdbBdb2RooPayloadMainContainerContext( const std::string&                             theTreeName,
                                               const std::string&                             theTreeDescription,
                                               const CdbCPtr<CdbRooConverterBase>& theConverterPtr ) :
            tree( theTreeName.c_str( ),
                  theTreeDescription.c_str( )),
            converterPtr(theConverterPtr)
        { }

            void add( const std::vector< BdbRef(BdbObject) >& theInputLegacyObjectRef,
                      const CdbRooRoObjectIdR&                theOutputObjectId )
        {
             inObjects.push_back( theInputLegacyObjectRef );
            outObjects.push_back( theOutputObjectId );
        }

      // Data members

        TTree tree;

        CdbCPtr<CdbRooConverterBase> converterPtr;

        std::vector< std::vector< BdbRef(BdbObject) > >  inObjects;
        std::vector< CdbRooRoObjectIdR >                 outObjects;
    };
    struct CdbBdb2RooPayloadExtraContainerContext {

        CdbBdb2RooPayloadExtraContainerContext( const std::string&                             theTreeName,
                                                const std::string&                             theTreeDescription,
                                                const CdbCPtr<CdbRooConverterBase>& theConverterPtr ) :
            tree( theTreeName.c_str( ),
                  theTreeDescription.c_str( )),
            converterPtr(theConverterPtr)
        { }

        void add( const BdbRef(BdbObject)& theInputLegacyObjectRef,
                  const CdbRooRoObjectIdR& theOutputObjectId,
                  const CdbRooRoObjectIdR& theMainObjectId )
        {
              inObjects.push_back( theInputLegacyObjectRef );
             outObjects.push_back( theOutputObjectId );
            mainObjects.push_back( theMainObjectId );
        }

      // Data members

        TTree tree;

        CdbCPtr<CdbRooConverterBase> converterPtr;

        std::vector< BdbRef(BdbObject) >   inObjects;
        std::vector< CdbRooRoObjectIdR >  outObjects;
        std::vector< CdbRooRoObjectIdR > mainObjects;
    };

    bool translateNextLine( const std::string&    theLine,
                            const CdbDatabasePtr& theDatabasePtr,
                            UShort_t&             theOriginId,
                            UShort_t&             theConditionId,
                            bool&                 isPartitionableFlag,
                            UShort_t&             thePartitionId,
                            UShort_t&             theClusterId,
                            UShort_t&             theIncrementId,
                            std::string&          theInputClassName )
    {
        const char* errorStr = "CdbBdb2RooPayloadConversionFwk::<anonymous>::translateNextLine() - ERROR.";

        Tokenize tokenize( theLine );

        std::string  physicalConditionId = tokenize( );
        std::string        placementInfo = tokenize( );
        std::string spaceFilledSeparator = tokenize( "\"" );
        std::string       inputClassName = tokenize( "\"" );

        if(  physicalConditionId.empty( ) ||
                   placementInfo.empty( ) ||
            spaceFilledSeparator.empty( ) ||
                  inputClassName.empty( )) {

            cerr << errorStr << endl
                 << "    The line read from the definition file has an illegal format." << endl
                 << "\"" << theLine << "\"" << endl;
            return false;
        }

        UShort_t originId    = 0;
        UShort_t conditionId = 0;
        {
            unsigned origin;
            unsigned local;

            if( 2 != sscanf( physicalConditionId.c_str( ), "%u::%u", &origin, &local )) {
                cerr << errorStr << endl
                     << "    Failed to translate the physical identifier of the condition found" << endl
                     << "    at the following line of the definition file." << endl
                     << "\"" << theLine << "\"" << endl;
                return false;
            }
            if(( origin > 0xFFFF ) || ( local > 0xFFFF )) {
                cerr << errorStr << endl
                     << "    The translated physical identifier of the condition has illegal format." << endl
                     << "    The identifier was found at the following line of the definition file." << endl
                     << "\"" << theLine << "\"" << endl;
                return false;
            }
            originId    = (UShort_t)origin;
            conditionId = (UShort_t)local;
        }

        bool isPartitionable = false;
        {
            CdbId physicalId( originId,
                              conditionId );

            CdbConditionPtr cPtr;
            if( CdbStatus::Success != CdbCondition::instance( cPtr,
                                                              physicalId )) {
                cerr << errorStr << endl
                     << "    Failed to find the condition with the condition physical identifier " << physicalId << endl
                     << "    in the input federation." << endl;
                return false;
            }
            isPartitionable = cPtr->isPartitionable( );
        }

        UShort_t partitionId = 0;
        UShort_t clusterId   = 0;
        UShort_t incrementId = 0;
        {
            unsigned partition;
            unsigned cluster;
            unsigned increment;

            if( 3 != sscanf( placementInfo.c_str( ), "[%u,%u,%u]", &partition, &cluster, &increment )) {
                cerr << errorStr << endl
                     << "    Failed to translate the condition placement parameters found" << endl
                     << "    at the following line of the definition file." << endl
                     << "\"" << theLine << "\"" << endl;
                return false;
            }
            if(( partition > 0xFFFF ) || ( cluster > 0xFFFF ) || ( increment > 0xFFFF )) {
                cerr << errorStr << endl
                     << "    The translated condition placement parameters have illegal format." << endl
                     << "    They were found at the following line of the definition file." << endl
                     << "\"" << theLine << "\"" << endl;
                return false;
            }
            partitionId = (UShort_t)partition;
            clusterId   = (UShort_t)cluster;
            incrementId = (UShort_t)increment;
        }

      // Derive a correct owner origin identifier in case if this is a PARTITIONABLE
      // condition. We need this correction in order to locate the correct MetaData.
      //
      //    NOTE: The MetaData of partitionable conditions belong to owners of partitions,
      //           not to a formal owner

        if( isPartitionable ) {

            CdbPartitionPtr partitionPtr;
            if( CdbStatus::Success != theDatabasePtr->findPartition( partitionPtr,
                                                                     partitionId )) {
                cerr << errorStr << endl
                     << "    Failed to obtain a pointer onto the CDB API object for the partition " <<  partitionId << endl;
                return false;
            }
            originId = partitionPtr->ownerId( );
        }

      // Done. Fill resulting variables

        theOriginId         = originId;
        theConditionId      = conditionId;
        isPartitionableFlag = isPartitionable;
        thePartitionId      = partitionId;
        theClusterId        = clusterId;
        theIncrementId      = incrementId;
        theInputClassName   = inputClassName;

        return true;
    }
}

////////////////////////////
// Static class's members //
////////////////////////////

bool CdbBdb2RooPayloadConversionFwk::verboseModeFlag       = true;
bool CdbBdb2RooPayloadConversionFwk::debugModeFlag         = false;
bool CdbBdb2RooPayloadConversionFwk::doNotCompressModeFlag = false;

/////////////////////
// Class's methods //
/////////////////////

bool
CdbBdb2RooPayloadConversionFwk::setVerbose( bool newModeFlag )
{
    bool previousFlag = verboseModeFlag;
    verboseModeFlag = newModeFlag;
    return previousFlag;
}

bool
CdbBdb2RooPayloadConversionFwk::setDebug( bool newModeFlag )
{
    bool previousFlag = debugModeFlag;
    debugModeFlag = newModeFlag;
    return previousFlag;
}

bool
CdbBdb2RooPayloadConversionFwk::setDoNotCompress( bool newModeFlag )
{
    bool previousFlag = doNotCompressModeFlag;
    doNotCompressModeFlag = newModeFlag;
    return previousFlag;
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::registerConverter( const CdbCPtr< CdbRooConverterBase >& thePtr )
{
    if( thePtr.isNull( )) return CdbStatus::IllegalParameters;

  // Choose the right dictionary to register the converter.
  //
  // IMPLEMENTATION NOTE:
  //
  //   We're using two dictionary to allow the same input class be used in two
  //   different modes. Note, that dictionaries are keying converters based on
  //   the input class only.

    if( thePtr->mergingConverter( )) return mergingConvertersDict( )->add( thePtr );
    return regularConvertersDict( )->add( thePtr );
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::registerConverters(const std::vector< CdbCPtr < CdbRooConverterDict > >& theDicts)
{
  vector< CdbCPtr< CdbRooConverterDict > >::const_iterator iter = theDicts.begin();

  while (iter != theDicts.end()) {

    vector< CdbCPtr< CdbRooConverterBase > > vect = (*iter)->converters();
    vector< CdbCPtr< CdbRooConverterBase > >::iterator converterIter = vect.begin();

    while (converterIter != vect.end()) {
      CdbStatus result = registerConverter(*converterIter);
      if (result != CdbStatus::Success) return result;
      converterIter++;
    }

    iter++;
  }

  return CdbStatus::Success;
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::testConversion( UShort_t theOriginId,
                                                UShort_t theConditionId,
                                                bool     isPartitionableFlag,
                                                UShort_t thePartitionId,
                                                UShort_t theClusterId,
                                                UShort_t theIncrementId )
{
    const char* infoStr = "CdbBdb2RooPayloadConversionFwk::testConversion(regular) - INFO.";

    cout << infoStr << endl
         << "    Just for the record, you've entered the following parameters:" << endl
         << endl
         << "         Origin identifier : "   << theOriginId << endl
         << "      Condition identifier : "   << theConditionId << endl
         << "          Is partitionable : "   << ( isPartitionableFlag ? "Yes" : "No" ) << endl
         << "      Partition identifier : "   << thePartitionId << endl
         << "        Cluster identifier : "   << theClusterId << endl
         << "          Increment number : "   << theIncrementId << endl
         << endl;

  // Initialize the context of the Objectivity/DB based CDB

    CdbBdbShared::forceLoad( );

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

  // Initialize the context of the ROOT I/O based CDB

    CdbRooReadonly::forceLoad( );

    CdbTransaction rooReadOnlyTransactionStartsHere( CdbRooReadonly::technology( ),
                                                     CdbRooReadonly::implementation( ));

  // Proceed to the actual conversion

    return convertMetadata( theOriginId,
                            theConditionId,
                            isPartitionableFlag,
                            thePartitionId,
                            theClusterId,
                            theIncrementId );
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::testConversion( UShort_t                     theOriginId,
                                                const std::vector<UShort_t>& theConditionId,
                                                bool                         isPartitionableFlag,
                                                UShort_t                     thePartitionId,
                                                UShort_t                     theClusterId,
                                                UShort_t                     theIncrementId )
{
    const char* infoStr = "CdbBdb2RooPayloadConversionFwk::testConversion(merging) - INFO.";

    cout << infoStr << endl
         << "    Just for the record, you've entered the following parameters:" << endl
         << endl
         << "          Origin identifier : " << theOriginId << endl
         << "      Condition identifiers : " << ::vector2string( theConditionId ) << endl
         << "           Is partitionable : " << ( isPartitionableFlag ? "Yes" : "No" ) << endl
         << "       Partition identifier : " << thePartitionId << endl
         << "         Cluster identifier : " << theClusterId << endl
         << "           Increment number : " << theIncrementId << endl
         << endl;

  // Initialize the context of the Objectivity/DB based CDB

    CdbBdbShared::forceLoad( );

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

  // Initialize the context of the ROOT I/O based CDB

    CdbRooReadonly::forceLoad( );

    CdbTransaction rooReadOnlyTransactionStartsHere( CdbRooReadonly::technology( ),
                                                     CdbRooReadonly::implementation( ));

  // Proceed to the actual conversion

    return convertMetadata( theOriginId,
                            theConditionId,
                            isPartitionableFlag,
                            thePartitionId,
                            theClusterId,
                            theIncrementId );
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::doConversion( const char* theDefinitionFile )
{
    const char* errorStr = "CdbBdb2RooPayloadConversionFwk::doConversion() - ERROR.";

    CdbStatus result = CdbStatus::Error;

    assert( 0 != theDefinitionFile );

  // Initialize the context of the Objectivity/DB based CDB

    CdbBdbShared::forceLoad( );

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

  // Initialize the context of the ROOT I/O based CDB

    CdbRooReadonly::forceLoad( );

    CdbTransaction rooReadOnlyTransactionStartsHere( CdbRooReadonly::technology( ),
                                                     CdbRooReadonly::implementation( ));

  // Find the API object for the Database. We'll need it wor two reasons:
  //
  // - to check if we're running against the MASTER CDB, which would be desired
  //   to have a complete database contents;
  //
  // - to retrieve owner identifiers of partitions, which is needed to derive
  //   correct parameters of MetaData objects.

    CdbDatabasePtr databasePtr;
    if( CdbStatus::Success != ( result = CdbDatabase::instance( databasePtr ))) {
        cerr << errorStr << endl
             << "    Failed to obtain a pointer onto the CDB API object for the Database." << endl;
        return result;
    }
    if( !databasePtr->localOrigin( )->isMaster( )) {
        cerr << errorStr << endl
             << "    The input CDB is not of the MASTER type. The converter can only be run" << endl
             << "    against a MASTER type CDB. Make sure that you have set the correct" << endl
             << "    value for the OO_FD_BOOT environment variable." << endl;
        return CdbStatus::Error;
    }

  // Read the definition file into the memory. Collect only those conditioons whose classes
  // are known to the converter.

    typedef std::map< std::string, std::vector< CdbBdb2RooPayloadMetaDataDescriptor > > DescriptorsOfAcceptedMetaData;

    DescriptorsOfAcceptedMetaData regularMetaDataDescr;
    DescriptorsOfAcceptedMetaData  mergedMetaDataDescr;

    UShort_t                mergedOriginId        = 0;
    std::vector< UShort_t > mergedConditionId    ( 0 );
    bool                    mergedIsPartitionable = false;
    UShort_t                mergedPartitionId     = 0;
    UShort_t                mergedClusterId       = 0;
    UShort_t                mergedIncrementId     = 0;
    std::string             mergedInputClassName  = "";

    unsigned int currentLineNumber = 0;
    std::string  currentLine       = "";

    enum {
        PARSER_IDLE,
        PARSER_REGULAR,
        PARSER_MERGING_BEGIN,
        PARSER_MERGING,
        PARSER_FAILED
    } parserState = PARSER_IDLE;

    ifstream definitionFileStream( theDefinitionFile );

    const int bufSize = 64 * 1024;
    char      buf[bufSize];

    while( definitionFileStream.getline( buf,
                                         bufSize, '\n' )) {

        ++currentLineNumber;
        currentLine = std::string( buf );

      // Translate the next line read from the input file.

        if( verboseModeFlag ) {
            cout << "translating line: \"" << currentLine << "\"" << endl;
        }
        if( "#REGULAR" == currentLine ) {

            if(( PARSER_IDLE == parserState ) || ( PARSER_REGULAR == parserState ) || ( PARSER_MERGING_BEGIN == parserState )) {

                parserState = PARSER_REGULAR;

            } else if( PARSER_MERGING == parserState ) {

              // Finalize a block of the currently merged conditions before switching
              // to a regular one.
              //
              // IMPLEMENTATION NOTES:
              //
              //    That's the "cut-and-paste" of the same code implemented
              //    below when switching from one block merged conditions to
              //    another one.
              //
              //    The code needs to be refactored to be more robust.

                if( mergedConditionId.size( ) > 1 ) {

                  // Check if the Dictionary of Merging Converters has an entry for the input
                  // persistent class. If not then just skip the whole block MetaData entries.

                    CdbCPtr<CdbRooConverterBase> converterPtr =
                        mergingConvertersDict( )->findByInputClassName( mergedInputClassName );

                    if( converterPtr.isNull( )) {

                        if( verboseModeFlag ) {
                            cout << "skipping merged block of MetaData for main class: " << mergedInputClassName << endl;
                        }

                    } else {

                        if( mergedMetaDataDescr.find( mergedInputClassName ) == mergedMetaDataDescr.end( )) {
                            std::vector< CdbBdb2RooPayloadMetaDataDescriptor > emptyVector;
                            mergedMetaDataDescr[ mergedInputClassName ] = emptyVector;
                        }
                        mergedMetaDataDescr[ mergedInputClassName ].push_back( CdbBdb2RooPayloadMetaDataDescriptor( mergedOriginId,
                                                                                                                    mergedConditionId,
                                                                                                                    mergedIsPartitionable,
                                                                                                                    mergedPartitionId,
                                                                                                                    mergedClusterId,
                                                                                                                    mergedIncrementId ));
                    }

                } else {

                    cerr << errorStr << endl
                         << "    Too few MetaData-s in a block of merged ones." << endl
                         << "    There must be at least 2 of them." << endl;

                    parserState = PARSER_FAILED;

                    break;
                }
                
                parserState = PARSER_REGULAR;

            } else {

                parserState = PARSER_FAILED;

                break;
            }

        } else if( "#MERGE" == currentLine ) {

            if(( PARSER_IDLE == parserState ) || ( PARSER_REGULAR == parserState ) || ( PARSER_MERGING_BEGIN == parserState )) {

                parserState = PARSER_MERGING_BEGIN;

            } else if( PARSER_MERGING == parserState ) {

              // Finalize a block of the currently merged conditions before switching
              // to the next merged one.
              //
              // IMPLEMENTATION NOTES:
              //
              //    That's the "cut-and-paste" of the same code implemented
              //    above when switching from merged to regular conditions.
              //
              //    The code needs to be refactored to be more robust.

                if( mergedConditionId.size( ) > 1 ) {

                  // Check if the Dictionary of Merging Converters has an entry for the input
                  // persistent class. If not then just skip the whole block MetaData entries.

                    CdbCPtr<CdbRooConverterBase> converterPtr =
                        mergingConvertersDict( )->findByInputClassName( mergedInputClassName );

                    if( converterPtr.isNull( )) {

                        if( verboseModeFlag ) {
                            cout << "skipping merged block of MetaData for main class: " << mergedInputClassName << endl;
                        }

                    } else {

                        if( mergedMetaDataDescr.find( mergedInputClassName ) == mergedMetaDataDescr.end( )) {
                            std::vector< CdbBdb2RooPayloadMetaDataDescriptor > emptyVector;
                            mergedMetaDataDescr[ mergedInputClassName ] = emptyVector;
                        }
                        mergedMetaDataDescr[ mergedInputClassName ].push_back( CdbBdb2RooPayloadMetaDataDescriptor( mergedOriginId,
                                                                                                                    mergedConditionId,
                                                                                                                    mergedIsPartitionable,
                                                                                                                    mergedPartitionId,
                                                                                                                    mergedClusterId,
                                                                                                                    mergedIncrementId ));
                    }

                } else {

                    cerr << errorStr << endl
                         << "    Too few MetaData-s in a block of merged ones." << endl
                         << "    There must be at least 2 of them." << endl;

                    parserState = PARSER_FAILED;

                    break;
                }
                
                parserState = PARSER_MERGING_BEGIN;

            } else {

                parserState = PARSER_FAILED;

                break;
            }

        } else if( "#END" == currentLine ) {

            if( PARSER_MERGING == parserState ) {

              // Finalize a block of the currently merged conditions before switching
              // to the next merged one.
              //
              // IMPLEMENTATION NOTES:
              //
              //    That's the "cut-and-paste" of the same code implemented
              //    above when switching from merged to regular conditions
              //    as well as when switching from one block merged conditions to
              //    another one.
              //
              //    The code needs to be refactored to be more robust.

                if( mergedConditionId.size( ) > 1 ) {

                  // Check if the Dictionary of Merging Converters has an entry for the input
                  // persistent class. If not then just skip the whole block MetaData entries.

                    CdbCPtr<CdbRooConverterBase> converterPtr =
                        mergingConvertersDict( )->findByInputClassName( mergedInputClassName );

                    if( converterPtr.isNull( )) {

                        if( verboseModeFlag ) {
                            cout << "skipping merged block of MetaData for main class: " << mergedInputClassName << endl;
                        }

                    } else {

                        if( mergedMetaDataDescr.find( mergedInputClassName ) == mergedMetaDataDescr.end( )) {
                            std::vector< CdbBdb2RooPayloadMetaDataDescriptor > emptyVector;
                            mergedMetaDataDescr[ mergedInputClassName ] = emptyVector;
                        }
                        mergedMetaDataDescr[ mergedInputClassName ].push_back( CdbBdb2RooPayloadMetaDataDescriptor( mergedOriginId,
                                                                                                                    mergedConditionId,
                                                                                                                    mergedIsPartitionable,
                                                                                                                    mergedPartitionId,
                                                                                                                    mergedClusterId,
                                                                                                                    mergedIncrementId ));
                    }

                } else {

                    cerr << errorStr << endl
                         << "    Too few MetaData-s in a block of merged ones." << endl
                         << "    There must be at least 2 of them." << endl;

                    parserState = PARSER_FAILED;

                    break;
                }
            }

            parserState = PARSER_IDLE;

            break;

        } else {

          // It must be the MetaData line then?

            UShort_t    originId        = 0;
            UShort_t    conditionId     = 0;
            bool        isPartitionable = false;
            UShort_t    partitionId     = 0;
            UShort_t    clusterId       = 0;
            UShort_t    incrementId     = 0;
            std::string inputClassName  = "";

            if( !::translateNextLine( currentLine,
                                      databasePtr,
                                      originId,
                                      conditionId,
                                      isPartitionable,
                                      partitionId,
                                      clusterId,
                                      incrementId,
                                      inputClassName )) {

                parserState = PARSER_FAILED;

                break;
            }

            if( PARSER_REGULAR == parserState ) {

              // Check if the Dictionary of Regular Converters has an entry for the input
              // persistent class. If not then just skip the MetaData entry.

                CdbCPtr<CdbRooConverterBase> converterPtr =
                    regularConvertersDict( )->findByInputClassName( inputClassName );

                if( converterPtr.isNull( )) {

                    if( verboseModeFlag ) {
                        cout << "no converter found for the input class \"" << inputClassName << "\", skipping line " << currentLine<< endl;
                    }

                } else {

                    if( regularMetaDataDescr.find( inputClassName ) == regularMetaDataDescr.end( )) {
                        std::vector< CdbBdb2RooPayloadMetaDataDescriptor > emptyVector;
                        regularMetaDataDescr[ inputClassName ] = emptyVector;
                    }
                    regularMetaDataDescr[ inputClassName ].push_back( CdbBdb2RooPayloadMetaDataDescriptor( originId,
                                                                                                           conditionId,
                                                                                                           isPartitionable,
                                                                                                           partitionId,
                                                                                                           clusterId,
                                                                                                           incrementId ));
                }

            } else if( PARSER_MERGING_BEGIN == parserState ) {

              // That's the very first MetaData in a block.
              // Set up the main context.

                mergedOriginId        = originId;

                mergedConditionId.resize( 0 );
                mergedConditionId.push_back( conditionId );

                mergedIsPartitionable = isPartitionable;
                mergedPartitionId     = partitionId;
                mergedClusterId       = clusterId;
                mergedIncrementId     = incrementId;
                mergedInputClassName  = inputClassName;

                parserState = PARSER_MERGING;

            } else if( PARSER_MERGING == parserState ) {

              // That's the next MetaData in a block.
              // Verify if its parameters are consistent with the current context.

                if(( mergedOriginId        != originId        ) ||
                   ( mergedIsPartitionable != isPartitionable ) ||
                   ( mergedPartitionId     != partitionId     ) ||
                   ( mergedClusterId       != clusterId       ) ||
                   ( mergedIncrementId     != incrementId     )) {

                    cerr << errorStr << endl
                         << "    Inconsistent MetaData in a block of merged ones." << endl;

                    parserState = PARSER_FAILED;

                    break;
                }

                mergedConditionId.push_back( conditionId );

            } else {

                parserState = PARSER_FAILED;

                break;
            }
        }
    }
    if( PARSER_IDLE != parserState ) {

        if( PARSER_FAILED == parserState ) {

            cerr << errorStr << endl
                 << "    Incorrect syntax of the definition file:" << endl
                 << "        \"" << theDefinitionFile << "\"" << endl
                 << "    at line #" << currentLineNumber << " containing:" << endl
                 << "        \"" << currentLine << "\"" << endl
                 << endl;

        } else {

            cerr << errorStr << endl
                 << "    Incorrect syntax at the end of the definition file:" << endl
                 << "        \"" << theDefinitionFile << "\"" << endl
                 << "    after line #" << currentLineNumber << " containing:" << endl
                 << "        \"" << currentLine << "\"" << endl
                 << "    Perhaps the #END command line is missing?" << endl
                 << endl;
        }
        return CdbStatus::Error;
    }

  // Process accepted regular MetaData objects.

    for( DescriptorsOfAcceptedMetaData::const_iterator itr = regularMetaDataDescr.begin( );
                                                       itr != regularMetaDataDescr.end( );
                                                     ++itr ) {

        if( verboseModeFlag )
            cout << "processing regular InputClass \"" << (*itr).first << "\"" << endl;

        std::vector< CdbBdb2RooPayloadMetaDataDescriptor > vectorOfDescriptors = (*itr).second;

        unsigned int num = vectorOfDescriptors.size( );
        assert( 0 != num );

        for( unsigned int i = 0; i < num; ++i ) {

            CdbBdb2RooPayloadMetaDataDescriptor descriptor = vectorOfDescriptors[i];

            if( verboseModeFlag )
                cout << "  processing regular MetaData " << descriptor.toString( ) << endl;

            if( CdbStatus::Success != ( result = convertMetadata( descriptor.originId,
                                                                  descriptor.conditionId[0],    // Main object only
                                                                  descriptor.isPartitionable,
                                                                  descriptor.partitionId,
                                                                  descriptor.clusterId,
                                                                  descriptor.incrementId ))) {
                cerr << errorStr << endl
                     << "    Failed when regular processing MetaData " << descriptor.toString( ) << endl;
                return result;
            }
        }
    }

  // Process accepted merged MetaData objects.

    for( DescriptorsOfAcceptedMetaData::const_iterator itr = mergedMetaDataDescr.begin( );
                                                       itr != mergedMetaDataDescr.end( );
                                                     ++itr ) {

        if( verboseModeFlag )
            cout << "processing merged InputClass \"" << (*itr).first << "\"" << endl;

        std::vector< CdbBdb2RooPayloadMetaDataDescriptor > vectorOfDescriptors = (*itr).second;

        unsigned int num = vectorOfDescriptors.size( );
        assert( 0 != num );

        for( unsigned int i = 0; i < num; ++i ) {

            CdbBdb2RooPayloadMetaDataDescriptor descriptor = vectorOfDescriptors[i];

            if( verboseModeFlag )
                cout << "  processing merged MetaData " << descriptor.toString( ) << endl;

            if( CdbStatus::Success != ( result = convertMetadata( descriptor.originId,
                                                                  descriptor.conditionId,   // Main + extra objects
                                                                  descriptor.isPartitionable,
                                                                  descriptor.partitionId,
                                                                  descriptor.clusterId,
                                                                  descriptor.incrementId ))) {
                cerr << errorStr << endl
                     << "    Failed when processing merged MetaData " << descriptor.toString( ) << endl;
                return result;
            }
        }
    }
    return CdbStatus::Success;
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::convertMetadata( UShort_t theOriginId,
                                                 UShort_t theConditionId,
                                                 bool     isPartitionableFlag,
                                                 UShort_t thePartitionId,
                                                 UShort_t theClusterId,
                                                 UShort_t theIncrementId )
{
    const char* errorStr = "CdbBdb2RooPayloadConversionFwk::::convertMetadata(regular) - ERROR.";

    if( verboseModeFlag ) {
        cout << "CdbBdb2RooPayloadConversionFwk::convertMetadata(regular) : PARAMETERS OF THE METHOD {" << endl
             << "    theOriginId         = " << theOriginId << endl
             << "    theConditionId      = " << theConditionId << endl
             << "    isPartitionableFlag = " << isPartitionableFlag << endl
             << "    thePartitionId      = " << thePartitionId << endl
             << "    theClusterId        = " << theClusterId << endl
             << "    theIncrementId      = " << theIncrementId << endl
             << "}" << endl;
    }

    CdbStatus result = CdbStatus::Error;

  // Find the output collection of original intervals to be processed

    CdbCPtr< CdbRooRoMetaDataR > mdPtr;
    if( CdbStatus::Success != ( result = CdbRooRoFileUtils::instance( ).findMetaData( mdPtr,
                                                                                      theConditionId,
                                                                                      isPartitionableFlag,
                                                                                      thePartitionId,
                                                                                      theClusterId,
                                                                                      theIncrementId,
                                                                                      theOriginId ))) {
        cerr << errorStr << endl
             << "    Failed to locate the input MetaData object." << endl;
        return result;
    }
    CdbCPtr< CdbRooRoOiCollectionR> oiCollectionPtr = mdPtr->originalIntervalsCollection( );

    if( debugModeFlag ) {
        cout << "CdbBdb2RooPayloadConversionFwk::convertMetadata(regular) : DUMP OF THE ORIGINAL COLLECTION {" << endl;
        oiCollectionPtr->dump( cout );
    }

  // Find and open the output file for updating

    CdbCPtr< TFile > databasePtr;
    if( CdbStatus::Success != ( result = CdbRooRoFileUtils::instance( ).findObjectsDatabase( databasePtr,
                                                                                             isPartitionableFlag,
                                                                                             thePartitionId,
                                                                                             theClusterId,
                                                                                             theIncrementId,
                                                                                             theOriginId,
                                                                                             "update" ))) {
        cerr << errorStr << endl
             << "    Failed to locate the file for output objects." << endl;
        return result;
    }
    if( doNotCompressModeFlag ) databasePtr->SetCompressionLevel( 0 );

  // Build a base name for container trees where with branches where we're going
  // to put converted objects.
  //
  // NOTE: In ROOT I/O the tree will be create in the last open/activated file. Hence we're
  //       assuming that it's been done by the previous operation.

    const std::string treeNameBase =
        CdbRooRoFileUtils::nameOfObjectsContainer( theConditionId,
                                                   isPartitionableFlag,
                                                   thePartitionId,
                                                   theClusterId,
                                                   theIncrementId,
                                                   theOriginId ) + ".";

  // Prepare a map with an executin plan for the conversion

    std::map< UShort_t, CdbBdb2RooPayloadContainerContext* > executionPlan;

  // Create a map of object which are processes to avoid double processing

    std::map< std::string, bool > mapOfProcessedObjects;

  // Process each interval found in the collection

    if( debugModeFlag )
        cout << endl
             << "Conversion pass  I: preparing an execution plan..." << endl
             << endl;

    UInt_t nIntervals = oiCollectionPtr->size( );
    assert( nIntervals > 0 );

    for( UInt_t idx = 1; idx < nIntervals; ++idx ) {

        CdbRooRoOiR oi;

        BdbTime beginOfDuration;    // not used by the current algorithm
        BdbTime   endOfDuration;    // not used by the current algorithm

        if( CdbStatus::Success != ( result = oiCollectionPtr->find( idx,
                                                                    oi,
                                                                    beginOfDuration,
                                                                    endOfDuration ))) {
            cerr << errorStr << endl
                 << "    Failed to find an 'origina' interval with index = " << idx << endl;
            return result;
        }

        if( debugModeFlag )
            cout << "  " << oi.object( ) << " <-- " << oi.legacyObjectAddress( ) << endl;

      // Check if the OID of the output object matches the current location of the "MetaData.<id>"
      // and "Objects.<id>" containers.

        if( ( oi.object( ).origin   ( ) != theOriginId    ) ||
            ( oi.object( ).condition( ) != theConditionId ) ||
            ( oi.object( ).cluster  ( ) != theClusterId   ) ||
            ( oi.object( ).partition( ) != thePartitionId ) ||
            ( oi.object( ).increment( ) != theIncrementId )) {

                cerr << errorStr << endl
                     << "    The output object address " << oi.object( ) << " doesn't match" << endl
                     << "    the location of the output container." << endl;
                return CdbStatus::Error;
        }

      // Skip the object if it's already been processed

        if( mapOfProcessedObjects.find( oi.object( ).toString( )) != mapOfProcessedObjects.end( )) continue;

        mapOfProcessedObjects[oi.object( ).toString( )] = true;

      // Get the input persistent object first and verify it exists

        BdbRef(BdbObject) legacyObjectRef;

        if( !translate_OID( legacyObjectRef,
                            oi.legacyObjectAddress( ))) {
            cerr << errorStr << endl
                 << "    Failed to translate the OID of the legacy object: " << oi.legacyObjectAddress( ) << endl;
            return CdbStatus::Error;
        }

      // Get the final class of that persistent object and verify if it derives
      // from the BdbObject class.

        const std::string legacyObjectTypeName = legacyObjectRef.typeName( );
        {
            if( !CdbBdbSchemaUtils::instance( ).isA( legacyObjectRef.typeN( ),
                                                     ooTypeN( BdbObject ))) {
                cerr << errorStr << endl
                     << "    The legacy object doesn't derive from BdbObject: " << oi.legacyObjectAddress( ) << endl;
                return CdbStatus::Error;
            }
        }

      // Create new container context if it doesn't exist yet, then put the found
      // original interval into this context.
      //
      // NOTE: the order of original intervals is important to match ROID-s
      //       generated during the metadata conversion phase.

        if( executionPlan.find( oi.object( ).container( )) == executionPlan.end( )) {

          // Build the name of the tree representing this container

            const std::string treeName = treeNameBase + CdbStringUtils::toString( oi.object( ).container( ));

          // Check if the corresponding converter is available

            CdbCPtr< CdbRooConverterBase > converterPtr = regularConvertersDict( )->findByInputClassName( legacyObjectTypeName );
            if( converterPtr.isNull( )) {
                cerr << errorStr << endl
                     << "    No converter avaialble for the legacy class: " << legacyObjectTypeName << endl;
                return CdbStatus::Error;
            }

          //
          // VERY IMPORTANT NOTE: Due to a way this "xyz!@#$^%" ROOT I/O works one should always
          //                      be sure that the proper "current file" is set before creating new
          //                      TTree-s which is what is going to happen below.
 
// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

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

// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

            executionPlan[oi.object( ).container( )] = new CdbBdb2RooPayloadContainerContext( treeName,
                                                                                              "A tree for 'payload' objects" ,
                                                                                              converterPtr );
        }
        executionPlan[oi.object( ).container( )]->add( legacyObjectRef,
                                                       oi.object( ));
    }

    if( debugModeFlag )
        cout << endl
             << "Conversion pass II: performing the actual conversion..." << endl
             << endl;

    for( std::map< UShort_t, CdbBdb2RooPayloadContainerContext* >::iterator itr = executionPlan.begin( );
                                                                            itr != executionPlan.end( );
                                                                          ++itr ) {

        CdbBdb2RooPayloadContainerContext* containerContext = (*itr).second;

      // Get the corresponding TBranch for that class
      //
      // DESIGN NOTE: A transient pointer passed to the constructor will be managed
      //              internally by the converter. See more details on how this
      //              converter is used insede an intervals conversion loop below.

        const char* branchName = "0";   // always use this branch name

        TBranch* branch = containerContext->converterPtr->branchConstructor( containerContext->tree,
                                                                             branchName );

        UInt_t nIntervals = containerContext->inObjects.size( );
        assert( nIntervals > 0 );

        for( UInt_t idx = 0; idx < nIntervals; ++idx ) {

            BdbRef(BdbObject)& legacyObjectRef = containerContext->inObjects [idx];
            CdbRooRoObjectIdR& outputObjectId  = containerContext->outObjects[idx];

            if( debugModeFlag )
                cout << "  " << outputObjectId << " <-- " << legacyObjectRef.sprint( ) << " : \"" << containerContext->converterPtr->inputClassName( ) << "\"" << endl;

          // Use the converter
          //
          // DESIGN NOTE:
          //
          //   1. The actual object will be created and stored at a location known to
          //   above constructed TBranch inside the converter. We just need to make sure
          //   that the object is saved (see TTree::Fill()) operation in the current TTree,
          //
          //   2. Remember that the converter manages the lifetime of the created transient objects.

            {
                std::vector< BdbRef(BdbObject) > extraObjects( 0 ); // no extra objects for now

                if( CdbStatus::Success != ( result = containerContext->converterPtr->conversion( legacyObjectRef,
                                                                                                 extraObjects ))) {
                    cerr << errorStr << endl
                         << "    Failed to convert the legacy object " << legacyObjectRef.sprint( ) << endl;
                    return result;
                }

              // Store the object in the Tree/Branch

                containerContext->tree.Fill( );

              // Make sure that the next available index in the branch matches the object index
              // in its OID.

                if( debugModeFlag )
                    cout << "  tree->GetNbranches( ) = " << containerContext->tree.GetNbranches( ) << endl
                         << "  tree->GetEntries  ( ) = " << containerContext->tree.GetEntries( ) << endl
                         << "branch->GetEntries  ( ) = " << branch->GetEntries( ) << endl;

                Long64_t objectIndexInBranch = branch->GetEntries( ) - 1;

                if(( objectIndexInBranch < 0 ) ||
                   ( objectIndexInBranch >= 0xFFFF ) ||
                   ( objectIndexInBranch != outputObjectId.index( ))) {

                    cerr << errorStr << endl
                         << "    The just converted object " << outputObjectId << " produced from the legacy object " << legacyObjectRef.sprint( ) << endl
                         << "    was writtent into a branch with a noon-valid index " << objectIndexInBranch << endl;

                    return CdbStatus::Error;
                }
            }
        }

      // Save the tree in the file

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

        Int_t numBytesWritten = databasePtr->WriteObject( &containerContext->tree,
                                                          containerContext->tree.GetName( ));
        if( numBytesWritten <= 0 ) {
            cerr << errorStr << endl
                 << "    Failed to save the tree \"" << containerContext->tree.GetName( ) << "\" in the file: " << databasePtr->GetName( ) << endl;
            return CdbStatus::Error;
        }
        if( debugModeFlag )
            cout << endl
                 << "Total of " << nIntervals << " converted and stored into the tree \"" << containerContext->tree.GetName( ) << "\"." << endl
                 << endl;

      // Destroy the context

        delete containerContext;
    }
    return CdbStatus::Success;
}

CdbStatus
CdbBdb2RooPayloadConversionFwk::convertMetadata( UShort_t                     theOriginId,
                                                 const std::vector<UShort_t>& theConditionId,
                                                 bool                         isPartitionableFlag,
                                                 UShort_t                     thePartitionId,
                                                 UShort_t                     theClusterId,
                                                 UShort_t                     theIncrementId )
{
    const char* infoStr  = "CdbBdb2RooPayloadConversionFwk::::convertMetadata(merging) - INFO.";
    const char* errorStr = "CdbBdb2RooPayloadConversionFwk::::convertMetadata(merging) - ERROR.";

    CdbStatus result = CdbStatus::Error;

    if( debugModeFlag ) {
        cout << infoStr << endl
             << "    Converting and merging the following conditions:" << endl
             << endl
             << "          Origin identifier : " << theOriginId << endl
             << "      Condition identifiers : " << ::vector2string( theConditionId ) << endl
             << "           Is partitionable : " << ( isPartitionableFlag ? "Yes" : "No" ) << endl
             << "       Partition identifier : " << thePartitionId << endl
             << "         Cluster identifier : " << theClusterId   << endl
             << "           Increment number : " << theIncrementId << endl
             << endl;
    }

  // Verify parameters

    const unsigned int nConditions = theConditionId.size( );
    if( nConditions < 2 ) {
        cerr << errorStr << endl
             << "    The total number of conditions to be merged is less than 2." << endl;
        return CdbStatus::IllegalParameters;
    }

  // Find and verify the output collections of original intervals to be processed
  //
  // NOTE: By "verify" we mean that all collections should have the same number
  //       of intervals. That's essential for the merging algorithm!

    std::vector< CdbCPtr< CdbRooRoOiCollectionR> > oiCollectionPtr( nConditions );

    for( unsigned int conditionIdx = 0; conditionIdx < nConditions; ++conditionIdx ) {

        CdbCPtr< CdbRooRoMetaDataR > mdPtr;
        if( CdbStatus::Success != ( result = CdbRooRoFileUtils::instance( ).findMetaData( mdPtr,
                                                                                          theConditionId[conditionIdx],
                                                                                          isPartitionableFlag,
                                                                                          thePartitionId,
                                                                                          theClusterId,
                                                                                          theIncrementId,
                                                                                          theOriginId ))) {
            cerr << errorStr << endl
                 << "    Failed to locate the input MetaData object for:" << endl;

            ::printMetDataParameters( theOriginId,
                                      theConditionId[conditionIdx],
                                      isPartitionableFlag,
                                      thePartitionId,
                                      theClusterId,
                                      theIncrementId );
            return result;
        }
        oiCollectionPtr[conditionIdx] = mdPtr->originalIntervalsCollection( );
    }

    for( unsigned int conditionIdx = 1; conditionIdx < nConditions; ++conditionIdx ) {
        if( oiCollectionPtr[0]->size( ) != oiCollectionPtr[conditionIdx]->size( )) {
            cerr << errorStr << endl
                 << "    MetData objects passed to the current procedure do not have the same" << endl
                 << "    number of original intervals. Merging is not possible for these conditions." << endl;
            return CdbStatus::Error;
        }
    }

  // Find and open the output file for updating
  //
  // IMPLEMENTATION NOTE: 

    CdbCPtr< TFile > databasePtr;
    if( CdbStatus::Success != ( result = CdbRooRoFileUtils::instance( ).findObjectsDatabase( databasePtr,
                                                                                             isPartitionableFlag,
                                                                                             thePartitionId,
                                                                                             theClusterId,
                                                                                             theIncrementId,
                                                                                             theOriginId,
                                                                                             "update" ))) {
        cerr << errorStr << endl
             << "    Failed to create/reopen a file for output objects." << endl;
        return result;
    }
    if( doNotCompressModeFlag ) databasePtr->SetCompressionLevel( 0 );

  // Build a base name for container trees where with branches where we're going
  // to put converted objects.
  //
  // NOTE: In ROOT I/O the tree will be create in the last open/activated file. Hence we're
  //       assuming that it's been done by the previous operation.

    std::vector< std::string > treeNameBase( nConditions );
    for( unsigned int conditionIdx = 0; conditionIdx < nConditions; ++conditionIdx ) {

        treeNameBase[conditionIdx] =
            CdbRooRoFileUtils::nameOfObjectsContainer( theConditionId[conditionIdx],
                                                       isPartitionableFlag,
                                                       thePartitionId,
                                                       theClusterId,
                                                       theIncrementId,
                                                       theOriginId ) + ".";
    }

  // Prepare two maps with executin plans for the conversion.
  //
  // NOTE: the element with index [0] is not used in the second plan. It's there just
  //       for the simplicity.

    std::map< UShort_t, CdbBdb2RooPayloadMainContainerContext* > mainExecutionPlan;

    std::vector< std::map< UShort_t, CdbBdb2RooPayloadExtraContainerContext* > > executionPlan( nConditions );

  // Create a map of object which are processes
  //
  // NOTE: When merging objects this map is used to track those cases when the same object
  //       is seen more then once and abort the procedure. Merging seems to be a non
  //       trivial thing in this case because the original identity of objects is going to be
  //       violated.

    std::map< std::string, bool > mapOfProcessedObjects;

  // Process each interval found in the collection

    if( debugModeFlag )
        cout << endl
             << "Conversion pass  I: preparing an execution plan..." << endl
             << endl;

    const UInt_t nIntervals = oiCollectionPtr[0]->size( );  // note, that this number also includes the interval
                                                            // with 0 index, which is used for special purposes
    assert( nIntervals > 0 );

    if( debugModeFlag )
        cout << "There are " << nIntervals - 1 << " intervals to be converted and merged." << endl;

    for( UInt_t idx = 1; idx < nIntervals; ++idx ) {

      // The validity intervals of found objects must also be synchronious. The corresponding
      // check will be done after the end of the evaluation loop.
      //
      // NOTE: This doesn't apply to the "duration" intervals as this parameter
      //       is meant to reflect when objects were actually stored in the database.
      //       CDB can't guarantee that two objects were stored at the same moment
      //       of time.

        std::vector< CdbRooRoOiR > oi( nConditions );

      // This is a vector of all objects to be merged.

        std::vector< BdbRef(BdbObject) > legacyObjectRef( nConditions );

      // Evaluate objects from all conditions
      //
      // IMPORTANT NOTE:
      //
      //   Note that extra objects are being processed first before the main one
      //   in the loop below. Otherwise the vector of the legacy objects defined
      //   above won't be usable. See details inseide the loop.

        for( int conditionIdx = nConditions - 1; conditionIdx >= 0; --conditionIdx ) {

            BdbTime beginOfDuration;    // not used by the current algorithm
            BdbTime   endOfDuration;    // not used by the current algorithm

            if( CdbStatus::Success != ( result = oiCollectionPtr[conditionIdx]->find( idx,
                                                                                      oi[conditionIdx],
                                                                                      beginOfDuration,
                                                                                      endOfDuration ))) {
                cerr << errorStr << endl
                     << "    Failed to find an 'origina' interval with index = " << idx << endl;
                return result;
            }

            if( debugModeFlag )
                cout << "[" << conditionIdx << "]" << "  " << oi[conditionIdx].object( ) << " <-- " << oi[conditionIdx].legacyObjectAddress( ) << endl;

          // Check if the OID of the output object matches the current location of the "MetaData.<id>"
          // and "Objects.<id>" containers.

            if( ( oi[conditionIdx].object( ).origin   ( ) != theOriginId                  ) ||
                ( oi[conditionIdx].object( ).condition( ) != theConditionId[conditionIdx] ) ||
                ( oi[conditionIdx].object( ).cluster  ( ) != theClusterId                 ) ||
                ( oi[conditionIdx].object( ).partition( ) != thePartitionId               ) ||
                ( oi[conditionIdx].object( ).increment( ) != theIncrementId               )) {

                    cerr << errorStr << endl
                         << "    The output object address " << oi[conditionIdx].object( ) << " doesn't match" << endl
                         << "    the location of the output container." << endl;
                    return CdbStatus::Error;
            }

          // Abort if the main object has already been processed. And we don't care about
          // extra conditions.

            if( 0 == conditionIdx ) {

                if( mapOfProcessedObjects.find( oi[conditionIdx].object( ).toString( )) != mapOfProcessedObjects.end( )) {

                    cerr << errorStr << endl
                         << "    An attempt to merge conditions with duplicate objects in the main" << endl
                         << "    condition has been detected. Aborting the operation." << endl;

                    ::printMetDataParameters( theOriginId,
                                              theConditionId[conditionIdx],
                                              isPartitionableFlag,
                                              thePartitionId,
                                              theClusterId,
                                              theIncrementId );

                    return CdbStatus::Error;
                }
            }
            mapOfProcessedObjects[oi[conditionIdx].object( ).toString( )] = true;

          // Get the input persistent object first and verify if it exists

            if( !translate_OID( legacyObjectRef[conditionIdx],
                                oi[conditionIdx].legacyObjectAddress( ))) {
                cerr << errorStr << endl
                     << "    Failed to translate the OID of the legacy object: " << oi[conditionIdx].legacyObjectAddress( ) << endl;
                return CdbStatus::Error;
            }

          // Get the final class of that persistent object and verify if it derives
          // from the BdbObject class.

            const std::string legacyObjectTypeName = legacyObjectRef[conditionIdx].typeName( );
            {
                if( !CdbBdbSchemaUtils::instance( ).isA( legacyObjectRef[conditionIdx].typeN( ),
                                                         ooTypeN( BdbObject ))) {
                    cerr << errorStr << endl
                         << "    The legacy object doesn't derive from BdbObject: " << oi[conditionIdx].legacyObjectAddress( ) << endl;
                    return CdbStatus::Error;
                }
            }

          // Create new container context if it doesn't exist yet, then put the found
          // original interval into this context.
          //
          // NOTES:
          //
          //   1. The order of original intervals is important to match ROID-s
          //      generated during the metadata conversion phase.
          //
          //   2. Also note that different execiton plans are used for the main and extra
          //      conditions.

            if( 0 == conditionIdx ) {

                if( mainExecutionPlan.find( oi[conditionIdx].object( ).container( )) == mainExecutionPlan.end( )) {

                  // Build the name of the tree representing this container

                    const std::string treeName = treeNameBase[conditionIdx] + CdbStringUtils::toString( oi[conditionIdx].object( ).container( ));

                  // Check if the corresponding converter is available

                    CdbCPtr< CdbRooConverterBase > converterPtr = mergingConvertersDict( )->findByInputClassName( legacyObjectTypeName );
                    if( converterPtr.isNull( )) {
                        cerr << errorStr << endl
                             << "    No MERGING converter avaialble for the legacy class: " << legacyObjectTypeName << endl;
                        return CdbStatus::Error;
                    }

                  //
                  // VERY IMPORTANT NOTE: Due to a way this "xyz!@#$^%" ROOT I/O works one should always
                  //                      be sure that the proper "current file" is set before creating new
                  //                      TTree-s which is what is going to happen below.

// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

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

// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

                    mainExecutionPlan[oi[conditionIdx].object( ).container( )] = new CdbBdb2RooPayloadMainContainerContext( treeName,
                                                                                                                           "A tree for 'payload' objects" ,
                                                                                                                           converterPtr );
                }

              // IMPORTANT NOTES:
              //
              //   1. When we're calling the "add" method below then we're relying on a fact that
              //   the main object is being processed after all extra objects have already been found.
              //   Otherwise the array with list of legacy objects won't be complete!!!
              //
              //   2. We're also "add"-ing records to the previously discovered containers for
              //   extra objects. We do it here since these extra objects should know where (a main object)
              //   their original contents is going to be moved to.

                mainExecutionPlan[oi[conditionIdx].object( ).container( )]->add( legacyObjectRef,
                                                                                 oi[conditionIdx].object( ));

                for( unsigned int extraConditionIdx = 1; extraConditionIdx < nConditions; ++extraConditionIdx ) {

                    const unsigned int containerId = oi[extraConditionIdx].object( ).container( );

                    executionPlan[extraConditionIdx][containerId]->add( legacyObjectRef[extraConditionIdx],
                                                                        oi[extraConditionIdx].object( ),
                                                                        oi[conditionIdx].object( ));
                }

            } else {

              // Precreate the execution plan, but do _NOT_ add the found object to it. Thsi will
              // be done after the main object will be found. That's because extra objects will be replaced
              // with the "ObjectHasMoved" entries reffering back to the new home for the extra objects'
              // original contents.

                if( executionPlan[conditionIdx].find( oi[conditionIdx].object( ).container( )) == executionPlan[conditionIdx].end( )) {

                  // Build the name of the tree representing this container

                    const std::string treeName = treeNameBase[conditionIdx] + CdbStringUtils::toString( oi[conditionIdx].object( ).container( ));

                  // No converter for extra objects because a special class known to the current
                  // application will be used to put dummy object into the extra conditions.

                    CdbCPtr< CdbRooConverterBase > converterPtr;

                  //
                  // VERY IMPORTANT NOTE: Due to a way this "xyz!@#$^%" ROOT I/O works one should always
                  //                      be sure that the proper "current file" is set before creating new
                  //                      TTree-s which is what is going to happen below.

// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

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

// cout << "gDirectory->GetName( ) = \"" << gDirectory->GetName( ) << "\"" << endl;
// cout << "gFile->GetName( )      = \"" << gFile->GetName( )      << "\"" << endl;

                    executionPlan[conditionIdx][oi[conditionIdx].object( ).container( )] = new CdbBdb2RooPayloadExtraContainerContext( treeName,
                                                                                                                                       "A tree for 'payload' objects" ,
                                                                                                                                       converterPtr );
                }

// This operation has moved up to the same place when the main object is added
// to its execution plan.
//
//                executionPlan[conditionIdx][oi[conditionIdx].object( ).container( )]->add( legacyObjectRef[conditionIdx],
//                                                                                           oi[conditionIdx].object( ));
            }
        }

      // Check if conditions are synchronious

        for( unsigned int conditionIdx = 1; conditionIdx < nConditions; ++conditionIdx ) {

          // The validity intervals must be equal

            if( oi[0].begin( ) != oi[conditionIdx].begin( )) {
                cerr << errorStr << endl
                     << "    Checked the begin validity time of objects in merged conditions and found" << endl
                     << "    that an attempt to merge asynchronious conditions is taking place." << endl;
                return CdbStatus::Error;
            }
            if( oi[0].end( ) != oi[conditionIdx].end( )) {
                cerr << errorStr << endl
                     << "    Checked the begin validity time of objects in merged conditions and found" << endl
                     << "    that an attempt to merge asynchronious conditions is taking place." << endl;
                return CdbStatus::Error;
            }
        }
    }

    if( debugModeFlag )
        cout << endl
             << "Conversion pass II: performing the actual conversion..." << endl
             << endl;

  // Process the main condition first

    for( std::map< UShort_t, CdbBdb2RooPayloadMainContainerContext* >::iterator itr = mainExecutionPlan.begin( );
                                                                                itr != mainExecutionPlan.end( );
                                                                              ++itr ) {

        CdbBdb2RooPayloadMainContainerContext* containerContext = (*itr).second;

      // Get the corresponding TBranch for that class
      //
      // DESIGN NOTE: A transient pointer passed to the constructor will be managed
      //              internally by the converter. See more details on how this
      //              converter is used insede an intervals conversion loop below.

        const char* branchName = "0";   // always use this branch name

        TBranch* branch = containerContext->converterPtr->branchConstructor( containerContext->tree,
                                                                             branchName );

        UInt_t nIntervals = containerContext->inObjects.size( );
        assert( nIntervals > 0 );

        for( UInt_t idx = 0; idx < nIntervals; ++idx ) {

            BdbRef(BdbObject) mainLegacyObjectRef = containerContext->inObjects[idx][0];

          // A vector of "extra" objects will have all but "main" ones. Teh main object is
          // supposed to be the first one in the input list, so we have to skip it.

            std::vector< BdbRef(BdbObject) >::iterator beginItr = containerContext->inObjects[idx].begin( );
            ++beginItr;
            std::vector< BdbRef(BdbObject) >::iterator   endItr = containerContext->inObjects[idx].end( );

            std::vector< BdbRef(BdbObject) > extraLegacyObjectRef( beginItr, endItr);

            CdbRooRoObjectIdR outputObjectId  = containerContext->outObjects[idx];

            if( debugModeFlag )
                cout << "  " << outputObjectId << " <-- " << mainLegacyObjectRef.sprint( ) << " : \"" << containerContext->converterPtr->inputClassName( ) << "\"" << endl;

          // Use the converter
          //
          // DESIGN NOTE:
          //
          //   1. The actual object will be created and stored at a location known to
          //   above constructed TBranch inside the converter. We just need to make sure
          //   that the object is saved (see TTree::Fill()) operation in the current TTree,
          //
          //   2. Remember that the converter manages the lifetime of the created transient objects.

            {
                if( CdbStatus::Success != ( result = containerContext->converterPtr->conversion( mainLegacyObjectRef,
                                                                                                 extraLegacyObjectRef ))) {
                    cerr << errorStr << endl
                         << "    Failed to convert the main legacy object " << mainLegacyObjectRef.sprint( ) << endl;
                    return result;
                }

              // Store the object in the Tree/Branch

                containerContext->tree.Fill( );

              // Make sure that the next available index in the branch matches the object index
              // in its OID.

                if( debugModeFlag )
                    cout << "  tree->GetNbranches( ) = " << containerContext->tree.GetNbranches( ) << endl
                         << "  tree->GetEntries  ( ) = " << containerContext->tree.GetEntries( ) << endl
                         << "branch->GetEntries  ( ) = " << branch->GetEntries( ) << endl;

                Long64_t objectIndexInBranch = branch->GetEntries( ) - 1;

                if(( objectIndexInBranch < 0 ) ||
                   ( objectIndexInBranch >= 0xFFFF ) ||
                   ( objectIndexInBranch != outputObjectId.index( ))) {

                    cerr << errorStr << endl
                         << "    The just converted object " << outputObjectId << " produced from the legacy object " << mainLegacyObjectRef.sprint( ) << endl
                         << "    was writtent into a branch with a noon-valid index " << objectIndexInBranch << endl;

                    return CdbStatus::Error;
                }
            }
        }

      // Save the tree in the file

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

        Int_t numBytesWritten = databasePtr->WriteObject( &containerContext->tree,
                                                          containerContext->tree.GetName( ));
        if( numBytesWritten <= 0 ) {
            cerr << errorStr << endl
                 << "    Failed to save the tree \"" << containerContext->tree.GetName( ) << "\" in the file: " << databasePtr->GetName( ) << endl;
            return CdbStatus::Error;
        }
        if( debugModeFlag )
            cout << endl
                 << "Total of " << nIntervals << " converted and stored into the tree \"" << containerContext->tree.GetName( ) << "\"." << endl
                 << endl;

      // Destroy the context

        delete containerContext;
    }

  // Process the extra conditions

    for( unsigned int conditionIdx = 1; conditionIdx < nConditions; ++conditionIdx ) {

        for( std::map< UShort_t, CdbBdb2RooPayloadExtraContainerContext* >::iterator itr = executionPlan[conditionIdx].begin( );
                                                                                     itr != executionPlan[conditionIdx].end( );
                                                                                   ++itr ) {

            CdbBdb2RooPayloadExtraContainerContext* containerContext = (*itr).second;

          // Get the corresponding TBranch for that class

            const char* branchName = "0";   // always use this branch name

            CdbRooObjectHasMovedR* dummyObjectPtr = 0;

            TBranch* branch = containerContext->tree.Branch( branchName,
                                                             "CdbRooObjectHasMovedR",
                                                             &dummyObjectPtr );

            UInt_t nIntervals = containerContext->inObjects.size( );
            assert( nIntervals > 0 );

            for( UInt_t idx = 0; idx < nIntervals; ++idx ) {

                BdbRef(BdbObject)& legacyObjectRef = containerContext->inObjects[idx];
                CdbRooRoObjectIdR& outputObjectId  = containerContext->outObjects[idx];
                CdbRooRoObjectIdR&   mainObjectId  = containerContext->mainObjects[idx];

                if( debugModeFlag )
                    cout << "  " << mainObjectId << " <== " << outputObjectId << " <-- " << legacyObjectRef.sprint( ) << " : \"" << legacyObjectRef.typeName( ) << "\"" << endl;

              // Store the same dummy object
              //
              // NOTE: We're not using any converters since we already have the final
              //       persistent class in the current framework.

                {
                    if( 0 != dummyObjectPtr ) delete dummyObjectPtr;
                    dummyObjectPtr = new CdbRooObjectHasMovedR( mainObjectId.toString( ));

                  // Store the object in the Tree/Branch and get rif of a temporary transient object

                    containerContext->tree.Fill( );

                    if( 0 != dummyObjectPtr ) delete dummyObjectPtr;
                    dummyObjectPtr = 0;

                  // Make sure that the next available index in the branch matches the object index
                  // in its OID.

                    if( debugModeFlag )
                        cout << "  tree->GetNbranches( ) = " << containerContext->tree.GetNbranches( ) << endl
                             << "  tree->GetEntries  ( ) = " << containerContext->tree.GetEntries( ) << endl
                             << "branch->GetEntries  ( ) = " << branch->GetEntries( ) << endl;

                    Long64_t objectIndexInBranch = branch->GetEntries( ) - 1;

                    if(( objectIndexInBranch < 0 ) ||
                       ( objectIndexInBranch >= 0xFFFF ) ||
                       ( objectIndexInBranch != outputObjectId.index( ))) {

                        cerr << errorStr << endl
                             << "    The just converted object " << outputObjectId << " produced from the legacy object " << legacyObjectRef.sprint( ) << endl
                             << "    was writtent into a branch with a noon-valid index " << objectIndexInBranch << endl;

                        return CdbStatus::Error;
                    }
                }
            }
            if( 0 != dummyObjectPtr ) delete dummyObjectPtr;
            dummyObjectPtr = 0;

          // Save the tree in the file

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

            Int_t numBytesWritten = databasePtr->WriteObject( &containerContext->tree,
                                                              containerContext->tree.GetName( ));
            if( numBytesWritten <= 0 ) {
                cerr << errorStr << endl
                     << "    Failed to save the tree \"" << containerContext->tree.GetName( ) << "\" in the file: " << databasePtr->GetName( ) << endl;
                return CdbStatus::Error;
            }
            if( debugModeFlag )
                cout << endl
                     << "Total of " << nIntervals << " converted and stored into the tree \"" << containerContext->tree.GetName( ) << "\"." << endl
                     << endl;

          // Destroy the context

            delete containerContext;
        }
    }
    return CdbStatus::Success;
}

CdbBdb2RooPayloadConvertersDict*
CdbBdb2RooPayloadConversionFwk::regularConvertersDict( )
{
    static CdbBdb2RooPayloadConvertersDict myDict;
    return &myDict;
}

CdbBdb2RooPayloadConvertersDict*
CdbBdb2RooPayloadConversionFwk::mergingConvertersDict( )
{
    static CdbBdb2RooPayloadConvertersDict myDict;
    return &myDict;
}

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

---