You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pinot.apache.org by "vrajat (via GitHub)" <gi...@apache.org> on 2024/11/14 10:24:32 UTC

[PR] Add a server API to list segments out of date with table config [pinot]

vrajat opened a new pull request, #14451:
URL: https://github.com/apache/pinot/pull/14451

   Fix for #14450 
   
   Instructions:
   1. The PR has to be tagged with at least one of the following labels (*):
      1. `feature`
      2. `bugfix`
      3. `performance`
      4. `ui`
      5. `backward-incompat`
      6. `release-notes` (**)
   2. Remove these instructions before publishing the PR.
    
   (*) Other labels to consider:
   - `testing`
   - `dependencies`
   - `docker`
   - `kubernetes`
   - `observability`
   - `security`
   - `code-style`
   - `extension-point`
   - `refactor`
   - `cleanup`
   
   (**) Use `release-notes` label for scenarios like:
   - New configuration options
   - Deprecation of configurations
   - Signature changes to public methods/interfaces
   - New plugins added or old plugins removed
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1862161067


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java:
##########
@@ -323,4 +323,12 @@ default void onConsumingToDropped(String segmentNameStr) {
    */
   default void onConsumingToOnline(String segmentNameStr) {
   }
+
+  /**
+   * Return list of segment names that need to be refreshed along with reason.
+   * @param tableConfig Table Config of the table
+   * @param schema Schema of the table
+   * @return List of {@link NeedRefreshResponse} with segment names and reason for refresh
+   */
+  List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema);

Review Comment:
   This is done



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish commented on PR #14451:
URL: https://github.com/apache/pinot/pull/14451#issuecomment-2509678253

   > The core feature in this PR is a function `BaseTableDataManager.getSegmentsForRefresh` which finds all the segments of a table that requires a refresh. It accepts a Table Config and Schema and compares that with the current state of a segment.
   > 
   > This functionality will be used by other components such as minion tasks to decide which segments have to be refreshed.
   > 
   > The function is exposed by the following server API:
   > 
   > ```
   > /tables/{tableName}/segments/needRefresh
   > 
   > Response is:
   > [
   >   {
   >      "segmentName":"...",
   >      "reason":"..."
   >   },
   > ...
   > ]
   > ```
   > 
   > **segmentName** is the name of the segment. **reason** contains a simple message for why the segment needs to be refreshed.
   > 
   > There is also a controller API that consolidates the responses from all servers.
   > 
   > ```
   > /segments/orders_OFFLINE/needRefresh
   > 
   > Response is:
   > {
   >   "<server name>" : [{
   >      "segmentName":"...",
   >      "reason":"..."
   >   }],
   > ...
   > ]
   > 
   > Example:
   > 
   > {
   >   "http://192.168.0.111:7500/tables/orders_OFFLINE/segments/needRefresh__1": [
   >     {
   >       "segmentName": "orders_OFFLINE_0",
   >       "reason": "column added"
   >     }
   >   ]
   > }
   > ```
   > 
   > Fix for #14450
   
   Please update description to getStaleSegments


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1855428433


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/NeedRefreshResponse.java:
##########
@@ -0,0 +1,62 @@
+/**
+ * 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.data.manager;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+
+/**
+ * Encapsulates the response to get list of segments that need to be refreshed. The response also contains the reason
+ * why a segment has to be refreshed.
+ */
+public class NeedRefreshResponse {
+  private final String _segmentName;
+  private final boolean _needRefresh;
+  private final String _reason;

Review Comment:
   Good idea to add reason as well !



##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,39 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    int failedParses = 0;
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint
+        serverResponses.put(streamResponse.getKey(),
+            JsonUtils.stringToObject(streamResponse.getValue(),
+                new TypeReference<List<NeedRefreshResponse>>() { }));
+      } catch (Exception e) {
+        failedParses++;
+        LOGGER.error("Unable to parse server {} response due to an error: ", streamResponse.getKey(), e);
+      }
+    }
+    if (failedParses != 0) {
+      LOGGER.error("Unable to parse server {} / {} response due to an error: ", failedParses, serverURLs.size());

Review Comment:
   Please add table name to this msg



##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,39 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    int failedParses = 0;
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint
+        serverResponses.put(streamResponse.getKey(),
+            JsonUtils.stringToObject(streamResponse.getValue(),
+                new TypeReference<List<NeedRefreshResponse>>() { }));
+      } catch (Exception e) {
+        failedParses++;
+        LOGGER.error("Unable to parse server {} response due to an error: ", streamResponse.getKey(), e);

Review Comment:
   Please add table name to this msg



##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -891,6 +892,31 @@ public String getTableReloadMetadata(
     }
   }
 
