Deprecate NoneShardSpec and drop support for automatic segment merge

core/src/main/java/org/apache/druid/guice/ConditionalMultibind.java

@@ -184,11 +184,7 @@ public ConditionalMultibind<T> addConditionBinding(
       Class<? extends T> target
   )
   {
-    final String value = properties.getProperty(property);
-    if (value == null) {
-      return this;
-    }
-    if (condition.apply(value)) {
+    if (matchCondition(property, condition)) {
       multibinder.addBinding().to(target);
     }
     return this;
@@ -209,11 +205,7 @@ public ConditionalMultibind<T> addConditionBinding(
       T target
   )
   {
-    final String value = properties.getProperty(property);
-    if (value == null) {
-      return this;
-    }
-    if (condition.apply(value)) {
+    if (matchCondition(property, condition)) {
       multibinder.addBinding().toInstance(target);
     }
     return this;
@@ -235,13 +227,18 @@ public ConditionalMultibind<T> addConditionBinding(
       TypeLiteral<T> target
   )
   {
-    final String value = properties.getProperty(property);
-    if (value == null) {
-      return this;
-    }
-    if (condition.apply(value)) {
+    if (matchCondition(property, condition)) {
       multibinder.addBinding().to(target);
     }
     return this;
   }
+
+  public boolean matchCondition(String property, Predicate<String> condition)
+  {
+    final String value = properties.getProperty(property);
+    if (value == null) {
+      return false;
+    }
+    return condition.apply(value);
+  }
 }

core/src/main/java/org/apache/druid/timeline/DataSegment.java

@@ -38,7 +38,7 @@
 import org.apache.druid.jackson.CommaListJoinSerializer;
 import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.query.SegmentDescriptor;
-import org.apache.druid.timeline.partition.NoneShardSpec;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
 import org.apache.druid.timeline.partition.ShardSpec;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -146,7 +146,7 @@ public DataSegment(
     // dataSource
     this.dimensions = prepareDimensionsOrMetrics(dimensions, DIMENSIONS_INTERNER);
     this.metrics = prepareDimensionsOrMetrics(metrics, METRICS_INTERNER);
-    this.shardSpec = (shardSpec == null) ? NoneShardSpec.instance() : shardSpec;
+    this.shardSpec = (shardSpec == null) ? new NumberedShardSpec(0, 1) : shardSpec;
     this.binaryVersion = binaryVersion;
     this.size = size;
   }
@@ -373,7 +373,7 @@ public Builder()
       this.loadSpec = ImmutableMap.of();
       this.dimensions = ImmutableList.of();
       this.metrics = ImmutableList.of();
-      this.shardSpec = NoneShardSpec.instance();
+      this.shardSpec = new NumberedShardSpec(0, 1);
       this.size = -1;
     }
 

core/src/main/java/org/apache/druid/timeline/partition/NoneShardSpec.java

@@ -30,6 +30,7 @@
 
 /**
  */
+@Deprecated
 public class NoneShardSpec implements ShardSpec
 {
   private static final NoneShardSpec INSTANCE = new NoneShardSpec();
@@ -40,11 +41,7 @@ public static NoneShardSpec instance()
     return INSTANCE;
   }
 
-  /**
-   * @deprecated use {@link #instance()} instead
-   */
-  @Deprecated
-  public NoneShardSpec()
+  private NoneShardSpec()
   {
     // empty
   }

core/src/main/java/org/apache/druid/timeline/partition/ShardSpec.java

@@ -33,8 +33,12 @@
  */
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes({
-                  @JsonSubTypes.Type(name = "none", value = NoneShardSpec.class),
-              })
+    @JsonSubTypes.Type(name = "none", value = NoneShardSpec.class),
+    @JsonSubTypes.Type(name = "single", value = SingleDimensionShardSpec.class),
+    @JsonSubTypes.Type(name = "linear", value = LinearShardSpec.class),
+    @JsonSubTypes.Type(name = "numbered", value = NumberedShardSpec.class),
+    @JsonSubTypes.Type(name = "hashed", value = HashBasedNumberedShardSpec.class)
+})
 public interface ShardSpec
 {
   <T> PartitionChunk<T> createChunk(T obj);

server/src/main/java/org/apache/druid/timeline/partition/SingleDimensionShardSpec.java → core/src/main/java/org/apache/druid/timeline/partition/SingleDimensionShardSpec.java

@@ -19,13 +19,16 @@
 
 package org.apache.druid.timeline.partition;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Range;
 import com.google.common.collect.RangeSet;
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.java.util.common.ISE;
 
+import javax.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
 
@@ -36,24 +39,22 @@
  */
 public class SingleDimensionShardSpec implements ShardSpec
 {
-  private String dimension;
-  private String start;
-  private String end;
-  private int partitionNum;
-
-  public SingleDimensionShardSpec()
-  {
-    this(null, null, null, -1);
-  }
-
+  private final String dimension;
+  @Nullable
+  private final String start;
+  @Nullable
+  private final String end;
+  private final int partitionNum;
+
+  @JsonCreator
   public SingleDimensionShardSpec(
-      String dimension,
-      String start,
-      String end,
-      int partitionNum
+      @JsonProperty("dimension") String dimension,
+      @JsonProperty("start") @Nullable String start,
+      @JsonProperty("end") @Nullable String end,
+      @JsonProperty("partitionNum") int partitionNum
   )
   {
-    this.dimension = dimension;
+    this.dimension = Preconditions.checkNotNull(dimension, "dimension");
     this.start = start;
     this.end = end;
     this.partitionNum = partitionNum;
@@ -65,33 +66,20 @@ public String getDimension()
     return dimension;
   }
 
-  public void setDimension(String dimension)
-  {
-    this.dimension = dimension;
-  }
-
+  @Nullable
   @JsonProperty("start")
   public String getStart()
   {
     return start;
   }
 
-  public void setStart(String start)
-  {
-    this.start = start;
-  }
-
+  @Nullable
   @JsonProperty("end")
   public String getEnd()
   {
     return end;
   }
 
-  public void setEnd(String end)
-  {
-    this.end = end;
-  }
-
   @Override
   @JsonProperty("partitionNum")
   public int getPartitionNum()
@@ -143,11 +131,6 @@ public boolean possibleInDomain(Map<String, RangeSet<String>> domain)
     return !rangeSet.subRangeSet(getRange()).isEmpty();
   }
 
-  public void setPartitionNum(int partitionNum)
-  {
-    this.partitionNum = partitionNum;
-  }
-
   @Override
   public <T> PartitionChunk<T> createChunk(T obj)
   {

docs/content/configuration/index.md

@@ -576,7 +576,7 @@ If you are running the indexing service in remote mode, the task logs must be st
 |`druid.indexer.logs.type`|Choices:noop, s3, azure, google, hdfs, file. Where to store task logs|file|
 
 You can also configure the Overlord to automatically retain the task logs in log directory and entries in task-related metadata storage tables only for last x milliseconds by configuring following additional properties.
-Caution: Automatic log file deletion typically works based on log file modification timestamp on the backing store, so large clock skews between druid processes and backing store nodes might result in un-intended behavior.  
+Caution: Automatic log file deletion typically works based on log file modification timestamp on the backing store, so large clock skews between druid processes and backing store nodes might result in un-intended behavior.
 
 |Property|Description|Default|
 |--------|-----------|-------|
@@ -719,7 +719,6 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.period`|The run period for the Coordinator. The Coordinator’s operates by maintaining the current state of the world in memory and periodically looking at the set of segments available and segments being served to make decisions about whether any changes need to be made to the data topology. This property sets the delay between each of these runs.|PT60S|
 |`druid.coordinator.period.indexingPeriod`|How often to send compact/merge/conversion tasks to the indexing service. It's recommended to be longer than `druid.manager.segments.pollDuration`|PT1800S (30 mins)|
 |`druid.coordinator.startDelay`|The operation of the Coordinator works on the assumption that it has an up-to-date view of the state of the world when it runs, the current ZK interaction code, however, is written in a way that doesn’t allow the Coordinator to know for a fact that it’s done loading the current state of the world. This delay is a hack to give it enough time to believe that it has all the data.|PT300S|
-|`druid.coordinator.merge.on`|Boolean flag for whether or not the Coordinator should try and merge small segments into a more optimal segment size.|false|
 |`druid.coordinator.load.timeout`|The timeout duration for when the Coordinator assigns a segment to a Historical process.|PT15M|
 |`druid.coordinator.kill.pendingSegments.on`|Boolean flag for whether or not the Coordinator clean up old entries in the `pendingSegments` table of metadata store. If set to true, Coordinator will check the created time of most recently complete task. If it doesn't exist, it finds the created time of the earlist running/pending/waiting tasks. Once the created time is found, then for all dataSources not in the `killPendingSegmentsSkipList` (see [Dynamic configuration](#dynamic-configuration)), Coordinator will ask the Overlord to clean up the entries 1 day or more older than the found created time in the `pendingSegments` table. This will be done periodically based on `druid.coordinator.period` specified.|false|
 |`druid.coordinator.kill.on`|Boolean flag for whether or not the Coordinator should submit kill task for unused segments, that is, hard delete them from metadata store and deep storage. If set to true, then for all whitelisted dataSources (or optionally all), Coordinator will submit tasks periodically based on `period` specified. These kill tasks will delete all segments except for the last `durationToRetain` period. Whitelist or All can be set via dynamic configuration `killAllDataSources` and `killDataSourceWhitelist` described later.|false|
@@ -1457,7 +1456,7 @@ See [cache configuration](#cache-configuration) for how to configure cache setti
 
 This section describes caching configuration that is common to Broker, Historical, and MiddleManager/Peon processes.
 
-Caching can optionally be enabled on the Broker, Historical, and MiddleManager/Peon processses. See [Broker](#broker-caching), 
+Caching can optionally be enabled on the Broker, Historical, and MiddleManager/Peon processses. See [Broker](#broker-caching),
 [Historical](#Historical-caching), and [Peon](#peon-caching) configuration options for how to enable it for different processes.
 
 Druid uses a local in-memory cache by default, unless a diffrent type of cache is specified.

docs/content/ingestion/hadoop.md

@@ -194,7 +194,7 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |jobProperties|Object|A map of properties to add to the Hadoop job configuration, see below for details.|no (default == null)|
 |indexSpec|Object|Tune how data is indexed. See below for more information.|no|
 |numBackgroundPersistThreads|Integer|The number of new background threads to use for incremental persists. Using this feature causes a notable increase in memory pressure and cpu usage but will make the job finish more quickly. If changing from the default of 0 (use current thread for persists), we recommend setting it to 1.|no (default == 0)|
-|forceExtendableShardSpecs|Boolean|Forces use of extendable shardSpecs. Experimental feature intended for use with the [Kafka indexing service extension](../development/extensions-core/kafka-ingestion.html).|no (default = false)|
+|forceExtendableShardSpecs|Boolean|Forces use of extendable shardSpecs. Hash-based partitioning always uses an extendable shardSpec. For single-dimension partitioning, this option should be set to true to use an extendable shardSpec. For partitioning, please check [Partitioning specification](#partitioning-specification). This option can be useful when you need to append more data to existing dataSource.|no (default = false)|
 |useExplicitVersion|Boolean|Forces HadoopIndexTask to use version.|no (default = false)|
 |logParseExceptions|Boolean|If true, log an error message when a parsing exception occurs, containing information about the row where the error occurred.|false|no|
 |maxParseExceptions|Integer|The maximum number of parse exceptions that can occur before the task halts ingestion and fails. Overrides `ignoreInvalidRows` if `maxParseExceptions` is defined.|unlimited|no|

docs/content/ingestion/native_tasks.md

@@ -170,7 +170,7 @@ that range if there's some stray data with unexpected timestamps.
 |--------|-----------|-------|---------|
 |type|The task type, this should always be `index_parallel`.|none|yes|
 |firehose|Specify a [Firehose](../ingestion/firehose.html) here.|none|yes|
-|appendToExisting|Creates segments as additional shards of the latest version, effectively appending to the segment set instead of replacing it. This will only work if the existing segment set has extendable-type shardSpecs (which can be forced by setting 'forceExtendableShardSpecs' in the tuning config).|false|no|
+|appendToExisting|Creates segments as additional shards of the latest version, effectively appending to the segment set instead of replacing it. This will only work if the existing segment set has extendable-type shardSpecs.|false|no|
 
 #### TuningConfig
 
@@ -186,7 +186,6 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |numShards|Directly specify the number of shards to create. If this is specified and 'intervals' is specified in the granularitySpec, the index task can skip the determine intervals/partitions pass through the data. numShards cannot be specified if maxRowsPerSegment is set.|null|no|
 |indexSpec|defines segment storage format options to be used at indexing time, see [IndexSpec](#indexspec)|null|no|
 |maxPendingPersists|Maximum number of persists that can be pending but not started. If this limit would be exceeded by a new intermediate persist, ingestion will block until the currently-running persist finishes. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|0 (meaning one persist can be running concurrently with ingestion, and none can be queued up)|no|
-|forceExtendableShardSpecs|Forces use of extendable shardSpecs. Experimental feature intended for use with the [Kafka indexing service extension](../development/extensions-core/kafka-ingestion.html).|false|no|
 |reportParseExceptions|If true, exceptions encountered during parsing will be thrown and will halt ingestion; if false, unparseable rows and fields will be skipped.|false|no|
 |pushTimeout|Milliseconds to wait for pushing segments. It must be >= 0, where 0 means to wait forever.|0|no|
 |segmentWriteOutMediumFactory|Segment write-out medium to use when creating segments. See [SegmentWriteOutMediumFactory](#segmentWriteOutMediumFactory).|Not specified, the value from `druid.peon.defaultSegmentWriteOutMediumFactory.type` is used|no|
@@ -377,7 +376,6 @@ An example of the result is
           "longEncoding": "longs"
         },
         "maxPendingPersists": 0,
-        "forceExtendableShardSpecs": false,
         "reportParseExceptions": false,
         "pushTimeout": 0,
         "segmentWriteOutMediumFactory": null,
@@ -541,7 +539,7 @@ that range if there's some stray data with unexpected timestamps.
 |--------|-----------|-------|---------|
 |type|The task type, this should always be "index".|none|yes|
 |firehose|Specify a [Firehose](../ingestion/firehose.html) here.|none|yes|
-|appendToExisting|Creates segments as additional shards of the latest version, effectively appending to the segment set instead of replacing it. This will only work if the existing segment set has extendable-type shardSpecs (which can be forced by setting 'forceExtendableShardSpecs' in the tuning config).|false|no|
+|appendToExisting|Creates segments as additional shards of the latest version, effectively appending to the segment set instead of replacing it. This will only work if the existing segment set has extendable-type shardSpecs.|false|no|
 
 #### TuningConfig
 
@@ -558,7 +556,6 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |partitionDimensions|The dimensions to partition on. Leave blank to select all dimensions. Only used with `forceGuaranteedRollup` = true, will be ignored otherwise.|null|no|
 |indexSpec|defines segment storage format options to be used at indexing time, see [IndexSpec](#indexspec)|null|no|
 |maxPendingPersists|Maximum number of persists that can be pending but not started. If this limit would be exceeded by a new intermediate persist, ingestion will block until the currently-running persist finishes. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|0 (meaning one persist can be running concurrently with ingestion, and none can be queued up)|no|
-|forceExtendableShardSpecs|Forces use of extendable shardSpecs. Experimental feature intended for use with the [Kafka indexing service extension](../development/extensions-core/kafka-ingestion.html).|false|no|
 |forceGuaranteedRollup|Forces guaranteeing the [perfect rollup](../ingestion/index.html#roll-up-modes). The perfect rollup optimizes the total size of generated segments and querying time while indexing time will be increased. If this is set to true, the index task will read the entire input data twice: one for finding the optimal number of partitions per time chunk and one for generating segments. Note that the result segments would be hash-partitioned. You can set `forceExtendableShardSpecs` if you plan to append more data to the same time range in the future. This flag cannot be used with `appendToExisting` of IOConfig. For more details, see the below __Segment pushing modes__ section.|false|no|
 |reportParseExceptions|DEPRECATED. If true, exceptions encountered during parsing will be thrown and will halt ingestion; if false, unparseable rows and fields will be skipped. Setting `reportParseExceptions` to true will override existing configurations for `maxParseExceptions` and `maxSavedParseExceptions`, setting `maxParseExceptions` to 0 and limiting `maxSavedParseExceptions` to no more than 1.|false|no|
 |pushTimeout|Milliseconds to wait for pushing segments. It must be >= 0, where 0 means to wait forever.|0|no|
@@ -617,4 +614,4 @@ the index task immediately pushes all segments created until that moment, cleans
 continues to ingest remaining data.
 
 To enable bulk pushing mode, `forceGuaranteedRollup` should be set in the TuningConfig. Note that this option cannot
-be used with either `forceExtendableShardSpecs` of TuningConfig or `appendToExisting` of IOConfig.
+be used with `appendToExisting` of IOConfig.

docs/content/tutorials/tutorial-batch.md

@@ -99,8 +99,7 @@ which has been configured to read the `quickstart/tutorial/wikiticker-2015-09-12
     "tuningConfig" : {
       "type" : "index",
       "maxRowsPerSegment" : 5000000,
-      "maxRowsInMemory" : 25000,
-      "forceExtendableShardSpecs" : true
+      "maxRowsInMemory" : 25000
     }
   }
 }

docs/content/tutorials/tutorial-compaction.md

@@ -85,8 +85,7 @@ We have included a compaction task spec for this tutorial datasource at `quickst
   "tuningConfig" : {
     "type" : "index",
     "maxRowsPerSegment" : 5000000,
-    "maxRowsInMemory" : 25000,
-    "forceExtendableShardSpecs" : true
+    "maxRowsInMemory" : 25000
   }
 }
 ```

docs/content/tutorials/tutorial-rollup.md

@@ -100,8 +100,7 @@ We'll ingest this data using the following ingestion task spec, located at `quic
     "tuningConfig" : {
       "type" : "index",
       "maxRowsPerSegment" : 5000000,
-      "maxRowsInMemory" : 25000,
-      "forceExtendableShardSpecs" : true
+      "maxRowsInMemory" : 25000
     }
   }
 }

