From: Anthony Geay <anthony.geay@edf.fr>
Date: Tue, 14 Apr 2020 21:08:27 +0000 (+0200)
Subject: [EDF21149] : Improvements for // spliter
X-Git-Tag: V9_5_0b1~12
X-Git-Url: http://git.salome-platform.org/gitweb/?a=commitdiff_plain;h=1c9761aeac9192f62f324ad7a04430f2482bc843;p=tools%2Fmedcoupling.git

[EDF21149] : Improvements for // spliter
---

diff --git a/src/INTERP_KERNEL/MCIdType.hxx b/src/INTERP_KERNEL/MCIdType.hxx
index 9c37ffbb1..f4befdd77 100644
--- a/src/INTERP_KERNEL/MCIdType.hxx
+++ b/src/INTERP_KERNEL/MCIdType.hxx
@@ -18,8 +18,7 @@
 //
 // Author : Anthony Geay (EDF R&D)
 
-#ifndef __IK_MCIDTYPE_HXX__
-#define __IK_MCIDTYPE_HXX__
+#pragma once
 
 #include <cstdint>
 #include <stddef.h>
@@ -35,6 +34,10 @@ typedef std::int64_t mcIdType;
 
 #endif
 
+template <class T> inline std::size_t ToSizeT(T val)
+{
+  return static_cast<std::size_t>(val);
+}
 template <class T> inline mcIdType ToIdType(T val)
 {
   return static_cast<mcIdType>(val);
@@ -43,6 +46,3 @@ template <class T> inline T FromIdType(mcIdType val)
 {
   return static_cast<T>(val);
 }
-
-
-#endif
diff --git a/src/MEDCoupling/MCAuto.hxx b/src/MEDCoupling/MCAuto.hxx
index 76e00de38..1d83037b9 100644
--- a/src/MEDCoupling/MCAuto.hxx
+++ b/src/MEDCoupling/MCAuto.hxx
@@ -24,6 +24,7 @@
 #include "InterpKernelException.hxx"
 
 #include <vector>
+#include <algorithm>
 
 namespace MEDCoupling
 {
@@ -31,6 +32,7 @@ namespace MEDCoupling
   class MCAuto
   {
   public:
+    static MCAuto TakeRef(T *ptr) { MCAuto ret(MCAuto(nullptr)); ret.takeRef(ptr); return ret; }
     MCAuto(const MCAuto& other):_ptr(0) { referPtr(other._ptr); }
     MCAuto(T *ptr=0):_ptr(ptr) { }
     ~MCAuto() { destroyPtr(); }
@@ -59,6 +61,16 @@ namespace MEDCoupling
     T *_ptr;
   };
 
+  template<class T>
+  std::vector<const T*> FromVecAutoToVecOfConst(const std::vector<MCAuto<T>>& inputVec)
+  {
+    std::size_t size(inputVec.size());
+    std::vector< const T *> ret(size);
+    typename std::vector< const T *>::iterator itArrays(ret.begin());
+    std::for_each(inputVec.begin(),inputVec.end(),[&itArrays](MCAuto<T> elt) { *itArrays++=elt; });
+    return ret;
+  }
+
   template<class T, class U>
   typename MEDCoupling::MCAuto<U> DynamicCast(typename MEDCoupling::MCAuto<T>& autoSubPtr) throw()
   {
diff --git a/src/MEDCoupling/MEDCouplingMemArray.hxx b/src/MEDCoupling/MEDCouplingMemArray.hxx
index 35f751249..f9a847cc4 100755
--- a/src/MEDCoupling/MEDCouplingMemArray.hxx
+++ b/src/MEDCoupling/MEDCouplingMemArray.hxx
@@ -18,8 +18,7 @@
 //
 // Author : Anthony Geay (EDF R&D)
 
-#ifndef __MEDCOUPLING_MEDCOUPLINGMEMARRAY_HXX__
-#define __MEDCOUPLING_MEDCOUPLINGMEMARRAY_HXX__
+#pragma once
 
 #include "MEDCoupling.hxx"
 #include "MCType.hxx"
@@ -542,9 +541,10 @@ namespace MEDCoupling
   class DataArrayDiscrete : public DataArrayTemplateClassic<T>
   {
   public:
-    typedef typename Traits<T>::ArrayType DataArrayType;
+    using DataArrayType = typename Traits<T>::ArrayType;
   public:
     static DataArrayType *New();
+    static MCAuto<DataArrayType> NewFromArray(const T *arrBegin, const T *arrEnd);
     T intValue() const;
     bool isEqual(const DataArrayDiscrete<T>& other) const;
     bool isEqualIfNotWhy(const DataArrayDiscrete<T>& other, std::string& reason) const;
@@ -552,6 +552,7 @@ namespace MEDCoupling
     bool isEqualWithoutConsideringStrAndOrder(const typename Traits<T>::ArrayType& other) const;
     void switchOnTupleEqualTo(T val, std::vector<bool>& vec) const;
     void switchOnTupleNotEqualTo(T val, std::vector<bool>& vec) const;
+    DataArrayIdType *occurenceRankInThis() const;
     DataArrayIdType *buildPermutationArr(const DataArrayDiscrete<T>& other) const;
     DataArrayIdType *indicesOfSubPart(const DataArrayDiscrete<T>& partOfThis) const;
     void checkMonotonic(bool increasing) const;
@@ -568,7 +569,7 @@ namespace MEDCoupling
     DataArrayIdType *findIdsEqual(T val) const;
     DataArrayIdType *transformWithIndArrR(const T *indArr2Bg, const T *indArrEnd) const;
     void splitByValueRange(const T *arrBg, const T *arrEnd,
-                                              DataArrayType *& castArr, DataArrayType *& rankInsideCast, DataArrayType *& castsPresent) const;
+                           DataArrayType *& castArr, DataArrayType *& rankInsideCast, DataArrayType *& castsPresent) const;
     bool isRange(T& strt, T& sttoopp, T& stteepp) const;
     DataArrayIdType *invertArrayO2N2N2O(mcIdType newNbOfElem) const;
     DataArrayIdType *invertArrayN2O2O2N(mcIdType oldNbOfElem) const;
@@ -621,6 +622,7 @@ namespace MEDCoupling
     DataArrayType *buildSubstractionOptimized(const DataArrayType *other) const;
     DataArrayType *buildUnion(const DataArrayType *other) const;
     DataArrayType *buildIntersection(const DataArrayType *other) const;
+    DataArrayIdType *indexOfSameConsecutiveValueGroups() const;
     DataArrayType *buildUnique() const;
     DataArrayType *buildUniqueNotSorted() const;
     DataArrayType *deltaShiftIndex() const;
@@ -644,6 +646,7 @@ namespace MEDCoupling
     //const MemArray<T>& accessToMemArray() const { return _mem; }
   public:
     static DataArrayIdType *FindPermutationFromFirstToSecond(const DataArrayType *ids1, const DataArrayType *ids2);
+    static DataArrayIdType *FindPermutationFromFirstToSecondDuplicate(const DataArrayType *ids1, const DataArrayType *ids2);
     static mcIdType *CheckAndPreparePermutation(const T *start, const T *end);
     static DataArrayType *BuildListOfSwitchedOn(const std::vector<bool>& v);
     static DataArrayType *BuildListOfSwitchedOff(const std::vector<bool>& v);
@@ -1058,5 +1061,3 @@ namespace MEDCoupling
       throw INTERP_KERNEL::Exception("DataArrayDouble::insertAtTheEnd : not available for DataArrayDouble with number of components different than 1 !");
   }
 }
-
-#endif
diff --git a/src/MEDCoupling/MEDCouplingMemArray.txx b/src/MEDCoupling/MEDCouplingMemArray.txx
index 69da58300..373e9201b 100755
--- a/src/MEDCoupling/MEDCouplingMemArray.txx
+++ b/src/MEDCoupling/MEDCouplingMemArray.txx
@@ -3634,6 +3634,19 @@ struct NotInRange
     return new typename Traits<T>::ArrayType;
   }
   
