// ----------------------------------------------------------------------
//
// HepHistoNtuple.cc - implementation of Histo n-tuple
//
// HepHistoNtuple is the implementation of the Histo n-tuple
// It is derived from HepNtuple (the generic interface).
// It is mainly a C++ wrapper around the CERNlib Histo
// Fortran histogramming libraries.
//
// No HepHistoNtuple objects are ever directly
// instantiated by the programmer. When creating new HepNtuple* objects,
// the manager is in charge of creating the appropriate HepHisto object.
//
// To use generate Histo n-tuple, HepHistoFileManager must be used:
//
//   int lunit = 10;
//   HepFileManager* m = new HepHistoFileManager("filename.rz", lunit);
//    :
//    :
//   HepNtuple & h(m->newHepNtuple("Title"));
//   [define HepNtuple]
//    :
//   [gather data]
//    :
//   h->capture();
//
//
// Note that the constructor can take an ID number which is used by
// the Histo implementation of HepNtuple. If omitted, the ID number is
// generated automatically (which is recommended). The ID number is
// available through the id() method.
//
//
// History:
//   20-Aug-1998 Philippe Canal Bare Creation
//
// ----------------------------------------------------------------------

#ifndef HEPTRACE_H
#include "HepTuple/HepTrace.h"
#endif

#ifndef HepHistoTuple_H
#include "HepTuple/HepHistoNtuple.h"
#endif

#include "HepTuple/HistoNameTag.h"

#include "ZMutility/iostream"
#include "ZMutility/itos.h"

#include <string>
USING( std::string )

#ifndef HEPCOLUMN_H
#include "HepTuple/Column.h"
#endif

#ifndef HEPFILEMANAGER_H
#include "HepTuple/HepFileManager.h"
#endif

#ifndef HEPCOLUMN_H
#include "HepTuple/Column.h"
#endif

#ifndef HEPBLOCK_H
#include "HepTuple/Block.h"
#endif

#include "histoscope.h"

#ifndef HEPHISTOCOLUMN_H
#include "HepTuple/HistoColumn.h"
#endif

// this is defined in nirvana/histo_util/histo.h 
// but is not available in the 'release' package
#define HS_MAX_CATEGORY_LENGTH 255
#define HS_MAX_TITLE_LENGTH 80
#define HS_MAX_NAME_LENGTH 80

ZM_BEGIN_NAMESPACE( zmht )	/*  namespace zmht  {  */

// helpers

// for automatic creation of !=
#include "ZMutility/algorithm"

static bool operator ==(hsBlockId a, hsBlockId b) {
  return ( a.id == b.id
	   && a.rank == b.rank );
}

static bool operator !=(hsBlockId a, hsBlockId b) {
  return ( a.id != b.id
	   || a.rank != b.rank );
}

static bool operator ==(struct _hsColId a, struct _hsColId b) {
  return ( a.id == b.id
	   && a.rank == b.rank);
}

static ColumnData_t ToColumnData_t( _ColumnType hs_type ) {
  switch( hs_type ) {
  case hsbool_t: return Bool_ct;
  case hsChar:   return Int1_ct;
  case hsuInt:   return Uint4_ct;
  case hsInt:    return Int4_ct;
  case hsFloat:  return Float4_ct;
  case hsDouble: return Float8_ct;
  case hsString: return Char32_ct; // extended to inf for Histoscope!
  default: return ColumnData_t_End;
  }
}

static ColumnType ToHistoType( ColumnData_t hep_type ) {
  switch( hep_type ) {  
  case Int1_ct:    return hsChar;
  case Int2_ct:    
  case Int4_ct:    return hsInt;
  case Int8_ct:    return hsLastType;
  case Uint1_ct:   
  case Uint2_ct:   
  case Uint4_ct:   
  case Uint8_ct:   return hsuInt;
  case Float4_ct:  return hsFloat;
  case Float8_ct:  return hsDouble;
  case Float16_ct: return hsLastType;
  case Bool_ct:    
  case Bool2_ct:   
  case Bool4_ct:   return hsbool_t;//compatible to Logical*2: *4 in HBOOK
  case Char_ct:    return hsChar;
  case Char4_ct:   
  case Char8_ct:   
  case Char12_ct:  
  case Char16_ct:  
  case Char20_ct:  
  case Char24_ct:  
  case Char28_ct:  
  case Char32_ct:  return hsString;
  default:         return hsLastType;
  }
}


// create a ntuple

