A new includeAllDimension flag for dimensionsSpec

benchmarks/src/test/java/org/apache/druid/benchmark/FlattenJSONBenchmarkUtil.java

@@ -55,7 +55,7 @@ public Parser getFlatParser()
   {
     JSONParseSpec spec = new JSONParseSpec(
         new TimestampSpec("ts", "iso", null),
-        new DimensionsSpec(null, null, null),
+        DimensionsSpec.EMPTY,
         null,
         null,
         null
@@ -70,7 +70,7 @@ public Parser getFieldDiscoveryParser()
 
     JSONParseSpec spec = new JSONParseSpec(
         new TimestampSpec("ts", "iso", null),
-        new DimensionsSpec(null, null, null),
+        DimensionsSpec.EMPTY,
         flattenSpec,
         null,
         null
@@ -114,7 +114,7 @@ public Parser getNestedParser()
     JSONPathSpec flattenSpec = new JSONPathSpec(true, fields);
     JSONParseSpec spec = new JSONParseSpec(
         new TimestampSpec("ts", "iso", null),
-        new DimensionsSpec(null, null, null),
+        DimensionsSpec.EMPTY,
         flattenSpec,
         null,
         null
@@ -158,7 +158,7 @@ public Parser getForcedPathParser()
     JSONPathSpec flattenSpec = new JSONPathSpec(false, fields);
     JSONParseSpec spec = new JSONParseSpec(
         new TimestampSpec("ts", "iso", null),
-        new DimensionsSpec(null, null, null),
+        DimensionsSpec.EMPTY,
         flattenSpec,
         null,
         null
@@ -200,7 +200,7 @@ public Parser getJqParser()
     JSONPathSpec flattenSpec = new JSONPathSpec(true, fields);
     JSONParseSpec spec = new JSONParseSpec(
         new TimestampSpec("ts", "iso", null),
-        new DimensionsSpec(null, null, null),
+        DimensionsSpec.EMPTY,
         flattenSpec,
         null,
         null

benchmarks/src/test/java/org/apache/druid/benchmark/ProtobufParserBenchmark.java

@@ -86,7 +86,7 @@ public void setup()
                         new StringDimensionSchema("id"),
                         new StringDimensionSchema("someOtherId"),
                         new StringDimensionSchema("isValid")
-                ), null, null),
+                )),
                 new JSONPathSpec(
                         true,
                         Lists.newArrayList(
@@ -106,7 +106,7 @@ public void setup()
                     new StringDimensionSchema("id"),
                     new StringDimensionSchema("someOtherId"),
                     new StringDimensionSchema("isValid")
-            ), null, null),
+            )),
             null,
             null,
             null

core/src/main/java/org/apache/druid/data/input/impl/DimensionsSpec.java

@@ -37,6 +37,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -46,8 +47,9 @@ public class DimensionsSpec
   private final List<DimensionSchema> dimensions;
   private final Set<String> dimensionExclusions;
   private final Map<String, DimensionSchema> dimensionSchemaMap;
+  private final boolean includeAllDimensions;
 
