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

Program:   Visualization Toolkit
Module:    vtkCompositeDataPipeline.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 "vtkCompositeDataPipeline.h"

#include "vtkAlgorithm.h"
#include "vtkAlgorithmOutput.h"
#include "vtkMultiGroupDataInformation.h"
#include "vtkMultiGroupDataSet.h"
#include "vtkInformation.h"
#include "vtkInformationDoubleKey.h"
#include "vtkInformationIntegerKey.h"
#include "vtkInformationIntegerVectorKey.h"
#include "vtkInformationKey.h"
#include "vtkInformationObjectBaseKey.h"
#include "vtkInformationStringKey.h"
#include "vtkInformationVector.h"
#include "vtkInformationIdTypeKey.h"
#include "vtkObjectFactory.h"
#include "vtkPolyData.h"
#include "vtkSmartPointer.h"
#include "vtkTemporalDataSet.h"

#include "vtkImageData.h"
#include "vtkRectilinearGrid.h"
#include "vtkStructuredGrid.h"
#include "vtkUniformGrid.h"

//----------------------------------------------------------------------------
#if defined (JB_DEBUG1)
  #ifndef WIN32
  #else
    #define OUTPUTTEXT(a) vtkOutputWindowDisplayText(a);
  #endif

  #undef vtkDebugMacro
  #define vtkDebugMacro(a)  \
  { \
    vtkOStreamWrapper::EndlType endl; \
    vtkOStreamWrapper::UseEndl(endl); \
    vtkOStrStreamWrapper vtkmsg; \
    const char *name = this->Algorithm->GetClassName(); \
    if (!strcmp(name, "vtkTemporalDataSetCache") || \
        !strcmp(name, "vtkContourFilter") || \
        !strcmp(name, "vtkOpenDXStructuredGridReader") || \
        !strcmp(name, "vtkPCellDataToPointData") || \
        !strcmp(name, "vtkProcessIdScalars") || \
        !strcmp(name, "vtkTemporalFractal") || \
        !strcmp(name, "vtkTemporalSphereSource") || \
        !strcmp(name, "vtkTemporalInterpolator") || \
        !strcmp(name, "vtkTemporalStreamTracer")) \
      { \
      vtkmsg << name << " : " a << endl; \
      OUTPUTTEXT(vtkmsg.str()); \
      vtkmsg.rdbuf()->freeze(0); \
      } \
  }
#endif
//----------------------------------------------------------------------------
/*
    if (!strcmp(name, "vtkTemporalDataSetCache") || \
        !strcmp(name, "vtkContourFilter")) \
        !strcmp(name, "vtkOpenDXStructuredGridReader") || \
        !strcmp(name, "vtkPCellDataToPointData") || \
        !strcmp(name, "vtkProcessIdScalars") || \
        !strcmp(name, "vtkTemporalFractal") || \
        !strcmp(name, "vtkTemporalSphereSource") || \
        !strcmp(name, "vtkTemporalInterpolator") || \
        !strcmp(name, "vtkTemporalStreamTracer")) \
      { \
*/
vtkCxxRevisionMacro(vtkCompositeDataPipeline, "1.63");
vtkStandardNewMacro(vtkCompositeDataPipeline);

vtkInformationKeyMacro(vtkCompositeDataPipeline,COMPOSITE_DATA_INFORMATION,ObjectBase);
vtkInformationKeyMacro(vtkCompositeDataPipeline,UPDATE_BLOCKS, ObjectBase);
vtkInformationKeyMacro(vtkCompositeDataPipeline,REQUIRES_TIME_DOWNSTREAM, Integer);

