Skip to content

Add logging for sketches on workers #16697

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LakshSingla merged 2 commits into from
Jul 9, 2024
Merged

Add logging for sketches on workers #16697

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c5276c8
Add logging for sketches on workers
adarshsanjeev Jul 7, 2024
31409e1
Refactor
adarshsanjeev Jul 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1772,6 +1772,7 @@ private ResultAndChannels<?> gatherResultKeyStatistics(final OutputChannels chan
@Override
public void onSuccess(final ClusterByStatisticsCollector result)
{
result.logSketches();
kernelManipulationQueue.add(
holder ->
holder.getStageKernelMap().get(stageDefinition.getId())
Expand Down
1 change: 1 addition & 0 deletions ...-stage-query/src/main/java/org/apache/druid/msq/kernel/GlobalSortMaxCountShuffleSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ public Either<Long, ClusterByPartitions> generatePartitionsForGlobalSort(
} else if (maxPartitions > maxNumPartitions) {
return Either.error((long) maxPartitions);
} else {
collector.logSketches();
final ClusterByPartitions generatedPartitions = collector.generatePartitionsWithMaxCount(maxPartitions);
if (generatedPartitions.size() <= maxNumPartitions) {
return Either.value(generatedPartitions);
Expand Down
1 change: 1 addition & 0 deletions ...tage-query/src/main/java/org/apache/druid/msq/kernel/GlobalSortTargetSizeShuffleSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ public Either<Long, ClusterByPartitions> generatePartitionsForGlobalSort(
if (expectedPartitions > maxNumPartitions) {
return Either.error(expectedPartitions);
} else {
collector.logSketches();
final ClusterByPartitions generatedPartitions = collector.generatePartitionsWithTargetWeight(targetSize);
if (generatedPartitions.size() <= maxNumPartitions) {
return Either.value(generatedPartitions);
Expand Down
5 changes: 5 additions & 0 deletions ...age-query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,11 @@ public interface ClusterByStatisticsCollector
*/
ClusterByPartitions generatePartitionsWithMaxCount(int maxNumPartitions);

/**
* Logs some information regarding the collector. This is useful in seeing which sketches were downsampled the most.
*/
void logSketches();

/**
* Returns an immutable, JSON-serializable snapshot of this collector.
*/
Expand Down
7 changes: 2 additions & 5 deletions ...query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsCollectorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,6 @@ public ClusterByStatisticsCollector clear()
@Override
public ClusterByPartitions generatePartitionsWithTargetWeight(final long targetWeight)
{
logSketches();

if (targetWeight < 1) {
throw new IAE("Target weight must be positive");
}
Expand Down Expand Up @@ -288,8 +286,6 @@ public ClusterByPartitions generatePartitionsWithTargetWeight(final long targetW
@Override
public ClusterByPartitions generatePartitionsWithMaxCount(final int maxNumPartitions)
{
logSketches();

if (maxNumPartitions < 1) {
throw new IAE("Must have at least one partition");
} else if (buckets.isEmpty()) {
Expand Down Expand Up @@ -331,7 +327,8 @@ public ClusterByPartitions generatePartitionsWithMaxCount(final int maxNumPartit
return ranges;
}

private void logSketches()
@Override
public void logSketches()
{
if (log.isDebugEnabled()) {
// Log all sketches
Expand Down