HepHistoNtuple::HepHistoNtuple(
    HepHistoFileManager * manager
  , const string& title
  , const int idReq
  , const int histo_id
  ) : HepNtuple( manager, title, idReq ),
      _nColumnCreated(0),
      _lastRwnArray(0),
      _lastArrayRead(0),
      _lastRowRead(-1),
      histo_id_(histo_id),isVersion5_(true)
{
  HEP_DEBUG( "HepHistoNtuple::constructor( \"" << title << "\", " << idReq << " )" );
  // set the default values.
  setColumnWise();
  setDiskResident();
  allowNewColumns();

}  //  HepHistoNtuple::HepHistoNtuple()


// clone an existing Tuple
HepHistoNtuple::HepHistoNtuple(
    HepNtuple * original
  , const string& title
  , const int     idReq
  ) : HepNtuple( original->manager(), title, idReq ),
      _nColumnCreated(0),
      _lastRwnArray(0),
      _lastArrayRead(0),
      _lastRowRead(-1)
{
  HEP_DEBUG( "HepHistoNtuple::constructor( \"" << original->title() << 
	 "\", " << title << "\", " << idReq << " )" );
  // make the Fortran object

  histo_id_ = hs_copy(((HepHistoNtuple*)original)->histo_id_, idReq, 
  		      (char*)title.c_str(), 
	       (char*)HepHistoFileManager::category(manager()->pwd()).c_str() );
  isVersion5_ = hs_type(histo_id_) != HS_NTUPLE;

}  //  HepHistoNtuple::HepHistoNtuple()

// Used for dummy construction ...
// the object is then not valid ...
HepHistoNtuple::HepHistoNtuple() : HepNtuple() {
  HEP_DEBUG( "HepHistoNtuple::constructor( )" );
} // HepHistoNtuple::HepHistoNtuple()


HepHistoNtuple::~HepHistoNtuple()  {
  HEP_DEBUG( "HepHistoNtuple::destructor( ) on ntuple #" << id_ );
} // HepHistoNtuple::~HepHistoNtuple()

////////////////////////////////////
//    isXxxxxEnabled() family     //
//                                //
// virtual function indicating    //
// which part of the interface    //
// is supported                   //
////////////////////////////////////

bool HepHistoNtuple::isRowWiseEnabled() const {
  return true;
}

bool HepHistoNtuple::isColumnWiseEnabled() const {
  return true;
}

bool HepHistoNtuple::isBlockWiseEnabled() const {
  return true;
}

bool HepHistoNtuple::isDiskResidentEnabled() const {
  return true;
}

bool HepHistoNtuple::isMemResidentEnabled() const {
  return true;
}

bool HepHistoNtuple::isSharedMemoryEnabled() const {
  return false;
}

bool HepHistoNtuple::isNwbuffEnabled(int nwbuff) const {
  return false;
}

bool HepHistoNtuple::isBuffLimitEnabled(int limit) const {
  return true;
}

bool HepHistoNtuple::isDemandBuffersEnabled() const {
  return false;
}

bool HepHistoNtuple::isCircularBufferEnabled() const {
  return false;
}

bool HepHistoNtuple::isCaseSensitive() const {
  return true;
}

//////////////////////////////////////////////////////
//  isTypeEnabled                                   //
//////////////////////////////////////////////////////

// These methods tell if a manager can handle a certain type in
// the given block.
// It relies on the enum ColumnData_t to be ordered correctly
// i.e.: numerical type first, then boolean, then character strings

bool HepHistoNtuple::isTypeEnabled(ColumnData_t type,Block *bk) {
  return isTypeEnabled(type);
} // HepHistoNtuple::isTypeEnabled

bool HepHistoNtuple::isTypeEnabled(ColumnData_t type) {
  switch(type) {
  case Uint8_ct:
  case Int8_ct: 
  case Float16_ct: return false;
  case Int1_ct:
  case Int2_ct: return true;
  case Char_ct:
  case Char4_ct:
  case Char8_ct:
  case Char12_ct:
  case Char16_ct:
  case Char20_ct:
  case Char24_ct:
  case Char28_ct:
  case Char32_ct: return true;
  case Ptr_ct: return false;
  default: return true;
  }
} // HepHistoNtuple::isTypeEnabled

//////////////////////////////////////////////////////
//  setStorageGeometry                              //
//////////////////////////////////////////////////////

bool HepHistoNtuple::setStorageGeometry(StorageGeometryName newStrategy) {
  if ( (newStrategy != storageGeometry()) &&
       isCreated() ) {
      ZMthrow(ZMxHepLockedItem("Unable to change storage geometry after first booking"));
      return false;
  }
  return HepNtuple::setStorageGeometry(newStrategy);
} // HepHistoNtuple::setStorageGeometry

