Merge.cpp

Go to the documentation of this file.

/*  Copyright (C) 2008 National Institute For Space Research (INPE) - Brazil.

    This file is part of the TerraLib - a Framework for building GIS enabled applications.

    TerraLib is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License,
    or (at your option) any later version.

    TerraLib is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with TerraLib. See COPYING. If not, write to
    TerraLib Team at <terralib-team@terralib.org>.
 */

/*!
 \file Merge.cpp
 */

#include "../common/progress/TaskProgress.h"
#include "../core/logger/Logger.h"
#include "../core/translator/Translator.h"
#include "../dataaccess/dataset/DataSet.h"
#include "../dataaccess/dataset/DataSetAdapter.h"
#include "../dataaccess/dataset/DataSetType.h"
#include "../dataaccess/dataset/DataSetTypeConverter.h"
#include "../dataaccess/dataset/ObjectIdSet.h"
#include "../dataaccess/datasource/DataSource.h"
#include "../dataaccess/datasource/DataSourceInfo.h"
#include "../dataaccess/datasource/DataSourceInfoManager.h"
#include "../dataaccess/datasource/DataSourceCapabilities.h"
#include "../dataaccess/datasource/DataSourceTransactor.h"
#include "../dataaccess/query_h.h"
#include "../dataaccess/utils/Utils.h"

#include "../datatype/Property.h"
#include "../datatype/SimpleData.h"
#include "../datatype/StringProperty.h"

#include "../geometry/Geometry.h"
#include "../geometry/GeometryCollection.h"
#include "../geometry/GeometryProperty.h"
#include "../geometry/MultiLineString.h"
#include "../geometry/MultiPoint.h"
#include "../geometry/MultiPolygon.h"
#include "../geometry/Utils.h"

#include "../memory/DataSet.h"
#include "../memory/DataSetItem.h"

#include "AlgorithmParams.h"
#include "ComplexData.h"
#include "Merge.h"
#include "Utils.h"

// BOOST
#include <boost/lexical_cast.hpp>
#include <boost/uuid/random_generator.hpp>
#include <boost/uuid/uuid_io.hpp>

// STL
#include <time.h>

const int BLOCKSIZE = 10000;

te::vp::Merge::Merge()
: m_targetDst(nullptr),
  m_targetDs(nullptr),
  m_originDst(nullptr),
  m_originDs(nullptr),
  m_outDsetName(""),
  m_isUpdate(false),
  m_targetSRID(0)
{
}

