/*=========================================================================

 Program:   Visualization Toolkit
 Module:    vtkPGenericIOReader.cxx

 Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
 All rights reserved.
 See Copyright.txt or http://www.kitware.com/Copyright.htm for details.

 This software is distributed WITHOUT ANY WARRANTY; without even
 the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 PURPOSE.  See the above copyright notice for more information.

 =========================================================================*/
#include "vtkPGenericIOReader.h"

// VTK includes
#include "vtkCallbackCommand.h"
#include "vtkCellArray.h"
#include "vtkCellType.h"
#include "vtkCommand.h"
#include "vtkCommunicator.h"
#include "vtkDataArray.h"
#include "vtkDataArraySelection.h"
#include "vtkDataObject.h"
#include "vtkIdList.h"
#include "vtkInformation.h"
#include "vtkInformationVector.h"
#include "vtkMPI.h"
#include "vtkMPICommunicator.h"
#include "vtkMPIController.h"
#include "vtkMultiProcessController.h"
#include "vtkObjectFactory.h"
#include "vtkPointData.h"
#include "vtkPoints.h"
#include "vtkSmartPointer.h"
#include "vtkStdString.h"
#include "vtkStreamingDemandDrivenPipeline.h"
#include "vtkStringArray.h"
#include "vtkType.h"
#include "vtkTypeUInt64Array.h"
#include "vtkUnstructuredGrid.h"

#include "vtkGenericIOUtilities.h"

// GenericIO includes
#include "GenericIOReader.h"
#include "GenericIOMPIReader.h"
#include "GenericIOPosixReader.h"
#include "GenericIOUtilities.h"

// C/C++ includes
#include <algorithm>
#include <cassert>
#include <cctype>
#include <set>
#include <stdexcept>
#include <vector>

// Uncomment the line below to get debugging information
//#define DEBUG

//------------------------------------------------------------------------------
class vtkGenericIOMetaData
{
public:

  int NumberOfElements;
  std::map< std::string, gio::VariableInfo > Information;
  std::map< std::string, int >  VariableGenericIOType;
  std::map< std::string, bool > VariableStatus;
  std::map< std::string, void* > RawCache;
  MPI_Comm MPICommunicator;
  std::set< int > RanksToLoad;

  /**
   * @brief Metadata constructor.
   */
  vtkGenericIOMetaData(){};

  /**
   * @brief Destructor
   */
  ~vtkGenericIOMetaData() { this->Clear();};

  /**
   * @brief Checks if the supplied rank should load data.
   * @param r the rank in query.
   * @return status true or false.
   */
  bool LoadRank( const int r )
  {
    return( (this->RanksToLoad.find(r) != this->RanksToLoad.end()) );
  }

  /**
   * @brief Get the raw MPI communicator from a Multi-process controller.
   * @param controller the multi-process controller
   */
  void InitCommunicator(vtkMultiProcessController *controller)
  {
     assert("pre: controller is NULL!" && (controller != NULL) );
     this->MPICommunicator =
         vtkGenericIOUtilities::GetMPICommunicator(controller);
  }

  /**
   * @brief Performs a quick sanity on the metadata
   * @return status false iff the metadata is somehow corrupted.
   */
  bool SanityCheck()
  {
  size_t N = this->Information.size();
  return( ( (this->VariableGenericIOType.size()==N) &&
             (this->VariableStatus.size()==N) &&
             (this->RawCache.size()==N)
      ) );
  }

  /**
   * @brief Checks if a variable exists
   * @param varName the name of the variable in query
   * @return status true or false
   */
  bool HasVariable(const std::string &varName)
  {
  bool status = (
   (this->Information.find(varName)!=this->Information.end()) &&
   (this->VariableGenericIOType.find(varName)!=
       this->VariableGenericIOType.end()) &&
   (this->VariableStatus.find(varName)!=this->VariableStatus.end()) &&
   (this->RawCache.find(varName)!=this->RawCache.end())
   );
  return( status );
  }