//////////////////////////////////////////////////////
//  setStorageLocation                              //
//////////////////////////////////////////////////////

bool HepHistoNtuple::setStorageLocation(unsigned char newLocation) {
  if ( (newLocation != resident() )
       && isCreated() ) {
    ZMthrow(ZMxHepLockedItem("Unable to change storage location after first booking"));
    return false;
  }
  if ( (newLocation == 'D' && manager()->fileName() == "" )
       || (newLocation == 'M' && manager()->fileName() != "" ) ) {
    ZMthrow(ZMxHepUnsupported("Cannot switch to residence. The HepHistoFileManager filename locks it"));
    return false;
  }
  return HepNtuple::setStorageLocation(newLocation);
} // HepHistoNtuple::setStorageLocation

//////////////////////////////////////////////////////
//  setNwbuffValue                                  //
//////////////////////////////////////////////////////

bool HepHistoNtuple::setNwbuffValue(int nwbuff) {
  ZMthrow(ZMxHepBadRequest("Histoscope can not set nwbuff"));
  return HepNtuple::setNwbuffValue(nwbuff);
} // HepHistoNtuple::setNwbuffValue

//////////////////////////////////////////////////////
//  setBuffLimitValue                               //
//////////////////////////////////////////////////////

bool HepHistoNtuple::setBuffLimitValue(int limit) {
  hs_cwn_column_chunk_size = limit / 1024;
  return HepNtuple::setBuffLimitValue(limit);
} // HepHistoNtuple::setBuffLimitValue

/*******************/
/*                 */
/* isNameAvailable */
/*                 */
/*******************/

//
// This version verifies that the block::column name in the nameTag 
// is unique in the HepTuple

bool
HepHistoNtuple::isNameAvailable ( const string& nameTag ) const
{
  TupleNameTag tag;
  parseNameTag(tag, nameTag);
  bool result;
  if ( tag.isValid() ) {
    if ( tag.block().length() != 0 ) {
      Block *bk = findBlock(tag.block());
      if ( bk == NULL ) {
	return false;
      }
      result = (bk->findColumn(tag.column()) == NULL);
    } else {
      result = (findColumn(tag.column()) == NULL);
    }
    return result;
  } else {
    return false;
  }
} // HepHistoNtuple::isNameAvailable

/****************/
/*              */
/* parseNametag */
/*              */
/****************/

//
// Parse the nametag for a column, providing block and column names.
// return a TupleNameTag with the method isValid(), block() and column()
// returning the proper result

const TupleNameTag&
HepHistoNtuple::parseNameTag (TupleNameTag& what,
			      const string& nametag,
			      bool withIndices) const {
  static HistoNameTag tag;
  tag.init(nametag, withIndices, what);
  return what;
} /* parseNameTag */

/****************/
/*              */
/* bookRowWise  */
/*              */
/****************/

// Boook a row wise N-Tuple in Histo.  If the n-tuple was already
// created we must delete the old ntuple and create a new one.
// since the user may have specified the id, keep the id constant

bool HepHistoNtuple::bookRowWise() {
  HEP_DEBUG( "HepHistoNtuple::bookRowWise" );

  int nrows=0, ncol=0;           // in case of deleting and recreating
  float *dat = 0;
  int id = histo_id_;
  int nColumns = HepNtuple::nColumns();

  if ( isCreated() ) {
    // somebody added a column after the histo ntuple was created
    // we must delete the old ntuple and create a new one.
    // since the user may have specified the id, keep the id constant
    ncol = hs_num_variables(id);
    nrows = hs_num_entries(id);
    dat = new float[nrows*ncol];
    hs_ntuple_contents(id , dat );
    hs_eliminate(id);
  } // if ( isCreated() )

  isCreated(true);
  _nColumnCreated = nColumns;

  const char **labels = new const char*[nColumns];
  Column* col;
  int colNum = 0;
  std::STL_VECTOR(Column*) colVec(nColumns);
  for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	iterateBlock != blocks.end();
	iterateBlock++ ) {
    Block *bk = (*iterateBlock).second;
    for ( Block::columnsVec::const_iterator iterateCol(bk->columns().begin());
	  iterateCol != bk->columns().end();
	  iterateCol++ ) {
      col = (*iterateCol);
      colVec[colNum] = col;
     
      labels[colNum] = col->name().c_str();
      // At the same time we verify that all column are floats
      if ( col->attributes().type != Float4_ct ) {
	ZMthrow(ZMxHepUnsupported(
		"Histo Row Wise N-Tuple can only contain column of floats"));
	return false;
      }
      colNum++;
    } // for each Columns
  } // for each Blocks
  
  histo_id_ = hs_create_rwntuple( HepNtuple::id(), (char*)title().c_str(),
				  (char*)HepHistoFileManager::category(*dir_).c_str(),
				  nColumns, (char**)labels );
  delete labels;

  if (dat && histo_id_ !=0) {
    // need to put the data back in
    float *d = new float[nColumns];

    // Initialize the data to the default values
    int i = 0;
    Column *col;
    for ( i = 0; i < nColumns; i++ ) {
      col = colVec[i];
      if ( col->attributes().defaultVal != 0 ) {
	memcpy(d+i,(col->attributes().defaultVal),sizeof(float));
      }
    } // for each columns

    for (int r=1; r<=nrows; r++) {
      memcpy(d, &dat[(r-1)*ncol], ncol*sizeof(float) );

      hs_fill_ntuple( histo_id_, d );
    } // for each row
    delete [] dat;
    delete [] d;
  } // if (dat) 

  return (histo_id_!=0);
} // HepHistoNtuple::bookRowWise