bool te::vp::Merge::executeMemory(te::vp::AlgorithmParams* mainParams)
{
  clock_t tStart = clock();
  TE_LOG_INFO(TE_TR("Merge (memory) operation started"));

  te::vp::ValidateAlgorithmParams(mainParams, te::vp::MEMORY);

  m_targetSource = mainParams->getInputParams()[0].m_inputDataSource;
  m_targetDst = mainParams->getInputParams()[0].m_inputDataSetType;
  m_targetDs = mainParams->getInputParams()[0].m_inputDataSet;

  m_originSource = mainParams->getInputParams()[1].m_inputDataSource;
  m_originDst = mainParams->getInputParams()[1].m_inputDataSetType;
  m_originDs = mainParams->getInputParams()[1].m_inputDataSet;
  
  m_outDsetName = mainParams->getOutputDataSetName();
  m_outDsrc = mainParams->getOutputDataSource();

  m_isUpdate = isUpdate(mainParams->getSpecificParams());

  TE_LOG_INFO(TE_TR("Operation Output: ") + (m_isUpdate?TE_TR("Update"):TE_TR("New Layer")));

  te::gm::GeometryProperty* gpInput = te::da::GetFirstGeomProperty(m_targetDst);

  m_targetSRID = gpInput->getSRID();

  m_properties = getPropNames(mainParams->getSpecificParams());

  ///////////////////////////////////////////////////////////////////////////////////////////////

  std::unique_ptr<te::da::DataSourceTransactor> transactor;

  try {

    std::unique_ptr<te::da::DataSetType> dst;

    te::common::TaskProgress task("Merging...");
    task.useTimer(true);
    if (m_isUpdate)
    {
      task.setTotalSteps((int)m_originDs->size());
    }
    else
    {
      task.setTotalSteps((int)m_targetDs->size() + (int)m_originDs->size());
    }

    
    //task.setTotalSteps((int)(m_selectedAttVec.size() * vectorMap.size()));

    if (m_isUpdate)
    {
      transactor = m_targetSource->getTransactor();

      updateInputDst(transactor.get());

      std::unique_ptr<te::da::DataSetType> aux = transactor->getDataSetType(m_targetDst->getName());

      std::unique_ptr<te::da::DataSetTypeConverter> targetConverter(new te::da::DataSetTypeConverter(aux.release(), m_targetSource->getCapabilities(), m_targetSource->getEncoding()));
      te::da::AssociateDataSetTypeConverterSRID(targetConverter.get(), m_targetSRID);

      te::da::DataSetType* dtClone = dynamic_cast<te::da::DataSetType*>(targetConverter->getResult()->clone());

      dst.reset(dtClone);
    }
    else
    {
      dst = getOutputDataSetType(mainParams);

      transactor = m_outDsrc->getTransactor();
    }

    MergeStrategy s = checkStrategy();
    int pkMax = -1;
    std::size_t pkPos=std::string::npos;
    int pkType=-1;
    std::string pkName;
    
    if (s == AUTOINCREMENT)
    {
      te::dt::Property* pkProp = dst->getPrimaryKey()->getProperties()[0];
      pkName = pkProp->getName();

      TE_LOG_INFO(TE_TR("Operation strategy: ") + TE_TR("AUTOINCREMENT"));
    }
    else if (s == TRYGETMAX)
    {
      te::dt::Property* pkProp = dst->getPrimaryKey()->getProperties()[0];
      pkMax = tryGetMax();
      pkPos = dst->getPropertyPosition(pkProp);
      pkType = getPropertyType(pkProp);

      TE_LOG_INFO(TE_TR("Operation strategy: ") + TE_TR("TRYGETMAX"));
    }
    else
    {
      TE_LOG_INFO(TE_TR("Operation strategy: ") + TE_TR("PUREMERGE"));
    }

    te::mem::DataSet* ds = new te::mem::DataSet(dst.get());

    te::gm::GeometryProperty* fgGrop = te::da::GetFirstGeomProperty(m_targetDst);
    te::gm::GeometryProperty* sgGrop = te::da::GetFirstGeomProperty(m_originDst);
    te::gm::GeometryProperty* outgGrop = te::da::GetFirstGeomProperty(dst.get());

    std::size_t fgPos = m_targetDst->getPropertyPosition(fgGrop);
    std::size_t sgPos = m_originDst->getPropertyPosition(sgGrop);
    std::size_t outgPos = dst->getPropertyPosition(outgGrop);

    transactor->begin();
    TE_LOG_INFO(TE_TR("Transaction Begin"));

    int count = 0;

    m_targetDs->moveBeforeFirst();
    m_originDs->moveBeforeFirst();

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

    bool isDtCreated = false; // When the operation is not a update operation

    if (!m_isUpdate)
    {
      TE_LOG_INFO(TE_TR("Start iterating Target Layer"));
      int targetCount = 0;
      while (m_targetDs->moveNext())
      {
        if (!task.isActive())
        {
          std::string msg = TE_TR("Operation canceled by the user");
          transactor->rollBack();
          throw te::common::Exception(msg, 1);
        }

        task.pulse();
        te::mem::DataSetItem* item = new te::mem::DataSetItem(ds);

        for (std::size_t i = 0; i < m_properties.size(); ++i)
        {
          if (m_properties[i].first.empty() || m_targetDs->isNull(m_properties[i].first))
            continue;

          if (!m_properties[i].first.empty())
          {
            if (isPrimaryKeyProperty(dst.get(), m_properties[i].first))
            {
              if (s != AUTOINCREMENT)
              {
                if (s == PUREMERGE)
                {
                  item->setValue(m_properties[i].first, m_targetDs->getValue(m_properties[i].first).release());
                }
                else if (s == TRYGETMAX)
                {
                  item->setValue(m_properties[i].first, m_targetDs->getValue(m_properties[i].first).release());
                }
              }
            }
            else
            {
              item->setValue(m_properties[i].first, m_targetDs->getValue(m_properties[i].first).release());
            }
          }
        }

        item->setGeometry(outgPos, m_targetDs->getGeometry(fgPos).release());

        ds->add(item);

        if (count == 0)
        {
          if (s == AUTOINCREMENT)
          {
            std::vector<std::string> pkNameVec;
            pkNameVec.push_back(pkName);

            std::unique_ptr<te::da::DataSet> adaptDs = te::da::HideColumns(ds, dst.get(), pkNameVec);

            transactor->createDataSet(dst.get(), op);
            transactor->add(dst->getName(), adaptDs.get(), op);

            isDtCreated = true;

          }
          else
          {
            transactor->createDataSet(dst.get(), op);
            transactor->add(dst->getName(), ds, op);

            isDtCreated = true;
          }

          delete ds;

          ds = new te::mem::DataSet(dst.get());
        }        

        if ((count % BLOCKSIZE) == 0)
        {
          TE_LOG_INFO(TE_TR("  Target Layer Iterating Count: ") + std::to_string(targetCount));

          if (!isDtCreated)
          {
            transactor->createDataSet(dst.get(), op);
            isDtCreated = true;
          }

          if (s == AUTOINCREMENT)
          {
            std::vector<std::string> pkNameVec;
            pkNameVec.push_back(pkName);

            std::unique_ptr<te::da::DataSet> adaptDs = te::da::HideColumns(ds, dst.get(), pkNameVec);

            transactor->add(dst->getName(), adaptDs.get(), op);
          }
          else
          {
            transactor->add(dst->getName(), ds, op);
          }

          delete ds;

          ds = new te::mem::DataSet(dst.get());
        }

        ++count;
        ++targetCount;
      }
      TE_LOG_INFO(TE_TR("  Target Layer Iterating Count: ") + std::to_string(targetCount));
      TE_LOG_INFO(TE_TR("Target Layer iterating finished"));
    }

    if (m_isUpdate)
      isDtCreated = true;

    int originCount = 0;
    TE_LOG_INFO(TE_TR("Start iterating Origin Layer"));
    while (m_originDs->moveNext() || m_originDs->isEmpty())
    {
      if (!task.isActive())
      {
        std::string msg = TE_TR("Operation canceled by the user");
        transactor->rollBack();
        throw te::common::Exception(msg, 1);
      }

      task.pulse();
      te::mem::DataSetItem* item = new te::mem::DataSetItem(ds);

      if (s == TRYGETMAX)
      {
        ++pkMax;

        if (pkType == te::dt::INT16_TYPE)
        {
          item->setInt16(pkPos, pkMax);
        }
        else if (pkType == te::dt::INT32_TYPE)
        {
          item->setInt32(pkPos, pkMax);
        }
        else
        {
          item->setInt64(pkPos, pkMax);
        }
      }

      for (std::size_t i = 0; i < m_properties.size(); ++i)
      {
        if (!m_properties[i].second.empty())
        {
          if (m_originDs->isNull(m_properties[i].second))
            continue;

          if (!m_properties[i].first.empty())
          {
            if (s != TRYGETMAX && isPrimaryKeyProperty(dst.get(), m_properties[i].first))
            {
              if (s != AUTOINCREMENT)
              {
                if (s == PUREMERGE)
                {
                  item->setValue(m_properties[i].first, m_originDs->getValue(m_properties[i].second).release());
                }
              }
            }
            else
            {
              if(!isPrimaryKeyProperty(dst.get(), m_properties[i].first))
                item->setValue(m_properties[i].first, m_originDs->getValue(m_properties[i].second).release());
            }
          }
          else
          {
            item->setValue(m_properties[i].second, m_originDs->getValue(m_properties[i].second).release());
          }
        }
      }

      std::unique_ptr<te::gm::Geometry> geom = m_originDs->getGeometry(sgPos);

      if (fgGrop->getSRID() != sgGrop->getSRID())
      {
        geom->transform(fgGrop->getSRID());
      }

      item->setGeometry(outgPos, geom.release());

      ds->add(item);

      if ((count % BLOCKSIZE) == 0)
      {

        TE_LOG_INFO("  "+TE_TR("Origin Layer Iterating Count: ") + std::to_string(originCount));

        if (!isDtCreated)
        {
          transactor->createDataSet(dst.get(), op);
          isDtCreated = true;
        }

        if (s == AUTOINCREMENT)
        {
          std::vector<std::string> pkNameVec;
          pkNameVec.push_back(pkName);

          std::unique_ptr<te::da::DataSet> adaptDs = te::da::HideColumns(ds, dst.get(), pkNameVec);

          transactor->add(dst->getName(), adaptDs.get(), op);
        }
        else
        {
          transactor->add(dst->getName(), ds, op);
        }

        delete ds;

        ds = new te::mem::DataSet(dst.get());
      }

      ++count;
      ++originCount;
    }
    TE_LOG_INFO("  "+TE_TR("Origin Layer Iterating Count: ") + std::to_string(originCount));
    TE_LOG_INFO(TE_TR("Origin Layer iterating finished"));
    TE_LOG_INFO(TE_TR("Total Count: ") + std::to_string(count));

    if (ds)
    {
      if (!isDtCreated)
      {
        transactor->createDataSet(dst.get(), op);
        isDtCreated = true;
      }

      if (s == AUTOINCREMENT)
      {
        std::vector<std::string> pkNameVec;
        pkNameVec.push_back(pkName);

        std::unique_ptr<te::da::DataSet> adaptDs = te::da::HideColumns(ds, dst.get(), pkNameVec);

        transactor->add(dst->getName(), adaptDs.get(), op);
      }
      else
      {
        transactor->add(dst->getName(), ds, op);
      }

      delete ds;
    }

    TE_LOG_INFO(TE_TR("Transaction Commit"));
    transactor->commit();

  }
  catch (const te::common::Exception& e)
  {
    if (e.code() == 1)
    {
      TE_LOG_INFO(e.what());
    }
    else
    {
      TE_LOG_ERROR(TE_TR("Exception: ") + e.what());      
    }

    TE_LOG_INFO(TE_TR("Transaction RollBack"));
    transactor->rollBack();
    throw;
  }
  catch (std::exception& e)
  {
    TE_LOG_ERROR(TE_TR("Exception: ") + e.what());
    TE_LOG_INFO(TE_TR("Transaction RollBack"));
    transactor->rollBack();
    throw;
  }

  int seconds = (clock() - tStart) / CLOCKS_PER_SEC;
  int hours, minutes;
  minutes = seconds / 60;
  hours = minutes / 60;
  std::string timeStr = std::to_string(int(hours)) + ":" + std::to_string(int(minutes % 60)) + ":" + std::to_string(int(seconds % 60));
  TE_LOG_INFO(TE_TR("Merge (memory) operation finished") +" - " +TE_TR("Time Spent: ") + timeStr);
  return true;
}

