rohityadav1993 commented on code in PR #11983:
URL: https://github.com/apache/pinot/pull/11983#discussion_r1551631254
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/merger/PartialUpsertMergerFactory.java:
##########
@@ -18,39 +18,50 @@
*/
package org.apache.pinot.segment.local.upsert.merger;
+import java.lang.reflect.InvocationTargetException;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.config.table.UpsertConfig;
public class PartialUpsertMergerFactory {
+
private PartialUpsertMergerFactory() {
}
- private static final AppendMerger APPEND_MERGER = new AppendMerger();
- private static final IncrementMerger INCREMENT_MERGER = new
IncrementMerger();
- private static final IgnoreMerger IGNORE_MERGER = new IgnoreMerger();
- private static final OverwriteMerger OVERWRITE_MERGER = new
OverwriteMerger();
- private static final MaxMerger MAX_MERGER = new MaxMerger();
- private static final MinMerger MIN_MERGER = new MinMerger();
- private static final UnionMerger UNION_MERGER = new UnionMerger();
-
- public static PartialUpsertMerger getMerger(UpsertConfig.Strategy strategy) {
- switch (strategy) {
- case APPEND:
- return APPEND_MERGER;
- case INCREMENT:
- return INCREMENT_MERGER;
- case IGNORE:
- return IGNORE_MERGER;
- case MAX:
- return MAX_MERGER;
- case MIN:
- return MIN_MERGER;
- case OVERWRITE:
- return OVERWRITE_MERGER;
- case UNION:
- return UNION_MERGER;
- default:
- throw new IllegalStateException("Unsupported partial upsert strategy:
" + strategy);
+ /**
+ * Initialise the default partial upsert merger or initialise a custom
implementation from a given class name in
+ * config
+ * @param primaryKeyColumns
+ * @param comparisonColumns
+ * @param upsertConfig
+ * @return
+ */
+ public static PartialUpsertMerger getPartialUpsertMerger(List<String>
primaryKeyColumns,
+ List<String> comparisonColumns, UpsertConfig upsertConfig) {
+ PartialUpsertMerger partialUpsertMerger;
+ String customRowMerger = upsertConfig.getPartialUpsertMergerClass();
+ // If a custom implementation is provided in config, initialize an
implementation and return.
+ if (StringUtils.isNotBlank(customRowMerger)) {
+ try {
+ Class<?> partialUpsertMergerClass = Class.forName(customRowMerger);
+ if
(!BasePartialUpsertMerger.class.isAssignableFrom(partialUpsertMergerClass)) {
Review Comment:
`BasePartialUpsertMerger ` defines a constructor which is be used to
initialize a custom merger class by overriding the constructor:
`(PartialUpsertMerger) partialUpsertMergerClass.getConstructor(List.class,
List.class, UpsertConfig.class)
.newInstance(primaryKeyColumns, comparisonColumns,
upsertConfig);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