+  /*!
+   * Returns a newly created array containing a copy of the input array defined by [ \a arrBegin, \a arrEnd )
+   */
+  template<class T>
+  MCAuto< typename Traits<T>::ArrayType > DataArrayDiscrete<T>::NewFromArray(const T *arrBegin, const T *arrEnd)
+  {
+    MCAuto< typename Traits<T>::ArrayType > ret(DataArrayDiscrete<T>::New());
+    std::size_t nbElts(std::distance(arrBegin,arrEnd));
+    ret->alloc(nbElts,1);
+    std::copy(arrBegin,arrEnd,ret->getPointer());
+    return ret;
+  }
+  
   /*!
    * Checks if values of \a this and another DataArrayInt are equal. For more info see
    * \ref MEDCouplingArrayBasicsCompare.
@@ -3706,6 +3719,46 @@ struct NotInRange
     switchOnTupleAlg(val,vec,std::not_equal_to<T>());
   }
 
+  /*!
+   * Compute for each element in \a this the occurence rank of that element. This method is typically useful of one-component array having a same element
+   * appearing several times. If each element in \a this appears once an 1 component array containing only 0 will be returned.
+   *
+   * \b Example:
+   * - \a this : [5, 3, 2, 1, 4, 5, 2, 1, 0, 11, 5, 4]
+   * - \a return is : [0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 2, 1] because at pos #0 of \a this (ie value 5) is the first occurrence ->0. At pos #10 of \a this (ie value 5 also) is the third occurrence of 5 -> 2.
+   *
+   * \return DataArrayInt * - a new instance of DataArrayInt with same number of tuples than \a this. The caller is to delete this
+   *          array using decrRef() as it is no more needed.
+   * \throw If either this not allocated or not with one component.
+   *
+   * \sa DataArrayInt::FindPermutationFromFirstToSecond
+   */
+  template<class T>
+  DataArrayIdType *DataArrayDiscrete<T>::occurenceRankInThis() const
+  {
+    constexpr char MSG0[] = "occurenceRankInThis :";
+    this->checkAllocated();
+    this->checkNbOfComps(1,MSG0);
+    MCAuto<DataArrayIdType> ret(DataArrayIdType::New());
+    ret->alloc(this->getNumberOfTuples(),1);
+    mcIdType *retPtr(ret->getPointer());
+    std::map<T,mcIdType> m;
+    for(const T *pt = this->begin() ; pt != this->end() ; ++pt, ++retPtr )
+    {
+      auto it = m.find(*pt);
+      if( it == m.end() )
+      {
+        *retPtr = 0;
+        m[*pt] = 1;
+      }
+      else
+      {
+        *retPtr = (*it).second++;
+      }
+    }
+    return ret.retn();
+  }
+
   /*!
    * Creates a new one-dimensional DataArrayInt of the same size as \a this and a given
    * one-dimensional arrays that must be of the same length. The result array describes
@@ -4529,7 +4582,7 @@ struct NotInRange
    *
    * \return - An array of size std::distance(valsBg,valsEnd)
    *
-   * \sa DataArrayInt::FindPermutationFromFirstToSecond
+   * \sa DataArrayInt::FindPermutationFromFirstToSecond , DataArrayInt::FindPermutationFromFirstToSecondDuplicate
    */
   template <class T>
   MCAuto<DataArrayIdType> DataArrayDiscrete<T>::findIdForEach(const T *valsBg, const T *valsEnd) const
@@ -5652,6 +5705,53 @@ struct NotInRange
     arrs[0]=dynamic_cast<const DataArrayType *>(this); arrs[1]=other;
     return DataArrayDiscrete<T>::BuildIntersection(arrs);
   }
+  /*!
+   * This method can be applied on allocated with one component DataArrayInt instance.
+   * Locate groups of all consecutive same values in \a this and return them into an indexed array of positions pointing to \a this starting with 0.
+   * Number of tuples of returned array is equal to size of \a this->buildUnique() + 1.
+   * Last value of returned array is equal to \a this->getNumberOfTuples()
+   * 
+   * \b Example:
+   * - \a this : [0, 1, 1, 2, 2, 3, 4, 4, 5, 5, 5, 11]
+   * - \a return is : [0, 1, 3, 5, 6, 8, 11, 12]
+   *
+   * \return a newly allocated array containing the indexed array of
+   * \throw if \a this is not allocated or if \a this has not exactly one component or if number of tuples is equal to 0.
+   * \sa DataArrayInt::buildUnique, MEDCouplingSkyLineArray::groupPacks
+   */
+  template <class T>
+  DataArrayIdType *DataArrayDiscrete<T>::indexOfSameConsecutiveValueGroups() const
+  {
+    this->checkAllocated();
+    if(this->getNumberOfComponents()!=1)
+      throw INTERP_KERNEL::Exception("DataArrayInt::indexOfSameConsecutiveValueGroups : only single component allowed !");
+    if(this->getNumberOfTuples()==0)
+      throw INTERP_KERNEL::Exception("DataArrayInt::indexOfSameConsecutiveValueGroups : number of tuples must be > 0 !");
+    const T *pt(this->begin());
+    const T *const ptEnd(this->end()) , * const ptBg(this->begin());
+    const T *oldPt(pt);
+    // first find nb of different values in this
+    std::size_t nbOfTuplesOut(0);
+    while( pt != ptEnd )
+    {
+      T val(*pt);
+      const T *endOfPack(std::find_if(pt+1,ptEnd,[val](T elt){ return val!=elt; }));
+      pt = endOfPack;
+      ++nbOfTuplesOut;
+    }
+    MCAuto<DataArrayIdType> ret(DataArrayIdType::New()); ret->alloc(nbOfTuplesOut+1,1);
+    mcIdType *retPtr(ret->getPointer()); *retPtr++ = 0;
+    pt = this->begin();
+    while( pt != ptEnd )
+    {
+      T val(*pt);
+      const T *endOfPack(std::find_if(pt+1,ptEnd,[val](T elt){ return val!=elt; }));
+      *retPtr++ = ToIdType( std::distance(ptBg,endOfPack) );
+      pt = endOfPack;
+      ++nbOfTuplesOut;
+    }
+    return ret.retn();
+  }
 
   /*!
    * This method can be applied on allocated with one component DataArrayInt instance.
@@ -5660,7 +5760,7 @@ struct NotInRange
    *
    * \return a newly allocated array that contain the result of the unique operation applied on \a this.
    * \throw if \a this is not allocated or if \a this has not exactly one component.
-   * \sa DataArrayInt::buildUniqueNotSorted
+   * \sa DataArrayInt::buildUniqueNotSorted, DataArrayInt::indexOfSameConsecutiveValueGroups
    */
   template <class T>
   typename Traits<T>::ArrayType *DataArrayDiscrete<T>::buildUnique() const
@@ -5668,12 +5768,12 @@ struct NotInRange
     this->checkAllocated();
     if(this->getNumberOfComponents()!=1)
       throw INTERP_KERNEL::Exception("DataArrayInt::buildUnique : only single component allowed !");
-    std::size_t nbOfElements=this->getNumberOfTuples();
-    MCAuto<DataArrayType> tmp=DataArrayType::New();
-    tmp->deepCopyFrom (*this);
-    T *data=tmp->getPointer();
-    T *last=std::unique(data,data+nbOfElements);
-    MCAuto<DataArrayType> ret=DataArrayType::New();
+    std::size_t nbOfElements(this->getNumberOfTuples());
+    MCAuto<DataArrayType> tmp(DataArrayType::New());
+    tmp->deepCopyFrom(*this);
+    T *data(tmp->getPointer());
+    T *last(std::unique(data,data+nbOfElements));
+    MCAuto<DataArrayType> ret(DataArrayType::New());
     ret->alloc(std::distance(data,last),1);
     std::copy(data,last,ret->getPointer());
     return ret.retn();
@@ -6636,7 +6736,7 @@ struct NotInRange
    *          array using decrRef() as it is no more needed.
    * \throw If either ids1 or ids2 is null not allocated or not with one components.
    *
-   * \sa DataArrayInt::findIdForEach
+   * \sa DataArrayInt::findIdForEach, DataArrayInt::FindPermutationFromFirstToSecondDuplicate, DataArrayInt::rankOfElementInThis
    */
   template<class T>
   DataArrayIdType *DataArrayDiscrete<T>::FindPermutationFromFirstToSecond(const DataArrayType *ids1, const DataArrayType *ids2)
@@ -6664,6 +6764,84 @@ struct NotInRange
     return p2.retn();
   }
 