//----------------------------------------------------------------------------
vtkCompositeDataPipeline::vtkCompositeDataPipeline()
{
  this->InLocalLoop = 0;
  this->SuppressResetPipelineInformation = 0;
  this->InformationCache = vtkInformation::New();

  this->GenericRequest = vtkInformation::New();

  this->DataObjectRequest = vtkInformation::New();
  this->DataObjectRequest->Set(vtkDemandDrivenPipeline::REQUEST_DATA_OBJECT());
  // The request is forwarded upstream through the pipeline.
  this->DataObjectRequest->Set(
    vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
      // Algorithms process this request after it is forwarded.
  this->DataObjectRequest->Set(vtkExecutive::ALGORITHM_AFTER_FORWARD(), 1);

  this->InformationRequest = vtkInformation::New();
  this->InformationRequest->Set(vtkDemandDrivenPipeline::REQUEST_INFORMATION());
  // The request is forwarded upstream through the pipeline.
  this->InformationRequest->Set(
    vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
  // Algorithms process this request after it is forwarded.
  this->InformationRequest->Set(vtkExecutive::ALGORITHM_AFTER_FORWARD(), 1);

  this->UpdateExtentRequest = vtkInformation::New();
  this->UpdateExtentRequest->Set(
    vtkStreamingDemandDrivenPipeline::REQUEST_UPDATE_EXTENT());
  // The request is forwarded upstream through the pipeline.
  this->UpdateExtentRequest->Set(
    vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
  // Algorithms process this request before it is forwarded.
  this->UpdateExtentRequest->Set(vtkExecutive::ALGORITHM_BEFORE_FORWARD(), 1);
                            
  this->DataRequest = vtkInformation::New();
  this->DataRequest->Set(REQUEST_DATA());
  // The request is forwarded upstream through the pipeline.
  this->DataRequest->Set(
    vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
  // Algorithms process this request after it is forwarded.
  this->DataRequest->Set(vtkExecutive::ALGORITHM_AFTER_FORWARD(), 1);
              
              
}

//----------------------------------------------------------------------------
vtkCompositeDataPipeline::~vtkCompositeDataPipeline()
{
  this->InformationCache->Delete();

  this->GenericRequest->Delete();
  this->DataObjectRequest->Delete();
  this->InformationRequest->Delete();
  this->UpdateExtentRequest->Delete();
  this->DataRequest->Delete();
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::ForwardUpstream(vtkInformation* request)
{
  vtkDebugMacro(<< "ForwardUpstream");

  // Do not forward upstream if the input is shared with another
  // executive.
  if(this->SharedInputInformation)
    {
    return 1;
    }

  if (!this->Algorithm->ModifyRequest(request, BeforeForward))
    {
    return 0;
    }

  // Forward the request upstream through all input connections.
  int result = 1;
  for(int i=0; i < this->GetNumberOfInputPorts(); ++i)
    {
    int nic = this->Algorithm->GetNumberOfInputConnections(i);
    vtkInformationVector* inVector = this->GetInputInformation()[i];
    for(int j=0; j < nic; ++j)
      {
      vtkInformation* info = inVector->GetInformationObject(j);
      // Get the executive producing this input.  If there is none, then
      // it is a NULL input.
      vtkExecutive* e;
      int producerPort;
      info->Get(vtkExecutive::PRODUCER(),e,producerPort);
      if(e)
        {
        int port = request->Get(FROM_OUTPUT_PORT());
        request->Set(FROM_OUTPUT_PORT(), producerPort);
        
        // if the input requires time them mark that
        if (this->Algorithm->GetInputPortInformation(i)
            ->Get(vtkAlgorithm::INPUT_REQUIRED_DATA_TYPE()) &&
            !strcmp("vtkTemporalDataSet",
                    this->Algorithm->GetInputPortInformation(i)
                    ->Get(vtkAlgorithm::INPUT_REQUIRED_DATA_TYPE())))
          {
          info->Set(REQUIRES_TIME_DOWNSTREAM(),1);
          vtkDebugMacro(<< "Set REQUIRES_TIME_DOWNSTREAM");
          }
        if(!e->ProcessRequest(request,
                              e->GetInputInformation(),
                              e->GetOutputInformation()))
          {
          result = 0;
          }
        info->Remove(REQUIRES_TIME_DOWNSTREAM());
        request->Set(FROM_OUTPUT_PORT(), port);
        }
      }
    }

  if (!this->Algorithm->ModifyRequest(request, AfterForward))
    {
    return 0;
    }

  return result;
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::ProcessRequest(vtkInformation* request,
                                             vtkInformationVector** inInfoVec,
                                             vtkInformationVector* outInfoVec)
{
  if(this->Algorithm && request->Has(REQUEST_DATA_OBJECT()))
    {
    vtkDebugMacro(<< "REQUEST_DATA_OBJECT()");
    // if we are up to date then short circuit
    if (this->PipelineMTime < this->DataObjectTime.GetMTime())
      {
      return 1;
      }

    // request Update inputs first if they are out of date
    if(!this->ForwardUpstream(request))
      {
      return 0;
      }
    
    // Make sure our output data type is up-to-date.
    int result = 1;
    if(this->PipelineMTime > this->DataObjectTime.GetMTime())
      {
      // Request data type from the algorithm.
      result = this->ExecuteDataObject(request,inInfoVec,outInfoVec);

      // Make sure the data object exists for all output ports.
      for(int i=0;
          result && i < outInfoVec->GetNumberOfInformationObjects(); ++i)
        {
        vtkInformation* info = outInfoVec->GetInformationObject(i);
        if(!info->Get(vtkDataObject::DATA_OBJECT()))
          {
          result = 0;
          }
        }

      if(result)
        {
        // Data object is now up to date.
        this->DataObjectTime.Modified();
        }
      }

    return result;
    }

  if(this->Algorithm && request->Has(REQUEST_INFORMATION()))
    {
    vtkDebugMacro(<< "REQUEST_INFORMATION()");
    int appendKey = 1;
    vtkInformationKey** keys = request->Get(vtkExecutive::KEYS_TO_COPY());
    if (keys)
      {
      int len = request->Length(vtkExecutive::KEYS_TO_COPY());
      for (int i=0; i<len; i++)
        {
        if (keys[i] == vtkCompositeDataPipeline::COMPOSITE_DATA_INFORMATION())
          {
          appendKey = 0;
          break;
          }
        }
      }
    if (appendKey)
      {
      request->Append(vtkExecutive::KEYS_TO_COPY(), 
                      vtkCompositeDataPipeline::COMPOSITE_DATA_INFORMATION());
      }
      
    return this->Superclass::ProcessRequest(request, inInfoVec,outInfoVec);
    }

  if(this->Algorithm && request->Has(REQUEST_DATA()))
    {
    vtkDebugMacro(<< "REQUEST_DATA()");

    // UPDATE_BLOCKS() is the key that tells filters downstream which
    // blocks of a composite dataset are available. This consumers will then
    // ask for these blocks if they have to loop (simple filters in the
    // middle)
    int appendKey = 1;
    vtkInformationKey** keys = request->Get(vtkExecutive::KEYS_TO_COPY());
    if (keys)
      {
      int len = request->Length(vtkExecutive::KEYS_TO_COPY());
      for (int i=0; i<len; i++)
        {
        if (keys[i] == vtkCompositeDataPipeline::UPDATE_BLOCKS())
          {
          appendKey = 0;
          break;
          }
        }
      }
    if (appendKey)
      {
      request->Append(vtkExecutive::KEYS_TO_COPY(), 
                      vtkCompositeDataPipeline::UPDATE_BLOCKS());
      }
    }

  // Let the superclass handle other requests.
  return this->Superclass::ProcessRequest(request, inInfoVec, outInfoVec);
}

//----------------------------------------------------------------------------
// Handle REQUEST_DATA_OBJECT
int vtkCompositeDataPipeline::ExecuteDataObject(
  vtkInformation* request, 
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec)
{
  vtkDebugMacro(<< "ExecuteDataObject");
  int result=1;

  // If the input is composite, allow algorithm to handle
  // REQUEST_DATA_OBJECT only if it can handle composite
  // datasets. Otherwise, the algorithm will get a chance to handle
  // REQUEST_DATA_OBJECT when it is being iterated over.
  int compositePort;
  int shouldIterate = this->ShouldIterateOverInput(compositePort) ||
    this->ShouldIterateTemporalData(request, inInfoVec, outInfoVec);
  if (!shouldIterate)
    {
    // Invoke the request on the algorithm.
    result = this->CallAlgorithm(request, vtkExecutive::RequestDownstream,
                                 inInfoVec, outInfoVec);
    if (!result)
      {
      return result;
      }
    }

  int i; 
  // Make sure a valid data object exists for all output ports.
  for(i=0; result && i < this->Algorithm->GetNumberOfOutputPorts(); ++i)
    {
    vtkDebugMacro(<< "ExecuteDataObject calling CheckCompositeData");
    result = this->CheckCompositeData(request, i, inInfoVec, outInfoVec);
    }

  return result;
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::ExecuteDataStart(
  vtkInformation* request,
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec)
{
  // If the last iteration of ExecuteData had to iterate over time values, but
  // the current does not, then we may need to replace the output.
  int hasIteratedTemporalData = 0;
  int isIteratingTemporalData = 0;
  for (int i = 0; i < outInfoVec->GetNumberOfInformationObjects(); i++)
    {
    vtkInformation *info = outInfoVec->GetInformationObject(i);
    isIteratingTemporalData |= info->Has(REQUIRES_TIME_DOWNSTREAM());
    const char *providedDataType
      = (this->Algorithm->GetOutputPortInformation(i)
         ->Get(vtkDataObject::DATA_TYPE_NAME()));
    if (   (strcmp(providedDataType, "vtkTemporalDataSet") != 0)
        && (info->Get(vtkDataObject::DATA_OBJECT())->IsA("vtkTemporalDataSet")))
      {
      hasIteratedTemporalData = 1;
      }
    }
  if (hasIteratedTemporalData && !isIteratingTemporalData)
    {
    this->SuppressResetPipelineInformation = 1;
    this->ExecuteDataObject(this->DataObjectRequest, inInfoVec, outInfoVec);
    this->SuppressResetPipelineInformation = 0;
    }

  this->Superclass::ExecuteDataStart(request, inInfoVec, outInfoVec);
}

//----------------------------------------------------------------------------
// Handle REQUEST_DATA
int vtkCompositeDataPipeline::ExecuteData(vtkInformation* request,
                                          vtkInformationVector** inInfoVec,
                                          vtkInformationVector* outInfoVec)
{
  vtkDebugMacro(<< "ExecuteData");
  int result = 1;

  for(int i=0; i < this->Algorithm->GetNumberOfOutputPorts(); ++i)
    {
    vtkInformation* info = outInfoVec->GetInformationObject(i);
    info->Remove(UPDATE_BLOCKS());
    }

  int compositePort;
  int composite = 
    this->ShouldIterateOverInput(compositePort);
  int temporal = 
    this->ShouldIterateTemporalData(request, inInfoVec, outInfoVec);
  compositePort = temporal ? -1 : compositePort;  
  if (temporal || composite)
    {
    this->ExecuteSimpleAlgorithm(
      request, inInfoVec, outInfoVec, compositePort);
    }
  else
    {
    vtkDebugMacro(<< "  Superclass::ExecuteData");
    result = this->Superclass::ExecuteData(request,inInfoVec,outInfoVec);
    }

  for(int j=0; j < this->Algorithm->GetNumberOfOutputPorts(); ++j)
    {
    vtkInformation* info = this->GetOutputInformation(j);
    vtkObject* dobj= info->Get(vtkDataObject::DATA_OBJECT());
    if (dobj)
      {
      info->Set(UPDATE_BLOCKS(), dobj);
      }
    }

  return result;
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::InputTypeIsValid(
  int port, int index,vtkInformationVector **inInfoVec)
{
  if (this->InLocalLoop)
    {
    return this->Superclass::InputTypeIsValid(port, index, inInfoVec);
    }
  if (!inInfoVec[port])
    {
    return 0;
    }  

  // If we will be iterating over the input on this port, assume that we
  // can handle any input type. The input type will be checked again during
  // each step of the iteration.
  int compositePort;
  if (this->ShouldIterateOverInput(compositePort))
    {
    if (compositePort == port)
      {
      return 1;
      }
    }

  // If the algorithm is requesting a vtkTemporalDataSet, then assume that the
  // upstream pipeline will be run multiple times and a vtkTemporalDataSet will
  // be created from the multiple results.
  vtkInformation *info = this->Algorithm->GetInputPortInformation(port);
  const char *requiredType
    = info->Get(vtkAlgorithm::INPUT_REQUIRED_DATA_TYPE());
  if (requiredType && (strcmp(requiredType, "vtkTemporalDataSet") == 0))
    {
    return 1;
    }

  // Otherwise, let superclass handle it.
  return this->Superclass::InputTypeIsValid(port, index, inInfoVec);
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::ShouldIterateOverInput(int& compositePort)
{
  compositePort = -1;
  // Find the first input that has a composite data that does not match
  // the required input type. We assume that that port input has to
  // be iterated over. We also require that this port has only one
  // connection.
  int numInputPorts = this->Algorithm->GetNumberOfInputPorts();
  for(int i=0; i < numInputPorts; ++i)
    {
    int numInConnections = this->Algorithm->GetNumberOfInputConnections(i);
    // If there is 1 connection
    if (numInConnections == 1)
      {
      vtkInformation* inPortInfo = 
        this->Algorithm->GetInputPortInformation(i);
      const char* inputType = 
        inPortInfo->Get(vtkAlgorithm::INPUT_REQUIRED_DATA_TYPE());
      if (inputType)
        {
        // the filter upstream will iterate
        if (strcmp(inputType, "vtkTemporalDataSet") == 0)
        {
          vtkDebugMacro(<< "ShouldIterateOverInput returns 0 (Temporal)");
          return 0;
        }
        vtkInformation* inInfo = this->GetInputInformation(i, 0);
        vtkDataObject* input = inInfo->Get(vtkDataObject::DATA_OBJECT());
        // If input does not match required input type
        if (input && !input->IsA(inputType))
          {
          // If input is composite
          if (vtkCompositeDataSet::SafeDownCast(input))
            {
            // Assume that we have to iterate over input
            compositePort = i;
            vtkDebugMacro(<< "ShouldIterateOverInput returns 1 (input composite)");
            return 1;
            }
          }
        }
      }
    }
  vtkDebugMacro(<< "ShouldIterateOverInput returns 0 (default)");
  return 0;
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::ShouldIterateTemporalData(
                                               vtkInformation *vtkNotUsed(request),
                                               vtkInformationVector** vtkNotUsed(inInfoVec),
                                               vtkInformationVector* outInfoVec)
{
  // Exit fast if no outputs exist
  if (!this->Algorithm->GetNumberOfOutputPorts())
    {
    vtkDebugMacro(<< "ShouldIterateTemporalData returns 0 (no outputs)");
    return 0;
    }

  // if the filter is a subclass of vtkTemporalDataSetAlgorithm
  // we do not need to loop, because it will input and output temporal data
  if (this->Algorithm->IsA("vtkTemporalDataSetAlgorithm"))
    {
    vtkDebugMacro(<< "ShouldIterateTemporalData returns 0 (vtkTemporalDataSetAlgorithm)");
    return 0;
    }

  // If the input to this is Temporal, the upstream will loop
  // we do not have to.
  int i, numInputPorts = this->Algorithm->GetNumberOfInputPorts();
  for (i=0; i<numInputPorts; ++i) 
    {
    vtkInformation* inPortInfo = 
      this->Algorithm->GetInputPortInformation(i);
    const char* inputType = 
      inPortInfo->Get(vtkAlgorithm::INPUT_REQUIRED_DATA_TYPE());
    if (inputType && (strcmp(inputType, "vtkTemporalDataSet") == 0))
      {
      vtkDebugMacro(<< "ShouldIterateTemporalData returns 0 (vtkTemporalDataSet input)");
      return 0;
      }
    }

  int numOut = outInfoVec->GetNumberOfInformationObjects();
  for (int out = 0; out < numOut; out++)
    {
    if (outInfoVec->GetInformationObject(out)->Has(REQUIRES_TIME_DOWNSTREAM()))
      {
      // Time was requested so answer yes.
      vtkDebugMacro(<< "ShouldIterateTemporalData returns 1 (REQUIRES_TIME_DOWNSTREAM)");
      return 1;
      }
    }

  return 0;
}

//----------------------------------------------------------------------------
vtkDataObject* vtkCompositeDataPipeline::ExecuteSimpleAlgorithmForBlock(
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec,
  vtkInformation* inInfo,
  vtkInformation* outInfo,
  vtkInformation* request,  
  vtkDataObject* dobj)
{
  vtkDebugMacro(<< "ExecuteSimpleAlgorithmForBlock");

  // if we have an input and it is composite, then recurse
  if (dobj && dobj->IsA("vtkMultiGroupDataSet"))
    {
    vtkMultiGroupDataSet* output = vtkMultiGroupDataSet::New();

    vtkMultiGroupDataSet* mbInput = vtkMultiGroupDataSet::SafeDownCast(dobj);
    unsigned int numGroups = mbInput->GetNumberOfGroups();
    output->SetNumberOfGroups(numGroups);
    for (unsigned int groupId=0; groupId<numGroups; groupId++)
      {
      unsigned int numBlocks = mbInput->GetNumberOfDataSets(groupId);
      output->SetNumberOfDataSets(groupId, numBlocks);
      for (unsigned int blockId=0; blockId<numBlocks; blockId++)
        {
        vtkDataObject* block = mbInput->GetDataSet(groupId, blockId);
        vtkDataObject* outBlock = 0;
        if (block)
          {
          outBlock = this->ExecuteSimpleAlgorithmForBlock(inInfoVec,
                                                          outInfoVec,
                                                          inInfo,
                                                          outInfo,
                                                          request,
                                                          block);
          }
        output->SetDataSet(groupId, blockId, outBlock);
        if (outBlock)
          {
          outBlock->Delete();
          }
        }
      }
    return output;
    }
  else
    {
    double time = 0;
    int hasTime = outInfo->Length(UPDATE_TIME_STEPS());
    if (hasTime)
      {
      time = outInfo->Get(UPDATE_TIME_STEPS())[0];
      }

    // There must be a bug somehwere. If this Remove()
    // is not called, the following Set() has the effect
    // of removing (!) the key.
    if (inInfo)
      {
      inInfo->Remove(vtkDataObject::DATA_OBJECT());
      inInfo->Set(vtkDataObject::DATA_OBJECT(), dobj);
      
      // Process the whole dataset
      this->CopyFromDataToInformation(dobj, inInfo);
      }

    request->Set(REQUEST_DATA_OBJECT());
    this->Superclass::ExecuteDataObject(
      request, this->GetInputInformation(),this->GetOutputInformation());
    request->Remove(REQUEST_DATA_OBJECT());
    
    request->Set(REQUEST_INFORMATION());

    // Make sure that pipeline informations is in sync with the data
    if (dobj)
      {
      dobj->CopyInformationToPipeline(request, 0, inInfo, 1);

      // This should not be needed but since a lot of image filters do:
      // img->GetScalarType(), it is necessary.
      dobj->GetProducerPort(); // make sure there is pipeline info.
      dobj->CopyInformationToPipeline
        (request, 0, dobj->GetPipelineInformation(), 1);
      }

    this->Superclass::ExecuteInformation(request,inInfoVec,outInfoVec);
    request->Remove(REQUEST_INFORMATION());
    
    int storedPiece = -1;
    int storedNumPieces = -1;
    for(int m=0; m < this->Algorithm->GetNumberOfOutputPorts(); ++m)
      {
      vtkInformation* info = this->GetOutputInformation(m);
      // Update the whole thing
      if (info->Has(
                    vtkStreamingDemandDrivenPipeline::WHOLE_EXTENT()))
        {
        int extent[6] = {0,-1,0,-1,0,-1};
        info->Get(
          vtkStreamingDemandDrivenPipeline::WHOLE_EXTENT(),
          extent);
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_EXTENT(), 
          extent, 
          6);
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_EXTENT_INITIALIZED(), 
          1);
        storedPiece = 
          info->Get(vtkStreamingDemandDrivenPipeline::UPDATE_PIECE_NUMBER());
        storedNumPieces= 
          info->Get(vtkStreamingDemandDrivenPipeline::UPDATE_NUMBER_OF_PIECES());
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_NUMBER_OF_PIECES(), 
          1);
        vtkDebugMacro(<< "UPDATE_PIECE_NUMBER() 0"  << " " << info);
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_PIECE_NUMBER(), 0);
        }
      }

    // if there was a time make sure that gets set in the RUE
    if (hasTime)
      {
      outInfo->Set(UPDATE_TIME_STEPS(),&time,1);
      }

    request->Set(REQUEST_UPDATE_EXTENT());
    this->CallAlgorithm(request, vtkExecutive::RequestUpstream,
                        inInfoVec, outInfoVec);
    request->Remove(REQUEST_UPDATE_EXTENT());
    
    request->Set(REQUEST_DATA());
    this->Superclass::ExecuteData(request,inInfoVec,outInfoVec);
    request->Remove(REQUEST_DATA());
    
    for(int m=0; m < this->Algorithm->GetNumberOfOutputPorts(); ++m)
      {
      vtkInformation* info = this->GetOutputInformation(m);
      if (storedPiece!=-1)
        {
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_NUMBER_OF_PIECES(), 
          storedNumPieces);
        vtkDebugMacro(<< "UPDATE_PIECE_NUMBER() 0"  << " " << info);
        info->Set(
          vtkStreamingDemandDrivenPipeline::UPDATE_PIECE_NUMBER(), 
          storedPiece);
        }
      }


    vtkDataObject* output = outInfo->Get(vtkDataObject::DATA_OBJECT());
    if (!output)
      {
      return 0;
      }
    vtkDataObject* outputCopy = output->NewInstance();
    outputCopy->ShallowCopy(output);
    return outputCopy;
    }
}

//----------------------------------------------------------------------------
// Execute a simple (non-composite-aware) filter multiple times, once per
// block. Collect the result in a composite dataset that is of the same
// structure as the input.
void vtkCompositeDataPipeline::ExecuteSimpleAlgorithm(
  vtkInformation* request,
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec,
  int compositePort)
{
  vtkDebugMacro(<< "ExecuteSimpleAlgorithm");

  this->ExecuteDataStart(request,inInfoVec,outInfoVec);

  vtkInformation* outInfo = 0;

  if (this->GetNumberOfOutputPorts() > 0)
    {
    outInfo = outInfoVec->GetInformationObject(0);
    }

  // Make sure a valid composite data object exists for all output ports.
  for(int i=0; i < this->Algorithm->GetNumberOfOutputPorts(); ++i)
    {
    this->CheckCompositeData(request, i, inInfoVec, outInfoVec);
    }

  // if we have no composite inputs then we are looping over time on a source
  if (compositePort==-1)
    {
    this->ExecuteSimpleAlgorithmTime(request, inInfoVec, outInfoVec);
    return;
    }

  // Loop using the first input on the first port.
  // This might not be valid for all cases but it is a decent
  // assumption to start with.
  // TODO: Loop over all inputs
  vtkInformation* inInfo = 0;
  vtkMultiGroupDataSet* updateInfo = 0;
  inInfo = this->GetInputInformation(compositePort, 0);
  vtkCompositeDataSet* input = vtkCompositeDataSet::SafeDownCast
    (inInfo->Get(vtkDataObject::DATA_OBJECT()));
  
  updateInfo = vtkMultiGroupDataSet::SafeDownCast
    (inInfo->Get(vtkCompositeDataPipeline::UPDATE_BLOCKS()));
  
  vtkSmartPointer<vtkCompositeDataSet> compositeOutput = 
    vtkCompositeDataSet::SafeDownCast
    (outInfo->Get(vtkDataObject::DATA_OBJECT()));
  
  // do we have a request for multiple time steps?
  int numTimeSteps = 0;
  double *times = 0;
  numTimeSteps = outInfo->Length(UPDATE_TIME_STEPS());
  if (numTimeSteps)
    {
    times = new double [numTimeSteps];
    memcpy(times,outInfo->Get(UPDATE_TIME_STEPS()),
           sizeof(double)*numTimeSteps);
    }

  if (input && updateInfo)
    {
    int outputInitialized = 0;

    vtkSmartPointer<vtkInformation> r = 
      vtkSmartPointer<vtkInformation>::New();

    r->Set(FROM_OUTPUT_PORT(), inInfo->GetPort(PRODUCER()));

    // The request is forwarded upstream through the pipeline.
    r->Set(vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
        
    // Algorithms process this request after it is forwarded.
    r->Set(vtkExecutive::ALGORITHM_AFTER_FORWARD(), 1);

    // iterate over time by default
    unsigned int numGroups = numTimeSteps;
    // if we have a composite input the use that instead
    vtkSmartPointer<vtkDataObject> prevInput = 0;
    numGroups = updateInfo->GetNumberOfGroups();
    prevInput = inInfo->Get(vtkDataObject::DATA_OBJECT());

    // Store the information (whole_extent and maximum_number_of_pieces)
    // before looping. Otherwise, executeinformation will cause
    // changes (because we pretend that the max. number of pieces is
    // one to process the whole block)
    this->PushInformation(inInfo);

    vtkDebugMacro(<< "EXECUTING " << this->Algorithm->GetClassName());;
    
    // True when the pipeline is iterating over the current (simple)
    // filter to produce composite output. In this case,
    // ExecuteDataStart() should NOT Initialize() the composite output.
    this->InLocalLoop = 1;

    for (unsigned int k=0; k<numGroups; k++)
      {
      // if it is a temporal input, set the time for each piece
      if (times)
        {
        outInfo->Set(UPDATE_TIME_STEPS(),times,numTimeSteps);
        }
      unsigned int numDataSets = updateInfo->GetNumberOfDataSets(k);
      for (unsigned l=0; l<numDataSets; l++)
        {
        if (updateInfo->GetDataSet(k,l))
          {
          r->Set(vtkMultiGroupDataSet::GROUP(), k);
          r->Set(vtkCompositeDataSet::INDEX(),    l);
          vtkDataObject* dobj = input->GetDataSet(r);

          vtkDataObject* outCopy = 0;
          if (dobj)
            {
            outCopy = this->ExecuteSimpleAlgorithmForBlock(inInfoVec,
                                                           outInfoVec,
                                                           inInfo,
                                                           outInfo,
                                                           r,
                                                           dobj);
            }
          if (!outputInitialized)
            {
            compositeOutput->PrepareForNewData();
            outputInitialized = 1;
            }
          compositeOutput->AddDataSet(r, outCopy);
          if (outCopy)
            {
            outCopy->Delete();
            }
          }
        }
      }
    // True when the pipeline is iterating over the current (simple)
    // filter to produce composite output. In this case,
    // ExecuteDataStart() should NOT Initialize() the composite output.
    this->InLocalLoop = 0;
    // Restore the extent information and force it to be
    // copied to the output. Composite sources should set
    // MAXIMUM_NUMBER_OF_PIECES to -1 anyway (and handle
    // piece requests properly).
    this->PopInformation(inInfo);
    if (times)
      {
      outInfo->Set(UPDATE_TIME_STEPS(),times,numTimeSteps);
      compositeOutput->GetInformation()
        ->Set(vtkDataObject::DATA_TIME_STEPS(),times,numTimeSteps);
      delete [] times;
      }
    r->Set(REQUEST_INFORMATION());
    this->CopyDefaultInformation(r, vtkExecutive::RequestDownstream,
                                 this->GetInputInformation(),
                                 this->GetOutputInformation());
        
    vtkDataObject* curInput = inInfo->Get(vtkDataObject::DATA_OBJECT());
    if (curInput != prevInput)
      {
      inInfo->Remove(vtkDataObject::DATA_OBJECT());
      inInfo->Set(vtkDataObject::DATA_OBJECT(), prevInput);
      }
    vtkDataObject* curOutput = outInfo->Get(vtkDataObject::DATA_OBJECT());
    if (curOutput != compositeOutput.GetPointer())
      {
      compositeOutput->SetPipelineInformation(outInfo);
      }
    }
  this->ExecuteDataEnd(request,inInfoVec,outInfoVec);
}


//----------------------------------------------------------------------------
// Execute a simple (non-composite-aware) filter multiple times, once per
// block. Collect the result in a composite dataset that is of the same
// structure as the input.
void vtkCompositeDataPipeline::ExecuteSimpleAlgorithmTime(
  vtkInformation* request,
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec)
{
  vtkDebugMacro(<< "ExecuteSimpleAlgorithmTime");

  vtkInformation* outInfo = 0;
  vtkSmartPointer<vtkInformation> originalInformation = 
    vtkSmartPointer<vtkInformation>::New();

  if (this->GetNumberOfOutputPorts() > 0)
    {
    outInfo = outInfoVec->GetInformationObject(0);
    originalInformation->CopyEntry(outInfo,TIME_STEPS(),0);
    originalInformation->CopyEntry(outInfo,TIME_RANGE(),0);
    }

  vtkSmartPointer<vtkCompositeDataSet> temporalOutput = 
    vtkCompositeDataSet::SafeDownCast
    (outInfo->Get(vtkDataObject::DATA_OBJECT()));
  
  // do we have a request for multiple time steps?
  int numTimeSteps = 0;
  double *times = 0;
  numTimeSteps = outInfo->Length(UPDATE_TIME_STEPS());
  times = new double [numTimeSteps];
  memcpy(times,outInfo->Get(UPDATE_TIME_STEPS()),
         sizeof(double)*numTimeSteps);

  int outputInitialized = 0;
  
  vtkSmartPointer<vtkInformation> r = 
    vtkSmartPointer<vtkInformation>::New();
  
  r->Set(FROM_OUTPUT_PORT(), request->Get(FROM_OUTPUT_PORT()));

  // The request is forwarded upstream through the pipeline.
  r->Set(vtkExecutive::FORWARD_DIRECTION(), vtkExecutive::RequestUpstream);
  
  // Algorithms process this request after it is forwarded.
  r->Set(vtkExecutive::ALGORITHM_AFTER_FORWARD(), 1);
  
  vtkDebugMacro(<<"EXECUTING: " << this->Algorithm->GetClassName());
  
  // Store the information (whole_extent and maximum_number_of_pieces)
  // before looping. Otherwise, executeinformation will cause
  // changes (because we pretend that the max. number of pieces is
  // one to process the whole block)
//  vtkInformation* inInfo = this->GetInputInformation(compositePort, 0);
//  this->PushInformation(inInfo);

  // True when the pipeline is iterating over the current (simple)
  // filter to produce composite output. In this case,
  // ExecuteDataStart() should NOT Initialize() the composite output.
  this->InLocalLoop = 1;
  
  for (unsigned int k=0; k< static_cast<unsigned int>(numTimeSteps); k++)
    {
    // if it is a temporal input, set the time for each piece
    outInfo->Set(UPDATE_TIME_STEPS(),times+k,1);

    r->Set(vtkMultiGroupDataSet::GROUP(), k);
    r->Set(vtkCompositeDataSet::INDEX(), 0);
    vtkDataObject* dobj = 0;
    
    vtkDataObject* outCopy = 
      this->ExecuteSimpleAlgorithmForBlock(inInfoVec,
                                           outInfoVec,
                                           0,
                                           outInfo,
                                           r,
                                           dobj);
    if (outCopy)
      {
      vtkDebugMacro(<<"Got Data from Block");
      if (!outputInitialized)
        {
        temporalOutput->PrepareForNewData();
        outputInitialized = 1;
        }
      temporalOutput->AddDataSet(r, outCopy);
      outCopy->Delete();
      }
    }

  // True when the pipeline is iterating over the current (simple)
  // filter to produce composite output. In this case,
  // ExecuteDataStart() should NOT Initialize() the composite output.
  this->InLocalLoop = 0;

  // Restore the extent information and force it to be
  // copied to the output. Composite sources should set
  // MAXIMUM_NUMBER_OF_PIECES to -1 anyway (and handle
  // piece requests properly).
//  this->PopInformation(inInfo);

  outInfo->Set(UPDATE_TIME_STEPS(),times,numTimeSteps);
  temporalOutput->GetInformation()
    ->Set(vtkDataObject::DATA_TIME_STEPS(),times,numTimeSteps);
  delete [] times;
  
  r->Set(REQUEST_INFORMATION());
  this->CopyDefaultInformation(r, vtkExecutive::RequestDownstream,
                               this->GetInputInformation(),
                               this->GetOutputInformation());

  outInfo->CopyEntry(originalInformation,TIME_STEPS(),0);
  outInfo->CopyEntry(originalInformation,TIME_RANGE(),0);
  
  vtkDataObject* curOutput = outInfo->Get(vtkDataObject::DATA_OBJECT());
  if (curOutput != temporalOutput.GetPointer())
    {
    temporalOutput->SetPipelineInformation(outInfo);
    }
  this->ExecuteDataEnd(request,inInfoVec,outInfoVec);
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::NeedToExecuteData(
  int outputPort,
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec)
{
  // Has the algorithm asked to be executed again?
  if(this->ContinueExecuting)
    {
    return 1;
    }

  // If no port is specified, check all ports.  This behavior is
  // implemented by the superclass.
  if(outputPort < 0)
    {
    return this->Superclass::NeedToExecuteData(outputPort,
                                               inInfoVec,outInfoVec);
    }

  // Does the superclass want to execute?
  if(this->vtkDemandDrivenPipeline::NeedToExecuteData(
       outputPort,inInfoVec,outInfoVec))
    {
    return 1;
    }

  // We need to check the requested update extent.  Get the output
  // port information and data information.  We do not need to check
  // existence of values because it has already been verified by
  // VerifyOutputInformation.
  vtkInformation* outInfo = outInfoVec->GetInformationObject(outputPort);
  vtkDataObject* dataObject = outInfo->Get(vtkDataObject::DATA_OBJECT());
  if (!vtkCompositeDataSet::SafeDownCast(dataObject))
    {
    return this->Superclass::NeedToExecuteData(outputPort,
                                               inInfoVec,outInfoVec);
    }
  vtkInformation* dataInfo = dataObject->GetInformation();

  // Check the unstructured extent.  If we do not have the requested
  // piece, we need to execute.
  int updateNumberOfPieces = outInfo->Get(UPDATE_NUMBER_OF_PIECES());
  int dataNumberOfPieces = dataInfo->Get(vtkDataObject::DATA_NUMBER_OF_PIECES());
  if(dataNumberOfPieces != updateNumberOfPieces)
    {
    return 1;
    }
  if (dataNumberOfPieces != 1)
    {
    int dataPiece = dataInfo->Get(vtkDataObject::DATA_PIECE_NUMBER());
    int updatePiece = outInfo->Get(UPDATE_PIECE_NUMBER());
    if (dataPiece != updatePiece)
      {
      return 1;
      }
    }

  if (this->NeedToExecuteBasedOnTime(outInfo, dataObject))
    {
    return 1;
    }

  // We do not need to execute.
  return 0;
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::NeedToExecuteBasedOnTime(
                                                      vtkInformation* outInfo,
                                                      vtkDataObject* dataObject)
{
  if (this->Superclass::NeedToExecuteBasedOnTime(outInfo, dataObject))
    {
    vtkDebugMacro(<<"NeedToExecuteBasedOnTime returns 1");
    return 1;
    }

  if (outInfo->Has(REQUIRES_TIME_DOWNSTREAM()))
    {
    // Check to see if there is anything new to do to make a proper
    // vtkTemporalDataSet.
    if (!dataObject->IsA("vtkTemporalDataSet"))
      {
      // Re-run the algorithm to generate a vtkTemporalDataSet.  This is not
      // very efficient, as we could just take the current data set and stuff it
      // into a vtkTemporalDataSet without re-running the algorithm.  In fact,
      // it would be nice if vtkCompositeDataPipeline would always use existing
      // data in its output to fill a vtkTemporalDataSet with a set at the same
      // time step.  However, that would require a lot of logic about whether
      // something else changed, and it is not a straightforward change.
      vtkDebugMacro(<<"NeedToExecuteBasedOnTime returns 1 (!vtkTemporalDataSet)");
      return 1;
      }
    }

  return 0;
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline::ForwardUpstream(
  int i, int j, vtkInformation* request)
{
  // Do not forward upstream if input information is shared.
  if(this->SharedInputInformation)
    {
    return 1;
    }

  if (!this->Algorithm->ModifyRequest(request, BeforeForward))
    {
    return 0;
    }

  int result = 1;
  if(vtkExecutive* e = this->GetInputExecutive(i, j))
    {
    vtkAlgorithmOutput* input = this->Algorithm->GetInputConnection(i, j);
    int port = request->Get(FROM_OUTPUT_PORT());
    request->Set(FROM_OUTPUT_PORT(), input->GetIndex());
    if(!e->ProcessRequest(request,
                          e->GetInputInformation(),
                          e->GetOutputInformation()))
      {
      result = 0;
      }
    request->Set(FROM_OUTPUT_PORT(), port);
    }

  if (!this->Algorithm->ModifyRequest(request, AfterForward))
    {
    return 0;
    }

  return result;
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::CopyDefaultInformation(
  vtkInformation* request, int direction,
  vtkInformationVector** inInfoVec,
  vtkInformationVector* outInfoVec)
{
  int hasUpdateBlocks = 0;
  if(direction == vtkExecutive::RequestDownstream)
    {
    vtkInformationKey** keys = request->Get(vtkExecutive::KEYS_TO_COPY());
    if (keys)
      {
      int len = request->Length(vtkExecutive::KEYS_TO_COPY());
      for (int i=0; i<len; i++)
        {
        if (keys[i] == vtkCompositeDataPipeline::UPDATE_BLOCKS())
          {
          hasUpdateBlocks = 1;
          break;
          }
        }
      if (hasUpdateBlocks)
        {
        request->Remove(vtkExecutive::KEYS_TO_COPY(),
                        vtkCompositeDataPipeline::UPDATE_BLOCKS());
        }
      }
    }

  this->Superclass::CopyDefaultInformation(request, direction, 
                                           inInfoVec, outInfoVec);

  if(request->Has(REQUEST_UPDATE_EXTENT()))
    {
    // Find the port that has a data that we will iterator over.
    // If there is one, make sure that we use piece extent for
    // that port. Composite data pipeline works with piece extents
    // only.
    int compositePort;
    if (this->ShouldIterateOverInput(compositePort))
      {
      // Get the output port from which to copy the extent.
      int outputPort = -1;
      if(request->Has(FROM_OUTPUT_PORT()))
        {
        outputPort = request->Get(FROM_OUTPUT_PORT());
        }
      
      // Setup default information for the inputs.
      if(outInfoVec->GetNumberOfInformationObjects() > 0)
        {
        // Copy information from the output port that made the request.
        // Since VerifyOutputInformation has already been called we know
        // there is output information with a data object.
        vtkInformation* outInfo =
          outInfoVec->GetInformationObject((outputPort >= 0)? outputPort : 0);

        // Loop over all connections on this input port.
        int numInConnections = 
          inInfoVec[compositePort]->GetNumberOfInformationObjects();
        for (int j=0; j<numInConnections; j++)
          {
          // Get the pipeline information for this input connection.
          vtkInformation* inInfo = 
            inInfoVec[compositePort]->GetInformationObject(j);
            
          if (outInfo->Has(UPDATE_TIME_STEPS()))
            {
            inInfo->CopyEntry(outInfo, UPDATE_TIME_STEPS());
            }
          if (outInfo->Has(FAST_PATH_OBJECT_ID()))
            {
            inInfo->CopyEntry(outInfo, FAST_PATH_OBJECT_ID());
            }
          if (outInfo->Has(FAST_PATH_ID_TYPE()))
            {
            inInfo->CopyEntry(outInfo, FAST_PATH_ID_TYPE());
            }
          if (outInfo->Has(FAST_PATH_OBJECT_TYPE()))
            {
            inInfo->CopyEntry(outInfo, FAST_PATH_OBJECT_TYPE());
            }
          vtkDebugMacro(<< "CopyEntry UPDATE_PIECE_NUMBER() " << outInfo->Get(UPDATE_PIECE_NUMBER()) << " " << outInfo);

          inInfo->CopyEntry(outInfo, UPDATE_PIECE_NUMBER());
          inInfo->CopyEntry(outInfo, UPDATE_NUMBER_OF_PIECES());
          inInfo->CopyEntry(outInfo, UPDATE_NUMBER_OF_GHOST_LEVELS());
          inInfo->CopyEntry(outInfo, UPDATE_EXTENT_INITIALIZED());
          }
        }
      }
    }

  if (hasUpdateBlocks)
    {
    request->Append(vtkExecutive::KEYS_TO_COPY(), 
                    vtkCompositeDataPipeline::UPDATE_BLOCKS());
    }
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::ResetPipelineInformation(int port,
                                                        vtkInformation* info)
{
  if (this->SuppressResetPipelineInformation)
    {
    return;
    }

  this->Superclass::ResetPipelineInformation(port, info);
  info->Remove(COMPOSITE_DATA_INFORMATION());
  info->Remove(UPDATE_BLOCKS());
  info->Remove(REQUIRES_TIME_DOWNSTREAM());
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::CopyFromDataToInformation(
  vtkDataObject* dobj, vtkInformation* inInfo)
{
  if (dobj->IsA("vtkImageData"))
    {
    inInfo->Set(
      WHOLE_EXTENT(), static_cast<vtkImageData*>(dobj)->GetExtent(), 6);
    }
  else if (dobj->IsA("vtkStructuredGrid"))
    {
    inInfo->Set(
      WHOLE_EXTENT(), static_cast<vtkStructuredGrid*>(dobj)->GetExtent(), 6);
    }
  else if (dobj->IsA("vtkRectilinearGrid"))
    {
    inInfo->Set(
      WHOLE_EXTENT(), static_cast<vtkRectilinearGrid*>(dobj)->GetExtent(), 6);
    }
  else if (dobj->IsA("vtkUniformGrid"))
    {
    inInfo->Set(
      WHOLE_EXTENT(), static_cast<vtkUniformGrid*>(dobj)->GetExtent(), 6);
    }
  else
    {
    inInfo->Set(MAXIMUM_NUMBER_OF_PIECES(), 1);
    }
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::PushInformation(vtkInformation* inInfo)
{
  vtkDebugMacro(<< "PushInformation " << inInfo);
  this->InformationCache->CopyEntry(inInfo, WHOLE_EXTENT());
  this->InformationCache->CopyEntry(inInfo, MAXIMUM_NUMBER_OF_PIECES());
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::PopInformation(vtkInformation* inInfo)
{
  vtkDebugMacro(<< "PopInformation " << inInfo);
  inInfo->CopyEntry(this->InformationCache, WHOLE_EXTENT());
  inInfo->CopyEntry(this->InformationCache, MAXIMUM_NUMBER_OF_PIECES());
}

//----------------------------------------------------------------------------
int vtkCompositeDataPipeline
::CheckCompositeData(vtkInformation *request,
                     int port, 
                     vtkInformationVector** inInfoVec,
                     vtkInformationVector* outInfoVec)
{
  // Check that the given output port has a valid data object.
  vtkInformation* outInfo = outInfoVec->GetInformationObject(port);

  // If this is a simple filter but has composite input, 
  // create a composite output.
  int compositePort;
  int temporaldownstream
    =request && this->ShouldIterateTemporalData(request, inInfoVec, outInfoVec);
  if (this->ShouldIterateOverInput(compositePort) || temporaldownstream)
    {
    // This assumes that the first output of the filter is the one
    // that will have the composite data.
    vtkDataObject* doOutput = 
      outInfo->Get(vtkDataObject::DATA_OBJECT());
    vtkCompositeDataSet* output = vtkCompositeDataSet::SafeDownCast(doOutput);
    vtkTemporalDataSet* temporal = vtkTemporalDataSet::SafeDownCast(doOutput);
    if (!output || (temporaldownstream && !temporal))
      {
      if (temporaldownstream)
        {
        vtkDebugMacro(<< "CheckCompositeData Created vtkTemporalDataSet output");
        output = vtkTemporalDataSet::New();
        }
      else
        {
        vtkDebugMacro(<< "CheckCompositeData Created vtkMultiGroupDataSet output");
        output = vtkMultiGroupDataSet::New();
        }
      output->SetPipelineInformation(outInfo);
      output->Delete();
      }
    return 1;
    }
  // Otherwise, create a simple output
  else
    {
    return this->Superclass::CheckDataObject(port, outInfoVec);
    }
}

//----------------------------------------------------------------------------
vtkDataObject* vtkCompositeDataPipeline::GetCompositeInputData(
  int port, int index, vtkInformationVector **inInfoVec)
{
  if (!inInfoVec[port])
    {
    return 0;
    }
  vtkInformation *info = inInfoVec[port]->GetInformationObject(index);
  if (!info)
    {
    return 0;
    }
  return info->Get(vtkDataObject::DATA_OBJECT());
}

//----------------------------------------------------------------------------
vtkDataObject* vtkCompositeDataPipeline::GetCompositeOutputData(int port)
{
  if(!this->OutputPortIndexInRange(port, "get data for"))
    {
    return 0;
    }

  // Check that the given output port has a valid data object.
  vtkDebugMacro(<< "GetCompositeOutputData calling CheckCompositeData ");

  this->CheckCompositeData(0, port, this->GetInputInformation(), this->GetOutputInformation());

  // Return the data object.
  if(vtkInformation* info = this->GetOutputInformation(port))
    {
    return info->Get(vtkDataObject::DATA_OBJECT());
    }
  return 0;
}

//----------------------------------------------------------------------------
void vtkCompositeDataPipeline::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
}