bool te::vp::Merge::isUpdate(const std::map<std::string, te::dt::AbstractData*>& specificParams)
{
  std::map<std::string, te::dt::AbstractData*>::const_iterator it = specificParams.begin();

  bool isUpdate = false;

  while (it != specificParams.end())
  {
    if (it->first != "ISUPDATE")
    {
      ++it;
      continue;
    }

    te::dt::AbstractData* abData = it->second;

    te::vp::ComplexData<bool>* cd = dynamic_cast<te::vp::ComplexData<bool>* >(abData);

    if (cd)
      isUpdate = cd->getValue();

    ++it;
  }

  return isUpdate;
}

std::vector<std::pair<std::string, std::string> > te::vp::Merge::getPropNames(const std::map<std::string, te::dt::AbstractData*>& specificParams)
{
  std::map<std::string, te::dt::AbstractData*>::const_iterator it = specificParams.begin();

  std::vector<std::pair<std::string, std::string> > propNames;

  while (it != specificParams.end())
  {
    if (it->first != "ATTRIBUTES")
    {
      ++it;
      continue;
    }

    te::dt::AbstractData* abData = it->second;

    te::vp::ComplexData<std::vector<std::pair<std::string, std::string> > >* cd = dynamic_cast<te::vp::ComplexData<std::vector<std::pair<std::string, std::string> > >* >(abData);

    if (cd)
      propNames = cd->getValue();

    ++it;
  }

  return propNames;
}