/******************/
/*                */
/* bookColumnWise */
/*                */
/******************/

// Do booking for column wise N-Tuple in Histo.
// If a block happen to have been changed since booked, we
// simply IGNORE it (actually we do not know about it ...)

#define HS_TAG(BLOCK,tag) ( (char*)					   \
			    (BLOCK->name()+TupleNameTag::NameTagDelimiter+ \
			     tag).c_str() )

bool HepHistoNtuple::bookColumnWise() {
  HEP_DEBUG( "HepHistoNtuple::bookColumnWise" );

  if ( ! isCreated() ) {
    histo_id_ = hs_create_cwntuple( HepNtuple::id(), (char*)title().c_str(), 
				    (char*)
                                  HepHistoFileManager::category(*dir_).c_str());
    if ( !histo_id_ ) {
      ZMthrow(ZMxHepCantMakeNtuple("Histoscope failed to created ntuple"));
      return false;
    }
    isCreated( true);
  }
  // Now the N-Tuple is created add all the blocks that have not
  // been created yet.

  Block* bk;
  for ( blocksMap::iterator iterateBlock = blocks.begin();
	iterateBlock != blocks.end();
	iterateBlock++ ) {
    bk = (*iterateBlock).second;
    if ( ! bk->isBooked() ) {

//  Check if the current column is indexed by some other column. If so,
//  verify that the index column span matched the specified array size.

      for (std::string::size_type i=0; i < bk->columns().size(); i++ ) {
        Column *c = bk->columns()[i];
        string colName = c->name();
        string indexName = c->getIndex();
	if( indexName.size() != 0 ) {
          int * pdim = new int[10];
          int numDim = c->getDimension(pdim);
          if( numDim != 0 ) {
            Column * ci = findColumn(indexName);
            ColumnAttribs CAI = ci->attributes();
            if( *(pdim+numDim-1) != CAI.indexHi ) 
              ZMthrow(ZMxHepIllFormedItem(colName+string("'s dimension [0,")+
                      itos(*pdim)+string("] does not match the span [0,")+
                      itos(CAI.indexHi)+string("] of its index, ")+indexName));
          }
        delete [] pdim;
        }
      }
      if ( nullBlockId == hs_cwn_add_block( histo_id_, 
					    (char*)bk->name().c_str() )

           ) {
	ZMthrow(ZMxHepCantMakeBlock(string("Unable to book ")
				    +bk->name()
				    +" through Histoscope"));
	return false;
      }		      
      bk->book();

      for (std::string::size_type i=0; i < bk->columns().size(); i++ ) {
	Column *c = bk->columns()[i];
	ColumnAttribs ca = c->attributes();
	int ndim;
	if ( ca.ndim == 0 ) {
	  ndim = 1;
	  ca.extents[0] = 1;
	} else {
	  ndim = ca.ndim;
	}
	hsColId newcol =  hs_cwn_add_column( histo_id_, 
					     HS_TAG(bk,ca.tag),
					     ToHistoType(ca.type),
					     ca.defaultVal,
					     ndim, 
					     ca.extents );
	if ( nullColId == newcol ) {
	  ZMthrow(ZMxHepCantMakeColumn(string("Unable to book ")
				       +c->attributes().tag
				       +" through Histoscope"));
	  return false;
	}
	if (! hs_cwn_set_column_packing( newcol, ca.nbits,
					 ca.rangeLo, ca.rangeHi ) ) {
	  // NOTE: error handling
	}
	c->internals( new HistoColInternals(newcol) );
      }
      // NOTE: HepNtuple ONLY let you use ONE index located in 
      // the SAME block (this COULD change in the near futur but
      // we take advantage of it here.
      for (std::string::size_type i=0; i < bk->columns().size(); i++ ) {
	Column *c = bk->columns()[i];
	ColumnAttribs ca = c->attributes();
	hsColId newcol = ((HistoColInternals*)(c->internals()))->id();
	hsColId index_id = hs_cwn_get_column(histo_id_, HS_TAG(bk,ca.index) );
	if (! hs_cwn_set_column_index( newcol, 0, index_id ) ) {
	  // NOTE: error handling
	}
      }
    } // end of if not block booked
  } // end of for each block loop

  return true;
} // end of HepHistoNtuple::bookColumnWise