+  /*!
+   * This method tries to find the permutation to apply to the first input \a ids1 to obtain the same array (without considering strings information) the second
+   * input array \a ids2.
+   * \a ids1 and \a ids2 are expected to be both a list of ids (both with number of components equal to one) not sorted and with values that can be negative.
+   * This method will throw an exception is no such permutation array can be obtained. It is typically the case if there is some ids in \a ids1 not in \a ids2 or
+   * inversely.
+   * The difference with DataArrayInt::FindPermutationFromFirstToSecond is that this method supports multiple same values in \a ids1 and \a ids2 whereas 
+   * DataArrayInt::FindPermutationFromFirstToSecond doesn't. It implies that this method my be slower than the DataArrayInt::FindPermutationFromFirstToSecond one.
+   * 
+   * In case of success both assertion will be true (no throw) :
+   * \c ids1->renumber(ret)->isEqual(ids2) where \a ret is the return of this method.
+   * \c ret->transformWithIndArr(ids2)->isEqual(ids1)
+   *
+   * \b Example:
+   * - \a ids1 : [5, 3, 2, 1, 4, 5, 2, 1, 0, 11, 5, 4]
+   * - \a ids2 : [0, 1, 1, 2, 2, 3, 4, 4, 5, 5, 5, 11]
+   * - \a return is : [8, 5, 3, 1, 6, 9, 4, 2, 0, 11, 10, 7] because ids2[8]==ids1[0], ids2[5]==ids1[1], ids2[3]==ids1[2], ids2[1]==ids1[3]...
+   *
+   * \return DataArrayInt * - a new instance of DataArrayInt. The caller is to delete this
+   *          array using decrRef() as it is no more needed.
+   * \throw If either ids1 or ids2 is null not allocated or not with one components.
+   *
+   * \sa DataArrayInt::findIdForEach, DataArrayInt::FindPermutationFromFirstToSecond, DataArrayInt::occurenceRankInThis
+   */
+  template<class T>
+  DataArrayIdType *DataArrayDiscrete<T>::FindPermutationFromFirstToSecondDuplicate(const DataArrayType *ids1, const DataArrayType *ids2)
+  {
+    if(!ids1 || !ids2)
+      throw INTERP_KERNEL::Exception("DataArrayInt::FindPermutationFromFirstToSecondDuplicate : the two input arrays must be not null !");
+    constexpr char MSG0[] = "DataArrayInt::FindPermutationFromFirstToSecondDuplicate :";
+    ids1->checkAllocated(); ids2->checkAllocated();
+    ids1->checkNbOfComps(1,MSG0); ids2->checkNbOfComps(1,MSG0);
+    mcIdType nbTuples(ids1->getNumberOfTuples());
+    if(nbTuples != ids2->getNumberOfTuples())
+      {
+        std::ostringstream oss; oss << "DataArrayInt::FindPermutationFromFirstToSecondDuplicate : first array has " << ids1->getNumberOfTuples() << " tuples and the second one " << ids2->getNumberOfTuples() << " tuples ! No chance to find a permutation between the 2 arrays !";
+        throw INTERP_KERNEL::Exception(oss.str().c_str());
+      }
+    MCAuto<DataArrayIdType> ret(DataArrayIdType::New()); ret->alloc(nbTuples,1);
+    MCAuto<DataArrayIdType> oids2(ids2->occurenceRankInThis());
+    std::map< std::pair<T,mcIdType>, mcIdType> m;
+    mcIdType pos(0);
+    const mcIdType *oids2Ptr(oids2->begin());
+    for(const T * it2 = ids2->begin() ; it2 != ids2->end() ; ++it2, ++oids2Ptr, ++pos)
+      m[{*it2,*oids2Ptr}] = pos;
+    mcIdType *retPtr(ret->getPointer());
+    //
+    std::map<T,mcIdType> mOccurence1; // see DataArrayInt::occurenceRankInThis : avoid to compute additionnal temporary array
+    //
+    for(const T * it1 = ids1->begin() ; it1 != ids1->end() ; ++it1, ++retPtr)
+    {
+      auto it = mOccurence1.find(*it1);
+      mcIdType occRk1;
+      if( it == mOccurence1.end() )
+      {
+        occRk1 = 0;
+        mOccurence1[*it1] = 1;
+      }
+      else
+      {
+        occRk1 = (*it).second++;
+      }
+      //
+      auto it2 = m.find({*it1,occRk1});
+      if(it2 != m.end())
+      {
+        *retPtr = (*it2).second;
+      }
+      else
+      {
+        std::ostringstream oss; oss << MSG0 << "At pos " << std::distance(ids1->begin(),it1) << " value is " << *it1 << " and occurence rank is " << occRk1 << ". No such item into second array !"; 
+        throw INTERP_KERNEL::Exception(oss.str());
+      }
+      
+    }
+    return ret.retn();
+  }
+
   /*!
    * Returns a C array which is a renumbering map in "Old to New" mode for the input array.
    * This map, if applied to \a start array, would make it sorted. For example, if
diff --git a/src/MEDCoupling/MEDCouplingSkyLineArray.cxx b/src/MEDCoupling/MEDCouplingSkyLineArray.cxx
index b6297553c..5fdc66fa8 100755
--- a/src/MEDCoupling/MEDCouplingSkyLineArray.cxx
+++ b/src/MEDCoupling/MEDCouplingSkyLineArray.cxx
@@ -312,6 +312,42 @@ std::string MEDCouplingSkyLineArray::simpleRepr() const
   return oss.str();
 }
 
+MEDCouplingSkyLineArray *MEDCouplingSkyLineArray::groupPacks(const DataArrayIdType *indexedPacks) const
+{
+  indexedPacks->checkAllocated();
+  if( indexedPacks->getNumberOfComponents() != 1 )
+    throw INTERP_KERNEL::Exception("MEDCouplingSkyLineArray::groupPacks : number of components must be 1 !");
+  std::size_t nbTuples(indexedPacks->getNumberOfTuples());
+  if( nbTuples == 0 )
+    throw INTERP_KERNEL::Exception("MEDCouplingSkyLineArray::groupPacks : number of tuples must be > 0 !");
+  const DataArrayIdType *index(this->getIndexArray());
+  MCAuto<DataArrayIdType> partIndex(index->selectByTupleIdSafe(indexedPacks->begin(),indexedPacks->end()));
+  MCAuto<MEDCouplingSkyLineArray> ret(MEDCouplingSkyLineArray::New(partIndex,this->getValuesArray()));
+  return ret.retn();
+}
+
+MEDCouplingSkyLineArray *MEDCouplingSkyLineArray::uniqueNotSortedByPack() const
+{
+  mcIdType nbPacks(this->getNumberOf());
+  MCAuto<DataArrayIdType> retIndex(DataArrayIdType::New()); retIndex->alloc(nbPacks+1,1);
+  const mcIdType *valuesPtr(this->_values->begin()),*indexPtr(this->_index->begin());
+  mcIdType *retIndexPtr(retIndex->getPointer()); *retIndexPtr = 0;
+  for(mcIdType i = 0 ; i < nbPacks ; ++i, ++retIndexPtr)
+  {
+    std::set<mcIdType> s(valuesPtr+indexPtr[i],valuesPtr+indexPtr[i+1]);
+    retIndexPtr[1] = retIndexPtr[0] + ToIdType(s.size());
+  }
+  MCAuto<DataArrayIdType> retValues(DataArrayIdType::New()); retValues->alloc(retIndex->back(),1);
+  mcIdType *retValuesPtr(retValues->getPointer());
+  for(mcIdType i = 0 ; i < nbPacks ; ++i)
+  {
+    std::set<mcIdType> s(valuesPtr+indexPtr[i],valuesPtr+indexPtr[i+1]);
+    retValuesPtr = std::copy(s.begin(),s.end(),retValuesPtr);
+  }
+  MCAuto<MEDCouplingSkyLineArray> ret(MEDCouplingSkyLineArray::New(retIndex,retValues));
+  return ret.retn();
+}
+
 /**
  * For a 2- or 3-level SkyLine array, return a copy of the absolute pack with given identifier.
  */
diff --git a/src/MEDCoupling/MEDCouplingSkyLineArray.hxx b/src/MEDCoupling/MEDCouplingSkyLineArray.hxx
index 44dd20066..a56d94b9a 100644
--- a/src/MEDCoupling/MEDCouplingSkyLineArray.hxx
+++ b/src/MEDCoupling/MEDCouplingSkyLineArray.hxx
@@ -26,6 +26,7 @@
 #include "NormalizedGeometricTypes"
 
 #include <vector>
