Jackie-Jiang commented on code in PR #10234: URL: https://github.com/apache/pinot/pull/10234#discussion_r1123834102
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ComparisonColumns.java: ########## @@ -0,0 +1,67 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.segment.local.upsert; + + +@SuppressWarnings({"rawtypes", "unchecked"}) +public class ComparisonColumns implements Comparable<ComparisonColumns> { + private final Comparable[] _values; + private final int _comparableIndex; + + public ComparisonColumns(int comparableIndex, Comparable[] values) { Review Comment: (nit) ```suggestion public ComparisonColumns(Comparable[] values, int comparableIndex) { ``` ########## pinot-spi/src/main/java/org/apache/pinot/spi/config/table/UpsertConfig.java: ########## @@ -154,10 +136,20 @@ public void setDefaultPartialUpsertStrategy(Strategy defaultPartialUpsertStrateg * same primary key, the record with the larger value of the time column is picked as the * latest update. * However, there are cases when users need to use another column to determine the order. - * In such case, you can use option comparisonColumn to override the column used for comparison. + * In such case, you can use option comparisonColumn to override the column used for comparison. When using + * multiple comparison columns, typically in the case of partial upserts, it is expected that input documents will + * each only have a singular non-null comparisonColumn. Multiple non-null values in an input document _will_ result + * in undefined behaviour. Typically, one comparisonColumn is allocated per distinct producer application of data + * in the case where there are multiple producers sinking to the same table. */ + public void setComparisonColumns(List<String> comparisonColumns) { Review Comment: Just realize that certain configs might be handled unexpected. Consider the following configs: ``` { "comparisonColumn": "a", "comparisonColumns": null } ``` ``` { "comparisonColumn": null, "comparisonColumns": ["a"] } ``` ``` { "comparisonColumn": null } ``` ``` { "comparisonColumns": null } ``` In order to handle all of them, we can modify these 2 methods in the following way: ``` public void setComparisonColumns(List<String> comparisonColumns) { if (CollectionUtils.isNotEmpty(comparisonColumns) { _comparisonColumns = comparisonColumns; } } public void setComparisonColumn(String comparisonColumn) { if (comparisonColumn != null) { _comparisonColumns = Collections.singletonList(comparisonColumn); } } ``` ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java: ########## @@ -558,15 +560,41 @@ private RecordInfo getRecordInfo(GenericRow row, int docId) { PrimaryKey primaryKey = row.getPrimaryKey(_schema.getPrimaryKeyColumns()); if (isUpsertEnabled()) { - Object upsertComparisonValue = row.getValue(_upsertComparisonColumn); - Preconditions.checkState(upsertComparisonValue instanceof Comparable, - "Upsert comparison column: %s must be comparable", _upsertComparisonColumn); - return new RecordInfo(primaryKey, docId, (Comparable) upsertComparisonValue); + if (_upsertComparisonColumns.size() > 1) { + return multiComparisonRecordInfo(primaryKey, docId, row); + } + Comparable comparisonValue = (Comparable) row.getValue(_upsertComparisonColumns.get(0)); + return new RecordInfo(primaryKey, docId, comparisonValue); } return new RecordInfo(primaryKey, docId, null); } + private RecordInfo multiComparisonRecordInfo(PrimaryKey primaryKey, int docId, GenericRow row) { + int numComparisonColumns = _upsertComparisonColumns.size(); + Comparable[] comparisonValues = new Comparable[numComparisonColumns]; + + int comparableIndex = -1; + for (int i = 0; i < numComparisonColumns; i++) { + String columnName = _upsertComparisonColumns.get(i); + + if (!row.isNullValue(columnName)) { + // Inbound records may only have exactly 1 non-null value in one of the comparison column i.e. comparison + // columns are mutually exclusive + Preconditions.checkState(comparableIndex == -1, + "Documents must have exactly 1 non-null comparison column value"); + + comparableIndex = i; + + Object comparisonValue = row.getValue(columnName); + Preconditions.checkState(comparisonValue instanceof Comparable, + "Upsert comparison column: %s must be comparable", columnName); + comparisonValues[i] = (Comparable) comparisonValue; + } + } Review Comment: We might need a precondition here to enforce one non-null value set, or it will throw exception in `ComparisonColumns.compareTo()` ``` Preconditions.checkState(comparableIndex != -1, "Documents must have exactly 1 non-null comparison column value"); ``` -- 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
