Add DruidInputSource (replacement for IngestSegmentFirehose) by jon-wei · Pull Request #8982 · apache/druid

jon-wei · 2019-12-03T08:46:10Z

Based on an initial patch by @jihoonson

This PR adds DruidInputSource and associated classes, a new replacement for IngestSegmentFirehose. This input source always has a fixed input format (DruidInputFormat). The row reading class is DruidSegmentReader and the input entity is DruidInputEntity.

The new input source has the same parameters as the existing IngestSegmentFirehose, and has similar code.

Additionally:

The PR adjusts the CompactionTask so that it uses DruidInputSource instead of the old firehose.
Fixes an issue where the value of inputSource.needsFormat() was not being respected.
The new input source avoids the issue described in ingestSegment firehose is not friendly to the sampler #8448, which is present in the old firehose

For testing, this patch relies on the existing CompactionTask tests for checking functionality of the new input source.

A test where an IngestSegmentFirehose is used for reindexing with a native batch task (non-parallel) has also been added to CompactionRunTaskTest. The existing ITParallelIndexTaskTest still uses a firehose instead of input source.

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
added unit tests or modified existing tests to cover new code paths.
added integration tests.
been tested in a test Druid cluster.

jon-wei · 2019-12-03T08:46:58Z

Marking WIP, need to update the docs, and will add an integration test to ITParallelIndexTaskTest that uses the new input source.

vogievetsky · 2019-12-03T20:42:29Z

