Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Remove "granularity" from IngestSegmentFirehose. #4110

Merged
merged 1 commit into from
Mar 24, 2017
Merged

Remove "granularity" from IngestSegmentFirehose. #4110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion docs/content/ingestion/update-existing-data.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ Here is what goes inside `ingestionSpec`:
|dataSource|String|Druid dataSource name from which you are loading the data.|yes|
|intervals|List|A list of strings representing ISO-8601 Intervals.|yes|
|segments|List|List of segments from which to read data from, by default it is obtained automatically. You can obtain list of segments to put here by making a POST query to coordinator at url /druid/coordinator/v1/metadata/datasources/segments?full with list of intervals specified in the request paylod e.g. ["2012-01-01T00:00:00.000/2012-01-03T00:00:00.000", "2012-01-05T00:00:00.000/2012-01-07T00:00:00.000"]. You may want to provide this list manually in order to ensure that segments read are exactly same as they were at the time of task submission, task would fail if the list provided by the user does not match with state of database when the task actually runs.|no|
|granularity|String|Defines the granularity of the query while loading data. Default value is "none". See [Granularities](../querying/granularities.html).|no|
|filter|JSON|See [Filters](../querying/filters.html)|no|
|dimensions|Array of String|Name of dimension columns to load. By default, the list will be constructed from parseSpec. If parseSpec does not have an explicit list of dimensions then all the dimension columns present in stored data will be read.|no|
|metrics|Array of String|Name of metric columns to load. By default, the list will be constructed from the "name" of all the configured aggregators.|no|
Expand Down
20 changes: 0 additions & 20 deletions indexing-hadoop/src/main/java/io/druid/indexer/hadoop/DatasourceIngestionSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import io.druid.common.utils.JodaUtils;
import io.druid.java.util.common.granularity.Granularities;
import io.druid.java.util.common.granularity.Granularity;
import io.druid.query.filter.DimFilter;
import io.druid.timeline.DataSegment;
Expand All @@ -38,7 +37,6 @@ public class DatasourceIngestionSpec
private final List<Interval> intervals;
private final List<DataSegment> segments;
private final DimFilter filter;
private final Granularity granularity;
private final List<String> dimensions;
private final List<String> metrics;
private final boolean ignoreWhenNoSegments;
Expand All @@ -50,7 +48,6 @@ public DatasourceIngestionSpec(
@JsonProperty("intervals") List<Interval> intervals,
@JsonProperty("segments") List<DataSegment> segments,
@JsonProperty("filter") DimFilter filter,
@JsonProperty("granularity") Granularity granularity,
@JsonProperty("dimensions") List<String> dimensions,
@JsonProperty("metrics") List<String> metrics,
@JsonProperty("ignoreWhenNoSegments") boolean ignoreWhenNoSegments
Expand All @@ -77,8 +74,6 @@ public DatasourceIngestionSpec(
this.segments = segments;

this.filter = filter;
this.granularity = granularity == null ? Granularities.NONE : granularity;

this.dimensions = dimensions;
this.metrics = metrics;

Expand Down Expand Up @@ -109,12 +104,6 @@ public DimFilter getFilter()
return filter;
}

@JsonProperty
public Granularity getGranularity()
{
return granularity;
}

@JsonProperty
public List<String> getDimensions()
{
Expand All @@ -141,7 +130,6 @@ public DatasourceIngestionSpec withDimensions(List<String> dimensions)
intervals,
segments,
filter,
granularity,
dimensions,
metrics,
ignoreWhenNoSegments
Expand All @@ -156,7 +144,6 @@ public DatasourceIngestionSpec withMetrics(List<String> metrics)
intervals,
segments,
filter,
granularity,
dimensions,
metrics,
ignoreWhenNoSegments
Expand All @@ -171,7 +158,6 @@ public DatasourceIngestionSpec withQueryGranularity(Granularity granularity)
intervals,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now this method is effectively just "clone", the method argument is unused.

segments,
filter,
granularity,
dimensions,
metrics,
ignoreWhenNoSegments
Expand All @@ -186,7 +172,6 @@ public DatasourceIngestionSpec withIgnoreWhenNoSegments(boolean ignoreWhenNoSegm
intervals,
segments,
filter,
granularity,
dimensions,
metrics,
ignoreWhenNoSegments
Expand Down Expand Up @@ -220,9 +205,6 @@ public boolean equals(Object o)
if (filter != null ? !filter.equals(that.filter) : that.filter != null) {
return false;
}
if (!granularity.equals(that.granularity)) {
return false;
}
if (dimensions != null ? !dimensions.equals(that.dimensions) : that.dimensions != null) {
return false;
}
Expand All @@ -237,7 +219,6 @@ public int hashCode()
result = 31 * result + intervals.hashCode();
result = 31 * result + (segments != null ? segments.hashCode() : 0);
result = 31 * result + (filter != null ? filter.hashCode() : 0);
result = 31 * result + granularity.hashCode();
result = 31 * result + (dimensions != null ? dimensions.hashCode() : 0);
result = 31 * result + (metrics != null ? metrics.hashCode() : 0);
result = 31 * result + (ignoreWhenNoSegments ? 1 : 0);
Expand All @@ -252,7 +233,6 @@ public String toString()
", intervals=" + intervals +
", segments=" + segments +
", filter=" + filter +
", granularity=" + granularity +
", dimensions=" + dimensions +
", metrics=" + metrics +
", ignoreWhenNoSegments=" + ignoreWhenNoSegments +
Expand Down
4 changes: 1 addition & 3 deletions indexing-hadoop/src/main/java/io/druid/indexer/hadoop/DatasourceRecordReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;
import com.google.common.io.Files;

import io.druid.data.input.InputRow;
import io.druid.data.input.MapBasedInputRow;
import io.druid.data.input.MapBasedRow;
Expand Down Expand Up @@ -111,8 +110,7 @@ public WindowedStorageAdapter apply(WindowedDataSegment segment)
adapters,
spec.getDimensions(),
spec.getMetrics(),
spec.getFilter(),
spec.getGranularity()
spec.getFilter()
);

}
Expand Down
3 changes: 1 addition & 2 deletions indexing-hadoop/src/test/java/io/druid/indexer/BatchDeltaIngestionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,8 +322,7 @@ private void testIngestion(
ImmutableList.of(new WindowedStorageAdapter(adapter, windowedDataSegment.getInterval())),
ImmutableList.of("host"),
ImmutableList.of("visited_sum", "unique_hosts"),
null,
Granularities.NONE
null
);

List<InputRow> rows = Lists.newArrayList();
Expand Down
6 changes: 1 addition & 5 deletions ...c/test/java/io/druid/indexer/HadoopIngestionSpecUpdateDatasourcePathSpecSegmentsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ public void testupdateSegmentListIfDatasourcePathSpecIsUsedWithJustDatasourcePat
PathSpec pathSpec = new DatasourcePathSpec(
jsonMapper,
null,
new DatasourceIngestionSpec(testDatasource, testDatasourceInterval, null, null, null, null, null, null, false),
new DatasourceIngestionSpec(testDatasource, testDatasourceInterval, null, null, null, null, null, false),
null
);
HadoopDruidIndexerConfig config = testRunUpdateSegmentListIfDatasourcePathSpecIsUsed(
Expand All @@ -119,7 +119,6 @@ public void testupdateSegmentListIfDatasourcePathSpecWithMatchingUserSegments()
null,
null,
null,
null,
false
),
null
Expand Down Expand Up @@ -148,7 +147,6 @@ public void testupdateSegmentListThrowsExceptionWithUserSegmentsMismatch() throw
null,
null,
null,
null,
false
),
null
Expand All @@ -174,7 +172,6 @@ public void testupdateSegmentListIfDatasourcePathSpecIsUsedWithJustDatasourcePat
null,
null,
null,
null,
false
),
null
Expand Down Expand Up @@ -206,7 +203,6 @@ public void testupdateSegmentListIfDatasourcePathSpecIsUsedWithMultiplePathSpec(
null,
null,
null,
null,
false
),
null
Expand Down
6 changes: 1 addition & 5 deletions indexing-hadoop/src/test/java/io/druid/indexer/hadoop/DatasourceIngestionSpecTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import io.druid.java.util.common.granularity.Granularities;
import io.druid.query.filter.SelectorDimFilter;
import io.druid.segment.TestHelper;
import io.druid.timeline.DataSegment;
Expand All @@ -49,7 +48,6 @@ public void testSingleIntervalSerde() throws Exception
null,
null,
new SelectorDimFilter("dim", "value", null),
Granularities.DAY,
Lists.newArrayList("d1", "d2"),
Lists.newArrayList("m1", "m2", "m3"),
false
Expand Down Expand Up @@ -86,7 +84,6 @@ public void testMultiIntervalSerde() throws Exception
null,
null,
null,
null,
false
);