/**********/
/*        */
/* book   */
/*        */
/**********/

bool HepHistoNtuple::book() {
  if ( ! isDefinitionChanged() ) {
    return true;
  }
  if ( !mayUse() ) {
    ZMthrow(ZMxHepUnmanagedItem("attempt to book an unmanaged ntuple"));
    return false;
  }
  if ( !isValid() ) {
    ZMthrow(ZMxHepVacuousItem("attempt to book an invalidated ntuple"));
    return false;
  }
  if ( isRowWise() ) {
    bookRowWise();
  } else {
    // isColumnWise()
    bookColumnWise();
  }
  // this should make isDefinitionChanged() false
  return HepNtuple::book();
} // HepHistoNtuple::book()

/*******************/
/*                 */
/* storeWholeRow   */
/*                 */
/*******************/

bool HepHistoNtuple::storeWholeRow() {
  HEP_DEBUG ( "HepHistoNtuple::storeWholeRow" );

  blocksMap::iterator iterateBlock;
  bool status = true;

  if ( isRowWise() ) {

    // First get the data in a single block
    float * d = NULL; 
    bool needToDelete_d = false;
    if (nBlocks() == 1) {
      Block *bk = findBlock(blockName(0));
      d = (float*)bk->capturedData();
    } else {
      // Humm somehow more than one block made it into this
      // row-wise ntuple.  We have to gather the information
      // in a chunk of memory. (row wise ntuple in histo have
      // only one "block" of columns)
      Block* bk;
      int ncol = 0;

      // Count the number of columns 
      // NOTE: isn't it the values of this->nColumns() ?
      for ( iterateBlock = blocks.begin();
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	bk = (*iterateBlock).second;
	ncol += bk->nColumns();
      }

      // Copy the columns data in one memory chunk
      d = new float[ncol];
      needToDelete_d = true;
      ncol = 0;
      for ( iterateBlock = blocks.begin();
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	bk = (*iterateBlock).second;
	for ( int i = 0; i < bk->nColumns() ; i++ ) {
	  // the array return by capturedData is linear collation of
	  // all the elements of the column in the block with no
	  // insurance that it is aligned (thus the memcpy instead
	  // of d[ncol+i] = (bk->capturedData())[i]
	  memcpy(&(d[ncol+i]),
		 (char*)bk->capturedData()+(i*sizeof(float)),sizeof(float));
	}
	ncol += bk->nColumns();
      }
    }

    if ( ! hs_fill_ntuple( histo_id_, d ) ) {
      ZMthrow(ZMxHepCantWriteFile("Unable to fill row wise ntuple"));
      return false;
    }

    if ( needToDelete_d ) delete [] d; /* in case of more than one block */

  } // end of if ( isRowWise() ) 
  else {
    
    for ( iterateBlock = blocks.begin();
	  iterateBlock != blocks.end();
	  iterateBlock++ ) {
      Block *bk = (*iterateBlock).second;

      HistoColInternals* internals;
      for ( Block::columnsVec::const_iterator iterateCol(bk->columns().begin());
	    iterateCol != bk->columns().end();
	    iterateCol++ ) {
	Column* c = (*iterateCol);
	internals = (HistoColInternals*)(c->internals());
	
	status &= hs_fill_cwn_column( internals->id(), c->getValueAddr() );
	
      }
    }
    hs_cwntuple_advance( histo_id_ );
   // isColumnWise()
  }

  columnsMap::iterator iterateCol = columns.begin();
  while ( ! (iterateCol == columns.end()) ) {
    ((*iterateCol).second)->setHasBeenStored();
    iterateCol++;
  }

  return status;
} // HepHistoNtuple::storeWholeRow

/*********************/
/*                   */
/* storeSingleBlock  */
/*                   */
/*********************/