std::unique_ptr<te::da::DataSetType> te::vp::Merge::getOutputDataSetType(
    te::vp::AlgorithmParams* /*mainParams*/)
{
  std::unique_ptr<te::da::DataSetType> dt(new te::da::DataSetType(m_outDsetName));

  for (std::size_t i = 0; i < m_properties.size(); ++i)
  {
    std::string fp = m_properties[i].first;
    std::string sp = m_properties[i].second;

    if (fp.empty() && sp.empty())
    {
      continue;
    }

    te::dt::Property* fProp = m_targetDst->getProperty(fp);
    te::dt::Property* sProp = m_originDst->getProperty(sp);

    te::dt::Property* newProp = nullptr;

    if ((!fp.empty()) && (!sp.empty()))
    {
      te::dt::SimpleProperty* fSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(fProp);
      te::dt::SimpleProperty* sSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(sProp);

      if (fSimpleProp->isRequired() != sSimpleProp->isRequired())
      {
        newProp = fProp->clone();

        te::dt::SimpleProperty* newSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(newProp);

        newSimpleProp->setRequired(false);
      }
      else
      {
        newProp = fProp->clone();
      }
    }
    else if (!fp.empty())
    {
      newProp = fProp->clone();

      te::dt::SimpleProperty* newSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(newProp);

      newSimpleProp->setRequired(false);
    }
    else if (!sp.empty())
    {
      newProp = sProp->clone();

      te::dt::SimpleProperty* newSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(newProp);

      newSimpleProp->setRequired(false);
      newSimpleProp->setAutoNumber(false);
    }
    else
    {
      throw te::common::Exception(TE_TR("An unexpected error occurred during the output dataset creation!"));
    }

    dt->add(newProp);
  }

  if (m_outDsrc->getType() != "OGR")
  {
    te::da::PrimaryKey* pk = m_targetDst->getPrimaryKey();

    std::vector<te::dt::Property*> pkProp = pk->getProperties();

    te::da::PrimaryKey* newPk = new te::da::PrimaryKey(dt->getName() + "_pk");

    for (std::size_t i = 0; i < pkProp.size(); ++i)
    {
      newPk->add(dt->getProperty(pkProp[i]->getName()));
    }

    dt->setPrimaryKey(newPk);
  }

  dt->add(te::da::GetFirstSpatialProperty(m_targetDst)->clone());

  return dt;
}