+#include <functional>
 
 namespace MEDCoupling
 {
@@ -63,6 +64,28 @@ namespace MEDCoupling
     static MEDCouplingSkyLineArray * New( const MEDCouplingSkyLineArray & other );
 
     static MEDCouplingSkyLineArray * BuildFromPolyhedronConn( const DataArrayIdType* c, const DataArrayIdType* cI );
+
+    static std::vector< MCAuto<DataArrayIdType> > RetrieveVecIndex(const std::vector< MCAuto<MEDCouplingSkyLineArray> >& vecSka)
+    {
+       auto fct = [](MEDCouplingSkyLineArray *ska) { return ska->getIndexArray(); };
+       return RetrieveVecOfSkyLineArrayGen(vecSka,fct);
+    }
+    
+    static std::vector< MCAuto<DataArrayIdType> > RetrieveVecValues(const std::vector< MCAuto<MEDCouplingSkyLineArray> >& vecSka)
+    {
+       auto fct = [](MEDCouplingSkyLineArray *ska) { return ska->getValuesArray(); };
+       return RetrieveVecOfSkyLineArrayGen(vecSka,fct);
+    }
+    
+    static std::vector< MCAuto<DataArrayIdType> > RetrieveVecOfSkyLineArrayGen(const std::vector< MCAuto<MEDCouplingSkyLineArray> >& vecSka, std::function<DataArrayIdType *(MEDCouplingSkyLineArray *)> fct)
+    {
+       std::size_t sz(vecSka.size());
+       std::vector< MCAuto<DataArrayIdType> > ret(sz);
+       std::vector< MCAuto<DataArrayIdType> >::iterator it(ret.begin());
+       std::for_each(vecSka.begin(),vecSka.end(),[&it,fct](MCAuto<MEDCouplingSkyLineArray> elt) { *it++ = MCAuto<DataArrayIdType>::TakeRef(fct(elt)); } );
+       return ret;
+    }
+
     std::string getClassName() const override { return std::string("MEDCouplingSkyLineArray"); }
     std::size_t getHeapMemorySizeWithoutChildren() const;
     std::vector<const BigMemoryObject *> getDirectChildrenWithNull() const;
@@ -84,6 +107,9 @@ namespace MEDCoupling
 
     std::string simpleRepr() const;
 
+    MEDCouplingSkyLineArray *groupPacks(const DataArrayIdType *indexedPacks) const;
+    MEDCouplingSkyLineArray *uniqueNotSortedByPack() const;
+
     void getSimplePackSafe(const mcIdType absolutePackId, std::vector<mcIdType> & pack) const;
     const mcIdType * getSimplePackSafePtr(const mcIdType absolutePackId, mcIdType & packSize) const;
     void findPackIds(const std::vector<mcIdType> & superPackIndices, const mcIdType *packBg, const mcIdType *packEnd,
diff --git a/src/MEDCoupling/MEDCouplingTraits.hxx b/src/MEDCoupling/MEDCouplingTraits.hxx
index 2d3f91ff4..fad3d7cf7 100644
--- a/src/MEDCoupling/MEDCouplingTraits.hxx
+++ b/src/MEDCoupling/MEDCouplingTraits.hxx
@@ -58,10 +58,10 @@ namespace MEDCoupling
     static const char FieldTypeName[];
     static const char NPYStr[];
     static const char ReprStr[];
-    typedef DataArrayDouble ArrayType;
-    typedef DataArrayDouble ArrayTypeCh;
-    typedef MEDCouplingFieldDouble FieldType;
-    typedef DataArrayDoubleTuple ArrayTuple;
+    using ArrayType = DataArrayDouble;
+    using ArrayTypeCh = DataArrayDouble;
+    using FieldType = MEDCouplingFieldDouble;
+    using ArrayTuple = DataArrayDoubleTuple;
   };
 
   template<>
@@ -71,10 +71,10 @@ namespace MEDCoupling
     static const char FieldTypeName[];
     static const char NPYStr[];
     static const char ReprStr[];
-    typedef DataArrayFloat ArrayType;
-    typedef DataArrayFloat ArrayTypeCh;
-    typedef MEDCouplingFieldFloat FieldType;
-    typedef DataArrayFloatTuple ArrayTuple;
+    using ArrayType = DataArrayFloat;
+    using ArrayTypeCh = DataArrayFloat;
+    using FieldType = MEDCouplingFieldFloat;
+    using ArrayTuple = DataArrayFloatTuple;
   };
   
   template<>
@@ -85,11 +85,11 @@ namespace MEDCoupling
     static const char NPYStr[];
     static const char ReprStr[];
     static const char VTKReprStr[];
-    typedef DataArrayInt32 ArrayType;
-    typedef DataArrayInt32 ArrayTypeCh;
-    typedef MEDCouplingFieldInt FieldType;
-    typedef DataArrayInt32Tuple ArrayTuple;
-    typedef DataArrayInt32Iterator IteratorType;
+    using ArrayType = DataArrayInt32;
+    using ArrayTypeCh = DataArrayInt32;
+    using FieldType = MEDCouplingFieldInt;
+    using ArrayTuple = DataArrayInt32Tuple;
+    using IteratorType = DataArrayInt32Iterator;
   };
 
   template<>
@@ -100,21 +100,21 @@ namespace MEDCoupling
     static const char NPYStr[];
     static const char ReprStr[];
     static const char VTKReprStr[];
-    typedef DataArrayInt64 ArrayType;
-    typedef DataArrayInt64 ArrayTypeCh;
-    //typedef MEDCouplingFieldInt64 FieldType;
-    typedef DataArrayInt64Tuple ArrayTuple;
-    typedef DataArrayInt64Iterator IteratorType;
+    using ArrayType = DataArrayInt64;
+    using ArrayTypeCh = DataArrayInt64;
+    //using FieldType = MEDCouplingFieldInt64;
+    using ArrayTuple = DataArrayInt64Tuple;
+    using IteratorType = DataArrayInt64Iterator;
   };
 
   template<>
   struct MEDCOUPLING_EXPORT Traits<char>
   {
     static const char ArrayTypeName[];
-    typedef DataArrayByte ArrayTypeCh;
-    typedef DataArrayChar ArrayType;
-    typedef DataArrayByteTuple ArrayTuple;
-    typedef DataArrayByteIterator IteratorType;
+    using ArrayTypeCh = DataArrayByte;
+    using ArrayType = DataArrayChar;
+    using ArrayTuple = DataArrayByteTuple;
+    using IteratorType = DataArrayByteIterator;
   };
 }
 
diff --git a/src/MEDCoupling_Swig/DataArrayInt.i b/src/MEDCoupling_Swig/DataArrayInt.i
index 32ff65d59..da27feb99 100644
--- a/src/MEDCoupling_Swig/DataArrayInt.i
+++ b/src/MEDCoupling_Swig/DataArrayInt.i
@@ -44,6 +44,7 @@
     bool isEqual(const ARRAY& other) const;
     bool isEqualWithoutConsideringStr(const ARRAY& other) const;
     bool isEqualWithoutConsideringStrAndOrder(const ARRAY& other) const;
+    DataArrayIdType *occurenceRankInThis() const;
     DataArrayIdType *buildPermutationArr(const ARRAY& other) const;
     ARRAY *sumPerTuple() const;
     void sort(bool asc=true);
@@ -126,11 +127,13 @@
     static ARRAY *BuildUnion(const std::vector<const ARRAY *>& arr);
     static ARRAY *BuildIntersection(const std::vector<const ARRAY *>& arr);
     static DataArrayIdType *FindPermutationFromFirstToSecond(const ARRAY *ids1, const ARRAY *ids2);
+    static DataArrayIdType *FindPermutationFromFirstToSecondDuplicate(const ARRAY *ids1, const ARRAY *ids2);
     DataArrayIdType *buildComplement(mcIdType nbOfElement) const;
     ARRAY *buildSubstraction(const ARRAY *other) const;
     ARRAY *buildSubstractionOptimized(const ARRAY *other) const;
     ARRAY *buildUnion(const ARRAY *other) const;
     ARRAY *buildIntersection(const ARRAY *other) const;
+    DataArrayIdType *indexOfSameConsecutiveValueGroups() const;
     ARRAY *buildUnique() const;
     ARRAY *buildUniqueNotSorted() const;
     ARRAY *deltaShiftIndex() const;
diff --git a/src/MEDCoupling_Swig/MEDCouplingBasicsTest7.py b/src/MEDCoupling_Swig/MEDCouplingBasicsTest7.py
index ab6252947..ed473e56b 100644
--- a/src/MEDCoupling_Swig/MEDCouplingBasicsTest7.py
+++ b/src/MEDCoupling_Swig/MEDCouplingBasicsTest7.py
@@ -755,6 +755,37 @@ class MEDCouplingBasicsTest7(unittest.TestCase):
         self.assertTrue(f_voro.getArray().isEqual(f.getArray(),1e-8))
         pass
 
