Fix ambiguity about IndexerSQLMetadataStorageCoordinator.getUsedSegmentsForInterval() returning only non-overshadowed or all used segments

benchmarks/pom.xml

@@ -104,6 +104,10 @@
       <artifactId>json-flattener</artifactId>
       <version>0.1.0</version>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.dataformat</groupId>
       <artifactId>jackson-dataformat-smile</artifactId>

benchmarks/src/main/java/org/apache/druid/benchmark/ExpressionAggregationBenchmark.java

@@ -100,6 +100,7 @@ public void setup()
                                                .interval(schemaInfo.getDataInterval())
                                                .version("1")
                                                .shardSpec(new LinearShardSpec(0))
+                                               .size(0)
                                                .build();
 
     final SegmentGenerator segmentGenerator = closer.register(new SegmentGenerator());

benchmarks/src/main/java/org/apache/druid/benchmark/ExpressionSelectorBenchmark.java

@@ -114,6 +114,7 @@ public void setup()
                                                .interval(schemaInfo.getDataInterval())
                                                .version("1")
                                                .shardSpec(new LinearShardSpec(0))
+                                               .size(0)
                                                .build();
 
     final SegmentGenerator segmentGenerator = closer.register(new SegmentGenerator());

benchmarks/src/main/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java

@@ -199,6 +199,7 @@ public void setup()
                                                  .interval(schemaInfo.getDataInterval())
                                                  .version("1")
                                                  .shardSpec(new LinearShardSpec(i))
+                                                 .size(0)
                                                  .build();
       final SegmentGenerator segmentGenerator = closer.register(new SegmentGenerator());
       LOG.info("Starting benchmark setup using cacheDir[%s], rows[%,d].", segmentGenerator.getCacheDir(), rowsPerSegment);

benchmarks/src/main/java/org/apache/druid/benchmark/query/SerializingQueryRunner.java

@@ -19,10 +19,11 @@
 
 package org.apache.druid.benchmark.query;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Function;
 import org.apache.druid.java.util.common.guava.Sequence;
 import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.jackson.JacksonUtils;
 import org.apache.druid.query.QueryPlus;
 import org.apache.druid.query.QueryRunner;
 import org.apache.druid.query.context.ResponseContext;
@@ -52,17 +53,12 @@ public Sequence<T> run(
   {
     return Sequences.map(
         baseRunner.run(queryPlus, responseContext),
-        new Function<T, T>()
-        {
-          @Override
-          public T apply(T input)
-          {
-            try {
-              return smileMapper.readValue(smileMapper.writeValueAsBytes(input), clazz);
-            }
-            catch (Exception e) {
-              throw new RuntimeException(e);
-            }
+        input -> {
+          try {
+            return JacksonUtils.readValue(smileMapper, smileMapper.writeValueAsBytes(input), clazz);
+          }
+          catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
           }
         }
     );

benchmarks/src/main/java/org/apache/druid/benchmark/query/SqlBenchmark.java

@@ -173,6 +173,7 @@ public void setup()
                                                .interval(schemaInfo.getDataInterval())
                                                .version("1")
                                                .shardSpec(new LinearShardSpec(0))
+                                               .size(0)
                                                .build();
 
     final PlannerConfig plannerConfig = new PlannerConfig();

benchmarks/src/main/java/org/apache/druid/benchmark/query/SqlVsNativeBenchmark.java

@@ -98,6 +98,7 @@ public void setup()
                                                .interval(schemaInfo.getDataInterval())
                                                .version("1")
                                                .shardSpec(new LinearShardSpec(0))
+                                               .size(0)
                                                .build();
 
     final SegmentGenerator segmentGenerator = closer.register(new SegmentGenerator());

core/src/main/java/org/apache/druid/common/config/JacksonConfigManager.java

@@ -27,6 +27,7 @@
 import org.apache.druid.audit.AuditInfo;
 import org.apache.druid.audit.AuditManager;
 import org.apache.druid.common.config.ConfigManager.SetResult;
+import org.apache.druid.java.util.common.jackson.JacksonUtils;
 
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicReference;
@@ -115,12 +116,7 @@ public T deserialize(byte[] bytes)
           return defaultVal;
         }
 
-        try {
-          return jsonMapper.readValue(bytes, clazz);
-        }
-        catch (IOException e) {
-          throw new RuntimeException(e);
-        }
+        return JacksonUtils.readValue(jsonMapper, bytes, clazz);
       }
     };
   }

