pluggable partial upsert merger #11983

rohityadav1993 · 2023-11-09T16:20:16Z

feature refactor release-notes

Implements: #11174
This PR introduces the possibility to define row mergers(currently only per column merging happens) for partial upsert. This allows a table to define complex logic for merging in partial upsert table. An analogy is painless script in ES.
This is the second PR for the feature described in this design doc.

Makes `PartialUpsertHandler` do row merging using `PartialUpsertMerger`.

Based on previous feedback comment.

Introduces new contract to merge rows by refactoring PartialUpsertMerger as:
public void merge(LazyRow prevRecord, GenericRow newRecord, Map<String, Object> mergerResult)
LazyRow introduced in add LazyRow abstraction for previously indexed record #11826 enables lazily reading previous column value only if needed by PartialUpsertMerger
PartialUpsertHandler initiliazes an implementation of PartialUpsertMerger using PartialUpsertMergerFactory.
Existing column level merger behaviour is unchanged.

Adds the possibility to define custom merging per table and specify as table config.

PartialUpsertMergerFactory initializes PartialUpsertColumnarMerger (this is existing columnar merging logic extracted out of PartialUpsertHandler) if column level merger strategies are defined.
If UpsertConfig. partialUpsertMergerClass is defined then PartialUpsertMergerFactory provides the implementation for same. e.g.

{
  "upsertConfig": {
    "mode": "PARTIAL",
    "partialUpsertMergerClass": "org.apache.pinot.segment.local.upsert.merger.PartialUpsertMyCustomMerger",
  },
  "tableIndexConfig": {
    "nullHandlingEnabled": true
  }
}

_Note: When custom partialUpsertMergerClasscu is provided then the column merger strategies will not be applied. The defaultPartialUpsertStrategy is also not applicable for any column and should be handled in the custom merge logic.

Refactors existing column mergers as implementations of `PartialUpsertColumnMerger`.

Based on previous feedback comment.

Current merger logic, which is per column merging of previous and new value is still the default behaviour and backward compatible with configs.
The existing mergers now implement PartialUpsertColumnMerger to differentiate from row merger: PartialUpsertMerger

Changes based on the last review:

Renamed the merger class config name to: partialUpsertMergerClass
Modified PartialUpsertHandler to iterate over merger result when partialUpsertMergerClass is provided.

Tests:
Unit test: Added

Functional test:

Created a test setup by replicating the quick start example: upsertPartialMeetupRsvp i.e. the partial upsert strategies were implemented as a custom partialUpsertMergerClass:

public class CustomMeetUpRSVPRowMerger extends BasePartialUpsertMerger {
  private UnionMerger _unionMerger;
  private AppendMerger _appendMerger;
  private IncrementMerger _incrementMerger;

  public CustomMeetUpRSVPRowMerger(List<String> primaryKeyColumns, List<String> comparisonColumns,
      UpsertConfig upsertConfig) {
    super(primaryKeyColumns, comparisonColumns, upsertConfig);
    _unionMerger = new UnionMerger();
    _appendMerger = new AppendMerger();
    _incrementMerger = new IncrementMerger();
  }

  @Override
  public void merge(LazyRow prevRecord, GenericRow newRecord, Map<String, Object> mergerResult) {
    // Merge the rsvp_count
    mergerResult.put("rsvp_count",
        _incrementMerger.merge(prevRecord.getValue("rsvp_count"), newRecord.getValue("rsvp_count")));

    // union group_name string array

    mergerResult.put("group_name",
        _unionMerger.merge(prevRecord.getValue("group_name"), newRecord.getValue("group_name")));

    // append venue_name string
    mergerResult.put("venue_name",
        _appendMerger.merge(prevRecord.getValue("venue_name"), newRecord.getValue("venue_name")));
  }
}

Table config:
upsertPartialCustomMergerMeetupRsvp_realtime_table_config.json

Observation:
Both partial upsert table with columnar strategies and custom merger class had same data:
upsertPartialMeetupRsvp_quer.txt
upsertPartialCustomMergerMeetupRsvp_query.txt