+    def testDAIOccurenceRankInThis(self):
+        arr=DataArrayInt([5,3,2,1,4,5,2,1,0,11,5,4])
+        self.assertTrue(arr.occurenceRankInThis().isEqual(DataArrayInt([0,0,0,0,0,1,1,1,0,0,2,1])))
+
+    def testDAIFindPermutationFromFirstToSecondDuplicate(self):
+        arr0 = DataArrayInt([5,3,2,1,4,5,2,1,0,11,5,4])
+        arr1 = DataArrayInt([0,1,1,2,2,3,4,4,5,5,5,11])
+        self.assertTrue(DataArrayInt.FindPermutationFromFirstToSecondDuplicate(arr0,arr1).isEqual(DataArrayInt([8,5,3,1,6,9,4,2,0,11,10,7])))
+        self.assertTrue(DataArrayInt.FindPermutationFromFirstToSecondDuplicate(arr1,arr0).isEqual(DataArrayInt([8,3,7,2,6,1,4,11,0,5,10,9])))
+        
+    def testDAIIndexOfSameConsecutiveValueGroups(self):
+        arr = DataArrayInt([0,1,1,2,2,3,4,4,5,5,5,11])
+        self.assertTrue(arr.indexOfSameConsecutiveValueGroups().isEqual(DataArrayInt([0,1,3,5,6,8,11,12])))
+
+    def testSkyLineGroupPacks(self):
+        arr = DataArrayInt([1,4,5,0,2,4,5,6,1,3,5,6,7,2,6,7,0,1,5,8,9,0,1,2,4,6,8,9,10,1,2,3,5,7,9,10,11,2,3,6,10,11,4,5,9,12,13,4,5,6,8,10,12,13,14,5,6,7,9,11,13,14,15,6,7,10,14,15,8,9,13,8,9,10,12,14,9,10,11,13,15,10,11,14])
+        arrI = DataArrayInt([0,3,8,13,16,21,29,37,42,47,55,63,68,71,76,81,84])
+        sk = MEDCouplingSkyLineArray(arrI,arr)
+        part = DataArrayInt([0,3,4,7,16])
+        sk2 = sk.groupPacks(part)
+        self.assertTrue(sk2.getValuesArray().isEqual(arr))
+        self.assertTrue(sk2.getIndexArray().isEqual(DataArrayInt([0,13,16,37,84])))
+
+    def testSkyLineUniqueNotSortedByPack(self):    
+        arrI = DataArrayInt([0,3,9,15,18,24,36,48,54])
+        arr = DataArrayInt([1,4,5,0,4,5,2,5,6,3,6,7,1,5,6,2,6,7,0,1,5,5,8,9,0,1,4,6,9,10,1,2,4,6,8,9,2,3,5,7,9,10,1,2,5,7,10,11,2,3,6,6,10,11])
+        sk = MEDCouplingSkyLineArray(arrI,arr)
+        sk2 = sk.uniqueNotSortedByPack()
+        self.assertTrue(sk2.getIndexArray().isEqual(DataArrayInt([0,3,8,13,16,21,29,37,42])))
+        self.assertTrue(sk2.getValuesArray().isEqual(DataArrayInt([1,4,5,0,2,4,5,6,1,3,5,6,7,2,6,7,0,1,5,8,9,0,1,2,4,6,8,9,10,1,2,3,5,7,9,10,11,2,3,6,10,11])))
+
     pass
 
 if __name__ == '__main__':
diff --git a/src/MEDCoupling_Swig/MEDCouplingCommon.i b/src/MEDCoupling_Swig/MEDCouplingCommon.i
index d18ff48a2..d2143962a 100644
--- a/src/MEDCoupling_Swig/MEDCouplingCommon.i
+++ b/src/MEDCoupling_Swig/MEDCouplingCommon.i
@@ -466,6 +466,8 @@ typedef long int mcIdType;
 %newobject MEDCoupling::MEDCouplingSkyLineArray::getSuperIndexArray;
 %newobject MEDCoupling::MEDCouplingSkyLineArray::getIndexArray;
 %newobject MEDCoupling::MEDCouplingSkyLineArray::getValuesArray;
+%newobject MEDCoupling::MEDCouplingSkyLineArray::groupPacks;
+%newobject MEDCoupling::MEDCouplingSkyLineArray::uniqueNotSortedByPack;
 
 %feature("unref") MEDCouplingPointSet "$this->decrRef();"
 %feature("unref") MEDCouplingMesh "$this->decrRef();"
@@ -1296,6 +1298,9 @@ namespace MEDCoupling
     
     void deleteSimplePack(const int i);
     void deleteSimplePacks(const DataArrayIdType* idx);