Expand Down Expand Up @@ -133,7 +130,6 @@ public void testMultiIntervalSerde() throws Exception
)
),
new SelectorDimFilter("dim", "value", null),
Granularities.DAY,
Lists.newArrayList("d1", "d2"),
Lists.newArrayList("m1", "m2", "m3"),
true
Expand All @@ -156,7 +152,7 @@ public void testOldJsonDeserialization() throws Exception
DatasourceIngestionSpec actual = MAPPER.readValue(jsonStr, DatasourceIngestionSpec.class);

Assert.assertEquals(
new DatasourceIngestionSpec("test", Interval.parse("2014/2015"), null, null, null, null, null, null, false),
new DatasourceIngestionSpec("test", Interval.parse("2014/2015"), null, null, null, null, null, false),
actual
);
}
Expand Down
1 change: 0 additions & 1 deletion indexing-hadoop/src/test/java/io/druid/indexer/hadoop/DatasourceRecordReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,6 @@ public void testSanity() throws Exception
null,
null,
null,
null,
segment.getDimensions(),
segment.getMetrics(),
false
Expand Down
1 change: 0 additions & 1 deletion indexing-hadoop/src/test/java/io/druid/indexer/path/DatasourcePathSpecTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,6 @@ public DatasourcePathSpecTest()
null,
null,
null,
null,
false
);