codecov-commenter · 2023-11-09T17:02:22Z

Codecov Report

Attention: Patch coverage is 86.11111% with 15 lines in your changes are missing coverage. Please review.

Project coverage is 62.08%. Comparing base (59551e4) to head (edf3ab8).
Report is 212 commits behind head on master.

Files	Patch %	Lines
...he/pinot/segment/local/utils/TableConfigUtils.java	68.75%	1 Missing and 4 partials ⚠️
...ocal/upsert/merger/PartialUpsertMergerFactory.java	66.66%	3 Missing ⚠️
...not/segment/local/upsert/PartialUpsertHandler.java	90.47%	1 Missing and 1 partial ⚠️
...ger/columnar/PartialUpsertColumnMergerFactory.java	87.50%	1 Missing and 1 partial ⚠️
...rg/apache/pinot/spi/config/table/UpsertConfig.java	33.33%	2 Missing ⚠️
...cal/upsert/merger/PartialUpsertColumnarMerger.java	96.29%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #11983      +/-   ##
============================================
+ Coverage     61.75%   62.08%   +0.33%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2465      +29     
  Lines        133233   134846    +1613     
  Branches      20636    20834     +198     
============================================
+ Hits          82274    83720    +1446     
- Misses        44911    44954      +43     
- Partials       6048     6172     +124