+
+    MEDCouplingSkyLineArray *groupPacks(const DataArrayIdType *indexedPacks) const;
+    MEDCouplingSkyLineArray *uniqueNotSortedByPack() const;
     
     %extend 
     {
diff --git a/src/MEDCoupling_Swig/MEDCouplingMemArray.i b/src/MEDCoupling_Swig/MEDCouplingMemArray.i
index 9e9c330db..daf6d12a9 100644
--- a/src/MEDCoupling_Swig/MEDCouplingMemArray.i
+++ b/src/MEDCoupling_Swig/MEDCouplingMemArray.i
@@ -129,6 +129,7 @@
 %newobject MEDCoupling::DataArrayInt32::buildSubstraction;
 %newobject MEDCoupling::DataArrayInt32::buildSubstractionOptimized;
 %newobject MEDCoupling::DataArrayInt32::buildIntersection;
+%newobject MEDCoupling::DataArrayInt32::indexOfSameConsecutiveValueGroups;
 %newobject MEDCoupling::DataArrayInt32::buildUnique;
 %newobject MEDCoupling::DataArrayInt32::buildUniqueNotSorted;
 %newobject MEDCoupling::DataArrayInt32::deltaShiftIndex;
@@ -137,10 +138,12 @@
 %newobject MEDCoupling::DataArrayInt32::findRangeIdForEachTuple;
 %newobject MEDCoupling::DataArrayInt32::findIdInRangeForEachTuple;
 %newobject MEDCoupling::DataArrayInt32::duplicateEachTupleNTimes;
+%newobject MEDCoupling::DataArrayInt32::occurenceRankInThis;
 %newobject MEDCoupling::DataArrayInt32::buildPermutationArr;
 %newobject MEDCoupling::DataArrayInt32::buildPermArrPerLevel;
 %newobject MEDCoupling::DataArrayInt32::getDifferentValues;
 %newobject MEDCoupling::DataArrayInt32::FindPermutationFromFirstToSecond;
+%newobject MEDCoupling::DataArrayInt32::FindPermutationFromFirstToSecondDuplicate;
 %newobject MEDCoupling::DataArrayInt32::CheckAndPreparePermutation;
 %newobject MEDCoupling::DataArrayInt32::__neg__;
 %newobject MEDCoupling::DataArrayInt32::__add__;
@@ -201,6 +204,7 @@
 %newobject MEDCoupling::DataArrayInt64::buildSubstraction;
 %newobject MEDCoupling::DataArrayInt64::buildSubstractionOptimized;
 %newobject MEDCoupling::DataArrayInt64::buildIntersection;
+%newobject MEDCoupling::DataArrayInt64::indexOfSameConsecutiveValueGroups;
 %newobject MEDCoupling::DataArrayInt64::buildUnique;
 %newobject MEDCoupling::DataArrayInt64::buildUniqueNotSorted;
 %newobject MEDCoupling::DataArrayInt64::deltaShiftIndex;
@@ -209,10 +213,12 @@
 %newobject MEDCoupling::DataArrayInt64::findRangeIdForEachTuple;
 %newobject MEDCoupling::DataArrayInt64::findIdInRangeForEachTuple;
 %newobject MEDCoupling::DataArrayInt64::duplicateEachTupleNTimes;
+%newobject MEDCoupling::DataArrayInt64::occurenceRankInThis;
 %newobject MEDCoupling::DataArrayInt64::buildPermutationArr;
 %newobject MEDCoupling::DataArrayInt64::buildPermArrPerLevel;
 %newobject MEDCoupling::DataArrayInt64::getDifferentValues;
 %newobject MEDCoupling::DataArrayInt64::FindPermutationFromFirstToSecond;
+%newobject MEDCoupling::DataArrayInt64::FindPermutationFromFirstToSecondDuplicate;
 %newobject MEDCoupling::DataArrayInt64::CheckAndPreparePermutation;
 %newobject MEDCoupling::DataArrayInt64::__neg__;
 %newobject MEDCoupling::DataArrayInt64::__add__;
diff --git a/src/ParaMEDMEM/CMakeLists.txt b/src/ParaMEDMEM/CMakeLists.txt
index 5a7c28443..6871cd11c 100644
--- a/src/ParaMEDMEM/CMakeLists.txt
+++ b/src/ParaMEDMEM/CMakeLists.txt
@@ -40,7 +40,9 @@ SET(paramedmem_SOURCES
   ProcessorGroup.cxx
   MPIProcessorGroup.cxx
   ParaMESH.cxx
+  CommInterface.cxx
   ParaUMesh.cxx
+  ParaSkyLineArray.cxx
   ComponentTopology.cxx
   MPIAccess.cxx
   InterpolationMatrix.cxx
diff --git a/src/ParaMEDMEM/CommInterface.cxx b/src/ParaMEDMEM/CommInterface.cxx
index 6184d7077..9aaebdaec 100644
--- a/src/ParaMEDMEM/CommInterface.cxx
+++ b/src/ParaMEDMEM/CommInterface.cxx
@@ -19,6 +19,8 @@
 
 #include "CommInterface.hxx"
 
+#include <numeric>
+
 namespace MEDCoupling
 {
   /*! \anchor CommInterface-det
@@ -56,11 +58,58 @@ namespace MEDCoupling
     \endverbatim
   */
 
-  CommInterface::CommInterface()
+  /*!
+   * Generalized AllGather collective communication.
+   * This method send input \a array to all procs.
+   */
+  void CommInterface::allGatherArrays(MPI_Comm comm, const DataArrayIdType *array, std::unique_ptr<mcIdType[]>& result, std::unique_ptr<mcIdType[]>& resultIndex) const
   {
+    int size;
+    this->commSize(comm,&size);
+    std::unique_ptr<mcIdType[]> nbOfElems(new mcIdType[size]);
+    mcIdType nbOfCellsRequested(array->getNumberOfTuples());
+    this->allGather(&nbOfCellsRequested,1,MPI_ID_TYPE,nbOfElems.get(),1,MPI_ID_TYPE,comm);
+    mcIdType nbOfCellIdsSum(std::accumulate(nbOfElems.get(),nbOfElems.get()+size,0));
+    result.reset(new mcIdType[nbOfCellIdsSum]);
+    std::unique_ptr<int[]> nbOfElemsInt( CommInterface::ToIntArray<mcIdType>(nbOfElems,size) );
+    std::unique_ptr<int[]> offsetsIn( CommInterface::ComputeOffset(nbOfElemsInt,size) );
+    this->allGatherV(array->begin(),nbOfCellsRequested,MPI_ID_TYPE,result.get(),nbOfElemsInt.get(),offsetsIn.get(),MPI_ID_TYPE,comm);
+    resultIndex = std::move(nbOfElems);
   }
 
-  CommInterface::~CommInterface()
+  /*!
+  * Generalized AllToAll collective communication.
+  */
+  void CommInterface::allToAllArrays(MPI_Comm comm, const std::vector< MCAuto<DataArrayIdType> >& arrays, std::vector< MCAuto<DataArrayIdType> >& arraysOut) const
   {
+    int size;
+    this->commSize(comm,&size);
+    if( arrays.size() != ToSizeT(size) )
+      throw INTERP_KERNEL::Exception("AllToAllArrays : internal error ! Invalid size of input array.");
+      
+    std::vector< const DataArrayIdType *> arraysBis(FromVecAutoToVecOfConst<DataArrayIdType>(arrays));
+    std::unique_ptr<mcIdType[]> nbOfElems2(new mcIdType[size]),nbOfElems3(new mcIdType[size]);
+    for(int curRk = 0 ; curRk < size ; ++curRk)
+    {
+      nbOfElems3[curRk] = arrays[curRk]->getNumberOfTuples();
+    }
+    this->allToAll(nbOfElems3.get(),1,MPI_ID_TYPE,nbOfElems2.get(),1,MPI_ID_TYPE,comm);
+    mcIdType nbOfCellIdsSum(std::accumulate(nbOfElems2.get(),nbOfElems2.get()+size,0));
+    MCAuto<DataArrayIdType> cellIdsFromProcs(DataArrayIdType::New());
+    cellIdsFromProcs->alloc(nbOfCellIdsSum,1);
+    std::unique_ptr<int[]> nbOfElemsInt( CommInterface::ToIntArray<mcIdType>(nbOfElems3,size) ),nbOfElemsOutInt( CommInterface::ToIntArray<mcIdType>(nbOfElems2,size) );
+    std::unique_ptr<int[]> offsetsIn( CommInterface::ComputeOffset(nbOfElemsInt,size) ), offsetsOut( CommInterface::ComputeOffset(nbOfElemsOutInt,size) );
+    {
+      MCAuto<DataArrayIdType> arraysAcc(DataArrayIdType::Aggregate(arraysBis));
+      this->allToAllV(arraysAcc->begin(),nbOfElemsInt.get(),offsetsIn.get(),MPI_ID_TYPE,
+                      cellIdsFromProcs->getPointer(),nbOfElemsOutInt.get(),offsetsOut.get(),MPI_ID_TYPE,comm);
+    }
+    std::unique_ptr<mcIdType[]> offsetsOutIdType( CommInterface::ComputeOffset(nbOfElems2,size) );
+    // build output arraysOut by spliting cellIdsFromProcs into parts
+    arraysOut.resize(size);
+    for(int curRk = 0 ; curRk < size ; ++curRk)
+    {
+      arraysOut[curRk] = DataArrayIdType::NewFromArray(cellIdsFromProcs->begin()+offsetsOutIdType[curRk],cellIdsFromProcs->begin()+offsetsOutIdType[curRk]+nbOfElems2[curRk]);
+    }
   }
 }
diff --git a/src/ParaMEDMEM/CommInterface.hxx b/src/ParaMEDMEM/CommInterface.hxx
index 05aba5083..d132d09db 100644
--- a/src/ParaMEDMEM/CommInterface.hxx
+++ b/src/ParaMEDMEM/CommInterface.hxx
@@ -32,8 +32,8 @@ namespace MEDCoupling
   class CommInterface
   {
   public:
-    CommInterface(){}
-    virtual ~CommInterface(){}
+    CommInterface() { }
+    virtual ~CommInterface() { }
     int worldSize() const {
       int size;
       MPI_Comm_size(MPI_COMM_WORLD, &size);
@@ -81,7 +81,7 @@ namespace MEDCoupling
                   void* recvbuf, int recvcount, MPI_Datatype recvtype,
                   MPI_Comm comm) const { return MPI_Allgather(sendbuf,sendcount, sendtype, recvbuf, recvcount, recvtype, comm); }
     int allGatherV(const void *sendbuf, int sendcount, MPI_Datatype sendtype, void *recvbuf, const int recvcounts[],
-                   const int displs[], MPI_Datatype recvtype, MPI_Comm comm) { return MPI_Allgatherv(sendbuf,sendcount,sendtype,recvbuf,recvcounts,displs,recvtype,comm); }
+                   const int displs[], MPI_Datatype recvtype, MPI_Comm comm) const { return MPI_Allgatherv(sendbuf,sendcount,sendtype,recvbuf,recvcounts,displs,recvtype,comm); }
     int allToAll(void* sendbuf, int sendcount, MPI_Datatype sendtype,
                  void* recvbuf, int recvcount, MPI_Datatype recvtype,
                  MPI_Comm comm) const { return MPI_Alltoall(sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype, comm); }
@@ -94,6 +94,9 @@ namespace MEDCoupling
                MPI_Op op, int root, MPI_Comm comm) const { return MPI_Reduce(sendbuf, recvbuf, count, datatype, op, root, comm); }
     int allReduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) const { return MPI_Allreduce(sendbuf, recvbuf, count, datatype, op, comm); }
   public:
+    void allGatherArrays(MPI_Comm comm, const DataArrayIdType *array, std::unique_ptr<mcIdType[]>& result, std::unique_ptr<mcIdType[]>& resultIndex) const;
+    void allToAllArrays(MPI_Comm comm, const std::vector< MCAuto<DataArrayIdType> >& arrays, std::vector< MCAuto<DataArrayIdType> >& arraysOut) const;
+  public:
 
     /*!
     * \a counts is expected to be an array of array length. This method returns an array of split array.
@@ -124,10 +127,11 @@ namespace MEDCoupling
     /*!
     * Helper of alltoallv and allgatherv
     */