+  @GET
+  @Path("segments/{tableNameWithType}/needRefresh")
+  @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType", action = Actions.Table.GET_METADATA)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Gets a list of segments that need to be refreshed from servers hosting the table", notes =
+      "Gets a list of segments that need to be refreshed from servers hosting the table")
+  public Map<String, List<NeedRefreshResponse>> getTableRefreshMetadata(

Review Comment:
   How will the user know if we got partial Vs full response. 
   Would it be useful to find out missing segments for which we never got a response?
   After the getSegmentsForRefreshFromServer, we can get the list of segments from ideal state and do a diff. If there's a way to propagate that list, we can just retry that list in the subsequent call.
   Or at least the client knows that it did not get a complete list.



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   Good point about acquireAll. Yes thats one way. If we have a large table/with a large number of segments, this check can take time. However by the time you get to the segment, that can be released? I guess acquireSegment will return null in that case. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "KKcorps (via GitHub)" <gi...@apache.org>.

KKcorps commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1853913780


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   acquireAllSegments also increases the ref count
   so we need to decrease it as well once we get the list



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1861536819


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java:
##########
@@ -323,4 +323,12 @@ default void onConsumingToDropped(String segmentNameStr) {
    */
   default void onConsumingToOnline(String segmentNameStr) {
   }
+
+  /**
+   * Return list of segment names that need to be refreshed along with reason.
+   * @param tableConfig Table Config of the table
+   * @param schema Schema of the table
+   * @return List of {@link NeedRefreshResponse} with segment names and reason for refresh
+   */
+  List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema);

Review Comment:
   `getDivergedSegments` or `getStaleSegments` ?



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1862165234


##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -891,6 +892,31 @@ public String getTableReloadMetadata(
     }
   }
 