core/src/main/java/org/apache/druid/indexer/TaskStatus.java

@@ -67,9 +67,11 @@ public static TaskStatus fromCode(String taskId, TaskState code)
     return new TaskStatus(taskId, code, -1, null, null);
   }
 
-  // The error message can be large, so truncate it to avoid storing large objects in zookeeper/metadata storage.
-  // The full error message will be available via a TaskReport.
-  private static String truncateErrorMsg(String errorMsg)
+  /**
+   * The error message can be large, so truncate it to avoid storing large objects in zookeeper/metadata storage.
+   * The full error message will be available via a TaskReport.
+   */
+  private static @Nullable String truncateErrorMsg(@Nullable String errorMsg)
   {
     if (errorMsg != null && errorMsg.length() > MAX_ERROR_MSG_LENGTH) {
       return errorMsg.substring(0, MAX_ERROR_MSG_LENGTH) + "...";
@@ -81,15 +83,15 @@ private static String truncateErrorMsg(String errorMsg)
   private final String id;
   private final TaskState status;
   private final long duration;
-  private final String errorMsg;
+  private final @Nullable String errorMsg;
   private final TaskLocation location;
 
   @JsonCreator
   protected TaskStatus(
       @JsonProperty("id") String id,
       @JsonProperty("status") TaskState status,
       @JsonProperty("duration") long duration,
-      @JsonProperty("errorMsg") String errorMsg,
+      @JsonProperty("errorMsg") @Nullable String errorMsg,
       @Nullable @JsonProperty("location") TaskLocation location
   )
   {
@@ -122,6 +124,7 @@ public long getDuration()
     return duration;
   }
 
+  @Nullable
   @JsonProperty("errorMsg")
   public String getErrorMsg()
   {

core/src/main/java/org/apache/druid/java/util/common/jackson/JacksonUtils.java

@@ -20,15 +20,32 @@
 package org.apache.druid.java.util.common.jackson;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
+import java.io.IOException;
 import java.util.Map;
 
-public class JacksonUtils
+public final class JacksonUtils
 {
   public static final TypeReference<Map<String, Object>> TYPE_REFERENCE_MAP_STRING_OBJECT = new TypeReference<Map<String, Object>>()
   {
   };
   public static final TypeReference<Map<String, String>> TYPE_REFERENCE_MAP_STRING_STRING = new TypeReference<Map<String, String>>()
   {
   };
+
+  /** Silences Jackson's {@link IOException}. */
+  public static <T> T readValue(ObjectMapper mapper, byte[] bytes, Class<T> valueClass)
+  {
+    try {
+      return mapper.readValue(bytes, valueClass);
+    }
+    catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private JacksonUtils()
+  {
+  }
 }

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

@@ -214,6 +214,7 @@ public DataSegment(
                                ? null
                                : prepareCompactionState(lastCompactionState);
     this.binaryVersion = binaryVersion;
+    Preconditions.checkArgument(size >= 0);
     this.size = size;
   }
 

indexing-hadoop/src/main/java/org/apache/druid/indexer/path/UsedSegmentLister.java → core/src/main/java/org/apache/druid/timeline/Partitions.java

@@ -17,28 +17,17 @@
  * under the License.
  */
 
-package org.apache.druid.indexer.path;
-
-import org.apache.druid.timeline.DataSegment;
-import org.joda.time.Interval;
-
-import java.io.IOException;
-import java.util.List;
+package org.apache.druid.timeline;
 
 /**
+ * This enum is used a parameter for several methods in {@link VersionedIntervalTimeline}, specifying whether only
+ * {@linkplain org.apache.druid.timeline.partition.PartitionHolder#isComplete() complete} partitions should be
+ * considered, or incomplete partitions as well.
  */
-public interface UsedSegmentLister
+public enum Partitions
 {
-  /**
-   * Get all segments which may include any data in the interval and are flagged as used.
-   *
-   * @param dataSource The datasource to query
-   * @param intervals  The intervals for which used segments are to be returned
-   *
-   * @return The DataSegments which include data in the requested intervals. These segments may contain data outside the requested interval.
-   *
-   * @throws IOException
-   */
-  List<DataSegment> getUsedSegmentsForIntervals(String dataSource, List<Interval> intervals)
-      throws IOException;
+  /** Specifies that only complete partitions should be considered. */
+  ONLY_COMPLETE,
+  /** Specifies that complete as well as incomplete partitions should be considered. */
+  INCOMPLETE_OK
 }