Flag	Coverage Δ
custom-integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration	`<0.01% <0.00%> (-0.01%)`	⬇️
integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration2	`0.00% <0.00%> (ø)`
java-11	`61.98% <86.11%> (+0.27%)`	⬆️
java-21	`61.93% <86.11%> (+0.30%)`	⬆️
skip-bytebuffers-false	`62.04% <86.11%> (+0.29%)`	⬆️
skip-bytebuffers-true	`61.87% <86.11%> (+34.14%)`	⬆️
temurin	`62.08% <86.11%> (+0.33%)`	⬆️
unittests	`62.08% <86.11%> (+0.33%)`	⬆️
unittests1	`46.71% <0.92%> (-0.18%)`	⬇️
unittests2	`28.08% <85.18%> (+0.34%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Jackie-Jiang · 2023-11-14T02:04:57Z

Please add some description to the PR, including what support is added, how to configure it etc.

rohityadav1993 · 2023-11-22T06:45:01Z

Please add some description to the PR, including what support is added, how to configure it etc.

@Jackie-Jiang, please review, added the necessary details.

deemoliu · 2023-11-22T00:26:31Z

...ocal/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java

@@ -106,8 +106,8 @@ public void init(TableConfig tableConfig, Schema schema, TableDataManager tableD
      Preconditions.checkArgument(partialUpsertStrategies != null,
          "Partial-upsert strategies must be configured for partial-upsert enabled table: %s", _tableNameWithType);
      _partialUpsertHandler =
-          new PartialUpsertHandler(schema, partialUpsertStrategies, upsertConfig.getDefaultPartialUpsertStrategy(),
-              _comparisonColumns);
+          new PartialUpsertHandler(schema, tableConfig.getValidationConfig().getTimeColumnName(), _comparisonColumns,


can we get _comparisonColumns from upsertConfig?

We should not pass in time column. We can pass in _comparisonColumns and upsertConfig

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

Jackie-Jiang · 2023-12-01T00:23:49Z

...ocal/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java

@@ -106,8 +106,8 @@ public void init(TableConfig tableConfig, Schema schema, TableDataManager tableD
      Preconditions.checkArgument(partialUpsertStrategies != null,
          "Partial-upsert strategies must be configured for partial-upsert enabled table: %s", _tableNameWithType);
      _partialUpsertHandler =
-          new PartialUpsertHandler(schema, partialUpsertStrategies, upsertConfig.getDefaultPartialUpsertStrategy(),
-              _comparisonColumns);
+          new PartialUpsertHandler(schema, tableConfig.getValidationConfig().getTimeColumnName(), _comparisonColumns,


We should not pass in time column. We can pass in _comparisonColumns and upsertConfig

Jackie-Jiang · 2023-12-01T00:25:55Z

.../java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java

@@ -307,7 +309,7 @@ protected GenericRow doUpdateRecord(GenericRow record, RecordInfo recordInfo) {
            int currentDocId = recordLocation.getDocId();
            if (currentQueryableDocIds == null || currentQueryableDocIds.contains(currentDocId)) {
              _reusePreviousRow.init(currentSegment, currentDocId);
-              _partialUpsertHandler.merge(_reusePreviousRow, record);
+              _partialUpsertHandler.merge(_reusePreviousRow, record, _reuseMergerResult);


I feel this reusable map should be maintained within the PartialUpsertHandler. I don't see why it should be part of the interface

The_reuseMergerResult map is defined in the context of a partition for a consuming segment. If we move it to PartialUpsertHandler then we need to modify PartialUpsertHandler to be initialised in BasePartitionUpsertMetadataManager instead of BaseTableUpsertMetadataManager. Moreover the _reuseMergerResult would have to be made threadsafe across consuming segments.

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

rohityadav1993 · 2024-03-09T20:59:13Z

The integration test failure is in delete table flow which is unrelated to the proposed changes.

...nt-local/src/main/java/org/apache/pinot/segment/local/upsert/merger/PartialUpsertMerger.java

Jackie-Jiang · 2024-03-11T23:58:09Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

+        // PartialUpsertColumnMerger already handles default merger but for any custom implementations
+        // non merged columns need to be applied with default merger
+        newRecord.putValue(column,
+            _defaultPartialUpsertMerger.merge(prevRecord.getValue(column), newRecord.getValue(column)));


The default strategy is columnar based. Do we want to apply it for other mergers? I feel it is more intuitive if all the merge logic is handled within the merger

If we do not handle default merge here, we may iterate over the merger result to reduce map lookups

Good point, this was done with intention to keep the logic in the merger class minimal. But the optmization on map iteration is a better option. Let me update this.

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/UpsertConfig.java

rohityadav1993 · 2024-03-18T17:33:28Z

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java

@@ -918,32 +918,43 @@ static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema sche
    UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
    assert upsertConfig != null;
    Map<String, UpsertConfig.Strategy> partialUpsertStrategies = upsertConfig.getPartialUpsertStrategies();
+    String partialUpsertMergerClass = upsertConfig.getPartialUpsertMergerClass();
+


@Jackie-Jiang , @deemoliu , we may want to disallow defualtMergerStrategy to be defined when using partialUpsertMergerClass. This will require us to modify UpsertConfig.class default initialization: private Strategy _defaultPartialUpsertStrategy = Strategy.OVERWRITE; And handle this initilization in POST /tableConfigs

rohityadav1993 · 2024-04-01T19:00:19Z

@Jackie-Jiang @deemoliu , please review the updated changes.

Jackie-Jiang · 2024-04-04T05:01:44Z

...ocal/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java

          "Partial-upsert strategies must be configured for partial-upsert enabled table: %s", _tableNameWithType);
      partialUpsertHandler =
-          new PartialUpsertHandler(schema, partialUpsertStrategies, upsertConfig.getDefaultPartialUpsertStrategy(),
-              comparisonColumns);
+          new PartialUpsertHandler(schema, comparisonColumns, upsertConfig);


(format) Please auto reformat all the changes with Pinot Style

Jackie-Jiang · 2024-04-04T05:40:04Z

...ocal/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java

          "Partial-upsert strategies must be configured for partial-upsert enabled table: %s", _tableNameWithType);
      partialUpsertHandler =
-          new PartialUpsertHandler(schema, partialUpsertStrategies, upsertConfig.getDefaultPartialUpsertStrategy(),
-              comparisonColumns);
+          new PartialUpsertHandler(schema, comparisonColumns, upsertConfig);


(format) Please auto-reformat all the changes with Pinot Style

Jackie-Jiang · 2024-04-04T05:40:31Z

...ocal/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java

+      String rowMergerCustomImplementation = upsertConfig.getPartialUpsertMergerClass();
+      Preconditions.checkArgument(
+          StringUtils.isNotBlank(rowMergerCustomImplementation) || partialUpsertStrategies != null,


(minor) These checks can be pushed down to the constructor of PartialUpsertHandler

If the checks are moved down to PartialUpsertHandler, the create table API response is success but the table goes in error state. This check would pro-actively fail table creation.

I've applied a commit to demonstrate the idea. The exception will be thrown from the constructor

Got it, thank you Jackie.

Jackie-Jiang · 2024-04-04T05:41:36Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

-            }
-          }
+  public void merge(LazyRow prevRecord, GenericRow newRecord, Map<String, Object> reuseMergerResult) {
+    reuseMergerResult.clear();


(minor) Clear the result on the caller side

Jackie-Jiang · 2024-04-04T05:42:01Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

+    // merger current row with previously indexed row
+    _partialUpsertMerger.merge(prevRecord, newRecord, reuseMergerResult);
+
+    if (_partialUpsertMerger instanceof PartialUpsertColumnarMerger) {


Do we need to differentiate PartialUpsertColumnarMerger and custom merger? The logic should be the same?

I wanted to keep the logic for PartialUpsertColumnarMerger(behaviour so far) to be unmodified and was doing null handling wrongly for custom merger. I have removed the differentiation and using putDefaultNullValue() for null handling based on another review comment. We will never have a null merger result from PartialUpsertColumnarMerger.

Jackie-Jiang · 2024-04-04T05:43:19Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java

+    } else {
+      // if column exists but mapped to a null value then merger result was a null value
+      newRecord.addNullValueField(column);
+      newRecord.putValue(column, null);


(MAJOR) This won't work. In order to set a value to null, you'll need to call putDefaultNullValue() with a default null value. Do you see a scenario where you want to explicitly set a value to null?

I might not be fully aware of the null value support feature, let me update it with putDefaultNullValue(). Thank you. I don't forsee a scenario for null value.

Jackie-Jiang · 2024-04-04T05:43:58Z

...l/src/main/java/org/apache/pinot/segment/local/upsert/merger/PartialUpsertMergerFactory.java

+      } catch (ClassNotFoundException
+               | NoSuchMethodException | InstantiationException | IllegalAccessException
+               | InvocationTargetException e) {


(minor) These can be merged as Exception e

Jackie-Jiang · 2024-04-04T05:44:35Z

...l/src/main/java/org/apache/pinot/segment/local/upsert/merger/PartialUpsertMergerFactory.java

+    if (StringUtils.isNotBlank(customRowMerger)) {
+      try {
+        Class<?> partialUpsertMergerClass = Class.forName(customRowMerger);
+        if (!BasePartialUpsertMerger.class.isAssignableFrom(partialUpsertMergerClass)) {


Why does it have to extend BasePartialUpsertMerger? Implementing PartialUpsertMerger should be good right?

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);

Jackie-Jiang

LGTM. Applied a commit for some minor cleanups

Jackie-Jiang added feature release-notes Referenced by PRs that need attention when compiling the next release notes Configuration Config changes (addition/deletion/change in behavior) upsert labels Nov 14, 2023

rohityadav1993 marked this pull request as ready for review November 14, 2023 08:14

rohityadav1993 marked this pull request as draft November 15, 2023 11:01

rohityadav1993 force-pushed the custom_merger branch 2 times, most recently from 3e4f005 to 7ec1c8a Compare November 15, 2023 14:47

rohityadav1993 changed the title ~~Custom merger~~ pluggable partial upsert merger Nov 17, 2023

rohityadav1993 marked this pull request as ready for review November 17, 2023 18:34

deemoliu reviewed Nov 22, 2023

View reviewed changes

Jackie-Jiang reviewed Dec 1, 2023

View reviewed changes

rohityadav1993 force-pushed the custom_merger branch from 7ec1c8a to a8d7723 Compare March 9, 2024 19:38

Jackie-Jiang reviewed Mar 12, 2024

View reviewed changes

rohityadav1993 commented Mar 18, 2024

View reviewed changes

rohityadav1993 force-pushed the custom_merger branch from c6bca7c to 6b6a184 Compare March 19, 2024 19:34

rohityadav1993 force-pushed the custom_merger branch from 6b6a184 to 519108f Compare April 2, 2024 20:22

rohityadav1993 added 6 commits April 3, 2024 11:00

refactor existing mergers to represent column mergers

bb71893

pluggable custom partial upsert merger

aae941b

specify default column merger for custom row merger

aa8fd9d

custom merger support in table config

982f781

address review comments

3a3e827

fix partial upsert merger class validation

db12052

rohityadav1993 force-pushed the custom_merger branch from 519108f to db12052 Compare April 3, 2024 05:31

Jackie-Jiang reviewed Apr 4, 2024

View reviewed changes

address review comments

917c9d2

rohityadav1993 force-pushed the custom_merger branch from 79dc192 to 917c9d2 Compare April 4, 2024 18:37

Minor cleanup

072b0e5

Jackie-Jiang approved these changes Apr 5, 2024

View reviewed changes

Fix linter

edf3ab8

Jackie-Jiang merged commit 01df14b into apache:master Apr 5, 2024
19 checks passed

rohityadav1993 mentioned this pull request Apr 30, 2024

fix merging null multi value in partial upsert #13031

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

pluggable partial upsert merger #11983

pluggable partial upsert merger #11983

rohityadav1993 commented Nov 9, 2023 •

edited

Loading

codecov-commenter commented Nov 9, 2023 •

edited

Loading

Jackie-Jiang commented Nov 14, 2023

rohityadav1993 commented Nov 22, 2023

deemoliu Nov 22, 2023

Jackie-Jiang Dec 1, 2023

Jackie-Jiang Dec 1, 2023

Jackie-Jiang Dec 1, 2023

rohityadav1993 Mar 8, 2024

rohityadav1993 commented Mar 9, 2024

Jackie-Jiang Mar 11, 2024

Jackie-Jiang Mar 11, 2024

rohityadav1993 Mar 14, 2024

rohityadav1993 Mar 18, 2024

rohityadav1993 commented Apr 1, 2024

Jackie-Jiang Apr 4, 2024

Jackie-Jiang Apr 4, 2024

Jackie-Jiang Apr 4, 2024

rohityadav1993 Apr 4, 2024

Jackie-Jiang Apr 5, 2024

rohityadav1993 Apr 5, 2024

Jackie-Jiang Apr 4, 2024

Jackie-Jiang Apr 4, 2024

rohityadav1993 Apr 4, 2024

Jackie-Jiang Apr 4, 2024

rohityadav1993 Apr 4, 2024

Jackie-Jiang Apr 4, 2024

Jackie-Jiang Apr 4, 2024

rohityadav1993 Apr 4, 2024

Jackie-Jiang left a comment

pluggable partial upsert merger #11983

pluggable partial upsert merger #11983

Conversation

rohityadav1993 commented Nov 9, 2023 • edited Loading

Makes PartialUpsertHandler do row merging using PartialUpsertMerger.

Adds the possibility to define custom merging per table and specify as table config.

Refactors existing column mergers as implementations of PartialUpsertColumnMerger.

codecov-commenter commented Nov 9, 2023 • edited Loading

Codecov Report

Jackie-Jiang commented Nov 14, 2023

rohityadav1993 commented Nov 22, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohityadav1993 commented Mar 9, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohityadav1993 commented Apr 1, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

rohityadav1993 commented Nov 9, 2023 •

edited

Loading

Makes `PartialUpsertHandler` do row merging using `PartialUpsertMerger`.

Refactors existing column mergers as implementations of `PartialUpsertColumnMerger`.

codecov-commenter commented Nov 9, 2023 •

edited

Loading