jihoonson commented on a change in pull request #10336:
URL: https://github.com/apache/druid/pull/10336#discussion_r482314233
##########
File path:
indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
##########
@@ -135,6 +148,49 @@ public void stopGracefully(TaskConfig taskConfig)
}
}
+ public static FilteringCloseableInputRowIterator inputSourceReader(
+ File tmpDir,
+ DataSchema dataSchema,
+ InputSource inputSource,
+ @Nullable InputFormat inputFormat,
+ Predicate<InputRow> rowFilter,
+ RowIngestionMeters ingestionMeters,
+ ParseExceptionHandler parseExceptionHandler
+ ) throws IOException
+ {
+ final List<String> metricsNames =
Arrays.stream(dataSchema.getAggregators())
+ .map(AggregatorFactory::getName)
+ .collect(Collectors.toList());
+ final InputSourceReader inputSourceReader =
dataSchema.getTransformSpec().decorate(
+ inputSource.reader(
+ new InputRowSchema(
+ dataSchema.getTimestampSpec(),
+ dataSchema.getDimensionsSpec(),
+ metricsNames
+ ),
Review comment:
Yeah, it would be useful. Or we can change `InputRowSchema` to have all
type information of metrics as well, so that we can check exceptions in parsing
and type casting in only `InputEntityReader`. It means, after this change, you
will not have to worry about parseExceptions in `IncrementalIndex` anymore.
Probably we don't need `ParseExceptionHandler` at all in that case. I haven't
done this refactoring here because it will be quite complicated and this PR is
already big enough.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
