mcvsubbu commented on a change in pull request #5394: URL: https://github.com/apache/incubator-pinot/pull/5394#discussion_r443148746
########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; Review comment: Please document if this method is to be called once during system startup, or once for every table. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { Review comment: the interface definition should go into pinot-spi ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; + + /** + * create a {@link IndexSegmentCallback} object to allow SegmentDataManager to create proper IndexSegment that supports + * either append/upsert mode + * + * In append-tables callback, this method will create a DefaultIndexSegmentCallback + * In upsert-tables callback, this method will create a UpsertDataManagerCallbackImpl + * + * The {@link IndexSegmentCallback} will be used in the constructor of IndexSegment + */ + IndexSegmentCallback getIndexSegmentCallback(); + + /** + * process the row after transformation in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the GenericRow has been transformed by RecordTransformer and before it is indexed by + * IndexSegmentImpl, to ensure we can provide other necessary data to the segment metadata. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will attach the offset object into the GenericRow object. + * + * The reason we need this particular logic is that in upsert table, we need to add offset data to the physical data + * this will help us to apply the update events from key coordinator to upsert table correctly as the offset + * is used as the index to identify which record's virtual column we want to update + * + * @param row the row of newly ingested and transformed data from upstream + * @param offset the offset of this particular pinot record + */ + void processTransformedRow(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * process the row after indexing in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the MutableSegmentImpl has done the indexing of the current row in its physical storage + * + * In append-tables callback, this method will do nothing Review comment: All realtime tables are append tables, so let us keep the terminology right. We can remove this comment, or restate it to say that the call needs to happen only if Upsert is ON for the table. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/IndexSegmentCallback.java ########## @@ -0,0 +1,77 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.indexsegment.IndexSegment; +import org.apache.pinot.core.segment.index.column.ColumnIndexContainer; +import org.apache.pinot.core.segment.index.metadata.SegmentMetadata; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.util.Map; + +/** + * Component inject to {@link IndexSegment} for handling extra logic for + * upsert-enabled pinot ingestion mode + */ [email protected] +public interface IndexSegmentCallback { + + /** + * Initialize the callback from {@link IndexSegment}. This happens in constructor of IndexSegment implementation class + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will initialize the necessary virtual columns for upsert table + * (offset mapping, $validFrom, $validUntil columns) + * + * this method ensure that virtual columns index (forward/reverted index) can be created properly and load + * any existing data for the virtual column from local storage + * + * @param segmentMetadata the metadata associated with the current segment + * @param columnIndexContainerMap mapping of necessary column name and the associate columnIndexContainer + */ + void init(SegmentMetadata segmentMetadata, Map<String, ColumnIndexContainer> columnIndexContainerMap); + + /** + * Perform operation from the callback for the given row after it has been processed and index. + * This method happens after indexing finished in MutableSegmentImpl.index(GenericRow, RowMetadata) method + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will build offset mapping and update virtual columns for the column if necessary + * + * This method is similar to + * {@link org.apache.pinot.core.data.manager.callback.DataManagerCallback#postIndexProcessing(GenericRow, StreamPartitionMsgOffset)} + * However, the other method don't have information to docID and it is necessary for upsert virtual columns + * to have these information to build the forward index and offset columns to build mapping between offset -> docId + * we also might need to update virtual column forward-index for the newly ingested row + * + * @param row the current pinot row we just indexed into the current IndexSegment + * @param docId the docId of this record + */ + void postProcessRecords(GenericRow row, int docId); + + /** + * Retrieve information related to an upsert-enable segment virtual column for debug purpose + * + * @param offset the offset of the record we are trying to get the virtual columnn data for + * @return string representation of the virtual column data information + */ + String getVirtualColumnInfo(StreamPartitionMsgOffset offset); Review comment: You should get the schema, tableconfig etc via the init(), so I am not sure why we need this vitual columninfo call. Also, why should Pinot code call this method? Let us remove the debug calls for now, and we can add back things we need once we have the overall setup right. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/IndexSegmentCallback.java ########## @@ -0,0 +1,77 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.indexsegment.IndexSegment; +import org.apache.pinot.core.segment.index.column.ColumnIndexContainer; +import org.apache.pinot.core.segment.index.metadata.SegmentMetadata; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.util.Map; + +/** + * Component inject to {@link IndexSegment} for handling extra logic for + * upsert-enabled pinot ingestion mode + */ [email protected] +public interface IndexSegmentCallback { + + /** + * Initialize the callback from {@link IndexSegment}. This happens in constructor of IndexSegment implementation class + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will initialize the necessary virtual columns for upsert table + * (offset mapping, $validFrom, $validUntil columns) + * + * this method ensure that virtual columns index (forward/reverted index) can be created properly and load + * any existing data for the virtual column from local storage + * + * @param segmentMetadata the metadata associated with the current segment + * @param columnIndexContainerMap mapping of necessary column name and the associate columnIndexContainer + */ + void init(SegmentMetadata segmentMetadata, Map<String, ColumnIndexContainer> columnIndexContainerMap); + + /** + * Perform operation from the callback for the given row after it has been processed and index. + * This method happens after indexing finished in MutableSegmentImpl.index(GenericRow, RowMetadata) method + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will build offset mapping and update virtual columns for the column if necessary + * + * This method is similar to + * {@link org.apache.pinot.core.data.manager.callback.DataManagerCallback#postIndexProcessing(GenericRow, StreamPartitionMsgOffset)} + * However, the other method don't have information to docID and it is necessary for upsert virtual columns + * to have these information to build the forward index and offset columns to build mapping between offset -> docId + * we also might need to update virtual column forward-index for the newly ingested row + * + * @param row the current pinot row we just indexed into the current IndexSegment + * @param docId the docId of this record + */ + void postProcessRecords(GenericRow row, int docId); Review comment: We should make ONE callback with whatever information needed. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; + + /** + * create a {@link IndexSegmentCallback} object to allow SegmentDataManager to create proper IndexSegment that supports + * either append/upsert mode + * + * In append-tables callback, this method will create a DefaultIndexSegmentCallback + * In upsert-tables callback, this method will create a UpsertDataManagerCallbackImpl + * + * The {@link IndexSegmentCallback} will be used in the constructor of IndexSegment + */ + IndexSegmentCallback getIndexSegmentCallback(); + + /** + * process the row after transformation in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the GenericRow has been transformed by RecordTransformer and before it is indexed by + * IndexSegmentImpl, to ensure we can provide other necessary data to the segment metadata. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will attach the offset object into the GenericRow object. + * + * The reason we need this particular logic is that in upsert table, we need to add offset data to the physical data + * this will help us to apply the update events from key coordinator to upsert table correctly as the offset + * is used as the index to identify which record's virtual column we want to update + * + * @param row the row of newly ingested and transformed data from upstream + * @param offset the offset of this particular pinot record + */ + void processTransformedRow(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * process the row after indexing in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the MutableSegmentImpl has done the indexing of the current row in its physical storage + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will emit an event to the message queue that will deliver the event to + * key coordinator. + * + * This method ensures that we can emit the metadata for an new entry that pinot just indexed to its internal storage + * and let key coordinator to be able to consume those events to process the updates correctly + * + * @param row the row we just index in the current segment + * @param offset the offset associated with the current row + */ + void postIndexProcessing(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * perform any necessary finalize operation after the consumer loop finished in LLRealtimeSegmentDataManager.consumeLoop(...) + * method. It happens after the consumerloop exits the loop and reached the end criteria. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will flush the queue producer to ensure all pending messages are deliverd + * to the queue between pinot server and pinot key-coordinator + * + * this method will ensure that pinot server can send all events to key coordinator eventually before a segment + * is committed. If this does not happen we might lose data in case of machine failure. + */ + void postConsumeLoop(); + + /** + * perform any necessary clean up operation when the SegmentDataManager called its destroy() method. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will notify segmentUpdater to remove any registered reference for this + * dataManagerCallback. + * + * this method will ensure that segmentUpdater can keep track of which dataManager is still alive in the current pinot + * server so it can dispatch appropriate update events to only the alive pinot data manager + */ + void destroy(); Review comment: onSegmentDestroy(). Also, you can use this callback for the previous one. Or, just key off of previous one. If I understand your proposal better, you are also getting a callback when the IndexSegment is loaded. In that case, you can remove this call as well as the previous one, and just use the callback when the index segment is loaded. It will also serve the use case when the segment goes from OFFLINE to ONLINE without going through CONSUMING state. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/IndexSegmentCallback.java ########## @@ -0,0 +1,77 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.indexsegment.IndexSegment; +import org.apache.pinot.core.segment.index.column.ColumnIndexContainer; +import org.apache.pinot.core.segment.index.metadata.SegmentMetadata; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.util.Map; + +/** + * Component inject to {@link IndexSegment} for handling extra logic for + * upsert-enabled pinot ingestion mode + */ [email protected] +public interface IndexSegmentCallback { + + /** + * Initialize the callback from {@link IndexSegment}. This happens in constructor of IndexSegment implementation class Review comment: You should not indicate which class should call this method. Instead, indicate on what condition you want this call to happen. Is it before the segment is loaded succesfsully? Or after the segment is loaded successfully ? If you start the name of the method as "on" then we can get the answer. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/IndexSegmentCallback.java ########## @@ -0,0 +1,77 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.indexsegment.IndexSegment; +import org.apache.pinot.core.segment.index.column.ColumnIndexContainer; +import org.apache.pinot.core.segment.index.metadata.SegmentMetadata; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.util.Map; + +/** + * Component inject to {@link IndexSegment} for handling extra logic for + * upsert-enabled pinot ingestion mode + */ [email protected] +public interface IndexSegmentCallback { + + /** + * Initialize the callback from {@link IndexSegment}. This happens in constructor of IndexSegment implementation class + * + * In append-tables callback, this method will do nothing Review comment: same comment as before, will not repeat. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; + + /** + * create a {@link IndexSegmentCallback} object to allow SegmentDataManager to create proper IndexSegment that supports + * either append/upsert mode + * + * In append-tables callback, this method will create a DefaultIndexSegmentCallback + * In upsert-tables callback, this method will create a UpsertDataManagerCallbackImpl + * + * The {@link IndexSegmentCallback} will be used in the constructor of IndexSegment + */ + IndexSegmentCallback getIndexSegmentCallback(); Review comment: It is not clear why we are returning a callback object from within another callback object, but let us get the coarse details right and let me understand the architecture before I propose anything. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; + + /** + * create a {@link IndexSegmentCallback} object to allow SegmentDataManager to create proper IndexSegment that supports + * either append/upsert mode + * + * In append-tables callback, this method will create a DefaultIndexSegmentCallback + * In upsert-tables callback, this method will create a UpsertDataManagerCallbackImpl + * + * The {@link IndexSegmentCallback} will be used in the constructor of IndexSegment + */ + IndexSegmentCallback getIndexSegmentCallback(); + + /** + * process the row after transformation in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the GenericRow has been transformed by RecordTransformer and before it is indexed by + * IndexSegmentImpl, to ensure we can provide other necessary data to the segment metadata. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will attach the offset object into the GenericRow object. + * + * The reason we need this particular logic is that in upsert table, we need to add offset data to the physical data + * this will help us to apply the update events from key coordinator to upsert table correctly as the offset + * is used as the index to identify which record's virtual column we want to update + * + * @param row the row of newly ingested and transformed data from upstream + * @param offset the offset of this particular pinot record + */ + void processTransformedRow(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * process the row after indexing in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the MutableSegmentImpl has done the indexing of the current row in its physical storage + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will emit an event to the message queue that will deliver the event to + * key coordinator. + * + * This method ensures that we can emit the metadata for an new entry that pinot just indexed to its internal storage + * and let key coordinator to be able to consume those events to process the updates correctly + * + * @param row the row we just index in the current segment + * @param offset the offset associated with the current row + */ + void postIndexProcessing(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * perform any necessary finalize operation after the consumer loop finished in LLRealtimeSegmentDataManager.consumeLoop(...) + * method. It happens after the consumerloop exits the loop and reached the end criteria. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will flush the queue producer to ensure all pending messages are deliverd + * to the queue between pinot server and pinot key-coordinator + * + * this method will ensure that pinot server can send all events to key coordinator eventually before a segment + * is committed. If this does not happen we might lose data in case of machine failure. + */ + void postConsumeLoop(); Review comment: consumeLoop() is an implementation. If we rename the method, and then this API name will have no meaning. Please indicate clearly _when_ you want this callback, and we can decide on a name. Do you want the callback to happen after successful CONSMING to ONLINE state transition (i.e. the completed segment has been loaded in memory, and we are ready to get rid of the CONSUMING segment memory for this segment). Or, do you really want it to be called even if we may catch up later on with some other server because we got a CATCHUP from the controller? (this can be VERY hard to verify correctness, and I would question why that is needed) ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { Review comment: A suggestion to make it more readable. All callback methods should be named starting with "on". So, `onSegmentConsumingToOnline` or `onSegmentOfflineToOnline`, or `onSegmentStartConsuming`, etc. ########## File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/callback/DataManagerCallback.java ########## @@ -0,0 +1,106 @@ +/** + * 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.core.data.manager.callback; + +import org.apache.pinot.core.data.manager.SegmentDataManager; +import org.apache.pinot.spi.annotations.InterfaceStability; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.stream.MessageBatch; +import org.apache.pinot.spi.stream.StreamPartitionMsgOffset; + +import java.io.IOException; + +/** + * Component inject to {@link org.apache.pinot.core.data.manager.SegmentDataManager} for handling extra logic for + * upsert-enabled pinot ingestion mode. + */ [email protected] +public interface DataManagerCallback { + + void init() throws IOException; + + /** + * create a {@link IndexSegmentCallback} object to allow SegmentDataManager to create proper IndexSegment that supports + * either append/upsert mode + * + * In append-tables callback, this method will create a DefaultIndexSegmentCallback + * In upsert-tables callback, this method will create a UpsertDataManagerCallbackImpl + * + * The {@link IndexSegmentCallback} will be used in the constructor of IndexSegment + */ + IndexSegmentCallback getIndexSegmentCallback(); + + /** + * process the row after transformation in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the GenericRow has been transformed by RecordTransformer and before it is indexed by + * IndexSegmentImpl, to ensure we can provide other necessary data to the segment metadata. + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will attach the offset object into the GenericRow object. + * + * The reason we need this particular logic is that in upsert table, we need to add offset data to the physical data + * this will help us to apply the update events from key coordinator to upsert table correctly as the offset + * is used as the index to identify which record's virtual column we want to update + * + * @param row the row of newly ingested and transformed data from upstream + * @param offset the offset of this particular pinot record + */ + void processTransformedRow(GenericRow row, StreamPartitionMsgOffset offset); + + /** + * process the row after indexing in LLRealtimeSegmentDataManager.processStreamEvents(...) method + * it happens after the MutableSegmentImpl has done the indexing of the current row in its physical storage + * + * In append-tables callback, this method will do nothing + * In upsert-tables callback, this method will emit an event to the message queue that will deliver the event to + * key coordinator. + * + * This method ensures that we can emit the metadata for an new entry that pinot just indexed to its internal storage + * and let key coordinator to be able to consume those events to process the updates correctly + * + * @param row the row we just index in the current segment + * @param offset the offset associated with the current row + */ + void postIndexProcessing(GenericRow row, StreamPartitionMsgOffset offset); Review comment: rename as `onRealtimeRowIndexed()`. Indicate whether this method should be called in LLC or HLC or both. Indicate whether the method should be called if the row is aggregated (i.e. in the case where `aggregateMetrics` is turned on and the row is not indexed, but aggregated. See https://docs.pinot.apache.org/basics/components/table#tableindexconfig) ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
