Add global strategy for partial upsert #7906

deemoliu · 2021-12-15T02:56:38Z

Description

Recently we got interesting use cases from industry about partial upsert.

Users have two event as follows, t is the timestamp column and t1<t2

{t1, a1, b1, c1, d1}
{t2, a2, nil, nil, nil}

user specified field "a" as Overwrite field, and "b", "c", "d" field are empty in the second event.
she expected merge result to be {a2, b1, c1, d1}
However the merge result was {a2, nil, nil, nil} which is the same as full upsert.

The reason of this issue is because she didn't specify the mergers for "b", "c", "d" fields. Thus these fields will use the default behavior, "Overwrite regardless null".

{
  "upsertConfig": {
    "mode": "PARTIAL",
    "partialUpsertStrategies":{
      "a": "OVERWRITE"
    }
  }
}

Her issue can be fixed with the following config, since the "overwrite" merger behavior is "Overwrite unless null".

{
  "upsertConfig": {
    "mode": "PARTIAL",
    "partialUpsertStrategies":{
      "a": "OVERWRITE",
      "b": "OVERWRITE",
      "c": "OVERWRITE",
      "d": "OVERWRITE"
    }
  }
}

In the PR, I added a global strategy. so that user can use "defaultPartialUpsertStrategy", user will not need to set partialUpsertStategy for fields "b", "c", "d" fields.

{
  "upsertConfig": {
    "mode": "PARTIAL",
    "defaultPartialUpsertStrategy": "OVERWRITE",
    "partialUpsertStrategies":{
      "a": "OVERWRITE"
    }
  }
}

NOTE: if we don't specify the overwrite. the overwrite behavior is "Overwrite regardless null".

Upgrade Notes

Does this PR prevent a zero down-time upgrade? (Assume upgrade order: Controller, Broker, Server, Minion)

Yes (Please label as backward-incompat, and complete the section below on Release Notes)

Does this PR fix a zero-downtime upgrade introduced earlier?

Yes (Please label this as backward-incompat, and complete the section below on Release Notes)

Does this PR otherwise need attention when creating release notes? Things to consider:

New configuration options
Deprecation of configurations
Signature changes to public methods/interfaces
New plugins added or old plugins removed

Yes (Please label this PR as release-notes and complete the section on Release Notes)

Release Notes

Documentation

codecov-commenter · 2021-12-15T10:05:45Z

Codecov Report

Merging #7906 (d863947) into master (1d1a7d3) will decrease coverage by 6.45%.
The diff coverage is 60.00%.

@@             Coverage Diff              @@
##             master    #7906      +/-   ##
============================================
- Coverage     71.40%   64.95%   -6.46%     
- Complexity     4223     4231       +8     
============================================
  Files          1597     1554      -43     
  Lines         82903    81100    -1803     
  Branches      12369    12178     -191     
============================================
- Hits          59201    52677    -6524     
- Misses        19689    24665    +4976     
+ Partials       4013     3758     -255

Flag	Coverage Δ
integration1	`?`
integration2	`?`
unittests1	`68.16% <60.00%> (+0.02%)`	⬆️
unittests2	`14.40% <0.00%> (+0.03%)`	⬆️

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

Impacted Files	Coverage Δ
...ata/manager/realtime/RealtimeTableDataManager.java	`11.64% <0.00%> (-57.46%)`	⬇️
...rg/apache/pinot/spi/config/table/UpsertConfig.java	`86.95% <75.00%> (-2.52%)`	⬇️
...not/segment/local/upsert/PartialUpsertHandler.java	`38.70% <100.00%> (+5.37%)`	⬆️
...a/org/apache/pinot/common/metrics/MinionMeter.java	`0.00% <0.00%> (-100.00%)`	⬇️
...g/apache/pinot/common/metrics/ControllerMeter.java	`0.00% <0.00%> (-100.00%)`	⬇️
.../apache/pinot/common/metrics/BrokerQueryPhase.java	`0.00% <0.00%> (-100.00%)`	⬇️
.../apache/pinot/common/metrics/MinionQueryPhase.java	`0.00% <0.00%> (-100.00%)`	⬇️
...he/pinot/common/messages/SegmentReloadMessage.java	`0.00% <0.00%> (-100.00%)`	⬇️
...pinot/core/data/manager/realtime/TimerService.java	`0.00% <0.00%> (-100.00%)`	⬇️
...pinot/minion/exception/TaskCancelledException.java	`0.00% <0.00%> (-100.00%)`	⬇️
... and 381 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 1d1a7d3...d863947. Read the comment docs.