+  @GET
+  @Path("segments/{tableNameWithType}/needRefresh")
+  @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType", action = Actions.Table.GET_METADATA)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Gets a list of segments that need to be refreshed from servers hosting the table", notes =
+      "Gets a list of segments that need to be refreshed from servers hosting the table")
+  public Map<String, List<NeedRefreshResponse>> getTableRefreshMetadata(

Review Comment:
   I have added an explicit class that tracks server -> response. The response also has an error message. Using null to denote error was icky. Check `TableStaleSegmentResponse`



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a server API to list segments out of date with table config [pinot]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #14451:
URL: https://github.com/apache/pinot/pull/14451#issuecomment-2484717463

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/14451?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   All modified and coverable lines are covered by tests :white_check_mark:
   > Project coverage is 0.00%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`04666ca`)](https://app.codecov.io/gh/apache/pinot/commit/04666ca17bdba7de018c28173fbb81aa3cda2a94?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   > Report is 1350 commits behind head on master.
   
   > :exclamation:  There is a different number of reports uploaded between BASE (59551e4) and HEAD (04666ca). Click for more details.
   > 
   > <details><summary>HEAD has 53 uploads less than BASE</summary>
   >
   >| Flag | BASE (59551e4) | HEAD (04666ca) |
   >|------|------|------|
   >|integration|7|1|
   >|integration2|3|1|
   >|temurin|12|1|
   >|java-21|7|1|
   >|skip-bytebuffers-true|3|1|
   >|skip-bytebuffers-false|7|0|
   >|unittests|5|0|
   >|unittests1|2|0|
   >|java-11|5|0|
   >|unittests2|3|0|
   >|integration1|2|0|
   >|custom-integration1|2|0|
   ></details>
   
   <details><summary>Additional details and impacted files</summary>
   
   
   ```diff
   @@              Coverage Diff              @@
   ##             master   #14451       +/-   ##
   =============================================
   - Coverage     61.75%    0.00%   -61.76%     
   =============================================
     Files          2436        3     -2433     
     Lines        133233        6   -133227     
     Branches      20636        0    -20636     
   =============================================
   - Hits          82274        0    -82274     
   + Misses        44911        6    -44905     
   + Partials       6048        0     -6048     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-0.01%)` | :arrow_down: |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
   | [java-11](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-21](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.63%)` | :arrow_down: |
   | [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-27.73%)` | :arrow_down: |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.76%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [unittests2](https://app.codecov.io/gh/apache/pinot/pull/14451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/14451?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   
   ----
   🚨 Try these New Features:
   
   - [Flaky Tests Detection](https://docs.codecov.com/docs/test-result-ingestion-beta) - Detect and resolve failed and flaky tests
   - [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis) - Avoid shipping oversized bundles


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1855827237


##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -891,6 +892,31 @@ public String getTableReloadMetadata(
     }
   }
 
+  @GET
+  @Path("segments/{tableNameWithType}/needRefresh")
+  @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType", action = Actions.Table.GET_METADATA)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Gets a list of segments that need to be refreshed from servers hosting the table", notes =
+      "Gets a list of segments that need to be refreshed from servers hosting the table")
+  public Map<String, List<NeedRefreshResponse>> getTableRefreshMetadata(

Review Comment:
   The message is a map of `server`->`list of responses`. Errors will be at the server level. So the map will have a null when there is an error. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1854163236


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();
+    for (SegmentDataManager segmentDataManager : segmentDataManagers) {
+      NeedRefreshResponse response = needRefresh(tableConfig, schema, segmentDataManager);
+      if (response.isNeedRefresh()) {
+        segmentsRequiringRefresh.add(response);
+      }
+    }
+
+    return segmentsRequiringRefresh;
+  }
+
+  protected NeedRefreshResponse needRefresh(TableConfig tableConfig, Schema schema,
+      SegmentDataManager segmentDataManager) {
+    String tableNameWithType = tableConfig.getTableName();
+    Map<String, FieldIndexConfigs> indexConfigsMap =
+        FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig, schema);
+
+    String segmentName = segmentDataManager.getSegmentName();
+    IndexSegment segment = segmentDataManager.getSegment();
+    SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+    Set<String> segmentPhysicalColumns = segment.getPhysicalColumnNames();
+
+    // Time column changed
+    String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+    if (timeColumn != null) {
+      if (segmentMetadata.getTimeColumn() == null || !segmentMetadata.getTimeColumn().equals(timeColumn)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: time column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "time column");
+      }
+    }
+
+    List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+    String sortedColumn = CollectionUtils.isNotEmpty(sortedColumns) ? sortedColumns.get(0) : null;
+
+    String partitionColumn = null;
+    ColumnPartitionConfig partitionConfig = null;
+    SegmentPartitionConfig segmentPartitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
+    // NOTE: Partition can only be enabled on a single column
+    if (segmentPartitionConfig != null && segmentPartitionConfig.getColumnPartitionMap().size() == 1) {
+      Map.Entry<String, ColumnPartitionConfig> entry =
+          segmentPartitionConfig.getColumnPartitionMap().entrySet().iterator().next();
+      partitionColumn = entry.getKey();
+      partitionConfig = entry.getValue();
+    }
+
+    Set<String> columnsInSegment = segmentMetadata.getAllColumns();
+
+    // Column is added
+    if (!columnsInSegment.containsAll(schema.getPhysicalColumnNames())) {
+      LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column added", tableNameWithType, segmentName);
+      return new NeedRefreshResponse(segmentName, true, "column added");
+    }
+
+    // Get Index configuration for the Table Config
+    Set<String> noDictionaryColumns =
+        FieldIndexConfigsUtil.columnsWithIndexDisabled(StandardIndexes.dictionary(), indexConfigsMap);
+    Set<String> bloomFilters =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.bloomFilter(), indexConfigsMap);
+    Set<String> jsonIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.json(), indexConfigsMap);
+    Set<String> invertedIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.inverted(), indexConfigsMap);
+    Set<String> nullValueVectorIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.nullValueVector(), indexConfigsMap);
+    Set<String> rangeIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.range(), indexConfigsMap);
+    Set<String> h3Indexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.h3(), indexConfigsMap);
+    Set<String> fstIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.fst(), indexConfigsMap);
+    Set<String> textIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.text(), indexConfigsMap);
+    List<StarTreeIndexConfig> starTreeIndexConfigsFromTableConfig =
+        tableConfig.getIndexingConfig().getStarTreeIndexConfigs();
+
+    // Get the index configuration for StarTree index from segment metadata as JsonNode.
+    List<StarTreeV2> starTreeIndexMetadata = segment.getStarTrees();
+
+    // Generate StarTree index builder config from the segment metadata.
+    List<StarTreeV2BuilderConfig> builderConfigFromSegmentMetadata = new ArrayList<>();
+    if (starTreeIndexMetadata != null) {
+      for (StarTreeV2 starTreeV2 : starTreeIndexMetadata) {
+        builderConfigFromSegmentMetadata.add(StarTreeV2BuilderConfig.fromMetadata(starTreeV2.getMetadata()));
+      }
+    }
+
+    // Generate StarTree index builder configs from the table config.
+    //TODO: RV This maybe using the wrong function. It is not using the table's schema
+    List<StarTreeV2BuilderConfig> builderConfigFromTableConfigs =
+        StarTreeBuilderUtils.generateBuilderConfigs(starTreeIndexConfigsFromTableConfig,
+            tableConfig.getIndexingConfig().isEnableDefaultStarTree(), segmentMetadata);
+
+    // TODO: RV Test
+    // Check if there is a mismatch between the StarTree index builder configs from the table config and the segment
+    // metadata.
+    if (!StarTreeBuilderUtils.areStarTreeBuilderConfigListsEqual(builderConfigFromTableConfigs,
+        builderConfigFromSegmentMetadata)) {
+      return new NeedRefreshResponse(segmentName, true, "startree index");
+    }
+
+    for (String columnName : segmentPhysicalColumns) {
+      ColumnMetadata columnMetadata = segmentMetadata.getColumnMetadataFor(columnName);
+      FieldSpec fieldSpecInSchema = schema.getFieldSpecFor(columnName);
+      DataSource source = segment.getDataSource(columnName);
+      Preconditions.checkNotNull(columnMetadata);
+      Preconditions.checkNotNull(source);
+
+      // Column is deleted
+      if (fieldSpecInSchema == null) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column deleted", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "column deleted: " + columnName);
+      }
+
+      // Field type changed
+      if (columnMetadata.getFieldType().compareTo(fieldSpecInSchema.getFieldType()) != 0) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: field type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "field type changed: " + columnName);
+      }
+
+      // Data type changed
+      if (!columnMetadata.getDataType().equals(fieldSpecInSchema.getDataType())) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: data type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "data type changed: " + columnName);
+      }
+
+      // SV/MV changed
+      if (columnMetadata.isSingleValue() != fieldSpecInSchema.isSingleValueField()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: single / multi value", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "single / multi value changed: " + columnName);
+      }
+
+      // TODO: detect if an index changes from Dictionary to Variable Length Dictionary or vice versa.
+      // TODO: RV TEST
+      boolean colHasDictionary = columnMetadata.hasDictionary();
+      // Encoding changed
+      if (colHasDictionary == noDictionaryColumns.contains(columnName)) {
+        // Check if dictionary update is needed
+        // 1. If the segment metadata has dictionary enabled and table has it disabled, its incompatible and refresh is
+        // needed.
+        // 2. If segment metadata has dictionary disabled, check if it has to be overridden. If not overridden,
+        // refresh is needed, since table has it enabled.
+        boolean incompatible = colHasDictionary || DictionaryIndexType.ignoreDictionaryOverride(
+            tableConfig.getIndexingConfig().isOptimizeDictionary(),
+            tableConfig.getIndexingConfig().isOptimizeDictionaryForMetrics(),
+            tableConfig.getIndexingConfig().getNoDictionarySizeRatioThreshold(),
+            tableConfig.getIndexingConfig().getNoDictionaryCardinalityRatioThreshold(), fieldSpecInSchema,
+            indexConfigsMap.get(columnName), columnMetadata.getCardinality(), columnMetadata.getTotalNumberOfEntries());
+        if (incompatible) {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: dictionary encoding,", tableNameWithType,
+              segmentName);
+          return new NeedRefreshResponse(segmentName, true, "dictionary encoding changed: " + columnName);
+        } else {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, no change as dictionary overrides applied to col: {}",
+              tableNameWithType, segmentName, columnName);
+        }
+      }
+
+      // Sorted column not sorted
+      if (columnName.equals(sortedColumn) && !columnMetadata.isSorted()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: sort column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "sort column changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getBloomFilter()) == bloomFilters.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: bloom filter changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "bloom filter changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getJsonIndex()) == jsonIndex.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: json index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "json index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getTextIndex()) == textIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: text index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "text index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getFSTIndex()) == fstIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: fst index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "fst index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getH3Index()) == h3Indexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: h3 index changed", tableNameWithType,