-  public static final DimensionsSpec EMPTY = new DimensionsSpec(null, null, null);
+  public static final DimensionsSpec EMPTY = new DimensionsSpec(null, null, null, false);
 
   public static List<DimensionSchema> getDefaultSchemas(List<String> dimNames)
   {
@@ -69,11 +71,22 @@ public static DimensionSchema convertSpatialSchema(SpatialDimensionSchema spatia
     return new NewSpatialDimensionSchema(spatialSchema.getDimName(), spatialSchema.getDims());
   }
 
+  public static Builder builder()
+  {
+    return new Builder();
+  }
+
+  public DimensionsSpec(List<DimensionSchema> dimensions)
+  {
+    this(dimensions, null, null, false);
+  }
+
   @JsonCreator
-  public DimensionsSpec(
+  private DimensionsSpec(
       @JsonProperty("dimensions") List<DimensionSchema> dimensions,
       @JsonProperty("dimensionExclusions") List<String> dimensionExclusions,
-      @Deprecated @JsonProperty("spatialDimensions") List<SpatialDimensionSchema> spatialDimensions
+      @Deprecated @JsonProperty("spatialDimensions") List<SpatialDimensionSchema> spatialDimensions,
+      @JsonProperty("includeAllDimensions") boolean includeAllDimensions
   )
   {
     this.dimensions = dimensions == null
@@ -101,11 +114,7 @@ public DimensionsSpec(
       this.dimensions.add(newSchema);
       dimensionSchemaMap.put(newSchema.getName(), newSchema);
     }
-  }
-
-  public DimensionsSpec(List<DimensionSchema> dimensions)
-  {
-    this(dimensions, null, null);
+    this.includeAllDimensions = includeAllDimensions;
   }
 
   @JsonProperty
@@ -120,6 +129,12 @@ public Set<String> getDimensionExclusions()
     return dimensionExclusions;
   }
 
+  @JsonProperty
+  public boolean isIncludeAllDimensions()
+  {
+    return includeAllDimensions;
+  }
+
   @Deprecated
   @JsonIgnore
   public List<SpatialDimensionSchema> getSpatialDimensions()
@@ -173,22 +188,23 @@ public boolean hasCustomDimensions()
   @PublicApi
   public DimensionsSpec withDimensions(List<DimensionSchema> dims)
   {
-    return new DimensionsSpec(dims, ImmutableList.copyOf(dimensionExclusions), null);
+    return new DimensionsSpec(dims, ImmutableList.copyOf(dimensionExclusions), null, includeAllDimensions);
   }
 
   public DimensionsSpec withDimensionExclusions(Set<String> dimExs)
   {
     return new DimensionsSpec(
         dimensions,
         ImmutableList.copyOf(Sets.union(dimensionExclusions, dimExs)),
-        null
+        null,
+        includeAllDimensions
     );
   }
 
   @Deprecated
   public DimensionsSpec withSpatialDimensions(List<SpatialDimensionSchema> spatials)
   {
-    return new DimensionsSpec(dimensions, ImmutableList.copyOf(dimensionExclusions), spatials);
+    return new DimensionsSpec(dimensions, ImmutableList.copyOf(dimensionExclusions), spatials, includeAllDimensions);
   }
 
   private void verify(List<SpatialDimensionSchema> spatialDimensions)
@@ -225,22 +241,16 @@ public boolean equals(Object o)
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
     DimensionsSpec that = (DimensionsSpec) o;
-
-    if (!dimensions.equals(that.dimensions)) {
-      return false;
-    }
-
-    return dimensionExclusions.equals(that.dimensionExclusions);
+    return includeAllDimensions == that.includeAllDimensions
+           && Objects.equals(dimensions, that.dimensions)
+           && Objects.equals(dimensionExclusions, that.dimensionExclusions);
   }
 
   @Override
   public int hashCode()
   {
-    int result = dimensions.hashCode();
-    result = 31 * result + dimensionExclusions.hashCode();
-    return result;
+    return Objects.hash(dimensions, dimensionExclusions, includeAllDimensions);
   }
 
   @Override
@@ -249,6 +259,51 @@ public String toString()
     return "DimensionsSpec{" +
            "dimensions=" + dimensions +
            ", dimensionExclusions=" + dimensionExclusions +
+           ", includeAllDimensions=" + includeAllDimensions +
            '}';
   }
+
+  public static final class Builder
+  {
+    private List<DimensionSchema> dimensions;
+    private List<String> dimensionExclusions;
+    private List<SpatialDimensionSchema> spatialDimensions;
+    private boolean includeAllDimensions;
+
+    public Builder setDimensions(List<DimensionSchema> dimensions)
+    {
+      this.dimensions = dimensions;
+      return this;
+    }
+
+    public Builder setDefaultSchemaDimensions(List<String> dimensions)
+    {
+      this.dimensions = getDefaultSchemas(dimensions);
+      return this;
+    }
+
+    public Builder setDimensionExclusions(List<String> dimensionExclusions)
+    {
+      this.dimensionExclusions = dimensionExclusions;
+      return this;
+    }
+
+    @Deprecated
+    public Builder setSpatialDimensions(List<SpatialDimensionSchema> spatialDimensions)
+    {
+      this.spatialDimensions = spatialDimensions;
+      return this;
+    }
+
+    public Builder setIncludeAllDimensions(boolean includeAllDimensions)
+    {
+      this.includeAllDimensions = includeAllDimensions;
+      return this;
+    }
+
+    public DimensionsSpec build()
+    {
+      return new DimensionsSpec(dimensions, dimensionExclusions, spatialDimensions, includeAllDimensions);
+    }
+  }
 }

core/src/main/java/org/apache/druid/data/input/impl/MapInputRowParser.java