  /**
   * @brief Clears the metadata
   */
  void Clear()
  {
  this->NumberOfElements = 0;
  this->VariableGenericIOType.clear();
  this->VariableStatus.clear();
  this->Information.clear();
  this->RanksToLoad.clear();

  std::map<std::string,void*>::iterator iter;
  for( iter=this->RawCache.begin(); iter != this->RawCache.end(); ++iter)
    {
    delete [] static_cast<char*>( iter->second );
    } // END for
  this->RawCache.clear();
  }

};

//------------------------------------------------------------------------------
vtkStandardNewMacro(vtkPGenericIOReader);

//------------------------------------------------------------------------------
vtkPGenericIOReader::vtkPGenericIOReader()
{
  this->SetNumberOfInputPorts(0);
  this->SetNumberOfOutputPorts(1);

  this->Controller        = vtkMultiProcessController::GetGlobalController();
  this->Reader            = NULL;
  this->FileName          = NULL;
  this->XAxisVariableName = NULL;
  this->YAxisVariableName = NULL;
  this->ZAxisVariableName = NULL;
  this->HaloIdVariableName= NULL;
  this->GenericIOType     = IOTYPEMPI;
  this->BlockAssignment   = ROUND_ROBIN;
  this->BuildMetaData     = false;
  this->AppendBlockCoordinates = true;

  this->MetaData  = new vtkGenericIOMetaData();
  this->MetaData->InitCommunicator( this->Controller );

  this->RequestInfoCounter = 0;
  this->RequestDataCounter = 0;

  this->RankInQuery        = 0;
  this->QueryRankNeighbors = 0;

  this->ArrayList = vtkStringArray::New();
  this->HaloList = vtkIdList::New();
  this->PointDataArraySelection = vtkDataArraySelection::New();
  this->SelectionObserver  = vtkCallbackCommand::New();
  this->SelectionObserver->SetCallback(
      &vtkPGenericIOReader::SelectionModifiedCallback);
  this->SelectionObserver->SetClientData( this );
  this->PointDataArraySelection->AddObserver(
      vtkCommand::ModifiedEvent,this->SelectionObserver);
}

//------------------------------------------------------------------------------
vtkPGenericIOReader::~vtkPGenericIOReader()
{
 if( this->Reader != NULL )
   {
   this->Reader->Close();
   delete this->Reader;
   }

 vtkGenericIOUtilities::SafeDeleteString(this->FileName);
 vtkGenericIOUtilities::SafeDeleteString(this->XAxisVariableName);
 vtkGenericIOUtilities::SafeDeleteString(this->YAxisVariableName);
 vtkGenericIOUtilities::SafeDeleteString(this->ZAxisVariableName);
 vtkGenericIOUtilities::SafeDeleteString(this->HaloIdVariableName);

 if( this->MetaData != NULL )
   {
   delete this->MetaData;
   }

 this->ArrayList->Delete();
 this->HaloList->Delete();

 this->PointDataArraySelection->RemoveObserver( this->SelectionObserver );
 this->SelectionObserver->Delete();
 this->PointDataArraySelection->Delete();

 this->Controller = NULL;
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::PrintSelf(std::ostream &os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os,indent);
  os << indent << "FileName: " << this->FileName << endl;
  os << indent << "x-axis: " << this->XAxisVariableName << endl;
  os << indent << "y-axis: " << this->YAxisVariableName << endl;
  os << indent << "z-axis: " << this->ZAxisVariableName << endl;
  os << indent << "GenericIOType: " << this->GenericIOType << endl;
  os << indent << "BlockAssignment: " << this->BlockAssignment << endl;
  os << indent << "ArrayList: " << endl;
  this->ArrayList->PrintSelf(os,indent.GetNextIndent());
  os << indent << "PointDataSelection: " << endl;
  this->PointDataArraySelection->PrintSelf(os,indent.GetNextIndent());
  if( Controller != NULL )
    {
    os << indent << "Controller: " << this->Controller << endl;
    }
  else
    {
    os << indent << "Controller: (null)" << endl;
    }

}

//------------------------------------------------------------------------------
int vtkPGenericIOReader::GetNumberOfPointArrays()
{
  return( this->PointDataArraySelection->GetNumberOfArrays() );
}

//------------------------------------------------------------------------------
const char* vtkPGenericIOReader::GetPointArrayName(int i)
{
  assert("pre: array index is out-of-bounds!" &&
            (i >= 0) && (i < this->GetNumberOfPointArrays()) );
  return( this->PointDataArraySelection->GetArrayName( i ) );
}

