tejaswini-imply commented on code in PR #13758:
URL: https://github.com/apache/druid/pull/13758#discussion_r1167046898
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java:
##########
@@ -573,7 +580,8 @@ private Map<String, TaskReport> getTaskCompletionReports()
getTaskCompletionRowStats(),
errorMsg,
segmentAvailabilityConfirmationCompleted,
- segmentAvailabilityWaitTimeMs
+ segmentAvailabilityWaitTimeMs,
+ JodaUtils.condenseIntervals(intervals)
Review Comment:
Rather taking intervals as input and condesing here, Could we create new
method `getIngestedIntervals()` which has the logic to condense there and maybe
we can maintain class wide variable `ingestedIntervals` like
`buildSegmentsMeters`, `ingestionState`. IMO it'll be consistent with existing
structure that way and we don't have to pass intervals everytime we write
reports and overloading `getTaskCompletionReports()` wouldn't be necessary as
well.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/config/ClusterConfigTest.java:
##########
@@ -37,11 +38,12 @@
public class ClusterConfigTest
{
@Test
+ @Ignore
public void testYaml()
{
ClusterConfig config =
ClusterConfig.loadFromResource("/config-test/test.yaml");
// Uncomment this line to see the full config with includes resolved.
- //System.out.println(config.resolveIncludes());
+ System.out.println(config.resolveIncludes());
Review Comment:
Could you please remove this debug line.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java:
##########
@@ -651,7 +654,8 @@ private Map<String, TaskReport> getTaskCompletionReports()
getTaskCompletionRowStats(),
errorMsg,
false, // not applicable for parallel subtask
- segmentAvailabilityWaitTimeMs
+ segmentAvailabilityWaitTimeMs,
+ ingestedIntervals
Review Comment:
Do we not need to condense the intervals here?
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java:
##########
@@ -640,7 +643,7 @@ private Map<String, Object> getTaskCompletionRowStats()
**
* @return
*/
- private Map<String, TaskReport> getTaskCompletionReports()
+ private Map<String, TaskReport> getTaskCompletionReports(List<Interval>
ingestedIntervals)
Review Comment:
Same comment applies here as suggested in `IndexTask` class.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java:
##########
@@ -695,7 +696,8 @@ private Map<String, TaskReport> getTaskCompletionReports()
getTaskCompletionRowStats(),
errorMsg,
segmentAvailabilityConfirmationCompleted,
- segmentAvailabilityWaitTimeMs
+ segmentAvailabilityWaitTimeMs,
+ JodaUtils.condenseIntervals(intervals)
Review Comment:
Same comments as in `IndexTask` class.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SequentialParallelIndexStatsReporter.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import org.apache.druid.indexing.common.task.IndexTask;
+
+import java.util.Collections;
+
+public class SequentialParallelIndexStatsReporter extends
ParallelIndexStatsReporter
+{
+ @Override
+ ParallelIndexStats report(
+ ParallelIndexSupervisorTask task,
+ Object runner,
+ boolean includeUnparseable,
+ boolean full
+ )
+ {
+ IndexTask currentSequentialTask = (IndexTask) runner;
+ return new ParallelIndexStats(
+ currentSequentialTask.doGetRowStats(full),
+ currentSequentialTask.doGetUnparseableEvents(full),
+ Collections.emptySet()
Review Comment:
Can you please elaborate why we are updating stats with empty intervals set
here? Shouldn't we report intervals for sequential task as well?
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java:
##########
@@ -246,7 +249,8 @@ private Map<String, TaskReport> getTaskCompletionReports()
getTaskCompletionRowStats(),
"",
false, // not applicable for parallel subtask
- segmentAvailabilityWaitTimeMs
+ segmentAvailabilityWaitTimeMs,
+ intervals
Review Comment:
Do we not need to condense the intervals here?
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentGenerateTask.java:
##########
@@ -235,7 +238,7 @@ private List<DataSegment> generateSegments(
/**
* Generate an IngestionStatsAndErrorsTaskReport for the task.
*/
- private Map<String, TaskReport> getTaskCompletionReports()
+ private Map<String, TaskReport> getTaskCompletionReports(List<Interval>
intervals)
Review Comment:
Same comment applies here as suggested in `IndexTask` class.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexStatsReporterFactory.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
+
+public class ParallelIndexStatsReporterFactory
+{
+ ParallelIndexStatsReporter create(ParallelIndexSupervisorTask task)
+ {
+ if (task.isParallelMode()) {
+ if (AbstractBatchIndexTask.isGuaranteedRollup(
Review Comment:
Maybe we could replace this with `task.isPerfectRollup()` for simplicity?
--
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]
