Fix indexMerger to respect the includeAllDimensions flag (#12428)

* Fix indexMerger to respect flag includeAllDimensions flag; jsonInputFormat should set keepNullColumns if useFieldDiscovery is set

* address comments

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

@@ -74,7 +74,11 @@ public JsonInputFormat(
     super(flattenSpec);
     this.featureSpec = featureSpec == null ? Collections.emptyMap() : featureSpec;
     this.objectMapper = new ObjectMapper();
-    this.keepNullColumns = keepNullColumns == null ? false : keepNullColumns;
+    if (keepNullColumns != null) {
+      this.keepNullColumns = keepNullColumns;
+    } else {
+      this.keepNullColumns = flattenSpec != null && flattenSpec.isUseFieldDiscovery();
+    }
     for (Entry<String, Boolean> entry : this.featureSpec.entrySet()) {
       Feature feature = Feature.valueOf(entry.getKey());
       objectMapper.configure(feature, entry.getValue());
@@ -88,6 +92,12 @@ public Map<String, Boolean> getFeatureSpec()
     return featureSpec;
   }
 
+  @JsonProperty
+  public boolean isKeepNullColumns()
+  {
+    return keepNullColumns;
+  }
+
   @Override
   public boolean isSplittable()
   {

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

@@ -41,7 +41,7 @@ public void testSerde() throws IOException
     final ObjectMapper mapper = new ObjectMapper();
     final JsonInputFormat format = new JsonInputFormat(
         new JSONPathSpec(
-            false,
+            true,
             ImmutableList.of(
                 new JSONPathFieldSpec(JSONPathFieldType.ROOT, "root_baz", "baz"),
                 new JSONPathFieldSpec(JSONPathFieldType.ROOT, "root_baz2", "baz2"),
@@ -52,7 +52,7 @@ public void testSerde() throws IOException
             )
         ),
         ImmutableMap.of(Feature.ALLOW_COMMENTS.name(), true, Feature.ALLOW_UNQUOTED_FIELD_NAMES.name(), false),
-        false
+        true
     );
     final byte[] bytes = mapper.writeValueAsBytes(format);
     final JsonInputFormat fromJson = (JsonInputFormat) mapper.readValue(bytes, InputFormat.class);
@@ -72,4 +72,37 @@ public void testEquals()
               .withIgnoredFields("objectMapper")
               .verify();
   }
+
+  @Test
+  public void test_unsetUseFieldDiscovery_unsetKeepNullColumnsByDefault()
+  {
+    final JsonInputFormat format = new JsonInputFormat(
+        new JSONPathSpec(false, null),
+        null,
+        null
+    );
+    Assert.assertFalse(format.isKeepNullColumns());
+  }
+
+  @Test
+  public void testUseFieldDiscovery_setKeepNullColumnsByDefault()
+  {
+    final JsonInputFormat format = new JsonInputFormat(
+        new JSONPathSpec(true, null),
+        null,
+        null
+    );
+    Assert.assertTrue(format.isKeepNullColumns());
+  }
+
+  @Test
+  public void testUseFieldDiscovery_doNotChangeKeepNullColumnsUserSets()
+  {
+    final JsonInputFormat format = new JsonInputFormat(
+        new JSONPathSpec(true, null),
+        null,
+        false
+    );
+    Assert.assertFalse(format.isKeepNullColumns());
+  }
 }