bool HepHistoNtuple::storeSingleBlock(Block* bk) {
  HEP_DEBUG( " HepHistoNtuple::storeSingleBlock " );

  for ( Block::columnsVec::const_iterator iterateCol(bk->columns().begin());
	iterateCol != bk->columns().end();
	iterateCol++ ) {
    (*iterateCol)->setHasBeenStored();
  }
  if ( isRowWise() ) {
    // Storing a single block have to mean store the whole row.
    return storeWholeRow();
  } else {
    bool status = true;

    // isColumnWise()
    HistoColInternals* internals;
    for ( Block::columnsVec::const_iterator iterateCol2(bk->columns().begin());
	  iterateCol2 != bk->columns().end();
	  iterateCol2++ ) {
      Column* c = (*iterateCol2);
      internals = (HistoColInternals*)(c->internals());
	
      status &= hs_fill_cwn_column( internals->id(),
				    c->getValueAddr() );
						       
    }
    return status;
  }
} // HepHistoNtuple::storeSingleBlock


/*********************/
/*                   */
/* read methods      */
/*                   */
/*********************/

// Instruct the manager to read from file/memory into the destination variable
// of all the columns of repectively the row, block, single column For the
// first three methods ALL the listed columns are expected to have destination
// variables (i,e. get the value at an address and send it to col->set(void
// *addr);

/*****************************/
/*                           */
/* helper Histo routine      */
/*                           */
/*****************************/

// read a Row Wise Ntuple row
// return the Histo error code.
// array must point to a array of float of size NVAR

Float4* HepHistoNtuple::readRWNrow ( int row ) {
  HEP_DEBUG( "HepHistoNtuple::readRWNrow" );

  string routineName ( "readRWNrow" );

  // verify if we already read this row 

  if ( row == _lastRowRead ) {
    return _lastRwnArray;
  }

  // get array for data

  Float4* array = _lastRwnArray;

  if ( array == NULL ) {
    _lastRwnArray = new Float4[nColumns()];
    array = _lastRwnArray;
  }

  // read the row

  hs_row_contents(histo_id_, row - 1, array );

  return array;
} // HepHistoNtuple::readRWNrow

bool HepHistoNtuple::restoreNtuple() {
  
  isVersion5_ = true;
  bool result = true;
  bool changed = isDefinitionChanged();
  int i;

  // Determine the type of ntuple

  switch( hs_type(histo_id_) ) {
  case HS_NTUPLE:
    isVersion5_ = false;
  case HS_RWNTUPLE:
    // Tell the HepNtuple object that it is a row wise ntuple.
    setRowWise();
    break;
  case HS_CWNTUPLE:
    // Tell the HepNtuple object that it is a column wise ntuple.
    setColumnWise(); break;
  default:
    ZMthrow(ZMxHepCantMakeNtuple("Wrong item type"));
    isValid(false);
    return false;
  }

  if ( isColumnWise() ) {
    char blockname[HS_MAX_NAME_LENGTH];
    for(i=0; i<hs_cwn_number_of_blocks(histo_id_); ++i) {
      if ( hs_cwn_block_name(histo_id_, i, blockname) != nullBlockId ) {
	Block * bk;
	bk = new Block(this,blockname,isCaseSensitive());
	result &= this->addBlock(bk);
      } else {
	result = false;
      }
    }
    for(i=0; i<hs_cwn_number_of_columns(histo_id_); ++i) {
      char columntag[2*HS_MAX_NAME_LENGTH+2];      
      hsColId hs_col = hs_cwn_column_tag( histo_id_, i, columntag );
      ColumnType type = hs_cwn_column_type( hs_col );

      TupleNameTag tag;
      parseNameTag( tag, columntag);
     
      if ( ! tag.isValid() ) {
	result = false;
	break;
      };

      ColumnAttribs ca(tag.column(), ToColumnData_t (type) );
      
      ca.ndim = hs_cwn_column_array_ndim( hs_col );
      if ( ca.ndim > 10 ) {
	ZMthrow(ZMxHepTooBig("HepHistoNtuple does not support more than 10 dimension in a column array"));
	result = false;
	continue;   // ignore this column if we can't use it.
      }
      hs_cwn_column_array_dimensions( hs_col, ca.extents );
      if ( ca.ndim == 1 && ca.extents[0] == 1 ) {
	// Ntuple expect ca.ndim = 0 for single elements columns
	ca.ndim = 0;
      }
      double low, high;
      hs_cwn_get_column_packing( hs_col, &(ca.nbits), &low, &high);
      ca.rangeLo = low;
      ca.rangeHi = high;

      // NOTE: indexLo and indexHi are irrevelant for histoscope

      // NOTE: VOLUNTARILY ignore the INDEX because NTuple
      // it self do not care ... 
      // NOTE: THIS HOWEVER need to be dealt with when 
      // duplicating to another manager

      Block *bk = findBlock( tag.block() );
      Column* newCol = Column::newColumn(bk,ca);
      newCol->internals( new HistoColInternals( hs_col ) );
 
      if ( newCol == 0 ) result = false;
      if ( result ) result &= addColumn(bk,newCol);
    }
    //  AFTER creating all the columns, set the lock for each one.
    blocksMap::iterator iterateBlock = blocks.begin();
    while ( ! (iterateBlock == blocks.end()) ) {
      Block *bk = (*iterateBlock).second;
      for ( Block::columnsVec::const_iterator iterateCol(bk->columns().begin());
	    iterateCol != bk->columns().end();
	    iterateCol++ ) {
	Column* col = *iterateCol;
	col->setHasBeenStored();
      }
      bk->book();
      iterateBlock++;
    }
  }
  else {
    //  The thing on the file is an old fashion Row Wise.

    char columntag[2*HS_MAX_NAME_LENGTH+2];      
    int nCol = hs_num_variables( histo_id_ );
    Block * bk= this->unspecifiedBlock();
    for (int j = 0; j < nCol; ++j ) {
      if ( hs_variable_name( histo_id_, j, columntag ) ) {
	ColumnAttribs ca(columntag, Float4_ct);
	Column* newCol = Column::newColumn(bk,ca);
      
	if ( newCol == 0 ) result = false;
	if ( result ) result &= addColumn(bk,newCol);
      } else {
	result = false;
      }
    }
  }

//  If all is well so far, fill in the number of rows and set the addresses
  if( result ) {
    int nrows = hs_num_entries(histo_id_);
    result &= setNrowsValue( nrows );
    _nColumnCreated = nColumns();
  } else {
    isValid(false);
  }
  if( result ) isCreated( true);
  
  _changedDefinition = changed;

  return result;
} // HepHistoNtuple::restoreNtuple()