te::vp::Merge::MergeStrategy te::vp::Merge::checkStrategy()
{
  te::da::PrimaryKey* targetPk = m_targetDst->getPrimaryKey();

  if ((m_outDsrc && m_outDsrc->getType() == "OGR") || m_targetSource->getType() == "OGR")
    return PUREMERGE;

  if (targetPk)
  {
    std::vector<te::dt::Property*> targetProps = targetPk->getProperties();

    if (targetProps.size() > 1)
    {
      return PUREMERGE;
    }
    else
    {
      te::dt::Property* prop = targetProps[0];

      if ((prop->getType() != te::dt::INT16_TYPE) && (prop->getType() != te::dt::INT32_TYPE) && (prop->getType() != te::dt::INT64_TYPE))
      {
        return PUREMERGE;
      }
      else
      {
        te::dt::SimpleProperty* simpleProp = dynamic_cast<te::dt::SimpleProperty*>(prop);

        if (simpleProp)
        {
          if (simpleProp->isAutoNumber())
          {
            return AUTOINCREMENT;
          }
          else
          {
            return TRYGETMAX;
          }
        }
        else
        {
          throw te::common::Exception(TE_TR("Merge error: Property type invalid."));
        }

      }
    }
  }
  else
    return PUREMERGE;

}

int te::vp::Merge::tryGetMax()
{
  te::dt::Property* pkProp = m_targetDst->getPrimaryKey()->getProperties()[0];

  te::da::FromItem* pFromItem = new te::da::DataSetName(m_targetDst->getName());

  te::da::From pFrom;
  pFrom.push_back(pFromItem);

  te::da::PropertyName pName(pkProp->getName());

  te::da::Max fMax(pName);

  te::da::Field* pField = new te::da::Field(fMax);

  te::da::Fields pFields;
  pFields.push_back(pField);

  te::da::Select pSelect(pFields, pFrom);

  std::unique_ptr<te::da::DataSet> result = m_targetSource->query(pSelect);

  result->moveFirst();

  std::string maxStr = result->getAsString(0);

  int maxInt = boost::lexical_cast<int>(maxStr);

  return maxInt;
}

