This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new c6fcf4325b6 feat: warn when hash/range ingest publishes oversized
segments (#20173)
c6fcf4325b6 is described below
commit c6fcf4325b6cb039620288eb7ea27101f143f9c5
Author: David Alexander <[email protected]>
AuthorDate: Tue Sep 1 08:44:58 2026 -0400
feat: warn when hash/range ingest publishes oversized segments (#20173)
* feat: warn when hash/range ingest publishes oversized segments
* Fix typo in documentation
---------
Co-authored-by: David Alexander <[email protected]>
---
docs/operations/metrics.md | 1 +
.../src/main/resources/defaultMetrics.json | 1 +
.../main/resources/defaultMetricDimensions.json | 1 +
.../common/task/AbstractBatchIndexTask.java | 14 +++-
.../druid/indexing/common/task/IndexTaskUtils.java | 13 ++++
.../parallel/ParallelIndexSupervisorTask.java | 27 +++++++
.../batch/parallel/PartialSegmentMergeTask.java | 10 ++-
.../SeekableStreamIndexTaskRunner.java | 1 +
.../indexing/common/task/IndexTaskUtilsTest.java | 44 ++++++++++++
.../AbstractMultiPhaseParallelIndexingTest.java | 6 +-
.../AbstractParallelIndexSupervisorTaskTest.java | 8 ++-
.../MultiPhaseParallelIndexingRowStatsTest.java | 84 +++++++++++++++++++++-
.../parallel/SinglePhaseParallelIndexingTest.java | 6 +-
.../indexer/report/IngestionStatsAndErrors.java | 20 +++++-
.../resources/loggingEmitterAllowedMetrics.json | 1 +
.../druid/indexer/report/TaskReportSerdeTest.java | 4 +-
.../resources/loggingEmitterAllowedMetrics.json | 1 +
17 files changed, 227 insertions(+), 15 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 07a4922945f..f478889f19b 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -230,6 +230,7 @@ If SQL is enabled, the Broker will emit the following
metrics for SQL.
|------|-----------|----------|------------|
|`ingest/count`|Count of `1` every time an ingestion job runs (includes
compaction jobs). Aggregate using dimensions. | `dataSource`, `taskId`,
`taskType`, `groupId`, `taskIngestionMode`, `tags` |Always `1`.|
|`ingest/segments/count`|Count of final segments created by job (includes
tombstones). | `dataSource`, `taskId`, `taskType`, `groupId`,
`taskIngestionMode`, `tags` |At least `1`.|
+|`ingest/segments/oversized`|Count of final segments created by job that were
detected as being oversized, meaning the number of rows in the segment exceeded
more than twice the maxRowsPerSegment in the spec. This is relevant only for
hash and range partitioned ingestions. | `dataSource`, `taskId`, `taskType`,
`groupId`, `taskIngestionMode`, `tags` |At least `1`.|
|`ingest/rows/published`|Number of rows successfully published by the job. |
`dataSource`, `taskId`, `taskType`, `groupId`, `taskIngestionMode`, `tags` |At
least `1`.|
|`ingest/tombstones/count`|Count of tombstones created by job. | `dataSource`,
`taskId`, `taskType`, `groupId`, `taskIngestionMode`, `tags` |Zero or more for
replace. Always zero for non-replace tasks (always zero for legacy replace, see
below).|
diff --git
a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
index e1af9b007bd..53492716265 100644
---
a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
+++
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
@@ -103,6 +103,7 @@
"ingest/count" : { "dimensions" : ["dataSource", "taskType"], "type" :
"count", "help": "Count of 1 every time an ingestion job runs (includes
compaction jobs). Aggregate using dimensions." },
"ingest/segments/count" : { "dimensions" : ["dataSource", "taskType"],
"type" : "count", "help": "Count of final segments created by job (includes
tombstones)." },
+ "ingest/segments/oversized" : { "dimensions" : ["dataSource", "taskType"],
"type" : "count", "help": "Count of final segments created by job that were
detected as being oversized, meaning the number of rows in the segment exceeded
more than 2x the maxRowsPerSegment in the spec. This is relevant only for hash
and range partitioned ingestions." },
"ingest/tombstones/count" : { "dimensions" : ["dataSource", "taskType"],
"type" : "count", "help": "Count of tombstones created by job." },
"ingest/rows/published": { "dimensions" : ["dataSource", "taskType"], "type"
: "count", "help": "Number of rows successfully published by the job." },
diff --git
a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
index 6eb84425ab7..0b86e8efc5d 100644
---
a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
+++
b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
@@ -55,6 +55,7 @@
"ingest/merge/time" : { "dimensions" : ["dataSource"], "type" : "timer" },
"ingest/merge/cpu" : { "dimensions" : ["dataSource"], "type" : "timer" },
"ingest/segments/count" : { "dimensions" : ["dataSource"], "type" : "count"
},
+ "ingest/segments/oversized" : { "dimensions" : ["dataSource"], "type" :
"count" },
"ingest/rows/published": { "dimensions" : ["dataSource"], "type" : "count" },
"ingest/realtime/segmentUpgrade/persisted" : { "dimensions" : ["dataSource",
"interval", "version"], "type" : "count" },
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
index bb1f90d1299..806e6121146 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
@@ -929,6 +929,16 @@ public abstract class AbstractBatchIndexTask extends
AbstractTask
return null;
}
+ /**
+ * Number of published segments whose row count exceeds {@code
maxRowsPerSegment} times the oversize ratio.
+ * Null when the check does not apply (for example dynamic partitioning, or
when no target is configured).
+ */
+ @Nullable
+ protected Long getTaskCompletionOversizedSegments()
+ {
+ return null;
+ }
+
protected TaskReport.ReportMap buildLiveIngestionStatsReport(
IngestionState ingestionState,
Map<String, Object> unparseableEvents,
@@ -947,6 +957,7 @@ public abstract class AbstractBatchIndexTask extends
AbstractTask
0L,
null,
null,
+ null,
null
)
)
@@ -1008,7 +1019,8 @@ public abstract class AbstractBatchIndexTask extends
AbstractTask
segmentAvailabilityWaitTimeMs,
Collections.emptyMap(),
segmentsRead,
- segmentsPublished
+ segmentsPublished,
+ getTaskCompletionOversizedSegments()
)
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
index a03f410982c..154a222c782 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
@@ -141,6 +141,19 @@ public class IndexTaskUtils
.sum();
}
+ /**
+ * Counts published segments whose row count exceeds {@code
maxRowsPerSegment * oversizedRatio}.
+ * Segments without {@link DataSegment#getTotalRows()} populated are ignored.
+ */
+ public static long getOversizedSegments(Collection<DataSegment> segments,
int maxRowsPerSegment, double oversizedRatio)
+ {
+ return segments.stream()
+ .map(DataSegment::getTotalRows)
+ .filter(Objects::nonNull)
+ .filter(rowCount -> rowCount > maxRowsPerSegment * oversizedRatio)
+ .count();
+ }
+
/**
* Adds the upgraded pending segment's {@code interval} and {@code version}
to a metric builder so that every
* segment-upgrade metric can be sliced by the specific segment being
re-announced. Mirrors
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
index 4793b77ac44..1e88c34fda5 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
@@ -42,6 +42,7 @@ import org.apache.druid.indexer.granularity.GranularitySpec;
import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.indexer.partitions.SecondaryPartitionType;
import org.apache.druid.indexer.report.IngestionStatsAndErrors;
import org.apache.druid.indexer.report.IngestionStatsAndErrorsTaskReport;
import org.apache.druid.indexer.report.TaskReport;
@@ -154,6 +155,10 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
// and fix
private static final long DEFAULT_NUM_SHARDS_WHEN_ESTIMATE_GOES_NEGATIVE =
7L;
+ // Ratio of the total rows of a segment to the max rows per segment, above
which the segment is considered oversized
+ // For range partitioning, max rows per segment is set to 1.5x target rows,
so the ratio comparison is effectively 3x target rows.
+ private static final double DEFAULT_OVERSIZE_RATIO = 2.0;
+
private final ParallelIndexIngestionSpec ingestionSchema;
/**
* Base name for the {@link SubTaskSpec} ID.
@@ -210,6 +215,7 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
private TaskReport.ReportMap completionReports;
private Long segmentsRead;
private Long segmentsPublished;
+ private Long oversizedSegments; // Number of segments whose row count
exceeds maxRowsPerSegment * DEFAULT_OVERSIZE_RATIO
private final boolean isCompactionTask;
@JsonCreator
@@ -1161,6 +1167,8 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
final Set<DataSegment> oldSegments = new HashSet<>();
final Set<DataSegment> newSegments = new HashSet<>();
final SegmentSchemaMapping segmentSchemaMapping = new
SegmentSchemaMapping(CentralizedDatasourceSchemaConfig.SCHEMA_VERSION);
+ final SecondaryPartitionType type =
ingestionSchema.getTuningConfig().getGivenOrDefaultPartitionsSpec().getType();
+ final Integer maxRowsPerSegment =
ingestionSchema.getTuningConfig().getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment();
reportsMap
.values()
@@ -1224,6 +1232,19 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
emitMetric(toolbox.getEmitter(), "ingest/tombstones/count",
tombStones.size());
emitMetric(toolbox.getEmitter(), "ingest/segments/count",
newSegments.size());
emitMetric(toolbox.getEmitter(), "ingest/rows/published",
IndexTaskUtils.getTotalRowCount(newSegments));
+ // If partitionsSpec is range or hash, we emit info about the size in
rows of generated partitions, to detect a hot partition.
+ if ((type == SecondaryPartitionType.RANGE || type ==
SecondaryPartitionType.HASH) && maxRowsPerSegment != null) {
+ oversizedSegments = IndexTaskUtils.getOversizedSegments(newSegments,
maxRowsPerSegment, DEFAULT_OVERSIZE_RATIO);
+ if (oversizedSegments > 0) {
+ LOG.warn(
+ "Published [%d] oversized segments with more than
(maxRowsPerSegment [%d] x ratio [%s]) rows.",
+ oversizedSegments,
+ maxRowsPerSegment,
+ DEFAULT_OVERSIZE_RATIO
+ );
+ emitMetric(toolbox.getEmitter(), "ingest/segments/oversized",
oversizedSegments);
+ }
+ }
} else {
throw new ISE("Failed to publish segments");
}
@@ -1276,6 +1297,12 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
);
}
+ @Override
+ protected Long getTaskCompletionOversizedSegments()
+ {
+ return oversizedSegments;
+ }
+
@Override
protected Map<String, Object> getTaskCompletionRowStats()
{
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentMergeTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentMergeTask.java
index 6bf9c12040f..96a1f841cd4 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentMergeTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentMergeTask.java
@@ -286,7 +286,8 @@ abstract class PartialSegmentMergeTask<S extends ShardSpec>
extends PerfectRollu
final List<String> metricNames =
Arrays.stream(dataSchema.getAggregators())
.map(AggregatorFactory::getName)
.collect(Collectors.toList());
- SegmentId segmentId = SegmentId.of(
+ final int numRows;
+ final SegmentId segmentId = SegmentId.of(
getDataSource(),
interval,
Preconditions.checkNotNull(AbstractBatchIndexTask.findVersion(
@@ -295,6 +296,9 @@ abstract class PartialSegmentMergeTask<S extends ShardSpec>
extends PerfectRollu
), "version for interval[%s]", interval),
0
);
+ try (QueryableIndex index =
toolbox.getIndexIO().loadIndex(mergedFileAndDimensionNames.lhs)) {
+ numRows = index.getNumRows();
+ }
final DataSegment segment = segmentPusher.push(
mergedFileAndDimensionNames.lhs,
@@ -302,6 +306,7 @@ abstract class PartialSegmentMergeTask<S extends ShardSpec>
extends PerfectRollu
.shardSpec(createShardSpec(toolbox, interval, bucketId))
.dimensions(mergedFileAndDimensionNames.rhs)
.metrics(metricNames)
+ .totalRows(numRows)
.projections(dataSchema.getProjectionNames())
.build(),
false
@@ -324,12 +329,13 @@ abstract class PartialSegmentMergeTask<S extends
ShardSpec> extends PerfectRollu
}
LOG.info("Built segment [%s] for interval [%s] (from [%d] input
segment(s) in [%,d]ms) of "
- + "size [%d] bytes and pushed ([%,d]ms) to deep storage [%s].",
+ + "size [%d] bytes, [%d] rows and pushed ([%,d]ms) to deep storage
[%s].",
segment.getId(),
interval,
segmentFilesToMerge.size(),
(mergeFinishTime - startTime) / 1000000,
segment.getSize(),
+ segment.getTotalRows(),
(pushFinishTime - mergeFinishTime) / 1000000,
segment.getLoadSpec()
);
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
index ef60d61e31d..4cf0e70639a 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -1230,6 +1230,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
handoffWaitMs,
getPartitionStats(),
null,
+ null,
null
)
),
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskUtilsTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskUtilsTest.java
index 77d5015a28e..5538289c3f1 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskUtilsTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskUtilsTest.java
@@ -19,20 +19,28 @@
package org.apache.druid.indexing.common.task;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.timeline.DataSegment;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.util.Arrays;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Stream;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
@@ -110,4 +118,40 @@ public class IndexTaskUtilsTest
IndexTaskUtils.setTaskDimensions(metricBuilder, abstractTask);
Assertions.assertNull(metricBuilder.getDimension(DruidMetrics.GROUP_ID));
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("oversizedSegmentCases")
+ public void testGetOversizedSegments(
+ String name,
+ List<Integer> rowCounts,
+ int maxRowsPerSegment,
+ double oversizedRatio,
+ long expected
+ )
+ {
+ final List<DataSegment> segments = rowCounts.stream()
+ .map(rowCount -> {
+ DataSegment segment = Mockito.mock(DataSegment.class);
+ Mockito.when(segment.getTotalRows()).thenReturn(rowCount);
+ return segment;
+ })
+ .collect(ImmutableList.toImmutableList());
+ Assertions.assertEquals(
+ expected,
+ IndexTaskUtils.getOversizedSegments(segments, maxRowsPerSegment,
oversizedRatio)
+ );
+ segments.forEach(segment -> {
+ Mockito.verify(segment, Mockito.times(1)).getTotalRows();
+ });
+ }
+ public static Stream<Arguments> oversizedSegmentCases()
+ {
+ return Stream.of(
+ Arguments.of("empty", List.<Integer>of(), 5, 2.0, 0L),
+ Arguments.of("equal to threshold is not oversized", List.of(10), 5,
2.0, 0L),
+ Arguments.of("over threshold", List.of(11), 5, 2.0, 1L),
+ Arguments.of("null totalRows ignored", Arrays.asList(11, null, 3), 5,
2.0, 1L),
+ Arguments.of("mixed", List.of(3, 11, 10, 21), 5, 2.0, 2L)
+ );
+ }
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
index 5baca01a71f..baa483f28f4 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
@@ -19,7 +19,6 @@
package org.apache.druid.indexing.common.task.batch.parallel;
-import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.data.input.InputFormat;
import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.LocalInputSource;
@@ -168,7 +167,10 @@ abstract class AbstractMultiPhaseParallelIndexingTest
extends AbstractParallelIn
TaskReport.ReportMap runTaskAndGetReports(Task task, TaskState
expectedTaskStatus)
{
runTaskAndVerifyStatus(task, expectedTaskStatus);
- return
FutureUtils.getUnchecked(getIndexingServiceClient().taskReportAsMap(task.getId()),
true);
+ // Live reports always omit oversizedSegments; use the completion report
written after publish.
+ final ParallelIndexSupervisorTask executedTask =
+ (ParallelIndexSupervisorTask)
getIndexingServiceClient().getTaskContainer(task.getId()).getTask();
+ return executedTask.getCompletionReports();
}
protected ParallelIndexSupervisorTask createTask(
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
index 1ca23a349da..345cf2ff70b 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
@@ -786,6 +786,7 @@ public class AbstractParallelIndexSupervisorTaskTest
extends IngestionTestBase
0L,
null,
null,
+ null,
null
)
)
@@ -795,7 +796,8 @@ public class AbstractParallelIndexSupervisorTaskTest
extends IngestionTestBase
protected TaskReport.ReportMap buildExpectedTaskReportParallel(
String taskId,
List<ParseExceptionReport> expectedUnparseableEvents,
- RowIngestionMetersTotals expectedTotals
+ RowIngestionMetersTotals expectedTotals,
+ Long oversizedSegments
)
{
Map<String, Object> unparseableEvents = ImmutableMap.of("buildSegments",
expectedUnparseableEvents);
@@ -812,7 +814,8 @@ public class AbstractParallelIndexSupervisorTaskTest
extends IngestionTestBase
0L,
null,
null,
- null
+ null,
+ oversizedSegments
)
)
);
@@ -861,6 +864,7 @@ public class AbstractParallelIndexSupervisorTaskTest
extends IngestionTestBase
.stream().map(ParseExceptionReport::getInput).collect(Collectors.toList());
List<String> actualInputs = actualParseExceptionReports
.stream().map(ParseExceptionReport::getInput).collect(Collectors.toList());
+ Assertions.assertEquals(expectedPayload.getOversizedSegments(),
actualPayload.getOversizedSegments());
Assertions.assertEquals(expectedInputs, actualInputs);
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/MultiPhaseParallelIndexingRowStatsTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/MultiPhaseParallelIndexingRowStatsTest.java
index 52852ad922b..ec5cb2af87a 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/MultiPhaseParallelIndexingRowStatsTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/MultiPhaseParallelIndexingRowStatsTest.java
@@ -66,6 +66,13 @@ public class MultiPhaseParallelIndexingRowStatsTest extends
AbstractMultiPhasePa
private static final Interval INTERVAL_TO_INDEX =
Intervals.of("2017-12/P1M");
+ // Same day and dim1 so range cannot split the hot key. Unique dim2 avoids
rollup.
+ // Hashing only on dim1 still maps all rows into one bucket.
+ private static final String SKEWED_DAY = "2017-12-1";
+ private static final int SKEWED_DIM1 = 0;
+ private static final int SKEWED_FILE_COUNT = 5;
+ private static final int SKEWED_ROWS_PER_FILE = 20;
+
private File inputDir;
public MultiPhaseParallelIndexingRowStatsTest()
@@ -93,6 +100,15 @@ public class MultiPhaseParallelIndexingRowStatsTest extends
AbstractMultiPhasePa
}
}
+ for (int i = 0; i < SKEWED_FILE_COUNT; i++) {
+ try (final Writer writer =
+ Files.newBufferedWriter(new File(inputDir, "skewed_" +
i).toPath(), StandardCharsets.UTF_8)) {
+ for (int j = 0; j < SKEWED_ROWS_PER_FILE; j++) {
+ writer.write(StringUtils.format("%s,%d,%d th test file\n",
SKEWED_DAY, SKEWED_DIM1, i * SKEWED_ROWS_PER_FILE + j));
+ }
+ }
+ }
+
for (int i = 0; i < 5; i++) {
try (final Writer writer =
Files.newBufferedWriter(new File(inputDir, "filtered_" +
i).toPath(), StandardCharsets.UTF_8)) {
@@ -143,7 +159,8 @@ public class MultiPhaseParallelIndexingRowStatsTest extends
AbstractMultiPhasePa
: buildExpectedTaskReportParallel(
task.getId(),
ImmutableList.of(),
- expectedTotals
+ expectedTotals,
+ null
);
TaskReport.ReportMap actualReports = runTaskAndGetReports(task,
TaskState.SUCCESS);
@@ -169,9 +186,72 @@ public class MultiPhaseParallelIndexingRowStatsTest
extends AbstractMultiPhasePa
TaskReport.ReportMap expectedReports = buildExpectedTaskReportParallel(
task.getId(),
ImmutableList.of(),
- new RowIngestionMetersTotals(200, 5630, 0, 0, 0)
+ new RowIngestionMetersTotals(200, 5630, 0, 0, 0),
+ 0L
);
TaskReport.ReportMap actualReports = runTaskAndGetReports(task,
TaskState.SUCCESS);
compareTaskReports(expectedReports, actualReports);
}
+
+ @Test
+ public void testHashPartitionRowStatsWithOversizedSegments()
+ {
+ final int maxNumConcurrentSubTasks = 10;
+ final int maxRowsPerSegment = 20;
+
+ ParallelIndexSupervisorTask task = createTask(
+ TIMESTAMP_SPEC,
+ DIMENSIONS_SPEC,
+ INPUT_FORMAT,
+ INTERVAL_TO_INDEX,
+ inputDir,
+ "skewed_*",
+ new HashedPartitionsSpec(maxRowsPerSegment, null,
ImmutableList.of(DIM1), null),
+ maxNumConcurrentSubTasks,
+ false,
+ false
+ );
+
+ final RowIngestionMetersTotals expectedTotals =
RowMeters.with().bytes(2790).totalProcessed(100);
+ final TaskReport.ReportMap expectedReports =
buildExpectedTaskReportParallel(
+ task.getId(),
+ ImmutableList.of(),
+ expectedTotals,
+ 1L
+ );
+
+ TaskReport.ReportMap actualReports = runTaskAndGetReports(task,
TaskState.SUCCESS);
+ compareTaskReports(expectedReports, actualReports);
+ }
+
+ @Test
+ public void testRangePartitionRowStatsWithOversizedSegments()
+ {
+ final int maxNumConcurrentSubTasks = 10;
+ final int targetRowsPerSegment = 20;
+
+ ParallelIndexSupervisorTask task = createTask(
+ TIMESTAMP_SPEC,
+ DIMENSIONS_SPEC,
+ INPUT_FORMAT,
+ INTERVAL_TO_INDEX,
+ inputDir,
+ "skewed_*",
+ new SingleDimensionPartitionsSpec(targetRowsPerSegment, null, DIM1,
false),
+ maxNumConcurrentSubTasks,
+ false,
+ false
+ );
+
+ final RowIngestionMetersTotals expectedTotals =
RowMeters.with().bytes(2790).totalProcessed(100);
+ final TaskReport.ReportMap expectedReports =
buildExpectedTaskReportParallel(
+ task.getId(),
+ ImmutableList.of(),
+ expectedTotals,
+ 1L
+ );
+
+ TaskReport.ReportMap actualReports = runTaskAndGetReports(task,
TaskState.SUCCESS);
+ compareTaskReports(expectedReports, actualReports);
+ }
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseParallelIndexingTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseParallelIndexingTest.java
index ff16bb094af..d208d0fd1ce 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseParallelIndexingTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseParallelIndexingTest.java
@@ -498,7 +498,8 @@ public class SinglePhaseParallelIndexingTest extends
AbstractParallelIndexSuperv
1L
)
),
- new RowIngestionMetersTotals(10, 335, 1, expectedThrownAwayByReason, 1)
+ new RowIngestionMetersTotals(10, 335, 1, expectedThrownAwayByReason,
1),
+ null
);
compareTaskReports(expectedReports, actualReports);
}
@@ -564,7 +565,8 @@ public class SinglePhaseParallelIndexingTest extends
AbstractParallelIndexSuperv
expectedReports = buildExpectedTaskReportParallel(
task.getId(),
expectedUnparseableEvents,
- expectedTotals
+ expectedTotals,
+ null
);
}
diff --git
a/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
b/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
index f766d0a324c..0e16cd1baf8 100644
---
a/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
+++
b/processing/src/main/java/org/apache/druid/indexer/report/IngestionStatsAndErrors.java
@@ -38,6 +38,7 @@ public class IngestionStatsAndErrors
private final Map<String, Long> recordsProcessed;
private final Long segmentsRead;
private final Long segmentsPublished;
+ private final Long oversizedSegments;
public IngestionStatsAndErrors(
@JsonProperty("ingestionState") IngestionState ingestionState,
@@ -48,7 +49,8 @@ public class IngestionStatsAndErrors
@JsonProperty("segmentAvailabilityWaitTimeMs") long
segmentAvailabilityWaitTimeMs,
@JsonProperty("recordsProcessed") Map<String, Long> recordsProcessed,
@Nullable @JsonProperty("segmentsRead") Long segmentsRead,
- @Nullable @JsonProperty("segmentsPublished") Long segmentsPublished
+ @Nullable @JsonProperty("segmentsPublished") Long segmentsPublished,
+ @Nullable @JsonProperty("oversizedSegments") Long oversizedSegments
)
{
this.ingestionState = ingestionState;
@@ -60,6 +62,7 @@ public class IngestionStatsAndErrors
this.recordsProcessed = recordsProcessed;
this.segmentsRead = segmentsRead;
this.segmentsPublished = segmentsPublished;
+ this.oversizedSegments = oversizedSegments;
}
@JsonProperty
@@ -122,6 +125,14 @@ public class IngestionStatsAndErrors
return segmentsPublished;
}
+ @JsonProperty
+ @Nullable
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public Long getOversizedSegments()
+ {
+ return oversizedSegments;
+ }
+
public static IngestionStatsAndErrors getPayloadFromTaskReports(
Map<String, TaskReport> taskReports
)
@@ -148,7 +159,8 @@ public class IngestionStatsAndErrors
Objects.equals(getSegmentAvailabilityWaitTimeMs(),
that.getSegmentAvailabilityWaitTimeMs()) &&
Objects.equals(getRecordsProcessed(), that.getRecordsProcessed()) &&
Objects.equals(getSegmentsRead(), that.getSegmentsRead()) &&
- Objects.equals(getSegmentsPublished(), that.getSegmentsPublished());
+ Objects.equals(getSegmentsPublished(), that.getSegmentsPublished())
&&
+ Objects.equals(getOversizedSegments(), that.getOversizedSegments());
}
@Override
@@ -163,7 +175,8 @@ public class IngestionStatsAndErrors
getSegmentAvailabilityWaitTimeMs(),
getRecordsProcessed(),
getSegmentsRead(),
- getSegmentsPublished()
+ getSegmentsPublished(),
+ getOversizedSegments()
);
}
@@ -180,6 +193,7 @@ public class IngestionStatsAndErrors
", recordsProcessed=" + recordsProcessed +
", segmentsRead=" + segmentsRead +
", segmentsPublished=" + segmentsPublished +
+ ", oversizedSegments=" + oversizedSegments +
'}';
}
}
diff --git a/processing/src/main/resources/loggingEmitterAllowedMetrics.json
b/processing/src/main/resources/loggingEmitterAllowedMetrics.json
index 9bcdb712952..e0d89fac679 100644
--- a/processing/src/main/resources/loggingEmitterAllowedMetrics.json
+++ b/processing/src/main/resources/loggingEmitterAllowedMetrics.json
@@ -37,6 +37,7 @@
"ingest/persists/time": [],
"ingest/rows/output": [],
"ingest/segments/count": [],
+ "ingest/segments/oversized": [],
"ingest/rows/published": [],
"ingest/sink/count": [],
"ingest/tombstones/count": [],
diff --git
a/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
b/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
index 67957e9a047..e1c0ac44029 100644
---
a/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
+++
b/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
@@ -273,6 +273,7 @@ public class TaskReportSerdeTest
1000L,
null,
null,
+ null,
null
)
);
@@ -311,7 +312,8 @@ public class TaskReportSerdeTest
1000L,
Collections.singletonMap("PartitionA", 5000L),
5L,
- 10L
+ 10L,
+ 2L
)
);
}
diff --git a/processing/src/test/resources/loggingEmitterAllowedMetrics.json
b/processing/src/test/resources/loggingEmitterAllowedMetrics.json
index 3d3200362f0..5818ed2d36a 100644
--- a/processing/src/test/resources/loggingEmitterAllowedMetrics.json
+++ b/processing/src/test/resources/loggingEmitterAllowedMetrics.json
@@ -37,6 +37,7 @@
"ingest/persists/time": [],
"ingest/rows/output": [],
"ingest/segments/count": [],
+ "ingest/segments/oversized": [],
"ingest/rows/published": [],
"ingest/sink/count": [],
"ingest/tombstones/count": [],
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
