kfaraz commented on code in PR #18855: URL: https://github.com/apache/druid/pull/18855#discussion_r2639701932
########## indexing-service/src/main/java/org/apache/druid/indexing/common/task/InputRowFilter.java: ########## @@ -0,0 +1,73 @@ +/* + * 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; + +import org.apache.druid.data.input.InputRow; +import org.apache.druid.segment.incremental.InputRowFilterResult; + +import java.util.function.Predicate; + +/** + * A filter for input rows during ingestion that returns a {@link InputRowFilterResult} per row. + * This is similar to {@link Predicate} but returns an {@link InputRowFilterResult} instead of a boolean. + */ +@FunctionalInterface +public interface InputRowFilter +{ + /** + * Tests whether the given row should be accepted. + * + * @return {@link InputRowFilterResult#ACCEPTED} if the row should be accepted, or another {@link InputRowFilterResult} value if the row should be rejected Review Comment: ```suggestion * @return {@link InputRowFilterResult#ACCEPTED} only if the row should be accepted, otherwise another {@link InputRowFilterResult} value. ``` ########## processing/src/main/java/org/apache/druid/segment/incremental/InputRowFilterResult.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.segment.incremental; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * Result of filtering an input row during ingestion. + */ +public enum InputRowFilterResult +{ + /** + * The row passed the filter and should be processed. + */ + ACCEPTED("accepted"), + /** + * The row was null or the input record was empty. + */ + NULL_OR_EMPTY_RECORD("null"), + + /** + * The row's timestamp is before the minimum message time (late message rejection). + */ + BEFORE_MIN_MESSAGE_TIME("beforeMinMessageTime"), + + /** + * The row's timestamp is after the maximum message time (early message rejection). + */ + AFTER_MAX_MESSAGE_TIME("afterMaxMessageTime"), + + /** + * The row was filtered out by a transformSpec filter or other row filter. + */ + FILTERED("filtered"), + + /** + * A backwards-compatible value for tracking filter reasons for ingestion tasks using older Druid versions without filter reason tracking. + */ + UNKNOWN("unknown"); + + private static final InputRowFilterResult[] REJECTED_VALUES = Arrays.stream(InputRowFilterResult.values()) + .filter(InputRowFilterResult::isRejected) + .toArray(InputRowFilterResult[]::new); + public static final int NUM_FILTER_RESULT = InputRowFilterResult.values().length; + + private final String reason; + + InputRowFilterResult(String reason) + { + this.reason = reason; + } + + /** + * Returns string value representation of this {@link InputRowFilterResult} for metric emission. + */ + public String getReason() + { + return reason; + } + + /** + * Returns true if this result indicates the row was rejected (thrown away). + * Returns false for {@link #ACCEPTED}. + */ + public boolean isRejected() + { + return this != ACCEPTED; + } + + /** + * Public utility for building a mutable frequency map over the possible rejection {@link InputRowFilterResult} values. + * Keys on {@link InputRowFilterResult#getReason()} rather than the enum name as the latter is more likely to change longer-term. + * It is also easier to have stats payload keys match what is being emitted in metrics. + */ + public static Map<String, Long> buildRejectedCounterMap() Review Comment: I don't think we need this utility method. ########## indexing-service/src/main/java/org/apache/druid/indexing/common/stats/TaskRealtimeMetricsMonitor.java: ########## @@ -23,16 +23,21 @@ import org.apache.druid.java.util.emitter.service.ServiceEmitter; import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; import org.apache.druid.java.util.metrics.AbstractMonitor; +import org.apache.druid.segment.incremental.InputRowFilterResult; import org.apache.druid.segment.incremental.RowIngestionMeters; import org.apache.druid.segment.incremental.RowIngestionMetersTotals; import org.apache.druid.segment.realtime.SegmentGenerationMetrics; +import java.util.HashMap; +import java.util.Map; + /** * Emits metrics from {@link SegmentGenerationMetrics} and {@link RowIngestionMeters}. */ public class TaskRealtimeMetricsMonitor extends AbstractMonitor { private static final EmittingLogger log = new EmittingLogger(TaskRealtimeMetricsMonitor.class); + private static final String REASON_DIMENSION = "reason"; Review Comment: Move this to `DruidMetrics`? ########## processing/src/main/java/org/apache/druid/segment/incremental/InputRowFilterResult.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.segment.incremental; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * Result of filtering an input row during ingestion. + */ +public enum InputRowFilterResult +{ + /** + * The row passed the filter and should be processed. + */ + ACCEPTED("accepted"), + /** + * The row was null or the input record was empty. + */ + NULL_OR_EMPTY_RECORD("null"), + + /** + * The row's timestamp is before the minimum message time (late message rejection). + */ + BEFORE_MIN_MESSAGE_TIME("beforeMinMessageTime"), + + /** + * The row's timestamp is after the maximum message time (early message rejection). + */ + AFTER_MAX_MESSAGE_TIME("afterMaxMessageTime"), + + /** + * The row was filtered out by a transformSpec filter or other row filter. + */ + FILTERED("filtered"), Review Comment: ```suggestion REJECTED_BY_FILTER("rejectedByFilter"), ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/common/stats/DropwizardRowIngestionMeters.java: ########## @@ -100,13 +104,27 @@ public void incrementUnparseable() @Override public long getThrownAway() { - return thrownAway.getCount(); + long totalThrownAway = 0; + for (InputRowFilterResult reason : InputRowFilterResult.rejectedValues()) { + totalThrownAway += thrownAwayByReason[reason.ordinal()].getCount(); + } + return totalThrownAway; } @Override - public void incrementThrownAway() + public void incrementThrownAway(InputRowFilterResult reason) { - thrownAway.mark(); + thrownAwayByReason[reason.ordinal()].mark(); + } + + @Override + public Map<String, Long> getThrownAwayByReason() + { + Map<String, Long> result = InputRowFilterResult.buildRejectedCounterMap(); Review Comment: Why can't we just initialize this to an empty hash map? It is odd to have the map be constructed elsewhere. Is it just so that we have 0 values for the missing reasons? ########## processing/src/main/java/org/apache/druid/segment/incremental/RowIngestionMetersTotals.java: ########## @@ -38,13 +42,49 @@ public RowIngestionMetersTotals( @JsonProperty("processedBytes") long processedBytes, @JsonProperty("processedWithError") long processedWithError, @JsonProperty("thrownAway") long thrownAway, + @JsonProperty("thrownAwayByReason") @Nullable Map<String, Long> thrownAwayByReason, + @JsonProperty("unparseable") long unparseable + ) + { + this( + processed, + processedBytes, + processedWithError, + Configs.valueOrDefault(thrownAwayByReason, getBackwardsCompatibleThrownAwayByReason(thrownAway)), + unparseable + ); + } + + public RowIngestionMetersTotals( + @JsonProperty("processed") long processed, Review Comment: Since this is not the `@JsonCreator` constructor, we don't need the `@JsonProperty` annotations here. ########## processing/src/main/java/org/apache/druid/segment/incremental/RowIngestionMetersTotals.java: ########## @@ -38,13 +42,49 @@ public RowIngestionMetersTotals( @JsonProperty("processedBytes") long processedBytes, @JsonProperty("processedWithError") long processedWithError, @JsonProperty("thrownAway") long thrownAway, + @JsonProperty("thrownAwayByReason") @Nullable Map<String, Long> thrownAwayByReason, + @JsonProperty("unparseable") long unparseable + ) + { + this( + processed, + processedBytes, + processedWithError, + Configs.valueOrDefault(thrownAwayByReason, getBackwardsCompatibleThrownAwayByReason(thrownAway)), + unparseable + ); + } + + public RowIngestionMetersTotals( + @JsonProperty("processed") long processed, + @JsonProperty("processedBytes") long processedBytes, + @JsonProperty("processedWithError") long processedWithError, + @JsonProperty("thrownAway") long thrownAway, + @JsonProperty("unparseable") long unparseable + ) + { + this( + processed, + processedBytes, + processedWithError, + getBackwardsCompatibleThrownAwayByReason(thrownAway), + unparseable + ); + } + + public RowIngestionMetersTotals( Review Comment: Do we need these extra constructors? ########## processing/src/main/java/org/apache/druid/segment/incremental/RowIngestionMetersTotals.java: ########## @@ -109,7 +156,19 @@ public String toString() ", processedBytes=" + processedBytes + ", processedWithError=" + processedWithError + ", thrownAway=" + thrownAway + + ", thrownAwayByReason=" + thrownAwayByReason + ", unparseable=" + unparseable + '}'; } + + /** + * For backwards compatibility, key by {@link InputRowFilterResult} in case of lack of thrownAwayByReason input during rolling Druid upgrades. + * This can occur when tasks running on older Druid versions return ingest statistic payloads to an overlord running on a newer Druid version. + */ + private static Map<String, Long> getBackwardsCompatibleThrownAwayByReason(long thrownAway) + { + Map<String, Long> results = InputRowFilterResult.buildRejectedCounterMap(); + results.put(InputRowFilterResult.UNKNOWN.getReason(), thrownAway); + return results; Review Comment: ```suggestion return Map.of(InputRowFilterResult.UNKNOWN.getReason(), thrownAway); ``` ########## processing/src/main/java/org/apache/druid/segment/incremental/InputRowFilterResult.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.segment.incremental; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** + * Result of filtering an input row during ingestion. + */ +public enum InputRowFilterResult +{ + /** + * The row passed the filter and should be processed. + */ + ACCEPTED("accepted"), + /** + * The row was null or the input record was empty. + */ + NULL_OR_EMPTY_RECORD("null"), + + /** + * The row's timestamp is before the minimum message time (late message rejection). Review Comment: ```suggestion * The timestamp of the row is before the minimum message time (late message rejection). ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java: ########## @@ -2144,26 +2146,35 @@ private void refreshMinMaxMessageTime() ); } - public boolean withinMinMaxRecordTime(final InputRow row) + /** + * Returns the filter result for a row. + * Returns {@link InputRowFilterResult#ACCEPTED} if the row should be accepted, + * or a rejection reason otherwise. + * This method is used as a {@link InputRowFilter} for the {@link StreamChunkParser}. Review Comment: I think this can be shortened, as the other lines don't add any extra info. ```suggestion * Returns {@link InputRowFilterResult#ACCEPTED} if the row should be accepted, * or a rejection reason otherwise. ``` -- 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]