int HepHistoNtuple::readWholeRow(int irow) {
  HEP_DEBUG( "HepHistoNtuple::readWholeRow(int)" );


  // read the data in
  if ( isRowWise() ) {
    Float4* array = readRWNrow(irow);
    if ( array != NULL ) {
      // loop through all columns in order and set the values
      int i = 0;
      for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	Block *bk = (*iterateBlock).second;
	for ( Block::columnsVec::const_iterator 
		iterateCol(bk->columns().begin());
	      iterateCol != bk->columns().end();
	      iterateCol++ ) {
	  Column* col = (*iterateCol);
	  col->read(array+i);
	  i++;
	} // Block::columnsVec loop
      } // blocksMap loop
    } // if ( array != NULL )
  } // if ( isRowWise() )
  else { // isColumnWise
    bool status = true;
    
    for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	  iterateBlock != blocks.end();
	  iterateBlock++ ) {
      Block *bk = (*iterateBlock).second;
      HistoColInternals* internals;
      for ( Block::columnsVec::const_iterator 
	      iterateCol(bk->columns().begin());
	    iterateCol != bk->columns().end();
	    iterateCol++ ) {
	Column* col = (*iterateCol);
	internals = (HistoColInternals*)(col->internals());
	
	status &= hs_cwn_column_value( internals->id(), 
				       irow-1, col->getValueAddr());
	col->read();
      } // Block::columnsVec loop
    } // blocksMap loop
    
    if ( ! status ) {
      ZMthrow(ZMxHepCantReadFile("Unable to read CWN"));
      return 0;
    }
  } // isColumnWise()

  return nBlocks();
} // HepHistoNtuple::readWholeRow

int HepHistoNtuple::readWholeBlock(Block* bk, int irow)  {
  HEP_DEBUG( "HepHistoNtuple::readWholeBlock" );

  // Check if column has been stored 
  // NOTE: the above is not done !?

  // read the data in
  if ( isRowWise() ) {
    Float4* array = readRWNrow(irow);
    if ( array != NULL ) {
      // loop through all columns of the Block in order and set the values
      int i = 0;
      for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	Block *current = (*iterateBlock).second;
	if ( bk == current ) {
	  for ( Block::columnsVec::const_iterator iterateCol(bk->columns().begin());
		iterateCol != bk->columns().end();
		iterateCol++ ) {
	    Column * col = (*iterateCol);
	    col->read(array+i);
	    i++;
	  } // Block::columnsVec loop
	} // if ( bk == current )
	i += current->nColumns();
      } // blocksMap loop
    } // if ( array != NULL )
  } // if ( isRowWise() )
  else { // isColumnWise
    bool status = true;
    
    HistoColInternals* internals;
    for ( Block::columnsVec::const_iterator 
	    iterateCol(bk->columns().begin());
	  iterateCol != bk->columns().end();
	  iterateCol++ ) {
      Column* col = (*iterateCol);
      internals = (HistoColInternals*)(col->internals());

      status &= hs_cwn_column_value( internals->id(), irow-1, col->getValueAddr());
      col->read();
    } // Block::columnsVec loop
    
    if ( ! status ) {
      ZMthrow(ZMxHepCantReadFile("Unable to read CWN"));
      return 0;
    }    
  } // isColumnWise()  

  return bk->nColumns();
} // HepHistoNtuple::readWholeBlock(


