Jackie-Jiang commented on code in PR #11826:
URL: https://github.com/apache/pinot/pull/11826#discussion_r1373544701
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java:
##########
@@ -64,53 +62,42 @@ public PartialUpsertHandler(Schema schema, Map<String,
UpsertConfig.Strategy> pa
* For example, overwrite merger will only override the prev value if the
new value is not null.
* Null values will override existing values if not configured. They can be
ignored by using ignoreMerger.
*
- * @param indexSegment the segment of the last derived full record during
ingestion.
- * @param docId the docId of the last derived full record during ingestion
in the segment.
+ * @param prevRecord wrapper for previous record, which lazily reads column
values of previous row and caches for
+ * re-reads.
* @param newRecord the new consumed record.
*/
- public void merge(IndexSegment indexSegment, int docId, GenericRow
newRecord) {
- for (String column : indexSegment.getColumnNames()) {
+ public void merge(LazyRow prevRecord, GenericRow newRecord) {
+ for (String column : prevRecord.getColumnNames()) {
if (!_primaryKeyColumns.contains(column)) {
PartialUpsertMerger merger = _column2Mergers.getOrDefault(column,
_defaultPartialUpsertMerger);
// Non-overwrite mergers
// (1) If the value of the previous is null value, skip merging and
use the new value
// (2) Else If the value of new value is null, use the previous value
(even for comparison columns).
// (3) Else If the column is not a comparison column, we applied the
merged value to it.
if (!(merger instanceof OverwriteMerger)) {
- try (PinotSegmentColumnReader pinotSegmentColumnReader = new
PinotSegmentColumnReader(indexSegment, column)) {
- if (!pinotSegmentColumnReader.isNull(docId)) {
- Object previousValue = pinotSegmentColumnReader.getValue(docId);
- if (newRecord.isNullValue(column)) {
- // Note that we intentionally want to overwrite any previous
_comparisonColumn value in the case of
- // using
- // multiple comparison columns. We never apply a merge
function to it, rather we just take any/all
- // non-null comparison column values from the previous record,
and the sole non-null comparison column
- // value from the new record.
- newRecord.putValue(column, previousValue);
- newRecord.removeNullValueField(column);
- } else if (!_comparisonColumns.contains(column)) {
- newRecord.putValue(column, merger.merge(previousValue,
newRecord.getValue(column)));
- }
+ if (!prevRecord.isNullValue(column)) {
Review Comment:
^^
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java:
##########
@@ -51,6 +52,9 @@
@ThreadSafe
public class ConcurrentMapPartitionUpsertMetadataManager extends
BasePartitionUpsertMetadataManager {
+ // Used to initialize a reference to previous row for merging in partial
upsert
+ protected final LazyRow _reusePreviousRow = new LazyRow();
Review Comment:
```suggestion
private final LazyRow _reusePreviousRow = new LazyRow();
```
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java:
##########
@@ -64,53 +62,42 @@ public PartialUpsertHandler(Schema schema, Map<String,
UpsertConfig.Strategy> pa
* For example, overwrite merger will only override the prev value if the
new value is not null.
* Null values will override existing values if not configured. They can be
ignored by using ignoreMerger.
*
- * @param indexSegment the segment of the last derived full record during
ingestion.
- * @param docId the docId of the last derived full record during ingestion
in the segment.
+ * @param prevRecord wrapper for previous record, which lazily reads column
values of previous row and caches for
+ * re-reads.
* @param newRecord the new consumed record.
*/
- public void merge(IndexSegment indexSegment, int docId, GenericRow
newRecord) {
- for (String column : indexSegment.getColumnNames()) {
+ public void merge(LazyRow prevRecord, GenericRow newRecord) {
+ for (String column : prevRecord.getColumnNames()) {
if (!_primaryKeyColumns.contains(column)) {
PartialUpsertMerger merger = _column2Mergers.getOrDefault(column,
_defaultPartialUpsertMerger);
// Non-overwrite mergers
// (1) If the value of the previous is null value, skip merging and
use the new value
// (2) Else If the value of new value is null, use the previous value
(even for comparison columns).
// (3) Else If the column is not a comparison column, we applied the
merged value to it.
if (!(merger instanceof OverwriteMerger)) {
- try (PinotSegmentColumnReader pinotSegmentColumnReader = new
PinotSegmentColumnReader(indexSegment, column)) {
- if (!pinotSegmentColumnReader.isNull(docId)) {
- Object previousValue = pinotSegmentColumnReader.getValue(docId);
- if (newRecord.isNullValue(column)) {
- // Note that we intentionally want to overwrite any previous
_comparisonColumn value in the case of
- // using
- // multiple comparison columns. We never apply a merge
function to it, rather we just take any/all
- // non-null comparison column values from the previous record,
and the sole non-null comparison column
- // value from the new record.
- newRecord.putValue(column, previousValue);
- newRecord.removeNullValueField(column);
- } else if (!_comparisonColumns.contains(column)) {
- newRecord.putValue(column, merger.merge(previousValue,
newRecord.getValue(column)));
- }
+ if (!prevRecord.isNullValue(column)) {
+ Object prevValue = prevRecord.getValue(column);
+ if (newRecord.isNullValue(column)) {
+ // Note that we intentionally want to overwrite any previous
_comparisonColumn value in the case of
+ // using
+ // multiple comparison columns. We never apply a merge function
to it, rather we just take any/all
+ // non-null comparison column values from the previous record,
and the sole non-null comparison column
+ // value from the new record.
+ newRecord.putValue(column, prevValue);
+ newRecord.removeNullValueField(column);
+ } else if (!_comparisonColumns.contains(column)) {
+ newRecord.putValue(column, merger.merge(prevValue,
newRecord.getValue(column)));
+ }
}
Review Comment:
^^
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java:
##########
@@ -0,0 +1,101 @@
+/**
+ * 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.segment.readers;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.spi.IndexSegment;
+
+
+/**
+ * <p>A wrapper class to read column values of a row for a given {@link
IndexSegment} and docId.<br>
+ * The advantage of having wrapper over segment and docId is column values are
read only when
+ * {@link LazyRow#getValue(String)} is invoked.
+ * This is useful to reduce the disk reads incurred due to loading the
complete previous row during merge step.
+ *
+ * <p>The LazyRow has an internal state and should not be used concurrently.
To reuse the LazyRow, create an instance
+ * using no arg constructor and re-initialise using {@link
LazyRow#init(IndexSegment, int)}
+ */
+public class LazyRow {
+ private final Map<String, Object> _fieldToValueMap = new HashMap<>();
+ private final Set<String> _nullValueFields = new HashSet<>();
+ private IndexSegment _segment;
+ private int _docId;
+
+ public LazyRow() {
+ }
+
+ public void init(IndexSegment segment, int docId) {
+ clear();
+ _segment = segment;
+ _docId = docId;
+ }
+
+ /**
+ * Computes a field's value in an indexed row.
+ * @param fieldName
+ * @return Returns value or null for persisted null values
+ */
+ @Nullable public Object getValue(String fieldName) {
Review Comment:
```suggestion
@Nullable
public Object getValue(String fieldName) {
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