Review Comment:
   Added `columnName` to all segments. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1861544111


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java:
##########
@@ -323,4 +323,12 @@ default void onConsumingToDropped(String segmentNameStr) {
    */
   default void onConsumingToOnline(String segmentNameStr) {
   }
+
+  /**
+   * Return list of segment names that need to be refreshed along with reason.
+   * @param tableConfig Table Config of the table
+   * @param schema Schema of the table
+   * @return List of {@link NeedRefreshResponse} with segment names and reason for refresh
+   */
+  List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema);

Review Comment:
   **getStaleSegments** sounds good to me . Thanks Rajat. So we need to change the name for other data structures too (like NeedRefreshInfo)? 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1861554378


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java:
##########
@@ -323,4 +323,12 @@ default void onConsumingToDropped(String segmentNameStr) {
    */
   default void onConsumingToOnline(String segmentNameStr) {
   }
+
+  /**
+   * Return list of segment names that need to be refreshed along with reason.
+   * @param tableConfig Table Config of the table
+   * @param schema Schema of the table
+   * @return List of {@link NeedRefreshResponse} with segment names and reason for refresh
+   */
+  List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema);

Review Comment:
   Yes. All of them have to be changed.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "KKcorps (via GitHub)" <gi...@apache.org>.

KKcorps commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1865194105


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   Cool, I am fine with that!



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1865604773


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,276 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<StaleSegment> getStaleSegments(TableConfig tableConfig, Schema schema) {
+    List<StaleSegment> staleSegments = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   Done



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish merged PR #14451:
URL: https://github.com/apache/pinot/pull/14451


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on PR #14451:
URL: https://github.com/apache/pinot/pull/14451#issuecomment-2506102938

   Two more outstanding items:
   * Add a test where server returns an error. Is there a way to mock that ? 
   * Add tests for startree index


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1861534450


##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,35 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint

Review Comment:
   The key used to be a URL instead of the server name. Fixed now. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1864869862


##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableStaleSegmentResponse.java:
##########
@@ -0,0 +1,67 @@
+/**
+ * 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.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import org.apache.pinot.segment.local.data.manager.StaleSegment;
+
+
+public class TableStaleSegmentResponse {
+  private final List<StaleSegment> _staleSegmentList;
+  private final boolean _isValidResponse;
+  private final String _errorMessage;
+
+  @JsonCreator
+  public TableStaleSegmentResponse(@JsonProperty("staleSegmentList") List<StaleSegment> staleSegmentList,
+      @JsonProperty("validResponse") boolean isValidResponse,
+      @JsonProperty("errorMessage") String errorMessage) {
+    _staleSegmentList = staleSegmentList;
+    _isValidResponse = isValidResponse;
+    _errorMessage = errorMessage;

Review Comment:
   Thanks for adding the error message. 



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,276 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<StaleSegment> getStaleSegments(TableConfig tableConfig, Schema schema) {
+    List<StaleSegment> staleSegments = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   if it makes sense to record the compute time, please log that. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1854163236


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();
+    for (SegmentDataManager segmentDataManager : segmentDataManagers) {
+      NeedRefreshResponse response = needRefresh(tableConfig, schema, segmentDataManager);
+      if (response.isNeedRefresh()) {
+        segmentsRequiringRefresh.add(response);
+      }
+    }
+
+    return segmentsRequiringRefresh;
+  }
+
+  protected NeedRefreshResponse needRefresh(TableConfig tableConfig, Schema schema,
+      SegmentDataManager segmentDataManager) {
+    String tableNameWithType = tableConfig.getTableName();
+    Map<String, FieldIndexConfigs> indexConfigsMap =
+        FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig, schema);
+
+    String segmentName = segmentDataManager.getSegmentName();
+    IndexSegment segment = segmentDataManager.getSegment();
+    SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+    Set<String> segmentPhysicalColumns = segment.getPhysicalColumnNames();
+
+    // Time column changed
+    String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+    if (timeColumn != null) {
+      if (segmentMetadata.getTimeColumn() == null || !segmentMetadata.getTimeColumn().equals(timeColumn)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: time column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "time column");
+      }
+    }
+
+    List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+    String sortedColumn = CollectionUtils.isNotEmpty(sortedColumns) ? sortedColumns.get(0) : null;
+
+    String partitionColumn = null;
+    ColumnPartitionConfig partitionConfig = null;
+    SegmentPartitionConfig segmentPartitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
+    // NOTE: Partition can only be enabled on a single column
+    if (segmentPartitionConfig != null && segmentPartitionConfig.getColumnPartitionMap().size() == 1) {
+      Map.Entry<String, ColumnPartitionConfig> entry =
+          segmentPartitionConfig.getColumnPartitionMap().entrySet().iterator().next();
+      partitionColumn = entry.getKey();
+      partitionConfig = entry.getValue();
+    }
+
+    Set<String> columnsInSegment = segmentMetadata.getAllColumns();
+
+    // Column is added
+    if (!columnsInSegment.containsAll(schema.getPhysicalColumnNames())) {
+      LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column added", tableNameWithType, segmentName);
+      return new NeedRefreshResponse(segmentName, true, "column added");
+    }
+
+    // Get Index configuration for the Table Config
+    Set<String> noDictionaryColumns =
+        FieldIndexConfigsUtil.columnsWithIndexDisabled(StandardIndexes.dictionary(), indexConfigsMap);
+    Set<String> bloomFilters =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.bloomFilter(), indexConfigsMap);
+    Set<String> jsonIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.json(), indexConfigsMap);
+    Set<String> invertedIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.inverted(), indexConfigsMap);
+    Set<String> nullValueVectorIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.nullValueVector(), indexConfigsMap);
+    Set<String> rangeIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.range(), indexConfigsMap);
+    Set<String> h3Indexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.h3(), indexConfigsMap);
+    Set<String> fstIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.fst(), indexConfigsMap);
+    Set<String> textIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.text(), indexConfigsMap);
+    List<StarTreeIndexConfig> starTreeIndexConfigsFromTableConfig =
+        tableConfig.getIndexingConfig().getStarTreeIndexConfigs();
+
+    // Get the index configuration for StarTree index from segment metadata as JsonNode.
+    List<StarTreeV2> starTreeIndexMetadata = segment.getStarTrees();
+
+    // Generate StarTree index builder config from the segment metadata.
+    List<StarTreeV2BuilderConfig> builderConfigFromSegmentMetadata = new ArrayList<>();
+    if (starTreeIndexMetadata != null) {
+      for (StarTreeV2 starTreeV2 : starTreeIndexMetadata) {
+        builderConfigFromSegmentMetadata.add(StarTreeV2BuilderConfig.fromMetadata(starTreeV2.getMetadata()));
+      }
+    }
+
+    // Generate StarTree index builder configs from the table config.
+    //TODO: RV This maybe using the wrong function. It is not using the table's schema
+    List<StarTreeV2BuilderConfig> builderConfigFromTableConfigs =
+        StarTreeBuilderUtils.generateBuilderConfigs(starTreeIndexConfigsFromTableConfig,
+            tableConfig.getIndexingConfig().isEnableDefaultStarTree(), segmentMetadata);
+
+    // TODO: RV Test
+    // Check if there is a mismatch between the StarTree index builder configs from the table config and the segment
+    // metadata.
+    if (!StarTreeBuilderUtils.areStarTreeBuilderConfigListsEqual(builderConfigFromTableConfigs,
+        builderConfigFromSegmentMetadata)) {
+      return new NeedRefreshResponse(segmentName, true, "startree index");
+    }
+
+    for (String columnName : segmentPhysicalColumns) {
+      ColumnMetadata columnMetadata = segmentMetadata.getColumnMetadataFor(columnName);
+      FieldSpec fieldSpecInSchema = schema.getFieldSpecFor(columnName);
+      DataSource source = segment.getDataSource(columnName);
+      Preconditions.checkNotNull(columnMetadata);
+      Preconditions.checkNotNull(source);
+
+      // Column is deleted
+      if (fieldSpecInSchema == null) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column deleted", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "column deleted: " + columnName);
+      }
+
+      // Field type changed
+      if (columnMetadata.getFieldType().compareTo(fieldSpecInSchema.getFieldType()) != 0) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: field type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "field type changed: " + columnName);
+      }
+
+      // Data type changed
+      if (!columnMetadata.getDataType().equals(fieldSpecInSchema.getDataType())) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: data type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "data type changed: " + columnName);
+      }
+
+      // SV/MV changed
+      if (columnMetadata.isSingleValue() != fieldSpecInSchema.isSingleValueField()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: single / multi value", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "single / multi value changed: " + columnName);
+      }
+
+      // TODO: detect if an index changes from Dictionary to Variable Length Dictionary or vice versa.
+      // TODO: RV TEST
+      boolean colHasDictionary = columnMetadata.hasDictionary();
+      // Encoding changed
+      if (colHasDictionary == noDictionaryColumns.contains(columnName)) {
+        // Check if dictionary update is needed
+        // 1. If the segment metadata has dictionary enabled and table has it disabled, its incompatible and refresh is
+        // needed.
+        // 2. If segment metadata has dictionary disabled, check if it has to be overridden. If not overridden,
+        // refresh is needed, since table has it enabled.
+        boolean incompatible = colHasDictionary || DictionaryIndexType.ignoreDictionaryOverride(
+            tableConfig.getIndexingConfig().isOptimizeDictionary(),
+            tableConfig.getIndexingConfig().isOptimizeDictionaryForMetrics(),
+            tableConfig.getIndexingConfig().getNoDictionarySizeRatioThreshold(),
+            tableConfig.getIndexingConfig().getNoDictionaryCardinalityRatioThreshold(), fieldSpecInSchema,
+            indexConfigsMap.get(columnName), columnMetadata.getCardinality(), columnMetadata.getTotalNumberOfEntries());
+        if (incompatible) {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: dictionary encoding,", tableNameWithType,
+              segmentName);
+          return new NeedRefreshResponse(segmentName, true, "dictionary encoding changed: " + columnName);
+        } else {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, no change as dictionary overrides applied to col: {}",
+              tableNameWithType, segmentName, columnName);
+        }
+      }
+
+      // Sorted column not sorted
+      if (columnName.equals(sortedColumn) && !columnMetadata.isSorted()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: sort column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "sort column changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getBloomFilter()) == bloomFilters.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: bloom filter changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "bloom filter changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getJsonIndex()) == jsonIndex.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: json index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "json index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getTextIndex()) == textIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: text index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "text index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getFSTIndex()) == fstIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: fst index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "fst index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getH3Index()) == h3Indexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: h3 index changed", tableNameWithType,

Review Comment:
   Added `columnName` to all debug logs



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1854166400


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   I am releasing the segments now. To acquire a segment, how do I get the segment name ? One option is to acquire all segments, convert that into a list of names and then reacquire one by one. Any other options ? 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "KKcorps (via GitHub)" <gi...@apache.org>.

KKcorps commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1853916253


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   Also, since we are running a for loop, wouldn't a better way be to acquire each segment one by one



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1855815497


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   Yes. Also new segments will be missed. That is a reasonable compromise ? 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a server API to list segments out of date with table config [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on PR #14451:
URL: https://github.com/apache/pinot/pull/14451#issuecomment-2477951638

   This API is for refresh while IIUC that PR is for reload. I should make it
   clear that this PR is for refresh in the title.
   
   On Fri, Nov 15, 2024 at 10:00 AM Yash Mayya ***@***.***>
   wrote:
   
   > What's the difference between this and #13789
   > <https://github.com/apache/pinot/pull/13789>?
   >
   > —
   > Reply to this email directly, view it on GitHub
   > <https://github.com/apache/pinot/pull/14451#issuecomment-2477939670>, or
   > unsubscribe
   > <https://github.com/notifications/unsubscribe-auth/AAMP7GRJ3H2O3PHX2PI2CD32AV2HDAVCNFSM6AAAAABRYRMF3KVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDINZXHEZTSNRXGA>
   > .
   > You are receiving this because you authored the thread.Message ID:
   > ***@***.***>
   >
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a server API to list segments out of date with table config [pinot]

Posted by "yashmayya (via GitHub)" <gi...@apache.org>.

yashmayya commented on PR #14451:
URL: https://github.com/apache/pinot/pull/14451#issuecomment-2477939670

   What's the difference between this and https://github.com/apache/pinot/pull/13789?


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.

swaminathanmanish commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1860973442


##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,35 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint

Review Comment:
   What does this TODO mean?



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java:
##########
@@ -323,4 +323,12 @@ default void onConsumingToDropped(String segmentNameStr) {
    */
   default void onConsumingToOnline(String segmentNameStr) {
   }
+
+  /**
+   * Return list of segment names that need to be refreshed along with reason.
+   * @param tableConfig Table Config of the table
+   * @param schema Schema of the table
+   * @return List of {@link NeedRefreshResponse} with segment names and reason for refresh
+   */
+  List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema);