int HepHistoNtuple::readSingleColumn(Column* col, int irow) {
  HEP_DEBUG( "HepHistoNtuple::readSingleColumn" );

  // Check if column has been stored
  // NOTE: the above is not done !?


  // read the data in
  if ( isRowWise() ) {
    Float4* array = readRWNrow(irow);
    if ( array != NULL ) {
      // find the column number
      int i = 0;
      int colNumber = 0;
      for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	Block *current = (*iterateBlock).second;
	for ( Block::columnsVec::const_iterator 
		iterateCol(current->columns().begin());
	      iterateCol != current->columns().end();
	      iterateCol++ ) {
	  Column * current = (*iterateCol);
	  if ( current == col ) {
	    colNumber = i;
	  }
	  i++;
	} // Block::columnsVec loop
      } // blocksMap loop

      // set the values
      col->read(&(array[colNumber]));
    }
  }
  else { // isColumnWise
    HistoColInternals* internals;
    internals = (HistoColInternals*)(col->internals());

    bool status = hs_cwn_column_value( internals->id(), 
				       irow-1, col->getValueAddr());

    if ( ! status ) {
      ZMthrow(ZMxHepCantReadFile("Unable to read CWN"));
      return 0;
    }

    // Send the column values to the destination
    col->read();

  } // if isColumnWise()

  return true;
} // readSingleColumn

int HepHistoNtuple::readSingleColumn(Column* col, int irow, void * val) {
  HEP_DEBUG( "HepHistoNtuple::readSingleColumn" );


  // read the data in
  if ( isRowWise() ) {
    Float4* array = readRWNrow(irow);
    if ( array != NULL ) {
      // find the column number
      int i = 0;
      int colNumber = 0;
      for ( blocksMap::const_iterator iterateBlock (blocks.begin());
	    iterateBlock != blocks.end();
	    iterateBlock++ ) {
	Block *bk = (*iterateBlock).second;
	for ( Block::columnsVec::const_iterator 
		iterateCol(bk->columns().begin());
	      iterateCol != bk->columns().end();
	      iterateCol++ ) {
	  Column * current = (*iterateCol);
	  if ( current == col ) {
	    colNumber = i;
	  }
	  i++;
	} // Block::columnsVec loop
      } // blocksMap loop

      // set the values
      *((Float4*)val) = array[colNumber];
    }
  }
  else { // isColumnWise
    HistoColInternals* internals;
    internals = (HistoColInternals*)(col->internals());

    bool status = hs_cwn_column_value( internals->id(), 
				       irow-1, val);

    if ( ! status ) {
      ZMthrow(ZMxHepCantReadFile("Unable to read CWN"));
      return 0;
    }

    // Grab the column value.
    // since we passed val to Histoscope, we do not need the following.
    // its already done !
    // col->copyToMem(val);

  } // isColumnWise()
 
  return 1;
} // readSingleColumn with value

int HepHistoNtuple::readSingleColumn(Column* col,
				     const string& indices,
				     int irow,
				     void * val) {
  HEP_DEBUG( "HepHistoNtuple::readSingleColumn" );

  if ( isRowWise() ) {
    ZMthrow(ZMxHepUnsupported(
	       "Histo Row Wise Ntuple do not handle array of Columns"));
    return 0;
  } else {
    HistoColInternals* internals;
    internals = (HistoColInternals*)(col->internals());

    bool status = hs_cwn_column_value( internals->id(), 
				       irow-1, col->getValueAddr());

    if ( ! status ) {
      ZMthrow(ZMxHepCantReadFile("Unable to read CWN"));
      return 0;
    }

    // Now transfer the value to val
    return col->copyToMem(val,indices);

  }
} // readSingleColumn with indices


ZM_END_NAMESPACE( zmht )	/*  }  // namespace zmht  */