public static List<PartitionSpec> getPartitionspecsGroupedByStorageDescriptor(Table table,
                                                                                Collection<Partition> partitions) {
    final String tablePath = table.getSd().getLocation();

    ImmutableListMultimap<StorageDescriptorKey, Partition> partitionsWithinTableDirectory =
        Multimaps.index(partitions, input -> {
          // if sd is not in the list of projected fields, all the partitions
          // can be just grouped in PartitionSpec object
          if (input.getSd() == null) {
            return StorageDescriptorKey.UNSET_KEY;
          }

          // if sd has skewed columns we better not group partition, since different partitions
          // could have different skewed info like skewed location
          if (input.getSd().getSkewedInfo() != null
              && input.getSd().getSkewedInfo().getSkewedColNames() != null
              && !input.getSd().getSkewedInfo().getSkewedColNames().isEmpty()) {
            return new StorageDescriptorKey(input.getSd());
          }

          // if partitions don't have the same number of buckets we can not group their SD,
          // this could lead to incorrect number of buckets
          if (input.getSd().getNumBuckets()
              != partitions.iterator().next().getSd().getNumBuckets()) {
            return new StorageDescriptorKey(input.getSd());
          }
          // if the partition is within table, use the tableSDKey to group it with other partitions
          // within the table directory
          if (input.getSd().getLocation() != null && input.getSd().getLocation()
              .startsWith(tablePath)) {
            return new StorageDescriptorKey(tablePath, input.getSd());
          }
          // if partitions are located outside table location we treat them as non-standard
          // and do not perform any grouping
          // if the location is not set partitions are grouped according to the rest of the SD fields
          return new StorageDescriptorKey(input.getSd());
        });

    List<PartitionSpec> partSpecs = new ArrayList<>();

    // Classify partitions based on shared SD properties.
    Map<StorageDescriptorKey, List<PartitionWithoutSD>> sdToPartList
        = new HashMap<>();
    // we don't expect partitions to exist outside directory in most cases
    List<Partition> partitionsOutsideTableDir = new ArrayList<>(0);
    for (StorageDescriptorKey key : partitionsWithinTableDirectory.keySet()) {
      boolean isUnsetKey = key.equals(StorageDescriptorKey.UNSET_KEY);
      // group the partitions together when
      // case I : sd is not set because it was not in the requested fields
      // case II : when sd.location is not set because it was not in the requested fields
      // case III : when sd.location is set and it is located within table directory
      if (isUnsetKey || key.baseLocation == null || key.baseLocation.equals(tablePath)) {
        for (Partition partition : partitionsWithinTableDirectory.get(key)) {

          PartitionWithoutSD partitionWithoutSD
              = new PartitionWithoutSD();
          partitionWithoutSD.setValues(partition.getValues());
          partitionWithoutSD.setCreateTime(partition.getCreateTime());
          partitionWithoutSD.setLastAccessTime(partition.getLastAccessTime());
          partitionWithoutSD.setRelativePath(
              (isUnsetKey || !partition.getSd().isSetLocation()) ? null : partition.getSd()
                  .getLocation().substring(tablePath.length()));
          partitionWithoutSD.setParameters(partition.getParameters());

          if (!sdToPartList.containsKey(key)) {
            sdToPartList.put(key, new ArrayList<>());
          }
          sdToPartList.get(key).add(partitionWithoutSD);
        }
      } else {
        // Lump all partitions outside the tablePath into one PartSpec.
        // if non-standard partitions need not be deDuped create PartitionListComposingSpec
        // this will be used mostly for keeping backwards compatibility with  some HMS APIs which use
        // PartitionListComposingSpec for non-standard partitions located outside table
        partitionsOutsideTableDir.addAll(partitionsWithinTableDirectory.get(key));
      }
    }
    // create sharedSDPartSpec for all the groupings
    for (Map.Entry<StorageDescriptorKey, List<PartitionWithoutSD>> entry : sdToPartList
        .entrySet()) {
      partSpecs.add(getSharedSDPartSpec(table, entry.getKey(), entry.getValue()));
    }
    if (!partitionsOutsideTableDir.isEmpty()) {
      PartitionSpec partListSpec = new PartitionSpec();
      partListSpec.setCatName(table.getCatName());
      partListSpec.setDbName(table.getDbName());
      partListSpec.setTableName(table.getTableName());
      partListSpec.setPartitionList(new PartitionListComposingSpec(partitionsOutsideTableDir));
      partSpecs.add(partListSpec);
    }
    return partSpecs;
  }
@Test
  public void testGetPartitionspecsGroupedBySDNullSD() throws MetaException {
    // Create database and table
    Table tbl = new TableBuilder()
        .setDbName(DB_NAME)
        .setTableName(TABLE_NAME)
        .addCol("id", "int")
        .setLocation("/foo")
        .build(null);
    Partition p1 = new PartitionBuilder()
        .setDbName("DB_NAME")
        .setTableName(TABLE_NAME)
        .addCol("a", "int")
        .addValue("val1")
        .setInputFormat("foo")
        .build(null);
    // Set SD to null
    p1.unsetSd();
    assertThat(p1.getSd(), is((StorageDescriptor)null));
    List<PartitionSpec> result =
        MetaStoreServerUtils.getPartitionspecsGroupedByStorageDescriptor(tbl, Collections.singleton(p1));
    assertThat(result.size(), is(1));
    PartitionSpec ps = result.get(0);
    assertThat(ps.getRootPath(), is((String)null));
    List<PartitionWithoutSD> partitions = ps.getSharedSDPartitionSpec().getPartitions();
    assertThat(partitions.size(), is(1));
    PartitionWithoutSD partition = partitions.get(0);
    assertThat(partition.getRelativePath(), is((String)null));
    assertThat(partition.getValues(), is(Collections.singletonList("val1")));
  }