Review Comment:
   
   Would a name like getSegmentsOutOfSync be better. We use both reload and refresh to bring the segment in sync with table config, so we can separate the action from current state. 
   



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();

Review Comment:
   @vrajat and I chatted . I think we can start with current code (acquireAllSegments) . These checks are local and in-memory and should be quick. We can record the start/end times for this operation. If needed we can optimize this further. 
   



##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -891,6 +892,31 @@ public String getTableReloadMetadata(
     }
   }
 
+  @GET
+  @Path("segments/{tableNameWithType}/needRefresh")
+  @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType", action = Actions.Table.GET_METADATA)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Gets a list of segments that need to be refreshed from servers hosting the table", notes =
+      "Gets a list of segments that need to be refreshed from servers hosting the table")
+  public Map<String, List<NeedRefreshResponse>> getTableRefreshMetadata(

Review Comment:
   Discussed offline. 
   1. For a successful entry we will have non-empty/empty list indicating segments that need to be refreshed
   2. For unsuccessful entry, we will have null response. 



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "KKcorps (via GitHub)" <gi...@apache.org>.

KKcorps commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1853838884


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1046,6 +1064,261 @@ public boolean needReloadSegments()
     return needReload;
   }
 
+  @Override
+  public List<NeedRefreshResponse> getSegmentsForRefresh(TableConfig tableConfig, Schema schema) {
+    List<NeedRefreshResponse> segmentsRequiringRefresh = new ArrayList<>();
+    List<SegmentDataManager> segmentDataManagers = acquireAllSegments();
+    for (SegmentDataManager segmentDataManager : segmentDataManagers) {
+      NeedRefreshResponse response = needRefresh(tableConfig, schema, segmentDataManager);
+      if (response.isNeedRefresh()) {
+        segmentsRequiringRefresh.add(response);
+      }
+    }
+
+    return segmentsRequiringRefresh;
+  }
+
+  protected NeedRefreshResponse needRefresh(TableConfig tableConfig, Schema schema,
+      SegmentDataManager segmentDataManager) {
+    String tableNameWithType = tableConfig.getTableName();
+    Map<String, FieldIndexConfigs> indexConfigsMap =
+        FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig, schema);
+
+    String segmentName = segmentDataManager.getSegmentName();
+    IndexSegment segment = segmentDataManager.getSegment();
+    SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+    Set<String> segmentPhysicalColumns = segment.getPhysicalColumnNames();
+
+    // Time column changed
+    String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+    if (timeColumn != null) {
+      if (segmentMetadata.getTimeColumn() == null || !segmentMetadata.getTimeColumn().equals(timeColumn)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: time column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "time column");
+      }
+    }
+
+    List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+    String sortedColumn = CollectionUtils.isNotEmpty(sortedColumns) ? sortedColumns.get(0) : null;
+
+    String partitionColumn = null;
+    ColumnPartitionConfig partitionConfig = null;
+    SegmentPartitionConfig segmentPartitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
+    // NOTE: Partition can only be enabled on a single column
+    if (segmentPartitionConfig != null && segmentPartitionConfig.getColumnPartitionMap().size() == 1) {
+      Map.Entry<String, ColumnPartitionConfig> entry =
+          segmentPartitionConfig.getColumnPartitionMap().entrySet().iterator().next();
+      partitionColumn = entry.getKey();
+      partitionConfig = entry.getValue();
+    }
+
+    Set<String> columnsInSegment = segmentMetadata.getAllColumns();
+
+    // Column is added
+    if (!columnsInSegment.containsAll(schema.getPhysicalColumnNames())) {
+      LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column added", tableNameWithType, segmentName);
+      return new NeedRefreshResponse(segmentName, true, "column added");
+    }
+
+    // Get Index configuration for the Table Config
+    Set<String> noDictionaryColumns =
+        FieldIndexConfigsUtil.columnsWithIndexDisabled(StandardIndexes.dictionary(), indexConfigsMap);
+    Set<String> bloomFilters =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.bloomFilter(), indexConfigsMap);
+    Set<String> jsonIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.json(), indexConfigsMap);
+    Set<String> invertedIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.inverted(), indexConfigsMap);
+    Set<String> nullValueVectorIndex =
+        FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.nullValueVector(), indexConfigsMap);
+    Set<String> rangeIndex = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.range(), indexConfigsMap);
+    Set<String> h3Indexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.h3(), indexConfigsMap);
+    Set<String> fstIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.fst(), indexConfigsMap);
+    Set<String> textIndexes = FieldIndexConfigsUtil.columnsWithIndexEnabled(StandardIndexes.text(), indexConfigsMap);
+    List<StarTreeIndexConfig> starTreeIndexConfigsFromTableConfig =
+        tableConfig.getIndexingConfig().getStarTreeIndexConfigs();
+
+    // Get the index configuration for StarTree index from segment metadata as JsonNode.
+    List<StarTreeV2> starTreeIndexMetadata = segment.getStarTrees();
+
+    // Generate StarTree index builder config from the segment metadata.
+    List<StarTreeV2BuilderConfig> builderConfigFromSegmentMetadata = new ArrayList<>();
+    if (starTreeIndexMetadata != null) {
+      for (StarTreeV2 starTreeV2 : starTreeIndexMetadata) {
+        builderConfigFromSegmentMetadata.add(StarTreeV2BuilderConfig.fromMetadata(starTreeV2.getMetadata()));
+      }
+    }
+
+    // Generate StarTree index builder configs from the table config.
+    //TODO: RV This maybe using the wrong function. It is not using the table's schema
+    List<StarTreeV2BuilderConfig> builderConfigFromTableConfigs =
+        StarTreeBuilderUtils.generateBuilderConfigs(starTreeIndexConfigsFromTableConfig,
+            tableConfig.getIndexingConfig().isEnableDefaultStarTree(), segmentMetadata);
+
+    // TODO: RV Test
+    // Check if there is a mismatch between the StarTree index builder configs from the table config and the segment
+    // metadata.
+    if (!StarTreeBuilderUtils.areStarTreeBuilderConfigListsEqual(builderConfigFromTableConfigs,
+        builderConfigFromSegmentMetadata)) {
+      return new NeedRefreshResponse(segmentName, true, "startree index");
+    }
+
+    for (String columnName : segmentPhysicalColumns) {
+      ColumnMetadata columnMetadata = segmentMetadata.getColumnMetadataFor(columnName);
+      FieldSpec fieldSpecInSchema = schema.getFieldSpecFor(columnName);
+      DataSource source = segment.getDataSource(columnName);
+      Preconditions.checkNotNull(columnMetadata);
+      Preconditions.checkNotNull(source);
+
+      // Column is deleted
+      if (fieldSpecInSchema == null) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: column deleted", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "column deleted: " + columnName);
+      }
+
+      // Field type changed
+      if (columnMetadata.getFieldType().compareTo(fieldSpecInSchema.getFieldType()) != 0) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: field type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "field type changed: " + columnName);
+      }
+
+      // Data type changed
+      if (!columnMetadata.getDataType().equals(fieldSpecInSchema.getDataType())) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: data type", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "data type changed: " + columnName);
+      }
+
+      // SV/MV changed
+      if (columnMetadata.isSingleValue() != fieldSpecInSchema.isSingleValueField()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: single / multi value", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "single / multi value changed: " + columnName);
+      }
+
+      // TODO: detect if an index changes from Dictionary to Variable Length Dictionary or vice versa.
+      // TODO: RV TEST
+      boolean colHasDictionary = columnMetadata.hasDictionary();
+      // Encoding changed
+      if (colHasDictionary == noDictionaryColumns.contains(columnName)) {
+        // Check if dictionary update is needed
+        // 1. If the segment metadata has dictionary enabled and table has it disabled, its incompatible and refresh is
+        // needed.
+        // 2. If segment metadata has dictionary disabled, check if it has to be overridden. If not overridden,
+        // refresh is needed, since table has it enabled.
+        boolean incompatible = colHasDictionary || DictionaryIndexType.ignoreDictionaryOverride(
+            tableConfig.getIndexingConfig().isOptimizeDictionary(),
+            tableConfig.getIndexingConfig().isOptimizeDictionaryForMetrics(),
+            tableConfig.getIndexingConfig().getNoDictionarySizeRatioThreshold(),
+            tableConfig.getIndexingConfig().getNoDictionaryCardinalityRatioThreshold(), fieldSpecInSchema,
+            indexConfigsMap.get(columnName), columnMetadata.getCardinality(), columnMetadata.getTotalNumberOfEntries());
+        if (incompatible) {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: dictionary encoding,", tableNameWithType,
+              segmentName);
+          return new NeedRefreshResponse(segmentName, true, "dictionary encoding changed: " + columnName);
+        } else {
+          LOGGER.debug("tableNameWithType: {}, segmentName: {}, no change as dictionary overrides applied to col: {}",
+              tableNameWithType, segmentName, columnName);
+        }
+      }
+
+      // Sorted column not sorted
+      if (columnName.equals(sortedColumn) && !columnMetadata.isSorted()) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: sort column", tableNameWithType, segmentName);
+        return new NeedRefreshResponse(segmentName, true, "sort column changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getBloomFilter()) == bloomFilters.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: bloom filter changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "bloom filter changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getJsonIndex()) == jsonIndex.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: json index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "json index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getTextIndex()) == textIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: text index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "text index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getFSTIndex()) == fstIndexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: fst index changed", tableNameWithType,
+            segmentName);
+        return new NeedRefreshResponse(segmentName, true, "fst index changed: " + columnName);
+      }
+
+      if (Objects.isNull(source.getH3Index()) == h3Indexes.contains(columnName)) {
+        LOGGER.debug("tableNameWithType: {}, segmentName: {}, change: h3 index changed", tableNameWithType,

Review Comment:
   nit: Can we improve this log to also have columnName for which the index was changed



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add a Server API to list segments that need to be refreshed for a table [pinot]

Posted by "vrajat (via GitHub)" <gi...@apache.org>.

vrajat commented on code in PR #14451:
URL: https://github.com/apache/pinot/pull/14451#discussion_r1855820871


##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,39 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    int failedParses = 0;
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint
+        serverResponses.put(streamResponse.getKey(),
+            JsonUtils.stringToObject(streamResponse.getValue(),
+                new TypeReference<List<NeedRefreshResponse>>() { }));
+      } catch (Exception e) {
+        failedParses++;
+        LOGGER.error("Unable to parse server {} response due to an error: ", streamResponse.getKey(), e);
+      }
+    }
+    if (failedParses != 0) {
+      LOGGER.error("Unable to parse server {} / {} response due to an error: ", failedParses, serverURLs.size());

Review Comment:
   I've remove this line. 



##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -397,6 +398,39 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  public Map<String, List<NeedRefreshResponse>> getSegmentsForRefreshFromServer(
+      String tableNameWithType, Set<String> serverInstances, BiMap<String, String> endpoints, int timeoutMs) {
+    LOGGER.debug("Getting list of segments for refresh from servers for table {}.", tableNameWithType);
+    List<String> serverURLs = new ArrayList<>();
+    for (String serverInstance : serverInstances) {
+      serverURLs.add(generateNeedRefreshSegmentsServerURL(tableNameWithType, endpoints.get(serverInstance)));
+    }
+    BiMap<String, String> endpointsToServers = endpoints.inverse();
+    CompletionServiceHelper completionServiceHelper =
+        new CompletionServiceHelper(_executor, _connectionManager, endpointsToServers);
+    CompletionServiceHelper.CompletionServiceResponse serviceResponse =
+        completionServiceHelper.doMultiGetRequest(serverURLs, tableNameWithType, true, timeoutMs);
+    Map<String, List<NeedRefreshResponse>> serverResponses = new HashMap<>();
+
+    int failedParses = 0;
+    for (Map.Entry<String, String> streamResponse : serviceResponse._httpResponses.entrySet()) {
+      try {
+        // TODO: RV - get the instance name instead of the endpoint
+        serverResponses.put(streamResponse.getKey(),
+            JsonUtils.stringToObject(streamResponse.getValue(),
+                new TypeReference<List<NeedRefreshResponse>>() { }));
+      } catch (Exception e) {
+        failedParses++;
+        LOGGER.error("Unable to parse server {} response due to an error: ", streamResponse.getKey(), e);

Review Comment:
   Done.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org