public static List<PartitionerConfig> getPartitionerConfigs(TableConfig tableConfig, Schema schema,
      Map<String, String> taskConfig) {
    SegmentPartitionConfig segmentPartitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
    if (segmentPartitionConfig == null) {
      return Collections.emptyList();
    }
    List<PartitionerConfig> partitionerConfigs = new ArrayList<>();
    Map<String, ColumnPartitionConfig> columnPartitionMap = segmentPartitionConfig.getColumnPartitionMap();
    for (Map.Entry<String, ColumnPartitionConfig> entry : columnPartitionMap.entrySet()) {
      String partitionColumn = entry.getKey();
      Preconditions.checkState(schema.hasColumn(partitionColumn),
          "Partition column: %s does not exist in the schema for table: %s", partitionColumn,
          tableConfig.getTableName());
      PartitionerConfig partitionerConfig =
          new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
              .setColumnName(partitionColumn).setColumnPartitionConfig(entry.getValue()).build();
      partitionerConfigs.add(partitionerConfig);
    }
    return partitionerConfigs;
  }
@Test
  public void testGetPartitionerConfigs() {
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable")
        .setSegmentPartitionConfig(
            new SegmentPartitionConfig(Collections.singletonMap("memberId", new ColumnPartitionConfig("murmur", 10))))
        .build();
    Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("memberId", DataType.LONG).build();
    Map<String, String> taskConfig = Collections.emptyMap();

    List<PartitionerConfig> partitionerConfigs = MergeTaskUtils.getPartitionerConfigs(tableConfig, schema, taskConfig);
    assertEquals(partitionerConfigs.size(), 1);
    PartitionerConfig partitionerConfig = partitionerConfigs.get(0);
    assertEquals(partitionerConfig.getPartitionerType(), PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG);
    assertEquals(partitionerConfig.getColumnName(), "memberId");
    ColumnPartitionConfig columnPartitionConfig = partitionerConfig.getColumnPartitionConfig();
    assertEquals(columnPartitionConfig.getFunctionName(), "murmur");
    assertEquals(columnPartitionConfig.getNumPartitions(), 10);

    // Table with multiple partition columns.
    Map<String, ColumnPartitionConfig> columnPartitionConfigMap = new HashMap<>();
    columnPartitionConfigMap.put("memberId", new ColumnPartitionConfig("murmur", 10));
    columnPartitionConfigMap.put("memberName", new ColumnPartitionConfig("HashCode", 5));
    TableConfig tableConfigWithMultiplePartitionColumns =
        new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable")
            .setSegmentPartitionConfig(new SegmentPartitionConfig(columnPartitionConfigMap)).build();
    Schema schemaWithMultipleColumns = new Schema.SchemaBuilder().addSingleValueDimension("memberId", DataType.LONG)
        .addSingleValueDimension("memberName", DataType.STRING).build();
    partitionerConfigs =
        MergeTaskUtils.getPartitionerConfigs(tableConfigWithMultiplePartitionColumns, schemaWithMultipleColumns,
            taskConfig);
    assertEquals(partitionerConfigs.size(), 2);

    // No partition column in table config
    TableConfig tableConfigWithoutPartitionColumn =
        new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
    assertTrue(MergeTaskUtils.getPartitionerConfigs(tableConfigWithoutPartitionColumn, schema, taskConfig).isEmpty());

    // Partition column does not exist in schema
    Schema schemaWithoutPartitionColumn = new Schema.SchemaBuilder().build();
    try {
      MergeTaskUtils.getPartitionerConfigs(tableConfig, schemaWithoutPartitionColumn, taskConfig);
      fail();
    } catch (IllegalStateException e) {
      // Expected
    }
  }