public static PartialUpsertMerger getPartialUpsertMerger(List<String> primaryKeyColumns,
      List<String> comparisonColumns, UpsertConfig upsertConfig) {
    String customMergerClassName = upsertConfig.getPartialUpsertMergerClass();
    // If a custom implementation is provided in config, initialize an implementation and return.
    if (StringUtils.isNotBlank(customMergerClassName)) {
      try {
        Class<?> partialUpsertMergerClass = Class.forName(customMergerClassName);
        return (PartialUpsertMerger) partialUpsertMergerClass.getConstructor(List.class, List.class, UpsertConfig.class)
            .newInstance(primaryKeyColumns, comparisonColumns, upsertConfig);
      } catch (Exception e) {
        throw new RuntimeException(
            String.format("Failed to instantiate partial upsert merger with class: %s", customMergerClassName), e);
      }
    }
    return new PartialUpsertColumnarMerger(primaryKeyColumns, comparisonColumns, upsertConfig);
  }
@Test
  public void testGetPartialUpsertMerger() {
    Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("pk", FieldSpec.DataType.STRING)
        .addSingleValueDimension("field1", FieldSpec.DataType.LONG).addMetric("field2", FieldSpec.DataType.LONG)
        .addDateTime("hoursSinceEpoch", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
        .setPrimaryKeyColumns(Arrays.asList("pk")).build();

    UpsertConfig upsertConfig = new UpsertConfig();
    Map<String, UpsertConfig.Strategy> partialUpsertStrategies = new HashMap<>();
    partialUpsertStrategies.put("field1", UpsertConfig.Strategy.OVERWRITE);
    upsertConfig.setPartialUpsertStrategies(partialUpsertStrategies);
    upsertConfig.setDefaultPartialUpsertStrategy(UpsertConfig.Strategy.IGNORE);
    upsertConfig.setPartialUpsertMergerClass(
        "org.apache.pinot.segment.local.upsert.merger.PartialUpsertColumnarMerger");

    PartialUpsertMerger partialUpsertMerger =
        PartialUpsertMergerFactory.getPartialUpsertMerger(schema.getPrimaryKeyColumns(),
            Collections.singletonList("hoursSinceEpoch"), upsertConfig);

    assertNotNull(partialUpsertMerger);
    assertTrue(partialUpsertMerger instanceof PartialUpsertColumnarMerger);
  }