@@ -33,22 +33,20 @@
 
 import javax.annotation.Nullable;
 import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 public class MapInputRowParser implements InputRowParser<Map<String, Object>>
 {
   private final ParseSpec parseSpec;
-  private final List<String> dimensions;
 
   @JsonCreator
   public MapInputRowParser(
       @JsonProperty("parseSpec") ParseSpec parseSpec
   )
   {
     this.parseSpec = parseSpec;
-    this.dimensions = parseSpec.getDimensionsSpec().getDimensionNames();
   }
 
   @Override
@@ -57,8 +55,7 @@ public List<InputRow> parseBatch(Map<String, Object> theMap)
     return ImmutableList.of(
         parse(
             parseSpec.getTimestampSpec(),
-            dimensions,
-            parseSpec.getDimensionsSpec().getDimensionExclusions(),
+            parseSpec.getDimensionsSpec(),
             theMap
         )
     );
@@ -69,29 +66,45 @@ public static InputRow parse(InputRowSchema inputRowSchema, Map<String, Object>
     return parse(inputRowSchema.getTimestampSpec(), inputRowSchema.getDimensionsSpec(), theMap);
   }
 
-  private static InputRow parse(
-      TimestampSpec timestampSpec,
+  /**
+   * Finds the final set of dimension names to use for {@link InputRow}.
+   * There are 3 cases here.
+   *
+   * 1) If {@link DimensionsSpec#isIncludeAllDimensions()} is set, the returned list includes _both_
+   *    {@link DimensionsSpec#getDimensionNames()} and the dimensions in the given map ({@code rawInputRow#keySet()}).
+   * 2) If isIncludeAllDimensions is not set and {@link DimensionsSpec#getDimensionNames()} is not empty,
+   *    the dimensions in dimensionsSpec is returned.
+   * 3) If isIncludeAllDimensions is not set and {@link DimensionsSpec#getDimensionNames()} is empty,
+   *    the dimensions in the given map is returned.
+   *
+   * In any case, the returned list does not include any dimensions in {@link DimensionsSpec#getDimensionExclusions()}.
+   */
+  private static List<String> findDimensions(
       DimensionsSpec dimensionsSpec,
-      Map<String, Object> theMap
-  ) throws ParseException
+      Map<String, Object> rawInputRow
+  )
   {
-    return parse(timestampSpec, dimensionsSpec.getDimensionNames(), dimensionsSpec.getDimensionExclusions(), theMap);
+    if (dimensionsSpec.isIncludeAllDimensions()) {
+      LinkedHashSet<String> dimensions = new LinkedHashSet<>(dimensionsSpec.getDimensionNames());
+      dimensions.addAll(Sets.difference(rawInputRow.keySet(), dimensionsSpec.getDimensionExclusions()));
+      return new ArrayList<>(dimensions);
+    } else {
+      if (!dimensionsSpec.getDimensionNames().isEmpty()) {
+        return dimensionsSpec.getDimensionNames();
+      } else {
+        return new ArrayList<>(Sets.difference(rawInputRow.keySet(), dimensionsSpec.getDimensionExclusions()));
+      }
+    }
   }
 
   @VisibleForTesting
   static InputRow parse(
       TimestampSpec timestampSpec,
-      List<String> dimensions,
-      Set<String> dimensionExclusions,
+      DimensionsSpec dimensionsSpec,
       Map<String, Object> theMap
   ) throws ParseException
   {
-    final List<String> dimensionsToUse;
-    if (!dimensions.isEmpty()) {
-      dimensionsToUse = dimensions;
-    } else {
-      dimensionsToUse = new ArrayList<>(Sets.difference(theMap.keySet(), dimensionExclusions));
-    }
+    final List<String> dimensionsToUse = findDimensions(dimensionsSpec, theMap);
 
     final DateTime timestamp;
     try {

core/src/main/java/org/apache/druid/data/input/impl/TimeAndDimsParseSpec.java

@@ -38,7 +38,7 @@ public TimeAndDimsParseSpec(
   {
     super(
         timestampSpec != null ? timestampSpec : new TimestampSpec(null, null, null),
-        dimensionsSpec != null ? dimensionsSpec : new DimensionsSpec(null, null, null)
+        dimensionsSpec != null ? dimensionsSpec : DimensionsSpec.EMPTY
     );
   }
 

core/src/test/java/org/apache/druid/data/input/impl/CSVParseSpecTest.java

@@ -21,7 +21,6 @@
 
 import org.junit.Test;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 
@@ -37,11 +36,7 @@ public void testComma()
             "auto",
             null
         ),
-        new DimensionsSpec(
-            DimensionsSpec.getDefaultSchemas(Arrays.asList("a,", "b")),
-            new ArrayList<>(),
-            new ArrayList<>()
-        ),
+        new DimensionsSpec(DimensionsSpec.getDefaultSchemas(Arrays.asList("a,", "b"))),
         ",",
         Collections.singletonList("a,"),
         false,