-    static std::unique_ptr<int []> ComputeOffset(const std::unique_ptr<int []>& counts, std::size_t sizeOfCounts)
+    template<class T>
+    static std::unique_ptr<T []> ComputeOffset(const std::unique_ptr<T []>& counts, std::size_t sizeOfCounts)
     {
-      std::unique_ptr<int []> ret(new int[sizeOfCounts]);
-      ret[0] = 0;
+      std::unique_ptr<T []> ret(new T[sizeOfCounts]);
+      ret[0] = static_cast<T>(0);
       for(std::size_t i = 1 ; i < sizeOfCounts ; ++i)
       {
         ret[i] = ret[i-1] + counts[i-1];
diff --git a/src/ParaMEDMEM/ParaSkyLineArray.cxx b/src/ParaMEDMEM/ParaSkyLineArray.cxx
new file mode 100644
index 000000000..586c0c4a8
--- /dev/null
+++ b/src/ParaMEDMEM/ParaSkyLineArray.cxx
@@ -0,0 +1,135 @@
+//
+// Copyright (C) 2020  CEA/DEN, EDF R&D
+//
+// This library 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 2.1 of the License, or (at your option) any later version.
+//
+// This library 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 this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
+//
+// See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
+//
+// Author : Anthony Geay (EDF R&D)
+
+#include "ParaSkyLineArray.hxx"
+#include "ProcessorGroup.hxx"
+#include "MPIProcessorGroup.hxx"
+#include "Topology.hxx"
+#include "BlockTopology.hxx"
+#include "CommInterface.hxx"
+#include "MEDCouplingMemArray.hxx"
+
+#include "mpi.h"
+
+#include <fstream>
+#include <sstream>
+#include <numeric>
+#include <memory>
+#include <vector>
+
+using namespace MEDCoupling;
+
+ParaSkyLineArray *ParaSkyLineArray::New(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds)
+{
+  return new ParaSkyLineArray(ska,globalIds);
+}
+
+MEDCouplingSkyLineArray *ParaSkyLineArray::getSkyLineArray() const
+{
+  return this->_ska.iAmATrollConstCast();
+}
+
+DataArrayIdType *ParaSkyLineArray::getGlobalIdsArray() const
+{
+  return this->_global_ids.iAmATrollConstCast();
+}
+
+ParaSkyLineArray::ParaSkyLineArray(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds)
+{
+  _ska.takeRef(ska);
+  _global_ids.takeRef(globalIds);
+  _ska.checkNotNull();
+  _global_ids.checkNotNull();
+  if(_ska->getNumberOf() != _global_ids->getNumberOfTuples())
+  {
+    std::ostringstream oss; oss << "ParaSkyLineArray constructor : mismatch between # globalIds (" << _global_ids->getNumberOfTuples() << ") and len of indices in SkyLineArray (" << _ska->getNumberOf() << ").";
+    throw INTERP_KERNEL::Exception(oss.str());
+  }
+}
+
+std::size_t ParaSkyLineArray::getHeapMemorySizeWithoutChildren() const
+{
+  return 0;
+}
+
+std::vector<const BigMemoryObject *> ParaSkyLineArray::getDirectChildrenWithNull() const
+{
+  return {_ska,_global_ids};
+}
+
+MCAuto<ParaSkyLineArray> ParaSkyLineArray::equiRedistribute(mcIdType nbOfEntities) const
+{
+  MPI_Comm comm(MPI_COMM_WORLD);
+  CommInterface ci;
+  int size;
+  ci.commSize(comm,&size);
+  std::vector< MCAuto<MEDCouplingSkyLineArray> > skToBeSent(size);
+  std::vector< MCAuto<DataArrayIdType> > idsCaptured(size);
+  for(int curRk = 0 ; curRk < size ; ++curRk)
+  {
+    mcIdType curStart(0),curEnd(0);
+    DataArrayIdType::GetSlice(0,nbOfEntities,1,curRk,size,curStart,curEnd);
+    MCAuto<DataArrayIdType> idsInGlobalIds(_global_ids->findIdsInRange(curStart,curEnd));
+    idsCaptured[curRk] = _global_ids->selectByTupleIdSafe(idsInGlobalIds->begin(),idsInGlobalIds->end());
+    {
+      DataArrayIdType *tmpValues(nullptr),*tmpIndex(nullptr);
+      DataArrayIdType::ExtractFromIndexedArrays(idsInGlobalIds->begin(),idsInGlobalIds->end(),this->_ska->getValuesArray(),this->_ska->getIndexArray(),tmpValues,tmpIndex);
+      MCAuto<DataArrayIdType> tmpValues2(tmpValues),tmpIndex2(tmpIndex);
+      skToBeSent[curRk] = MEDCouplingSkyLineArray::New(tmpIndex,tmpValues);
+    }
+  }
+  // communication : 3 arrays are going to be all2allized : ids, SkyLine index and SyLine values
+  MCAuto<DataArrayIdType> aggregatedIds,indices,values;
+  {
+    std::vector< MCAuto<DataArrayIdType> > myRkIdsCaptured;
+    ci.allToAllArrays(comm,idsCaptured,myRkIdsCaptured);
+    aggregatedIds = DataArrayIdType::Aggregate(FromVecAutoToVecOfConst<DataArrayIdType>(myRkIdsCaptured));
+  }
+  {
+    std::vector< MCAuto<DataArrayIdType> > myRkSkIndex;
+    std::vector< MCAuto<DataArrayIdType> > indexToBeSent(MEDCouplingSkyLineArray::RetrieveVecIndex(skToBeSent));
+    ci.allToAllArrays(comm,indexToBeSent,myRkSkIndex);
+    indices = DataArrayIdType::AggregateIndexes(FromVecAutoToVecOfConst<DataArrayIdType>(myRkSkIndex));
+  }
+  {
+    std::vector< MCAuto<DataArrayIdType> > myRkSkValues;
+    std::vector< MCAuto<DataArrayIdType> > valuesToBeSent(MEDCouplingSkyLineArray::RetrieveVecValues(skToBeSent));
+    ci.allToAllArrays(comm,valuesToBeSent,myRkSkValues);
+    values = DataArrayIdType::Aggregate(FromVecAutoToVecOfConst<DataArrayIdType>(myRkSkValues));
+  }
+  // Reorder results coming from other procs
+  MCAuto<DataArrayIdType> aggregatedIdsSort(aggregatedIds->deepCopy()); aggregatedIdsSort->sort();
+  MCAuto<DataArrayIdType> idsIntoAggregatedIds(DataArrayIdType::FindPermutationFromFirstToSecondDuplicate(aggregatedIdsSort,aggregatedIds));
+  MCAuto<DataArrayIdType> indicesSorted,valuesSorted;
+  {
+    DataArrayIdType *indicesSortedTmp(nullptr),*valuesSortedTmp(nullptr);
+    DataArrayIdType::ExtractFromIndexedArrays(idsIntoAggregatedIds->begin(),idsIntoAggregatedIds->end(),values,indices,valuesSortedTmp,indicesSortedTmp);
+    indicesSorted = indicesSortedTmp; valuesSorted=valuesSortedTmp;
+  }
+  MCAuto<DataArrayIdType> idxOfSameIds(aggregatedIdsSort->indexOfSameConsecutiveValueGroups());
+  //
+  MCAuto<DataArrayIdType> globalIdsOut(aggregatedIdsSort->buildUnique());
+  MCAuto<MEDCouplingSkyLineArray> skOut(MEDCouplingSkyLineArray::New(indicesSorted,valuesSorted));
+  skOut = skOut->groupPacks(idxOfSameIds);//group partial packs coming from different procs
+  skOut = skOut->uniqueNotSortedByPack();//remove duplicates
+  MCAuto<ParaSkyLineArray> ret(ParaSkyLineArray::New(skOut,globalIdsOut));
+  return ret.retn();
+}
\ No newline at end of file
diff --git a/src/ParaMEDMEM/ParaSkyLineArray.hxx b/src/ParaMEDMEM/ParaSkyLineArray.hxx
new file mode 100644
index 000000000..7e3ba1962
--- /dev/null
+++ b/src/ParaMEDMEM/ParaSkyLineArray.hxx
@@ -0,0 +1,55 @@
+// Copyright (C) 2020  CEA/DEN, EDF R&D
+//
+// This library 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 2.1 of the License, or (at your option) any later version.
+//
+// This library 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 this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
+//
+// See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
+//
+// Author : Anthony Geay (EDF R&D)
+
+#pragma once
+
+#include "MEDCouplingSkyLineArray.hxx"
+#include "ProcessorGroup.hxx"
+#include "MEDCouplingMemArray.hxx"
+
+#include <string>
+#include <vector>
+
+namespace MEDCoupling
+{
+  /*!
+   * Parallel representation of a SkyLineArray
+   *
+   * This class is very specific to the requirement of parallel code computations.
+   */
+  class ParaSkyLineArray : public RefCountObject
+  {
+  public:
+    static ParaSkyLineArray *New(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds);
+    MCAuto<ParaSkyLineArray> equiRedistribute(mcIdType nbOfEntities) const;
+    MEDCouplingSkyLineArray *getSkyLineArray() const;
+    DataArrayIdType *getGlobalIdsArray() const;
+    virtual ~ParaSkyLineArray() { }
+  private:
+    ParaSkyLineArray(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds);
+  protected:
+    std::string getClassName() const override { return "ParaSkyLineArray"; }
+    std::size_t getHeapMemorySizeWithoutChildren() const override;
+    std::vector<const BigMemoryObject *> getDirectChildrenWithNull() const override;
+  private:
+    MCAuto<MEDCouplingSkyLineArray> _ska;
+    MCAuto<DataArrayIdType> _global_ids;
+  };
+}
diff --git a/src/ParaMEDMEM/ParaUMesh.cxx b/src/ParaMEDMEM/ParaUMesh.cxx
index 0b939fd17..812a7e4d8 100644
--- a/src/ParaMEDMEM/ParaUMesh.cxx
+++ b/src/ParaMEDMEM/ParaUMesh.cxx
@@ -35,7 +35,6 @@
 #include <memory>
 #include <vector>
 
-using namespace std;
 using namespace MEDCoupling;
 
 ParaUMesh::ParaUMesh(MEDCouplingUMesh *mesh, DataArrayIdType *globalCellIds, DataArrayIdType *globalNodeIds)
@@ -125,27 +124,33 @@ MCAuto<ParaUMesh> ParaUMesh::redistributeCells(const DataArrayIdType *globalCell
   CommInterface ci;
   int size;
   ci.commSize(comm,&size);
-  std::unique_ptr<mcIdType[]> nbOfElems(new mcIdType[size]),nbOfElems2(new mcIdType[size]);
-  mcIdType nbOfNodeIdsLoc(globalCellIds->getNumberOfTuples());
-  ci.allGather(&nbOfNodeIdsLoc,1,MPI_ID_TYPE,nbOfElems.get(),1,MPI_ID_TYPE,comm);
+  std::unique_ptr<mcIdType[]> nbOfElems(new mcIdType[size]);
+  mcIdType nbOfCellsRequested(globalCellIds->getNumberOfTuples());
+  ci.allGather(&nbOfCellsRequested,1,MPI_ID_TYPE,nbOfElems.get(),1,MPI_ID_TYPE,comm);
   mcIdType nbOfCellIdsSum(std::accumulate(nbOfElems.get(),nbOfElems.get()+size,0));
   std::unique_ptr<mcIdType[]> allGlobalCellIds(new mcIdType[nbOfCellIdsSum]);
   std::unique_ptr<int[]> nbOfElemsInt( CommInterface::ToIntArray<mcIdType>(nbOfElems,size) );
   std::unique_ptr<int[]> offsetsIn( CommInterface::ComputeOffset(nbOfElemsInt,size) );
-  ci.allGatherV(globalCellIds->begin(),nbOfNodeIdsLoc,MPI_ID_TYPE,allGlobalCellIds.get(),nbOfElemsInt.get(),offsetsIn.get(),MPI_ID_TYPE,comm);
+  ci.allGatherV(globalCellIds->begin(),nbOfCellsRequested,MPI_ID_TYPE,allGlobalCellIds.get(),nbOfElemsInt.get(),offsetsIn.get(),MPI_ID_TYPE,comm);
   mcIdType offset(0);
+  // Prepare ParaUMesh parts to be sent : compute for each proc the contribution of current rank.
+  std::vector< MCAuto<DataArrayIdType> > globalCellIdsToBeSent(size),globalNodeIdsToBeSent(size);
+  std::vector< MCAuto<MEDCouplingUMesh> > meshPartsToBeSent(size);
   for(int curRk = 0 ; curRk < size ; ++curRk)
   {
     MCAuto<DataArrayIdType> globalCellIdsOfCurProc(DataArrayIdType::New());
     globalCellIdsOfCurProc->useArray(allGlobalCellIds.get()+offset,false,DeallocType::CPP_DEALLOC,nbOfElems[curRk],1);
     offset += nbOfElems[curRk];
-    // prepare all2all session
+    // the key call is here : compute for rank curRk the cells to be sent
     MCAuto<DataArrayIdType> globalCellIdsCaptured(_cell_global->buildIntersection(globalCellIdsOfCurProc));// OK for the global cellIds
     MCAuto<DataArrayIdType> localCellIdsCaptured(_node_global->findIdForEach(globalCellIdsCaptured->begin(),globalCellIdsCaptured->end()));
     MCAuto<MEDCouplingUMesh> meshPart(_mesh->buildPartOfMySelf(localCellIdsCaptured->begin(),localCellIdsCaptured->end(),true));
     MCAuto<DataArrayIdType> o2n(meshPart->zipCoordsTraducer());// OK for the mesh
     MCAuto<DataArrayIdType> n2o(o2n->invertArrayO2N2N2O(meshPart->getNumberOfNodes()));
     MCAuto<DataArrayIdType> globalNodeIdsPart(_node_global->selectByTupleIdSafe(n2o->begin(),n2o->end())); // OK for the global nodeIds
+    meshPartsToBeSent[curRk] = meshPart;
+    globalCellIdsToBeSent[curRk] = globalCellIdsCaptured;
+    globalNodeIdsToBeSent[curRk] = globalNodeIdsPart;
   }
-  
+  // Receive 
 }
diff --git a/src/ParaMEDMEM_Swig/ParaMEDMEMCommon.i b/src/ParaMEDMEM_Swig/ParaMEDMEMCommon.i
index 995df1d24..452834e76 100644
--- a/src/ParaMEDMEM_Swig/ParaMEDMEMCommon.i
+++ b/src/ParaMEDMEM_Swig/ParaMEDMEMCommon.i
@@ -35,6 +35,7 @@
 #include "ICoCoMEDField.hxx"
 #include "ComponentTopology.hxx"
 #include "ParaUMesh.hxx"
+#include "ParaSkyLineArray.hxx"
 
 using namespace INTERP_KERNEL;
 using namespace MEDCoupling;
@@ -58,6 +59,11 @@ using namespace ICoCo;
 %include "ICoCoMEDField.hxx"
 
 %newobject MEDCoupling::ParaUMesh::getCellIdsLyingOnNodes;
+%newobject MEDCoupling::ParaSkyLineArray::equiRedistribute;
+%newobject MEDCoupling::ParaSkyLineArray::getSkyLineArray;
+%newobject MEDCoupling::ParaSkyLineArray::getGlobalIdsArray;
+
+%feature("unref") ParaSkyLineArray "$this->decrRef();"
 
 %nodefaultctor;
 
@@ -136,6 +142,41 @@ namespace MEDCoupling
       }
     }
   };
+
+  class ParaSkyLineArray : public RefCountObject
+  {
+  public:
+    static ParaSkyLineArray *New(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds);
+    %extend
+    {
+      ParaSkyLineArray(MEDCouplingSkyLineArray *ska, DataArrayIdType *globalIds)
+      {
+        return ParaSkyLineArray::New(ska,globalIds);
+      }
+
+      ParaSkyLineArray *equiRedistribute(mcIdType nbOfEntities) const
+      {
+        MCAuto<ParaSkyLineArray> ret(self->equiRedistribute(nbOfEntities));
+        return ret.retn();
+      }
+
+      MEDCouplingSkyLineArray *getSkyLineArray() const
+      {
+        MEDCouplingSkyLineArray *ret(self->getSkyLineArray());
+        if(ret)
+          ret->incrRef();
+        return ret;
+      }
+      
+      DataArrayIdType *getGlobalIdsArray() const
+      {
+        DataArrayIdType *ret(self->getGlobalIdsArray());
+        if(ret)
+          ret->incrRef();
+        return ret;
+      }
+    }
+  };
 }
 
 /* This object can be used only if MED_ENABLE_FVM is defined*/