Expand Down
8 changes: 2 additions & 6 deletions indexing-service/src/main/java/io/druid/indexing/firehose/IngestSegmentFirehoseFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
import io.druid.indexing.common.TaskToolboxFactory;
import io.druid.indexing.common.actions.SegmentListUsedAction;
import io.druid.indexing.common.task.NoopTask;
import io.druid.java.util.common.granularity.Granularities;
import io.druid.java.util.common.parsers.ParseException;
import io.druid.query.filter.DimFilter;
import io.druid.segment.IndexIO;
Expand Down Expand Up @@ -282,12 +281,9 @@ public WindowedStorageAdapter apply(final PartitionChunk<DataSegment> input)
)
);

return new IngestSegmentFirehose(adapters, dims, metricsList, dimFilter, Granularities.NONE);
return new IngestSegmentFirehose(adapters, dims, metricsList, dimFilter);
}
catch (IOException e) {
throw Throwables.propagate(e);
}
catch (SegmentLoadingException e) {
catch (IOException | SegmentLoadingException e) {
throw Throwables.propagate(e);
}

Expand Down
7 changes: 3 additions & 4 deletions server/src/main/java/io/druid/segment/realtime/firehose/IngestSegmentFirehose.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
import io.druid.data.input.Firehose;
import io.druid.data.input.InputRow;
import io.druid.data.input.MapBasedInputRow;
import io.druid.java.util.common.granularity.Granularity;
import io.druid.java.util.common.granularity.Granularities;
import io.druid.java.util.common.guava.Sequence;
import io.druid.java.util.common.guava.Sequences;
import io.druid.java.util.common.guava.Yielder;
Expand Down Expand Up @@ -59,8 +59,7 @@ public IngestSegmentFirehose(
final List<WindowedStorageAdapter> adapters,
final List<String> dims,
final List<String> metrics,
final DimFilter dimFilter,
final Granularity granularity
final DimFilter dimFilter
)
{
Sequence<InputRow> rows = Sequences.concat(
Expand All @@ -77,7 +76,7 @@ public Sequence<InputRow> apply(WindowedStorageAdapter adapter)
Filters.toFilter(dimFilter),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe could be further simplified by not calling concat() a few lines above.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How else would this be turned into a Sequence<InputRow> rather than Sequence<Sequence<InputRow>>?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, seems there is no way.

adapter.getInterval(),
VirtualColumns.EMPTY,
granularity,
Granularities.ALL,
false
), new Function<Cursor, Sequence<InputRow>>()
{
Expand Down
3 changes: 1 addition & 2 deletions server/src/test/java/io/druid/segment/realtime/firehose/IngestSegmentFirehoseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,7 @@ public void testSanity() throws Exception
ImmutableList.of(wsa, wsa),
ImmutableList.of("host"),
ImmutableList.of("visited_sum", "unique_hosts"),
null,
Granularities.NONE
null
);

int count = 0;
Expand Down