docs/content/tutorials/tutorial-transform-spec.md

@@ -115,8 +115,7 @@ We will ingest the sample data using the following spec, which demonstrates the
     "tuningConfig" : {
       "type" : "index",
       "maxRowsPerSegment" : 5000000,
-      "maxRowsInMemory" : 25000,
-      "forceExtendableShardSpecs" : true
+      "maxRowsInMemory" : 25000
     }
   }
 }

examples/quickstart/tutorial/compaction-init-index.json

@@ -56,8 +56,7 @@
     },
     "tuningConfig" : {
       "type" : "index",
-      "maxRowsPerSegment" : 1000,
-      "forceExtendableShardSpecs" : true
+      "maxRowsPerSegment" : 1000
     }
   }
 }

examples/quickstart/tutorial/compaction-keep-granularity.json

@@ -5,7 +5,6 @@
   "tuningConfig" : {
     "type" : "index",
     "maxRowsPerSegment" : 5000000,
-    "maxRowsInMemory" : 25000,
-    "forceExtendableShardSpecs" : true
+    "maxRowsInMemory" : 25000
   }
 }

examples/quickstart/tutorial/deletion-index.json

@@ -57,8 +57,7 @@
     "tuningConfig" : {
       "type" : "index",
       "maxRowsPerSegment" : 5000000,
-      "maxRowsInMemory" : 25000,
-      "forceExtendableShardSpecs" : true
+      "maxRowsInMemory" : 25000
     }
   }
 }