Add coordinator API for unused segments #14846

adarshsanjeev · 2023-08-16T10:55:09Z

There is a current issue due to inconsistent metadata between worker and controller in MSQ. A controller can receive one set of segments, which are then marked as unused by, say, a compaction job. The worker would be unable to get the segment information as MetadataResource.

This PR resolves this by allowing the worker to fetch the segment information even if it is unused.

Release Notes

Adds includeUnused as an optional parameter to the org.apache.druid.server.http.MetadataResource#getSegment coordinator API. The API will also return unused segments if the parameter is set.

This PR has:

kfaraz · 2023-08-16T11:44:46Z

Do we really want to fetch the information of unused segments in the worker task? Wouldn't the task just discard the segment once it realises that the fetched segment is unused?

adarshsanjeev · 2023-08-17T02:37:21Z

The controller already has a picture of the metadata containing only used segments at that point and assigns segments to the workers. On the worker side, we just want to fetch the segment if it has not been deleted completely, even if it is not used anymore.

cryptoe

Thanks for the PR.
Left some comments.

cryptoe · 2023-08-17T09:58:47Z

server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClient.java

+  /**
+   * Fetches segment metadata for the given dataSource and segmentId.
+   */
+  ListenableFuture<DataSegment> fetchSegment(String dataSource, String segmentId);


I feel we should rename the fetchUsedSegment call and rename that to fetchSegment taking a boolean "includeUnused".

The callers in most places can set this variable to false.

We can javadoc the behavior changes with this parameter.

WDYT ?

I made it a separate API instead of a parameter since this probably shouldn't be used in most other places, so adding an extra parameter for every current call seemed untidy. I am okay with changing it to a parameter since it would cut down on the interface though.

Since there are 3 places in production+ test code where you would have to make the changes. I guess it should be fine?

cryptoe · 2023-08-17T10:05:23Z

server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java

+   *
+   * @return DataSegment segment corresponding to given id
+   */
+  DataSegment retrieveSegmentForId(String id);


I think we should merge this method with DataSegment retrieveUsedSegmentForId(String id); by taking the same boolean parameter as discussed above.

abhishekagarwal87 · 2023-08-17T10:07:51Z

@kfaraz - Here is a situation. Let's say you launch an MSQ job to read data from a druid table. At this time T0, controller has a view of the used segments that make up the table. Controller will take the ids of these segments and distribute to worker. As workers operating on the segment at time T1, the segment is no longer available. Because the interval, this segment corresponds to, has been compacted. Worker could have discarded the segment if the segment just falls out of the retention window. But here, workers should instead find those new segments. Instead of taking that route, we are going to let the job run on the previous snapshot of the system. For that to happen, MSQ workers need to find those segments that have been marked unused.

kfaraz · 2023-08-17T14:36:37Z

Thanks for the clarification, @abhishekagarwal87 !

Instead of taking that route, we are going to let the job run on the previous snapshot of the system.

If the segments of the previous snapshot (v1) have already been marked unused, that implies that a new version (v2) has completely overshadowed v1. At this point, if we continue with v1 in an MSQ replace job, wouldn't that mean that we would overwrite v2 and potentially miss out on some data that was present in v2 but not in v1?

Also, to avoid this, would a better alternative be to fetch the list of used and non-overshadowed segments in the controller task. IIUC, in the case where new segments of v2 are created (that result in overshadowing of v1) after the controller task has already started should cause the MSQ workers to fail anyway.

cc: @cryptoe

adarshsanjeev · 2023-08-18T03:13:06Z

@kfaraz

Wouldn't that mean that we would overwrite v2 and potentially miss out on some data that was present in v2 but not in v1?Would a better alternative be to fetch the list of used and non-overshadowed segments in the controller task

Yes, it would. Only the controller decides which version to use (which it does by fetching the list of used and non-overshadowed segments). This change is only to the getSegmentById API, which the worker uses with the id from the list of segments which have already been fetched. This is currently the case as well, however, this is causing an issue where the worker is not able to access any segments marked as unused after the controller fetches the segment. This PR changes this behaviour.

the case where new segments of v2 are created (that result in overshadowing of v1) after the controller task has already started should cause the MSQ workers to fail anyway.

That is currently the case, however, this is causing an issue where some tasks like compaction/reindex can stop any MSQ tasks from running. Is there a reason to prefer failing the task over providing the results from the snapshot that the MSQ task was started? I think the snapshot approach makes sense.

cryptoe · 2023-08-18T04:48:54Z

If the segments of the previous snapshot (v1) have already been marked unused, that implies that a new version (v2) has completely overshadowed v1. At this point, if we continue with v1 in an MSQ replace job, wouldn't that mean that we would overwrite v2 and potentially miss out on some data that was present in v2 but not in v1?

Adding on to what @adarshsanjeev said, There are 3 distinct usecases:

Select case --> select * from foo which is okay to be run on the snapshot version of the coordinator since this is also the case of the current query engine.
Insert into new DS -> which is insert into bar select * from foo which is same as a select on point 1.
Insert/Replace into same DS -> insert/replace into foo select * from foo The controller takes a lock on the target data source ie foo hence we cannot have a case where we miss out on data on the same DS as compaction/reindex triggered will cause lock contention.