yupeng9 · 2021-12-15T18:14:15Z

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

@@ -87,6 +93,10 @@ public HashFunction getHashFunction() {
    return _partialUpsertStrategies;
  }

+  public Strategy getGlobalUpsertStrategy() {


I assume this strategy applies to partial upsert only, right?

yupeng9 · 2021-12-15T18:15:48Z

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

+    if (schema != null) {
+      for (String dimensionName : schema.getDimensionNames()) {
+        if (!schema.getPrimaryKeyColumns().contains(dimensionName) && !_column2Mergers.containsKey(dimensionName)) {
+          _column2Mergers.put(dimensionName, PartialUpsertMergerFactory.getMerger(globalUpsertStrategy));


I think globalUpsertStrategy can be null? Shall we always assign a default value?

Jackie-Jiang · 2021-12-16T01:43:42Z

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

@@ -53,21 +53,27 @@
  @JsonPropertyDescription("Partial update strategies.")
  private final Map<String, Strategy> _partialUpsertStrategies;

+  @JsonPropertyDescription("global upsert strategy")
+  private final Strategy _globalUpsertStrategy;


Probably more clear if rename to _defaultPartialUpsertStrategy?

Jackie-Jiang · 2021-12-16T01:44:31Z

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

    } else {
      _partialUpsertStrategies = null;
+      _globalUpsertStrategy = Strategy.OVERWRITE;


Probably set to null for full upsert?

Jackie-Jiang · 2021-12-16T01:45:23Z

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

      @JsonProperty("comparisonColumn") @Nullable String comparisonColumn,
      @JsonProperty("hashFunction") @Nullable HashFunction hashFunction) {
    Preconditions.checkArgument(mode != null, "Upsert mode must be configured");
    _mode = mode;

    if (mode == Mode.PARTIAL) {
      _partialUpsertStrategies = partialUpsertStrategies != null ? partialUpsertStrategies : new HashMap<>();
+      _globalUpsertStrategy = globalUpsertStrategy != null ? globalUpsertStrategy : Strategy.OVERWRITE;


This is changing the existing behavior. I do agree OVERWRITE makes more sense to partial upsert, but not sure if we want to introduce backward incompatibility here

Thanks @Jackie-Jiang for review. If we use Ignore (#7907) as default behavior for global mergers, do you think it will solve the backward compatible issue.

I'm not sure how much adoption of partial uspert today, so I feel it's ok to get to best possible default option, whenever we can

Jackie-Jiang · 2021-12-16T01:46:59Z

pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeTest.java

@@ -251,7 +249,8 @@ public void testSerDe()
    {
      // with upsert config
      UpsertConfig upsertConfig =
-          new UpsertConfig(UpsertConfig.Mode.FULL, null, "comparison", UpsertConfig.HashFunction.NONE);
+          new UpsertConfig(UpsertConfig.Mode.FULL, null, UpsertConfig.Strategy.OVERWRITE, "comparison",


For FULL upsert, pass in null as the default strategy as that does not apply? Same for other places

I think so, strategy applies to partial only

icefury71 · 2022-01-13T22:17:07Z

Quick comment: the existing code does not define any default strategy for any column (eg: OVERWRITE). So instead of introducing a new property, can we simply define a default strategy (eg: within UpsertConfig). (of course null handling is still needed). CC @yupeng9 @Jackie-Jiang

deemoliu · 2022-01-14T21:04:29Z

Quick comment: the existing code does not define any default strategy for any column (eg: OVERWRITE). So instead of introducing a new property, can we simply define a default strategy (eg: within UpsertConfig). (of course null handling is still needed). CC @yupeng9 @Jackie-Jiang

Hi @icefury71 thanks for the comment. Yes, there is no default strategy for columns that not specified in upsertConfig. The current behavior for columns not specified is "OVERWRITE even if the fieldValue of the new record is null". In this PR, i updated the default behavior (which is represented by "globalUpsertStrategy") to use "OVERWRITE unless the fieldValue of new record is null", which is the same behavior of the OVERWRITE merger.
I think @Jackie-Jiang mentioned updating this will introduce backward incompatibility and @yupeng9 mentioned we should get the best possible default option.

@icefury71 @yupeng9 @Jackie-Jiang what do you think?

Jackie-Jiang

Please also update the PR description to reflect the new config key defaultPartialUpsertStrategy

Jackie-Jiang · 2022-01-15T02:08:27Z

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

    _helixManager = helixManager;
    _tableNameWithType = tableNameWithType;
    for (Map.Entry<String, UpsertConfig.Strategy> entry : partialUpsertStrategies.entrySet()) {
      _column2Mergers.put(entry.getKey(), PartialUpsertMergerFactory.getMerger(entry.getValue()));
    }
+
+    if (schema != null) {


Schema should never be null here

Jackie-Jiang · 2022-01-15T02:13:54Z

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

    _helixManager = helixManager;
    _tableNameWithType = tableNameWithType;
    for (Map.Entry<String, UpsertConfig.Strategy> entry : partialUpsertStrategies.entrySet()) {
      _column2Mergers.put(entry.getKey(), PartialUpsertMergerFactory.getMerger(entry.getValue()));
    }
+
+    if (schema != null) {
+      for (String dimensionName : schema.getDimensionNames()) {


We should probably include all physical columns (including date time columns) except for primary key columns and comparison column (main time column if no comparison column is configured)

… primary key columns and comparison column

Jackie-Jiang · 2022-01-18T18:31:23Z

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

    _helixManager = helixManager;
    _tableNameWithType = tableNameWithType;
    for (Map.Entry<String, UpsertConfig.Strategy> entry : partialUpsertStrategies.entrySet()) {
      _column2Mergers.put(entry.getKey(), PartialUpsertMergerFactory.getMerger(entry.getValue()));
    }
+    // For all physical columns (including date time columns) except for primary key columns and comparison column.
+    // If no comparison column is configured, use main time column as the comparison time.
+    for (String columnName : schema.getColumnNames()) {


Suggested change

for (String columnName : schema.getColumnNames()) {

for (String columnName : schema.getPhysicalColumnNames()) {

Jackie-Jiang · 2022-01-18T18:33:32Z

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

-      Map<String, UpsertConfig.Strategy> partialUpsertStrategies) {
+  public PartialUpsertHandler(HelixManager helixManager, String tableNameWithType, Schema schema,
+      Map<String, UpsertConfig.Strategy> partialUpsertStrategies, UpsertConfig.Strategy defaultPartialUpsertStrategy,
+      String comparisonColumn) {


Annotate it as nullable

Jackie-Jiang · 2022-01-18T18:35:28Z

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

+            _column2Mergers.put(columnName, PartialUpsertMergerFactory.getMerger(defaultPartialUpsertStrategy));
+          }
+        } else {
+          if (!schema.getDateTimeNames().contains(columnName)) {


The main time column is configured within the table config validationConfig

Jackie-Jiang

LGTM otherwise

Jackie-Jiang · 2022-01-19T17:43:19Z

...core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java

+        partialUpsertHandler = new PartialUpsertHandler(_helixManager, _tableNameWithType, schema,
+            upsertConfig.getPartialUpsertStrategies(), upsertConfig.getDefaultPartialUpsertStrategy(),
+            tableConfig.getValidationConfig().getTimeColumnName(), upsertConfig.getComparisonColumn());


This can be simplified to avoid extra handling in the method (comparisonColumn is non-null this way)

Suggested change

partialUpsertHandler = new PartialUpsertHandler(_helixManager, _tableNameWithType, schema,

upsertConfig.getPartialUpsertStrategies(), upsertConfig.getDefaultPartialUpsertStrategy(),

tableConfig.getValidationConfig().getTimeColumnName(), upsertConfig.getComparisonColumn());

String comparisonColumn = upsertConfig.getComparisonColumn();

if (comparisonColumn == null) {

comparisonColumn = tableConfig.getValidationConfig().getTimeColumnName();

}

partialUpsertHandler = new PartialUpsertHandler(_helixManager, _tableNameWithType, schema,

upsertConfig.getPartialUpsertStrategies(), upsertConfig.getDefaultPartialUpsertStrategy(), comparisonColumn);

yupeng9 · 2022-01-20T00:44:04Z

the compatibility regression seems broken from previous PRs. and the error is not related to this one.

* Add global strategy for partial upsert * fix UT setup * try fix lint * fix tests * handle empty globalUpsertStrategy * update defaultValue for full upsert to be null * update _globalUpsertStrategy to _defaultPartialUpsertStrategy * try fix lint * fix checkstyle * add taskConfig test setup code * include all physical columns (including date time columns) except for primary key columns and comparison column * fix partial upsert handler merge tests * Annotate comparison column as nullable, use main time column * simplified partialUpsertHandler (comparison column is non-null) * fix checkstyle

yupeng9 reviewed Dec 15, 2021

View reviewed changes

Jackie-Jiang reviewed Dec 16, 2021

View reviewed changes

deemoliu added 7 commits January 14, 2022 13:19

Add global strategy for partial upsert

665aa10

fix UT setup

64702d8

try fix lint

2092c9d

fix tests

d73cfa3

handle empty globalUpsertStrategy

c2194a0

update defaultValue for full upsert to be null

5fc69ef

update _globalUpsertStrategy to _defaultPartialUpsertStrategy

d0d18b5

deemoliu force-pushed the qiaochu/global-strategy branch from 413fcca to d0d18b5 Compare January 14, 2022 21:49

deemoliu added 3 commits January 14, 2022 13:52

try fix lint

ac89963

fix checkstyle

6e13c9b

add taskConfig test setup code

30f5705

Jackie-Jiang reviewed Jan 15, 2022

View reviewed changes

deemoliu added 2 commits January 15, 2022 21:30

include all physical columns (including date time columns) except for…

e288884

… primary key columns and comparison column

fix partial upsert handler merge tests

e6afe17

Jackie-Jiang reviewed Jan 18, 2022

View reviewed changes

Annotate comparison column as nullable, use main time column

728e784

yupeng9 approved these changes Jan 19, 2022

View reviewed changes

Jackie-Jiang approved these changes Jan 19, 2022

View reviewed changes

deemoliu added 2 commits January 19, 2022 10:10

simplified partialUpsertHandler (comparison column is non-null)

a44d1a1

fix checkstyle

d863947

yupeng9 merged commit d317c59 into apache:master Jan 20, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add global strategy for partial upsert #7906

Add global strategy for partial upsert #7906

deemoliu commented Dec 15, 2021 •

edited

Loading

codecov-commenter commented Dec 15, 2021 •

edited

Loading

yupeng9 Dec 15, 2021

yupeng9 Dec 15, 2021

Jackie-Jiang Dec 16, 2021

Jackie-Jiang Dec 16, 2021

Jackie-Jiang Dec 16, 2021

deemoliu Jan 12, 2022

yupeng9 Jan 12, 2022

Jackie-Jiang Dec 16, 2021

yupeng9 Jan 12, 2022

icefury71 commented Jan 13, 2022

deemoliu commented Jan 14, 2022 •

edited

Loading

Jackie-Jiang left a comment

Jackie-Jiang Jan 15, 2022

Jackie-Jiang Jan 15, 2022

Jackie-Jiang Jan 18, 2022

Jackie-Jiang Jan 18, 2022

Jackie-Jiang Jan 18, 2022

Jackie-Jiang left a comment

Jackie-Jiang Jan 19, 2022

yupeng9 commented Jan 20, 2022

	for (String columnName : schema.getColumnNames()) {
	for (String columnName : schema.getPhysicalColumnNames()) {

Add global strategy for partial upsert #7906

Add global strategy for partial upsert #7906

Conversation

deemoliu commented Dec 15, 2021 • edited Loading

Description

Upgrade Notes

Release Notes

Documentation

codecov-commenter commented Dec 15, 2021 • edited Loading

Codecov Report

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

icefury71 commented Jan 13, 2022

deemoliu commented Jan 14, 2022 • edited Loading

Jackie-Jiang left a 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

Choose a reason for hiding this comment

yupeng9 commented Jan 20, 2022

deemoliu commented Dec 15, 2021 •

edited

Loading

codecov-commenter commented Dec 15, 2021 •

edited

Loading

deemoliu commented Jan 14, 2022 •

edited

Loading