//------------------------------------------------------------------------------
int vtkPGenericIOReader::GetPointArrayStatus(const char *name)
{
  return( this->PointDataArraySelection->ArrayIsEnabled(name)  );
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::SetPointArrayStatus(
          const char *name, int status)
{
  if( status )
    {
#ifdef DEBUG
    std::cout << "\t[INFO]: Enabling array " << name << std::endl;
    std::cout.flush();
#endif
    this->PointDataArraySelection->EnableArray( name );
    assert(this->PointDataArraySelection->ArrayIsEnabled(name));
    }
  else
    {
#ifdef DEBUG
    std::cout << "\t[INFO]: Disabling array " << name << std::endl;
    std::cout.flush();
#endif
    this->PointDataArraySelection->DisableArray( name );
    assert(!this->PointDataArraySelection->ArrayIsEnabled(name));
    }
}

//------------------------------------------------------------------------------
vtkIdType vtkPGenericIOReader::GetRequestedHaloId(vtkIdType i)
{
  assert("pre: array index out of bounds" &&
         (i >= 0 && this->HaloList->GetNumberOfIds() > i));
  return this->HaloList->GetId(i);
}

//------------------------------------------------------------------------------
vtkIdType vtkPGenericIOReader::GetNumberOfRequestedHaloIds()
{
  return this->HaloList->GetNumberOfIds();
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::SetNumberOfRequestedHaloIds(vtkIdType numIds)
{
  this->HaloList->SetNumberOfIds(numIds);
  this->Modified();
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::AddRequestedHaloId(vtkIdType haloId)
{
  this->SetRequestedHaloId(this->GetNumberOfRequestedHaloIds(),haloId);
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::ClearRequestedHaloIds()
{
  this->HaloList->Reset();
  this->Modified();
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::SetRequestedHaloId(vtkIdType i, vtkIdType haloId)
{
  *this->HaloList->WritePointer(i,1) = haloId;
  this->Modified();
}

//------------------------------------------------------------------------------
bool vtkPGenericIOReader::ReaderParametersChanged()
{
  assert("pre: internal reader is NULL!" && (this->Reader != NULL) );

  if(this->Reader->GetFileName() != std::string(this->FileName) )
    {
#ifdef DEBUG
    std::cout << "\t[INFO]: File name has changed!\n";
    std::cout.flush();
#endif
    return true;
    }

  bool status = false;
  switch(this->Reader->GetIOStrategy())
    {
    case gio::GenericIOBase::FileIOMPI:
      status = (this->GenericIOType!=IOTYPEMPI)? true : false;
#ifdef DEBUG
      if(status == true)
        {
        std::cout << "\t[INFO]: I/O strategy changed from MPI\n";
        std::cout.flush();
        }
#endif
      break;
    case gio::GenericIOBase::FileIOPOSIX:
      status = (this->GenericIOType!=IOTYPEPOSIX)? true : false;
#ifdef DEBUG
      if(status == true)
        {
        std::cout << "\t[INFO]: I/O strategy changed from POSIX\n";
        std::cout.flush();
        }
#endif
      break;
    default:
      assert("pre: Code should not reach here!" && false);
      throw std::runtime_error("Invalid I/O strategy!\n");
    } // END switch on I/O strategy

  if( status == true )
    {
    /* short-circuit here */
    return( status );
    }

  switch(this->Reader->GetBlockAssignmentStrategy())
    {
    case gio::RR_BLOCK_ASSIGNMENT:
      status = (this->BlockAssignment!=ROUND_ROBIN)? true : false;
#ifdef DEBUG
      if(status == true)
        {
        std::cout << "\t[INFO]: I/O block assignment changed to Round-Robin\n";
        std::cout.flush();
        }
#endif
      break;
    case gio::RCB_BLOCK_ASSIGNMENT:
      status = (this->BlockAssignment!=RCB)? true : false;
#ifdef DEBUG
      if(status == true)
        {
        std::cout << "\t[INFO]: I/O block assignment changed to RCB\n";
        std::cout.flush();
        }
#endif
      break;
    default:
      assert("pre: Code should not reach here!" && false);
      throw std::runtime_error("Invalid BlockAssignment strategy!\n");
    } // END switch on BlockAssignment

  return( status );
}

//------------------------------------------------------------------------------
gio::GenericIOReader* vtkPGenericIOReader::GetInternalReader()
{
  if( this->Reader != NULL )
    {
    if( this->ReaderParametersChanged() )
      {
#ifdef DEBUG
      std::cout << "\t[INFO]: Deleting Reader instance...\n";
      std::cout.flush();
#endif
      this->Reader->Close();
      delete this->Reader;
      this->Reader=NULL;
      } // END if the reader parameters
    else
      {
      return( this->Reader );
      }
    } // END if the reader is not NULL

  this->BuildMetaData = true; // signal to re-build metadata

  assert("pre: Reader should be NULL" && (this->Reader==NULL));
  gio::GenericIOReader *r = NULL;
  bool posix              = (this->GenericIOType==IOTYPEMPI)? false : true;
  int distribution        = (this->BlockAssignment==RCB)?
                                gio::RCB_BLOCK_ASSIGNMENT :
                                gio::RR_BLOCK_ASSIGNMENT;

  r = vtkGenericIOUtilities::GetReader(
        vtkGenericIOUtilities::GetMPICommunicator(this->Controller),
        posix,distribution,std::string(this->FileName));
  assert("post: Internal GenericIO reader should not be NULL!" && (r!=NULL) );

  return( r );
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::LoadMetaData()
{
  if( !this->BuildMetaData )
    {
#ifdef DEBUG
    std::cout << "\t[INFO]: No need to update metadata!\n";
    std::cout.flush();
#endif
    return;
    }

  this->MetaData->Clear();

#ifdef DEBUG
    std::cout << "\t[INFO]: Reading header to build metadata!\n";
    std::cout << "\t[INFO]: Filename: " << this->FileName << std::endl;
    std::cout.flush();
#endif
  this->Reader->OpenAndReadHeader();
  this->MetaData->NumberOfElements = this->Reader->GetNumberOfElements();

  this->ArrayList->SetNumberOfValues(0);

  for(int i=0; i < this->Reader->GetNumberOfVariablesInFile(); ++i)
    {
    std::string vname = this->Reader->GetVariableName( i );

    this->ArrayList->InsertNextValue( vname.c_str() );

    this->MetaData->Information[vname] =
        this->Reader->GetFileVariableInfo(i);

    this->MetaData->VariableGenericIOType[vname]=
        gio::GenericIOUtilities::DetectVariablePrimitiveType(
              this->MetaData->Information[vname] );

    this->MetaData->VariableStatus[vname]= false;
    this->MetaData->RawCache[vname]      = NULL;

    if( !this->PointDataArraySelection->ArrayExists(vname.c_str()) )
      {
      this->PointDataArraySelection->AddArray(vname.c_str());
      this->PointDataArraySelection->DisableArray( vname.c_str() );
      }
    } // END for all variables in the file

  this->BuildMetaData = false; /* signal that the metadata is build */

  assert("pre: metadata is corrupt!" && (this->MetaData->SanityCheck()));
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::LoadRawVariableData(std::string varName)
{
#ifdef DEBUG
  std::cout << "[INFO]: Loading variable: " << varName << std::endl;
  std::cout.flush();
#endif

  // Sanity check
  assert("pre: no variable in metadata with given name" &&
          this->MetaData->HasVariable(varName));

  if(this->MetaData->VariableStatus[varName])
    {
    // variable has already been loaded
#ifdef DEBUG
    std::cout << "\t[INFO]: Variable appears to be already loaded!\n";
    std::cout.flush();
#endif
    return;
    }

  this->MetaData->RawCache[varName]=
  gio::GenericIOUtilities::AllocateVariableArray(
    this->MetaData->Information[varName],this->MetaData->NumberOfElements);

  this->Reader->AddVariable(
    this->MetaData->Information[varName],this->MetaData->RawCache[varName]);

  this->MetaData->VariableStatus[varName] = true;

#ifdef DEBUG
  std::cout << "\t[INFO]: Variable [" << varName << "] is now loaded!\n";
  std::cout.flush();
#endif
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::LoadRawData()
{
  assert("pre: metadata is corrupt!" && (this->MetaData->SanityCheck()));

  std::string xaxis = std::string(this->XAxisVariableName);
  xaxis = vtkGenericIOUtilities::trim(xaxis);

  std::string yaxis = std::string(this->YAxisVariableName);
  yaxis = vtkGenericIOUtilities::trim(yaxis);

  std::string zaxis = std::string(this->ZAxisVariableName);
  zaxis = vtkGenericIOUtilities::trim(zaxis);

  this->LoadRawVariableData( xaxis );
  this->LoadRawVariableData( yaxis );
  this->LoadRawVariableData( zaxis );

  if (this->HaloList->GetNumberOfIds() > 0)
    {
    std::string haloIds = std::string(this->HaloIdVariableName);
    haloIds = vtkGenericIOUtilities::trim(haloIds);
    this->LoadRawVariableData(haloIds);
    }

#ifdef DEBUG
  std::cout << "\t==========\n";
  std::cout << "\tNUMBER OF ARRAYS: "
      << this->PointDataArraySelection->GetNumberOfArrays() << std::endl;
  std::cout.flush();
#endif

  int arrayIdx = 0;
  for(;arrayIdx < this->PointDataArraySelection->GetNumberOfArrays(); ++arrayIdx)
    {
    const char *name = this->PointDataArraySelection->GetArrayName(arrayIdx);
#ifdef DEBUG
    std::cout << "\tARRAY " << name << " is ";
#endif
    if( this->PointDataArraySelection->ArrayIsEnabled(name) )
      {
#ifdef DEBUG
      std::cout << "ENABLED\n";
      std::cout.flush();
#endif
      std::string varName = std::string( name );
      this->LoadRawVariableData( varName );
      } // END if the array is enabled
    else
      {
#ifdef DEBUG
      std::cout << "DISBLED\n";
      std::cout.flush();
#endif
      }
    } // END for all arrays

#ifdef DEBUG
  std::cout << "\t[INFO]: Reading data...";
#endif

  this->Reader->ReadData();

#ifdef DEBUG
  std::cout << "[DONE]\n";
  std::cout.flush();
#endif
}


//------------------------------------------------------------------------------
void vtkPGenericIOReader::GetPointFromRawData(
        int xType, void* xBuffer, int yType, void* yBuffer, int zType, void* zBuffer,
        vtkIdType idx, double pnt[3])
{
  void* buffer[3] = {xBuffer, yBuffer, zBuffer};

  int type[3] = {xType, yType, zType};

  for( int i=0; i < 3; ++i )
    {
    assert("pre: raw buffer is NULL!" && (buffer[i] != NULL) );

    pnt[i] = vtkGenericIOUtilities::GetDoubleFromRawBuffer(type[i],buffer[i],idx);
    } // END for all dimensions
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::LoadCoordinates(vtkUnstructuredGrid *grid,
                                          std::set< vtkIdType >& pointsInSelectedHalos)
{
  assert("pre: grid is NULL!" && (grid != NULL) );

  if( this->QueryRankNeighbors && (this->BlockAssignment==RCB) &&
      !this->MetaData->LoadRank(this->Controller->GetLocalProcessId()))
    {
    return;
    }

  std::string xaxis = std::string(this->XAxisVariableName);
  xaxis = vtkGenericIOUtilities::trim(xaxis);

  std::string yaxis = std::string(this->YAxisVariableName);
  yaxis = vtkGenericIOUtilities::trim(yaxis);

  std::string zaxis = std::string(this->ZAxisVariableName);
  zaxis = vtkGenericIOUtilities::trim(zaxis);

  if( !this->MetaData->HasVariable(xaxis) ||
       !this->MetaData->HasVariable(yaxis) ||
       !this->MetaData->HasVariable(zaxis))
    {
    vtkErrorMacro(<< "Don't have one or more coordinate arrays!\n");
    return;
    }

  int xType = this->MetaData->VariableGenericIOType[xaxis];
  void* xBuffer = this->MetaData->RawCache[xaxis];
  int yType = this->MetaData->VariableGenericIOType[yaxis];
  void* yBuffer = this->MetaData->RawCache[yaxis];
  int zType = this->MetaData->VariableGenericIOType[zaxis];
  void* zBuffer = this->MetaData->RawCache[zaxis];

  vtkCellArray *cells = vtkCellArray::New();
  cells->Allocate(cells->EstimateSize(this->MetaData->NumberOfElements,1));

  vtkPoints *pnts = vtkPoints::New();
  pnts->SetDataTypeToDouble();
  pnts->SetNumberOfPoints( this->MetaData->NumberOfElements );


  int nparticles = this->MetaData->NumberOfElements;
  double pnt[3];
  vtkIdType idx = 0;
  if (this->HaloList->GetNumberOfIds() == 0)
    {
    for( ;idx < nparticles; ++idx)
      {
      this->GetPointFromRawData(xType,xBuffer,yType,yBuffer,zType,zBuffer, idx, pnt);
      pnts->SetPoint(idx,pnt);
      cells->InsertNextCell(1,&idx);
      } // END for all points
    }
  else
    {
    std::string haloVarName = std::string(this->HaloIdVariableName);
    haloVarName = vtkGenericIOUtilities::trim(haloVarName);
    int haloType = this->MetaData->VariableGenericIOType[haloVarName];
    void* haloBuffer = this->MetaData->RawCache[haloVarName];
    vtkIdType numPointsSoFar = 0;
    for (; idx < nparticles; ++idx)
      {
      vtkIdType haloId = vtkGenericIOUtilities::GetIdFromRawBuffer(haloType,haloBuffer,idx);
      bool isInRequestedHalo = false;
      for (vtkIdType j = 0; j < this->GetNumberOfRequestedHaloIds(); ++j)
        {
        if (haloId == this->HaloList->GetId(j))
          {
          isInRequestedHalo = true;
          pointsInSelectedHalos.insert(idx);
          break;
          }
        }
      if (isInRequestedHalo)
        {
        this->GetPointFromRawData(xType,xBuffer,yType,yBuffer,zType,zBuffer,idx,pnt);
        pnts->SetPoint(numPointsSoFar,pnt);
        cells->InsertNextCell(1,&numPointsSoFar);
        ++numPointsSoFar;
        }
      }
    // set the number of points correctly
    pnts->SetNumberOfPoints(numPointsSoFar);
    }

  grid->SetPoints(pnts);
  pnts->Delete();

  grid->SetCells(VTK_VERTEX,cells);
  cells->Delete();

  grid->Squeeze();
}

namespace {
template< typename T >
void GetOnlyDataInHalo(vtkDataArray* allData, vtkDataArray* haloData, std::set< vtkIdType > pointsInHalo)
{
  T* data = (T*) allData->GetVoidPointer(0);
  T* filteredData = (T*) haloData->GetVoidPointer(0);
  vtkIdType i = 0;
  for (std::set< vtkIdType >::iterator itr = pointsInHalo.begin();
       itr != pointsInHalo.end(); ++itr)
    {
    filteredData[i++] = data[*itr];
    }
}
}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::LoadData(vtkUnstructuredGrid *grid,
                                   const std::set< vtkIdType >& pointsInSelectedHalos)
{
  assert("pre: grid is NULL!" && (grid != NULL) );

  if( this->QueryRankNeighbors && (this->BlockAssignment==RCB) &&
      !this->MetaData->LoadRank(this->Controller->GetLocalProcessId()))
    {
    return;
    }

  vtkPointData *PD = grid->GetPointData();
  int arrayIdx = 0;
  for(;arrayIdx < this->PointDataArraySelection->GetNumberOfArrays(); ++arrayIdx)
    {
    const char *name = this->PointDataArraySelection->GetArrayName( arrayIdx );
    if( this->PointDataArraySelection->ArrayIsEnabled(name) )
      {
      std::string varName = std::string( name );
      vtkSmartPointer< vtkDataArray > dataArray;
      dataArray.TakeReference(
          vtkGenericIOUtilities::GetVtkDataArray(
              varName,
              this->MetaData->VariableGenericIOType[ varName ],
              this->MetaData->RawCache[ varName ],
              this->MetaData->NumberOfElements
              ));
      if (this->HaloList->GetNumberOfIds() != 0)
        {
        vtkSmartPointer< vtkDataArray > onlyDataInHalo;
        onlyDataInHalo.TakeReference(dataArray->NewInstance());
        onlyDataInHalo->SetNumberOfTuples(grid->GetNumberOfPoints());
        onlyDataInHalo->SetName(dataArray->GetName());
        switch (dataArray->GetDataType()) {
        vtkTemplateMacro(GetOnlyDataInHalo<VTK_TT>(
                           dataArray,onlyDataInHalo,pointsInSelectedHalos));
          }
        dataArray = onlyDataInHalo;
        }

      PD->AddArray( dataArray );
      } // END if the array is enabled
    } // END for all arrays
  if (this->AppendBlockCoordinates && this->Reader->IsSpatiallyDecomposed())
    {
    vtkSmartPointer< vtkTypeUInt64Array > dataArray = vtkSmartPointer< vtkTypeUInt64Array >::New();
    dataArray->SetNumberOfComponents(3);
    dataArray->SetNumberOfTuples(this->MetaData->NumberOfElements);
    dataArray->SetName("gio_block_indices");
    int nextBlockIdx = 0;
    int nextBlockStart = 0;
    unsigned long long coords[3];
    // since the compiler can't tell if they're the same....
    assert (sizeof(unsigned long long) == sizeof(uint64_t));
    for (int i = 0; i < this->MetaData->NumberOfElements; ++i)
      {
      if (i == nextBlockStart)
        {
        this->Reader->GetBlockCoords(nextBlockIdx,(uint64_t*)coords);
        nextBlockStart += this->Reader->GetNumberOfElementsInBlock(nextBlockIdx);
        ++nextBlockIdx;
        }
      dataArray->SetTypedTuple(i,coords);
      }
      if (this->HaloList->GetNumberOfIds() != 0)
        {
        vtkSmartPointer< vtkTypeUInt64Array > onlyDataInHalo;
        onlyDataInHalo.TakeReference(dataArray->NewInstance());
        onlyDataInHalo->SetNumberOfComponents(3);
        onlyDataInHalo->SetNumberOfTuples(grid->GetNumberOfPoints());
        onlyDataInHalo->SetName(dataArray->GetName());
        vtkIdType i = 0;
        for (std::set< vtkIdType >::iterator itr = pointsInSelectedHalos.begin();
             itr != pointsInSelectedHalos.end(); ++itr, ++i)
          {
          vtkTypeUInt64 data[3];
          dataArray->GetTypedTuple(*itr,data);
          onlyDataInHalo->SetTypedTuple(i,data);
          }
        dataArray = onlyDataInHalo;
        }

    PD->AddArray( dataArray );
    }

}

//------------------------------------------------------------------------------
void vtkPGenericIOReader::FindRankNeighbors()
{
  if( !this->QueryRankNeighbors || this->BlockAssignment != RCB)
    {
    return;
    }

  this->MetaData->RanksToLoad.clear();

  // Sanity checks
  assert("pre: rank in query is out-of bounds!" &&
        (this->RankInQuery >= 0) &&
        (this->RankInQuery < this->Controller->GetNumberOfProcesses()) );
  assert("pre: reader should not be NULL!" && (this->Reader!=NULL) );
  assert("pre: block assignment is not RCB!" &&
    (this->Reader->GetBlockAssignmentStrategy()==gio::RCB_BLOCK_ASSIGNMENT) );

  std::vector<int> neiRanks;
  if( this->Controller->GetLocalProcessId() == this->RankInQuery )
    {
#ifdef DEBUG
    std::cout << "[INFO]: Loading neighbors for process P[";
    std::cout << this->Controller->GetLocalProcessId();
    std::cout << "]\n";
    std::cout.flush();
#endif

    int numNeis = this->Reader->GetNumberOfRankNeighbors();
    std::vector< gio::RankNeighbor > rankNeighbors;
    rankNeighbors.resize(numNeis);
    this->Reader->GetRankNeighbors( &rankNeighbors[0] );

    neiRanks.resize(numNeis);
    for(int nei=0; nei < numNeis; ++nei)
      {
#ifdef DEBUG
     std::cout << "\t Neighboring rank P[";
     std::cout << rankNeighbors[ nei ].RankID;
     std::cout << "] Orientation {";
     std::cout << gio::NEIGHBOR_ORIENTATION[rankNeighbors[nei].Orient[0]];
     std::cout << ", ";
     std::cout << gio::NEIGHBOR_ORIENTATION[rankNeighbors[nei].Orient[1]];
     std::cout << ", ";
     std::cout << gio::NEIGHBOR_ORIENTATION[rankNeighbors[nei].Orient[2]];
     std::cout << "}\n";
     std::cout.flush();
#endif
      neiRanks[ nei ] = rankNeighbors[ nei ].RankID;
      } // END for all neis
    rankNeighbors.clear();

    MPI_Bcast(&numNeis,1,MPI_INT,this->RankInQuery,this->MetaData->MPICommunicator);
    MPI_Bcast(&neiRanks[0],numNeis,MPI_INT,
              this->RankInQuery,this->MetaData->MPICommunicator);
    } // END if
  else
    {
    int numNeis = -1;
    MPI_Bcast(&numNeis,1,MPI_INT,this->RankInQuery,this->MetaData->MPICommunicator);
    neiRanks.resize(numNeis);
    MPI_Bcast(&neiRanks[0],numNeis,MPI_INT,
              this->RankInQuery,this->MetaData->MPICommunicator);
    } // END else

  this->MetaData->RanksToLoad.insert( this->RankInQuery );
  for(unsigned int i=0; i < neiRanks.size(); ++i )
    {
    this->MetaData->RanksToLoad.insert( neiRanks[i] );
    }
  neiRanks.clear();
}

//----------------------------------------------------------------------------
void vtkPGenericIOReader::SelectionModifiedCallback(
    vtkObject*, unsigned long, void* clientdata, void*)
{
  assert("pre: clientdata is NULL!" && (clientdata != NULL) );
  static_cast<vtkPGenericIOReader*>(clientdata)->Modified();
}

//------------------------------------------------------------------------------
int vtkPGenericIOReader::RequestInformation(
      vtkInformation *vtkNotUsed(request),
      vtkInformationVector **vtkNotUsed(inputVector),
      vtkInformationVector *outputVector)
{
#ifdef DEBUG
  std::cout << "[INFO]: Call RequestInformation()\n";
  std::cout.flush();
#endif

  ++this->RequestInfoCounter;

  // tell the pipeline that this dataset is distributed
  outputVector->GetInformationObject(0)->Set(
      CAN_HANDLE_PIECE_REQUEST(),
      1
      );
  outputVector->GetInformationObject(0)->Set(
      vtkDataObject::DATA_NUMBER_OF_PIECES(),
      this->Controller->GetNumberOfProcesses()
      );

  this->Reader = this->GetInternalReader();
  assert("pre: internal reader is NULL!" && (this->Reader != NULL) );

  this->LoadMetaData();
  return 1;
}

//------------------------------------------------------------------------------
int vtkPGenericIOReader::RequestData(
    vtkInformation *vtkNotUsed(request),
    vtkInformationVector **vtkNotUsed(inputVector),
    vtkInformationVector *outputVector)
{
#ifdef DEBUG
  std::cout << "[INFO]: Call RequestData()\n";
  std::cout.flush();
#endif

  // STEP 0: get the output grid
  vtkInformation *outInfo     = outputVector->GetInformationObject(0);
  vtkUnstructuredGrid *output =
      vtkUnstructuredGrid::SafeDownCast(
          outInfo->Get(vtkDataObject::DATA_OBJECT()));
  assert("pre: output grid is NULL!" && (output != NULL) );
  std::set< vtkIdType > pointsInSelectedHalos;

  // STEP 1: Load raw data
  this->LoadRawData();

  // STEP 2: Load coordinates
  this->LoadCoordinates(output,pointsInSelectedHalos);
  MPI_Barrier(this->MetaData->MPICommunicator);

  // STEP 3: Load data
  this->LoadData(output,pointsInSelectedHalos);
  MPI_Barrier(this->MetaData->MPICommunicator);

  // STEP 4: Clear variables
  this->Reader->ClearVariables();
  return 1;
}