Jackie-Jiang commented on code in PR #11826: URL: https://github.com/apache/pinot/pull/11826#discussion_r1372075906
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; Review Comment: (convention) Move the `private` fields after the `private final` fields ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; + private int _docId; + private final Map<String, Object> _fieldToValueMap = new HashMap<>(); + private final Set<String> _nullValueFields = new HashSet<>(); + + + public LazyRow() { + } + + public LazyRow(IndexSegment segment, int docId) { + _segment = segment; + _docId = docId; + } Review Comment: Suggest removing them and only allow empty constructor. In order to use it, `init()` must be called first ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; + private int _docId; + private final Map<String, Object> _fieldToValueMap = new HashMap<>(); + private final Set<String> _nullValueFields = new HashSet<>(); + + + public LazyRow() { + } + + public LazyRow(IndexSegment segment, int docId) { + _segment = segment; + _docId = docId; + } + Review Comment: (format) Remove extra empty line ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; + private int _docId; + private final Map<String, Object> _fieldToValueMap = new HashMap<>(); + private final Set<String> _nullValueFields = new HashSet<>(); + + + public LazyRow() { + } + + public LazyRow(IndexSegment segment, int docId) { + _segment = segment; + _docId = docId; + } + + + public void init(IndexSegment segment, int docId) { + this.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 + */ + public Object getValue(String fieldName) { Review Comment: Mark it as nullable ```suggestion @Nullable public Object getValue(String fieldName) { ``` ########## 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: We can directly call `prevRecord.getValue(column)` and check if the `prevValue` is `null` to reduce one extra lookup ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java: ########## @@ -86,6 +87,9 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps // If upsertTTL enabled, we will keep track of largestSeenComparisonValue to compute expired segments. protected volatile double _largestSeenComparisonValue; + // Used to initialize a reference to previous row for merging in partial upsert + protected LazyRow _reusePreviousRow; Review Comment: This can be moved to `ConcurrentMapPartitionUpsertMetadataManager`, and can be made `final` ########## 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: (format) Wrong indentation ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; + private int _docId; + private final Map<String, Object> _fieldToValueMap = new HashMap<>(); + private final Set<String> _nullValueFields = new HashSet<>(); + Review Comment: (format) Remove extra empty line ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/LazyRow.java: ########## @@ -0,0 +1,107 @@ +/** + * 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 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 IndexSegment _segment; + private int _docId; + private final Map<String, Object> _fieldToValueMap = new HashMap<>(); + private final Set<String> _nullValueFields = new HashSet<>(); + + + public LazyRow() { + } + + public LazyRow(IndexSegment segment, int docId) { + _segment = segment; + _docId = docId; + } + + + public void init(IndexSegment segment, int docId) { + this.clear(); Review Comment: (nit) ```suggestion clear(); ``` -- 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]