kfaraz · 2023-08-18T04:53:18Z

Thanks for the clarification, @adarshsanjeev , @cryptoe .
I was thinking about the 3rd case and it makes sense for it to have a lock, thus preventing any overwrites while this task is already in progress.

So I assume this PR is to ensure that cases 1 and 2 work correctly?

cryptoe · 2023-08-18T04:57:25Z

So I assume this PR is to ensure that cases 1 and 2 work correctly?

Exactly 🚀

cryptoe

Left minor comments.
Changes LGTM!!

cryptoe · 2023-08-21T10:57:23Z

server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientImpl.java

  {
    final String path = StringUtils.format(
-        "/druid/coordinator/v1/metadata/datasources/%s/segments/%s",
+        "/druid/coordinator/v1/metadata/datasources/%s/segments/%s%s",


This looks better to read. We can adjust %s to true/false based on the flag passed. Wdyt?

Suggested change

"/druid/coordinator/v1/metadata/datasources/%s/segments/%s%s",

"/druid/coordinator/v1/metadata/datasources/%s/segments/%s?includeUnused=%s",

I added it this way for consistency with the other APIs. The current pattern is to add a flag without value and check if it has been passed or not. ("includeOvershadowedStatus", "full", etc). Do we want to start moving away from this?

Something like

druid/server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientImpl.java

Line 57 in 82a8529

"/druid/coordinator/v1/datasources/%s/handoffComplete?interval=%s&partitionNumber=%d&version=%s",

is more readable.

I agree with that. Changing to this pattern.

cryptoe · 2023-08-21T10:57:33Z

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java

+
+    taskContextOverridesBuilder.put(
+        MultiStageQueryContext.CTX_IS_REINDEX,
+        MSQControllerTask.isReindexTask(task)


Lets rename this to isReplaceTask ?

It's only for a replace task which reads from the datasource it is replacing, so wouldn't reindex be better?

replaceInputDataSourceTask? How does that sound ?

…egment

adarshsanjeev added 2 commits August 16, 2023 16:01

Add coordinator API for unused segments

d23ae3a

Update docs

037ba93

github-actions bot added the Area - Documentation label Aug 16, 2023

kfaraz added the Area - Metadata label Aug 16, 2023

Add tests

e590e32

cryptoe reviewed Aug 17, 2023

View reviewed changes

adarshsanjeev added 2 commits August 18, 2023 13:47

Address review comments

989e83d

Add reindex check

dfc143a

cryptoe reviewed Aug 21, 2023

View reviewed changes

cryptoe approved these changes Aug 21, 2023

View reviewed changes

adarshsanjeev added 2 commits August 22, 2023 10:58

Merge remote-tracking branch 'origin/master' into msq-worker-unused-s…

99cf3c6

…egment

Change parameter

352bd6d

abhishekagarwal87 merged commit dfb5a98 into apache:master Aug 23, 2023
74 checks passed

LakshSingla added this to the 28.0 milestone Oct 12, 2023

LakshSingla mentioned this pull request Nov 4, 2023

[DRAFT] 28.0.0 release notes #15326

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add coordinator API for unused segments #14846

Add coordinator API for unused segments #14846

adarshsanjeev commented Aug 16, 2023

kfaraz commented Aug 16, 2023

adarshsanjeev commented Aug 17, 2023

cryptoe left a comment

cryptoe Aug 17, 2023

adarshsanjeev Aug 17, 2023

cryptoe Aug 18, 2023 •

edited

Loading

cryptoe Aug 17, 2023

abhishekagarwal87 commented Aug 17, 2023

kfaraz commented Aug 17, 2023

adarshsanjeev commented Aug 18, 2023

cryptoe commented Aug 18, 2023

kfaraz commented Aug 18, 2023

cryptoe commented Aug 18, 2023

cryptoe left a comment

cryptoe Aug 21, 2023

adarshsanjeev Aug 22, 2023

cryptoe Aug 22, 2023

adarshsanjeev Aug 22, 2023

cryptoe Aug 21, 2023

adarshsanjeev Aug 22, 2023

cryptoe Aug 22, 2023

	"/druid/coordinator/v1/metadata/datasources/%s/segments/%s%s",
	"/druid/coordinator/v1/metadata/datasources/%s/segments/%s?includeUnused=%s",

Add coordinator API for unused segments #14846

Add coordinator API for unused segments #14846

Conversation

adarshsanjeev commented Aug 16, 2023

Release Notes

kfaraz commented Aug 16, 2023

adarshsanjeev commented Aug 17, 2023

cryptoe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cryptoe Aug 18, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abhishekagarwal87 commented Aug 17, 2023

kfaraz commented Aug 17, 2023

adarshsanjeev commented Aug 18, 2023

cryptoe commented Aug 18, 2023

kfaraz commented Aug 18, 2023

cryptoe commented Aug 18, 2023

cryptoe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cryptoe Aug 18, 2023 •

edited

Loading