vrajat commented on code in PR #14110: URL: https://github.com/apache/pinot/pull/14110#discussion_r1807284193
########## pinot-common/src/main/java/org/apache/pinot/common/cursors/AbstractResponseStore.java: ########## @@ -0,0 +1,186 @@ +/** + * 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.common.cursors; + +import java.util.ArrayList; +import java.util.List; +import org.apache.pinot.common.metrics.BrokerMeter; +import org.apache.pinot.common.metrics.BrokerMetrics; +import org.apache.pinot.common.response.CursorResponse; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.spi.cursors.ResponseSerde; +import org.apache.pinot.spi.cursors.ResponseStore; +import org.apache.pinot.spi.env.PinotConfiguration; + + +public abstract class AbstractResponseStore implements ResponseStore { + + protected BrokerMetrics _brokerMetrics; + + /** + * Initialize the store. + * @param config Configuration of the store. + * @param brokerMetrics Metrics utility to track cursor metrics. + * @param responseSerde The Serde object to use to serialize/deserialize the responses + */ + public abstract void init(PinotConfiguration config, BrokerMetrics brokerMetrics, ResponseSerde responseSerde) + throws Exception; + + /** + * Write a CursorResponse + * @param requestId Request ID of the response + * @param response The response to write + * @throws Exception Thrown if there is any error while writing the response + */ + protected abstract void writeResponse(String requestId, CursorResponse response) + throws Exception; + + /** + * Write a @link{ResultTable} to the store + * @param requestId Request ID of the response + * @param resultTable The @link{ResultTable} of the query + * @throws Exception Thrown if there is any error while writing the result table. + * @return Returns the number of bytes written + */ + protected abstract long writeResultTable(String requestId, ResultTable resultTable) + throws Exception; + + /** + * Read the response (excluding the @link{ResultTable}) from the store + * @param requestId Request ID of the response + * @return CursorResponse (without the @link{ResultTable}) + * @throws Exception Thrown if there is any error while reading the response + */ + public abstract CursorResponse readResponse(String requestId) + throws Exception; + + /** + * Read the @link{ResultTable} of a query response + * @param requestId Request ID of the query + * @return @link{ResultTable} of the query + * @throws Exception Thrown if there is any error while reading the result table + */ + protected abstract ResultTable readResultTable(String requestId) + throws Exception; + + protected abstract boolean deleteResponseImpl(String requestId) + throws Exception; + + /** + * Stores the response in the store. @link{CursorResponse} and @link{ResultTable} are stored separately. + * @param response Response to be stored + * @throws Exception Thrown if there is any error while storing the response. + */ + public void storeResponse(CursorResponse response) + throws Exception { + String requestId = response.getRequestId(); + + try { + long bytesWritten = writeResultTable(requestId, response.getResultTable()); + + // Remove the resultTable from the response as it is serialized in a data file. + response.setResultTable(null); + response.setBytesWritten(bytesWritten); + writeResponse(requestId, response); + _brokerMetrics.addMeteredGlobalValue(BrokerMeter.CURSOR_RESULT_STORE_SIZE, bytesWritten); + } catch (Exception e) { + _brokerMetrics.addMeteredGlobalValue(BrokerMeter.CURSOR_WRITE_EXCEPTION, 1); + deleteResponse(requestId); + throw e; + } + } + + /** + * Reads the response from the store and populates it with a slice of the @link{ResultTable} + * @param requestId Request ID of the query + * @param offset Offset of the result slice + * @param numRows Number of rows required in the slice + * @return A CursorResponse with a slice of the @link{ResultTable} + * @throws Exception Thrown if there is any error during the operation. + */ + public CursorResponse handleCursorRequest(String requestId, int offset, int numRows) + throws Exception { + + CursorResponse response; + ResultTable resultTable; + + try { + response = readResponse(requestId); + } catch (Exception e) { + _brokerMetrics.addMeteredGlobalValue(BrokerMeter.CURSOR_READ_EXCEPTION, 1); + throw e; + } + + int totalTableRows = response.getNumRowsResultSet(); + + if (totalTableRows == 0 && offset == 0) { + // If sum records is 0, then result set is empty. + response.setResultTable(null); + response.setOffset(0); + response.setNumRows(0); + return response; + } else if (offset >= totalTableRows) { + throw new RuntimeException("Offset " + offset + " is greater than totalRecords " + totalTableRows); + } + + long fetchStartTime = System.currentTimeMillis(); + try { + resultTable = readResultTable(requestId); + } catch (Exception e) { + _brokerMetrics.addMeteredGlobalValue(BrokerMeter.CURSOR_READ_EXCEPTION, 1); + throw e; + } + + int sliceEnd = offset + numRows; + if (sliceEnd > totalTableRows) { + sliceEnd = totalTableRows; + numRows = sliceEnd - offset; + } + + response.setResultTable( + new ResultTable(resultTable.getDataSchema(), resultTable.getRows().subList(offset, sliceEnd))); Review Comment: This implementation does read the whole result table into memory. Does LinkedIn implementation support seeking within the result table ? Do you have suggestions on how can the interface be changed to also support seek ? -- 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