bool te::vp::Merge::isPrimaryKeyProperty(const te::da::DataSetType* dst, const std::string& p)
{
  te::da::PrimaryKey* pk = dst->getPrimaryKey();
  if (!pk)
    return false;

  std::vector<te::dt::Property*> props = pk->getProperties();

  for (std::size_t i = 0; i < props.size(); ++i)
  {
    if (props[i]->getName() == p)
      return true;
  }

  return false;
}

void te::vp::Merge::updateInputDst(te::da::DataSourceTransactor* transactor)
{
  te::da::PrimaryKey* pk = m_targetDst->getPrimaryKey();
  std::vector<te::dt::Property*> pkProps = pk->getProperties();
  std::vector<std::string> pkPropsNames;

  for (auto i : pkProps)
  {
    pkPropsNames.push_back(i->getName());
  }

  for (std::size_t i = 0; i < m_properties.size(); ++i)
  {
    if (std::find(pkPropsNames.begin(), pkPropsNames.end(), m_properties[i].first) != pkPropsNames.end())
      continue;

    std::string fp = m_properties[i].first;
    std::string sp = m_properties[i].second;

    if (fp.empty() && sp.empty())
    {
      continue;
    }

    te::dt::Property* fProp = m_targetDst->getProperty(fp);
    te::dt::Property* sProp = m_originDst->getProperty(sp);

    te::dt::Property* newProp = nullptr;

    if ((!fp.empty()) && (!sp.empty()))
    {
      te::dt::SimpleProperty* fSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(fProp);
      te::dt::SimpleProperty* sSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(sProp);

      if (fSimpleProp->isRequired() != sSimpleProp->isRequired())
      {
        te::dt::Property* auxProp = fSimpleProp->clone();
        te::dt::SimpleProperty* auxSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(auxProp);
        auxSimpleProp->setRequired(false);

        transactor->changePropertyDefinition(m_targetDst->getName(), fProp->getName(), auxSimpleProp);
      }
    }
    else if (!fp.empty())
    {
      te::dt::SimpleProperty* fSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(fProp);

      if (fSimpleProp->isRequired())
      {
        te::dt::Property* auxProp = fSimpleProp->clone();
        te::dt::SimpleProperty* auxSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(auxProp);
        auxSimpleProp->setRequired(false);

        transactor->changePropertyDefinition(m_targetDst->getName(), fProp->getName(), auxSimpleProp);
      }
    }
    else if (!sp.empty())
    {
      newProp = sProp->clone();

      te::dt::SimpleProperty* newSimpleProp = dynamic_cast<te::dt::SimpleProperty*>(newProp);

      newSimpleProp->setRequired(false);
      newSimpleProp->setAutoNumber(false);

      transactor->addProperty(m_targetDst->getName(), newProp);
    }
    else
    {
      throw te::common::Exception(TE_TR("An unexpected error occurred during the output dataset update!"));
    }
  }
}

int te::vp::Merge::getPropertyType(const te::dt::Property* p)
{
  if (p->getType() == te::dt::INT16_TYPE || p->getType() == te::dt::UINT16_TYPE)
    return te::dt::INT16_TYPE;
  else if (p->getType() == te::dt::INT32_TYPE || p->getType() == te::dt::UINT32_TYPE)
    return te::dt::INT32_TYPE;
  else if (p->getType() == te::dt::INT64_TYPE || p->getType() == te::dt::UINT64_TYPE)
    return te::dt::INT64_TYPE;
  else
    throw; //Throw!!!

}

bool te::vp::Merge::executeQuery(te::vp::AlgorithmParams* /*mainParams*/)
{
  return true;
}