Tested this (as a user) from the console (via #8828 ) and it works

jon-wei · 2019-12-04T03:38:52Z

Added a doc entry for the new input source. We'll need a larger rework of the docs for the new input sources/input formats, but that should be handled in another PR.

Adjusted ITParallelIndexTaskTest so that it also runs reingestion using DruidInputSource

clintropolis · 2019-12-04T08:18:32Z

+{
+    "type": "druid",
+    "dataSource": "wikipedia",
+    "interval": "2013-01-01/2013-01-02"


nit: It would be nice if either this example or a 2nd example included dimensions, metrics, and filter

Added a second example with more explanation of what the example specs do

clintropolis · 2019-12-04T08:21:10Z

+|property|description|required?|
+|--------|-----------|---------|
+|type|This should be "druid".|yes|
+|dataSource|A String defining the data source to fetch rows from, very similar to a table in a relational database|yes|


nit: this part 'very similar to a table in a relational database' feels sort of out of place to me

Reworded this part and other parts of this table

clintropolis · 2019-12-04T08:44:31Z

+        ),
+        source.getIntervalFilter()
+    );
+    final Sequence<Map<String, Object>> sequence = Sequences.concat(


This logic is pretty beastly to follow and I think could definitely stand to be broken down into a couple of methods. I suggest one that makes a Sequence from a Cursor, and one that makes a CloseableIterator from a Sequence. Then this part can be simplified to something like:

final Sequence<Cursor> cursors = storageAdapter.getAdapter().makeCursors( Filters.toFilter(dimFilter), storageAdapter.getInterval(), VirtualColumns.EMPTY, Granularities.ALL, false, null ); final Sequence<Map<String, Object>> sequence = Sequences.concat(Sequences.map(cursors, this::cursorToSequence)); return sequenceToCloseableIterator(segmentFile, sequence);

which makes this much more approachable imo

Adjusted this to use the suggested structure

Nice refactoring!

clintropolis · 2019-12-04T08:46:38Z

+      public void close()
+      {
+        if (!segmentFile.delete()) {
+          // log


is this a placeholder for a log message?

Filled in a warning log message

clintropolis · 2019-12-04T08:53:46Z

+    Interval lastInterval = null;
+    for (Interval interval : timeline.keySet()) {
+      if (lastInterval != null) {
+        if (interval.overlaps(lastInterval)) {


nit: i know this isn't really new code, but these if statements can be collapsed

Collapsed the ifs

clintropolis · 2019-12-04T08:54:50Z

+    // segments to olders.
+
+    // timelineSegments are sorted in order of interval
+    int[] index = {0};


hmm... so ugly, though I guess better than AtomicInteger 😢. This should be final at least I think? Alternatively, should this just be implemented how getUniqueDimensions is so that they are more consistent?

I tried changing it to be the same as the dimensions version, but because there is only a presence check for the metrics version (no "exclusions" like with dimensions), the map.put becomes a style failure (tells me to use computeIfAbsent instead)

Made this final

It could be MutableInt, but I think the original code is ok since it's just reusing the existing code.

clintropolis · 2019-12-04T08:58:33Z

+  {
+    Preconditions.checkNotNull(interval);
+
+    // This call used to use the TaskActionClient, so for compatibility we use the same retry configuration


Consider this a non-blocking discussion point, but should this maybe switch to using RetryUtils?

I left this for now

I don't know why this has its own retry logic too.. I think it's worth to consider it for future refactoring.

clintropolis · 2019-12-04T09:08:18Z

+  }
+
+  @VisibleForTesting
+  private static List<String> getUniqueDimensions(


nit: Would this method and getUniqueMetrics maybe more appropriately live in VersionedIntervalTimeline?

This method does not seem to be covered by any of the tests in org.apache.druid.indexing

I moved these to VersionedIntervalTimeline, IngestSegmentFirehoseFactory.testGetUniqueDimensionsAndMetrics should be covering these now

ccaominh · 2019-12-04T18:59:50Z

+  }
+
+  @VisibleForTesting
+  private static List<String> getUniqueDimensions(


This method does not seem to be covered by any of the tests in org.apache.druid.indexing

ccaominh · 2019-12-04T19:00:03Z

+  }
+
+  @VisibleForTesting
+  private static List<String> getUniqueMetrics(List<TimelineObjectHolder<String, DataSegment>> timelineSegments)


This method does not seem to be covered by any of the tests in org.apache.druid.indexing

I moved these to VersionedIntervalTimeline, IngestSegmentFirehoseFactory.testGetUniqueDimensionsAndMetrics should be covering these now

ccaominh · 2019-12-04T19:01:28Z

+                          List<String> dimVals = new ArrayList<>(valsSize);
+                          for (int i = 0; i < valsSize; ++i) {
+                            dimVals.add(selector.lookupName(vals.get(i)));
+                          }
+                          theEvent.put(dim, dimVals);


This part does not seem to be covered by any of the tests in org.apache.druid.indexing

I adjusted CompactionRunTaskTest so that there's a multivalued dimension (this case handles that), and added a row correctness check to testRun there

ccaominh · 2019-12-04T19:02:10Z

+            // This segment won't fit in the current non-empty split, so this split is done.
+            splits.add(new InputSplit<>(currentSplit));
+            currentSplit = new ArrayList<>();
+            bytesInCurrentSplit = 0;


May be worth having a unit test for this method. Looks like this part is not covered by any of the tests in org.apache.druid.indexing.

I added testDruidInputSourceCreateSplitsWithIndividualSplits to CompactionTaskParallelRunTest

ccaominh

LGTM 👍

ccaominh · 2019-12-05T17:54:00Z

+|dataSource|A String defining the Druid datasource to fetch rows from|yes|
+|interval|A String representing the ISO-8601 interval, which defines the time range to fetch the data over.|yes|
+|dimensions|A list of Strings containing the names of dimension columns to select from the Druid datasource. If the list is empty, no dimensions are returned. If null, all dimensions are returned. |no|
+|metrics|The list of Strings containg the names of metric columns to select. If the list is empty, no metrics are returned. If null, all metrics are returned.|no|


typo: containg -> containing

(travis failed on docs spellcheck)

Fixed spelling

jihoonson · 2019-12-05T19:18:43Z

  }

+  @VisibleForTesting
+  public static List<String> getUniqueDimensions(


The place of these methods seems a bit unintuitive to me. Their first parameter is a list of TimelineObjectHolders because I think it's easier to use in DruidInputSource and IngestSegmentFirehose. But, in fact, a list of DataSegments would be enough. Is there any other place that needs to use these methods other than DruidInputSource or IngestSegmentFirehose? If not, I guess a new util class such as CompactionUtils or DataSegments might be better?

I moved these to a new class ReingestionTimelineUtils and added some javadocs

The existing comments within the method mention ordering based on recency of segments, so I kept the input as TimelineObjectHolders

Hm, so I had suggested moving them to VersionedIntervalTimeline because they seem like generic timeline operations, not specific to re-ingesting segments. Is having a ReingestionTimelineUtils better than their original home if they are still going to live somewhere specific to re-ingesting segments?

Hmm, my only view on this is it should be somewhere other than IngestSegmentFirehoseFactory since that's headed for eventual deprecation.

It could go in VersionedIntervalTimeline although that class is a bit weird (it accepts generic types but also has methods that specifically use DataSegment). I feel like the utils class is fine for now

My stance is similar to @jon-wei. It could be used for more general use cases, but I don't see them now. Maybe we can move them from ReingestionTimelineUtils to somewhere else if we need in the future.

jihoonson · 2019-12-05T19:20:30Z

+    // segments to olders.
+
+    // timelineSegments are sorted in order of interval
+    int[] index = {0};


It could be MutableInt, but I think the original code is ok since it's just reusing the existing code.

jihoonson

@jon-wei thank you for making this as a PR!

jihoonson · 2019-12-05T19:26:37Z

+  {
+    Preconditions.checkNotNull(interval);
+
+    // This call used to use the TaskActionClient, so for compatibility we use the same retry configuration


I don't know why this has its own retry logic too.. I think it's worth to consider it for future refactoring.

jihoonson · 2019-12-05T19:29:09Z

+    }
+
+    @Override
+    public void remove()


nit: the Iterator interface has the default implementation for remove() throwing UnsupportedOperationException.

Removed the redundant override

jihoonson · 2019-12-05T19:30:41Z

+        ),
+        source.getIntervalFilter()
+    );
+    final Sequence<Map<String, Object>> sequence = Sequences.concat(


Nice refactoring!

clintropolis

👍

jihoonson and others added 4 commits December 2, 2019 13:52

Add Druid input source and format

65618b0

Inherit dims/metrics from segment

0186fef

Add ingest segment firehose reindexing test

5e20ef5

Remove unnecessary module

11d263d

jon-wei added Feature Area - Batch Ingestion Area - Streaming Ingestion WIP labels Dec 3, 2019

jon-wei added 3 commits December 3, 2019 16:42

Fix unit tests, checkstyle

1d66b54

Add doc entry

9459782

Fix dimensionExclusions handling, add parallel index integration test

b6a6b48

jon-wei removed the WIP label Dec 4, 2019

Add spelling exclusion

9c5e1d8

clintropolis reviewed Dec 4, 2019

View reviewed changes

ccaominh reviewed Dec 4, 2019

View reviewed changes

jon-wei added 3 commits December 4, 2019 19:52

Address some PR comments

32adef5

Checkstyle

10940a0

wip

7f31c0f

jihoonson removed the Area - Streaming Ingestion label Dec 5, 2019

Address rest of PR comments

8b68ff9

ccaominh approved these changes Dec 5, 2019

View reviewed changes

ccaominh reviewed Dec 5, 2019

View reviewed changes

jihoonson reviewed Dec 5, 2019

View reviewed changes

Address PR comments

65c0d0d

clintropolis approved these changes Dec 5, 2019

View reviewed changes

jihoonson approved these changes Dec 5, 2019

View reviewed changes

jon-wei merged commit c949a25 into apache:master Dec 6, 2019

jon-wei added this to the 0.17.0 milestone Dec 17, 2019

Conversation

jon-wei commented Dec 3, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jon-wei commented Dec 3, 2019

Uh oh!

vogievetsky commented Dec 3, 2019

Uh oh!

jon-wei commented Dec 4, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ccaominh left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ccaominh Dec 5, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jon-wei commented Dec 3, 2019 •

edited

Loading

ccaominh left a comment •

edited

Loading

ccaominh Dec 5, 2019 •

edited

Loading