jackjlli commented on code in PR #14110:
URL: https://github.com/apache/pinot/pull/14110#discussion_r1806858062
##########
pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java:
##########
@@ -100,6 +103,9 @@
public class PinotClientRequest {
private static final Logger LOGGER =
LoggerFactory.getLogger(PinotClientRequest.class);
+ @Inject
+ PinotConfiguration _brokerConf;
Review Comment:
You don't have to inject the broker config here as it's already in the
`BaseBrokerRequestHandler` class.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/ResponseStoreResource.java:
##########
@@ -0,0 +1,190 @@
+/**
+ * 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.broker.api.resources;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiKeyAuthDefinition;
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import io.swagger.annotations.Authorization;
+import io.swagger.annotations.SecurityDefinition;
+import io.swagger.annotations.SwaggerDefinition;
+import java.util.Collection;
+import javax.inject.Inject;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.container.AsyncResponse;
+import javax.ws.rs.container.Suspended;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.common.cursors.AbstractResponseStore;
+import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.response.BrokerResponse;
+import org.apache.pinot.common.response.CursorResponse;
+import org.apache.pinot.core.auth.Actions;
+import org.apache.pinot.core.auth.Authorize;
+import org.apache.pinot.core.auth.ManualAuthorization;
+import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.glassfish.jersey.server.ManagedAsync;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;
+
+
+/**
+ *
+ */
+@Api(tags = "ResponseStore", authorizations = {@Authorization(value =
SWAGGER_AUTHORIZATION_KEY)})
+@SwaggerDefinition(securityDefinition =
@SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
+ HttpHeaders.AUTHORIZATION, in =
ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY,
+ description = "The format of the key is ```\"Basic <token>\" or \"Bearer
<token>\"```")))
+@Path("/responseStore")
+public class ResponseStoreResource {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ResponseStoreResource.class);
+
+ @Inject
+ private PinotConfiguration _brokerConf;
Review Comment:
Same here, you might not need to inject the broker config here.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java:
##########
@@ -437,6 +456,22 @@ private BrokerResponse executeSqlQuery(ObjectNode
sqlRequestJson, HttpRequesterI
if (forceUseMultiStage) {
sqlNodeAndOptions.setExtraOptions(ImmutableMap.of(Request.QueryOptionKey.USE_MULTISTAGE_ENGINE,
"true"));
}
+ if (getCursor != null && getCursor) {
+ if (numRows == null) {
+ numRows =
_brokerConf.getProperty(CommonConstants.CursorConfigs.QUERY_RESULT_SIZE,
+ CommonConstants.CursorConfigs.DEFAULT_QUERY_RESULT_SIZE);
+ }
+
+ if (numRows > CommonConstants.CursorConfigs.MAX_QUERY_RESULT_SIZE) {
Review Comment:
Why do we want to introduce the `MAX_QUERY_RESULT_SIZE` here if there is no
such restriction on the generic queries?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/cursors/ResponseStoreCleaner.java:
##########
@@ -0,0 +1,206 @@
+/**
+ * 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.controller.cursors;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hc.client5.http.classic.methods.HttpDelete;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.http.MultiHttpRequest;
+import org.apache.pinot.common.http.MultiHttpRequestResponse;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.response.CursorResponse;
+import org.apache.pinot.common.response.broker.CursorResponseNative;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.api.resources.InstanceInfo;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import
org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.TimeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ResponseStoreCleaner extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ResponseStoreCleaner.class);
+ private static final int TIMEOUT_MS = 3000;
+ private static final String QUERY_RESULT_STORE = "%s://%s:%d/responseStore";
+ private static final String DELETE_QUERY_RESULT =
"%s://%s:%d/responseStore/%s";
+ public static final String CLEAN_AT_TIME =
"response.store.cleaner.clean.at.ms";
+ private final ControllerConf _controllerConf;
+ private final Executor _executor;
+ private final PoolingHttpClientConnectionManager _connectionManager;
+
+ public ResponseStoreCleaner(ControllerConf config, PinotHelixResourceManager
pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerMetrics
controllerMetrics, Executor executor,
+ PoolingHttpClientConnectionManager connectionManager) {
+ super("ResultStoreCleaner", getFrequencyInSeconds(config),
getInitialDelayInSeconds(config),
+ pinotHelixResourceManager, leadControllerManager, controllerMetrics);
+ _controllerConf = config;
+ _executor = executor;
+ _connectionManager = connectionManager;
+ }
+
+ private static long getInitialDelayInSeconds(ControllerConf config) {
+ long initialDelay = config.getPeriodicTaskInitialDelayInSeconds();
+ String resultStoreCleanerTaskInitialDelay =
+
config.getProperty(CommonConstants.CursorConfigs.RESULT_STORE_CLEANER_INITIAL_DELAY);
+ if (resultStoreCleanerTaskInitialDelay != null) {
+ initialDelay =
TimeUnit.SECONDS.convert(TimeUtils.convertPeriodToMillis(resultStoreCleanerTaskInitialDelay),
+ TimeUnit.MILLISECONDS);
+ }
+ return initialDelay;
+ }
+
+ private static long getFrequencyInSeconds(ControllerConf config) {
+ long frequencyInSeconds = 0;
+ String resultStoreCleanerTaskPeriod =
+
config.getProperty(CommonConstants.CursorConfigs.RESULT_STORE_CLEANER_FREQUENCY_PERIOD);
+ if (resultStoreCleanerTaskPeriod != null) {
+ frequencyInSeconds =
TimeUnit.SECONDS.convert(TimeUtils.convertPeriodToMillis(resultStoreCleanerTaskPeriod),
+ TimeUnit.MILLISECONDS);
+ }
+
+ return frequencyInSeconds;
+ }
+
+ @Override
+ protected void processTables(List<String> tableNamesWithType, Properties
periodicTaskProperties) {
+ long cleanAtMs = System.currentTimeMillis();
+ String cleanAtMsStr = periodicTaskProperties.getProperty(CLEAN_AT_TIME);
+ if (cleanAtMsStr != null) {
+ cleanAtMs = Long.parseLong(cleanAtMsStr);
+ }
+ doClean(cleanAtMs);
+ }
+
+ public void doClean(long currentTime) {
+ List<InstanceConfig> brokerList =
_pinotHelixResourceManager.getAllBrokerInstanceConfigs();
+ Map<String, InstanceInfo> brokers = brokerList.stream().collect(
+ Collectors.toMap(x -> getInstanceKey(x.getHostName(), x.getPort()),
+ x -> new InstanceInfo(x.getInstanceName(), x.getHostName(),
Integer.parseInt(x.getPort()))));
+
+ try {
+ Map<String, List<CursorResponseNative>> brokerResponses =
getAllQueryResults(brokers, Collections.emptyMap());
Review Comment:
IIUC, it seems every controller will try to fetch all the results from all
the brokers, is it kind of a waste of resources to do that? Could we distribute
the cleanup workloads to controllers instead?
##########
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:
Does that mean every time a read request will fetch the whole result table
into memory first and then pick the page from the memory?
##########
pinot-common/src/main/java/org/apache/pinot/common/response/CursorResponse.java:
##########
@@ -0,0 +1,132 @@
+/**
+ * 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.response;
+
+
+public interface CursorResponse extends BrokerResponse {
+
+ void setBrokerHost(String brokerHost);
+
+ /**
+ * get hostname of the processing broker
+ * @return String containing the hostname
+ */
+ String getBrokerHost();
+
+ void setBrokerPort(int brokerPort);
+
+ /**
+ * get port of the processing broker
+ * @return int containing the port.
+ */
+ int getBrokerPort();
+
+ /**
+ * Set the starting offset of result table slice
+ * @param offset Offset of the result table slice
+ */
+ void setOffset(int offset);
+
+ /**
+ * Set the number of rows in the result table slice.
+ * @param numRows Number of rows in the result table slice
+ */
+ void setNumRows(int numRows);
+
+ /**
+ * Current offset in the query result.
+ * Starts from 0.
+ * @return current offset.
+ */
+ int getOffset();
Review Comment:
nit: put the methods for the same object together, i.e. swapping the
`setNumRows()` and `getOffset()` methods
--
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
