kfaraz commented on code in PR #17741:
URL: https://github.com/apache/druid/pull/17741#discussion_r1964783655
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -552,19 +552,26 @@ private void emitSummaryMetrics(final
MSQTaskReportPayload msqTaskReportPayload,
long totalProcessedBytes = 0;
if (msqTaskReportPayload.getCounters() != null) {
- totalProcessedBytes = msqTaskReportPayload.getCounters()
- .copyMap()
- .entrySet()
- .stream()
- .filter(entry -> stagesReport == null ||
stagesToInclude.contains(entry.getKey()))
- .flatMap(counterSnapshotsMap ->
counterSnapshotsMap.getValue().values().stream())
- .flatMap(counterSnapshots ->
counterSnapshots.getMap().entrySet().stream())
- .filter(entry -> entry.getKey().startsWith("input"))
- .mapToLong(entry -> {
- ChannelCounters.Snapshot snapshot = (ChannelCounters.Snapshot)
entry.getValue();
- return snapshot.getBytes() == null ? 0L :
Arrays.stream(snapshot.getBytes()).sum();
- })
- .sum();
+ totalProcessedBytes =
+ msqTaskReportPayload.getCounters()
+ .copyMap()
+ .entrySet()
+ .stream()
+ .filter(entry -> stagesReport == null ||
stagesToInclude.contains(entry.getKey()))
+ .flatMap(counterSnapshotsMap ->
counterSnapshotsMap.getValue()
+
.values()
+
.stream())
+ .flatMap(counterSnapshots ->
counterSnapshots.getMap()
+
.entrySet()
+
.stream())
+ .filter(entry ->
entry.getKey().startsWith("input"))
+ .mapToLong(entry -> {
+ ChannelCounters.Snapshot snapshot =
(ChannelCounters.Snapshot) entry.getValue();
+ return snapshot.getBytes() == null
+ ? 0L
+ :
Arrays.stream(snapshot.getBytes()).sum();
+ })
+ .sum();
Review Comment:
Is the reformat really needed here and in other places in this file?
Maybe we can limit the diff in this PR to only the modified code.
##########
server/src/test/java/org/apache/druid/server/compaction/NewestSegmentFirstPolicyTest.java:
##########
@@ -1235,7 +1415,17 @@ public void
testIteratorReturnsSegmentsAsSegmentsWasCompactedAndHaveDifferentFil
partitionsSpec,
null,
null,
- mapper.readValue(mapper.writeValueAsString(new
TransformSpec(new SelectorDimFilter("dim1", "foo", null), null)), new
TypeReference<>() {}),
+ mapper.readValue(
+ mapper.writeValueAsString(new TransformSpec(
+ new SelectorDimFilter(
+ "dim1",
+ "foo",
+ null
+ ), null
+ )), new TypeReference<>()
+ {
+ }
+ ),
Review Comment:
This can be simplified by avoiding the unnecessary serde.
##########
processing/src/test/java/org/apache/druid/timeline/DataSegmentTest.java:
##########
@@ -345,10 +351,10 @@ public void testWithLastCompactionState()
final CompactionState compactionState = new CompactionState(
new DynamicPartitionsSpec(null, null),
new
DimensionsSpec(DimensionsSpec.getDefaultSchemas(ImmutableList.of("bar",
"foo"))),
- ImmutableList.of(ImmutableMap.of("type", "count", "name", "count")),
- ImmutableMap.of("filter", ImmutableMap.of("type", "selector",
"dimension", "dim1", "value", "foo")),
- Collections.singletonMap("test", "map"),
- Collections.singletonMap("test2", "map2")
+ ImmutableList.of(new CountAggregatorFactory("count")),
+ new CompactionTransformSpec(new SelectorDimFilter("dim1", "foo",
null)),
+ MAPPER.convertValue(Collections.singletonMap("test", "map"),
IndexSpec.class),
Review Comment:
Since we are touching these lines anyway, `Map.of` and `List.of` could be
used for less verbosity.
```suggestion
MAPPER.convertValue(Map.of("test", "map"), IndexSpec.class),
```
##########
processing/src/test/java/org/apache/druid/timeline/DataSegmentTest.java:
##########
@@ -129,10 +135,10 @@ public void testV1Serialization() throws Exception
new DimensionsSpec(
DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim1",
"bar", "foo"))
),
- ImmutableList.of(ImmutableMap.of("type", "count", "name",
"count")),
- ImmutableMap.of("filter", ImmutableMap.of("type", "selector",
"dimension", "dim1", "value", "foo")),
- ImmutableMap.of(),
- ImmutableMap.of()
+ ImmutableList.of(new CountAggregatorFactory("count")),
+ new CompactionTransformSpec(new SelectorDimFilter("dim1", "foo",
null)),
+ MAPPER.convertValue(ImmutableMap.of(), IndexSpec.class),
Review Comment:
Maybe use `IndexSpec.DEFAULT` instead?
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java:
##########
@@ -2193,18 +2179,21 @@ private void compactSegments(
clientCompactionTaskQuery.getDimensionsSpec().getDimensions()
),
metricsSpec,
- transformSpec,
- ImmutableMap.of(
- "bitmap",
- ImmutableMap.of("type", "roaring"),
- "dimensionCompression",
- "lz4",
- "metricCompression",
- "lz4",
- "longEncoding",
- "longs"
+ clientCompactionTaskQuery.getTransformSpec(),
+ jsonMapper.convertValue(
+ ImmutableMap.of(
+ "bitmap",
+ ImmutableMap.of("type", "roaring"),
+ "dimensionCompression",
+ "lz4",
+ "metricCompression",
+ "lz4",
+ "longEncoding",
+ "longs"
+ ),
+ IndexSpec.class
),
- ImmutableMap.of()
+ jsonMapper.convertValue(ImmutableMap.of(),
GranularitySpec.class)
Review Comment:
Might be nice to add a `GranularitySpec.DEFAULT` in a later PR.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
