fix issues with nested data conversion (#13407)

core/src/main/java/org/apache/druid/java/util/common/parsers/ObjectFlatteners.java

@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
@@ -250,15 +251,23 @@ public class ObjectFlatteners
      */
     default Map<String, Object> toMap(T obj)
     {
-      return (Map<String, Object>) toPlainJavaType(obj);
+      final Object mapOrNull = toPlainJavaType(obj);
+      if (mapOrNull == null) {
+        return Collections.emptyMap();
+      }
+      return (Map<String, Object>) mapOrNull;
     }
 
     /**
      * Recursively traverse "json" object using a {@link JsonProvider}, converting to Java {@link Map} and {@link List},
      * potentially transforming via {@link #finalizeConversionForMap} as we go
      */
+    @Nullable
     default Object toPlainJavaType(Object o)
     {
+      if (o == null) {
+        return null;
+      }
       final JsonProvider jsonProvider = getJsonProvider();
       if (jsonProvider.isMap(o)) {
         Map<String, Object> actualMap = new HashMap<>();
@@ -287,7 +296,7 @@ public class ObjectFlatteners
         return finalizeConversionForMap(actualList);
       }
       // unknown, just pass it through
-      return o;
+      return finalizeConversionForMap(o);
     }
 
     /**

core/src/test/java/org/apache/druid/java/util/common/parsers/ObjectFlattenersTest.java

@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import org.junit.Assert;
 import org.junit.Test;
@@ -32,12 +33,14 @@ import java.util.Map;
 public class ObjectFlattenersTest
 {
   private static final String SOME_JSON = "{\"foo\": null, \"bar\": 1}";
+
+  private static final ObjectFlatteners.FlattenerMaker FLATTENER_MAKER = new JSONFlattenerMaker(true);
   private static final ObjectFlattener FLATTENER = ObjectFlatteners.create(
       new JSONPathSpec(
           true,
           ImmutableList.of(new JSONPathFieldSpec(JSONPathFieldType.PATH, "extract", "$.bar"))
       ),
-      new JSONFlattenerMaker(true)
+      FLATTENER_MAKER
   );
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
@@ -62,4 +65,13 @@ public class ObjectFlattenersTest
     Assert.assertNull(flat.get("foo"));
     Assert.assertEquals(1, flat.get("bar"));
   }
+
+  @Test
+  public void testToMapNull() throws JsonProcessingException
+  {
+    JsonNode node = OBJECT_MAPPER.readTree("null");
+    Map<String, Object> flat = FLATTENER.toMap(node);
+    Assert.assertNull(FLATTENER_MAKER.toPlainJavaType(node));
+    Assert.assertEquals(ImmutableMap.of(), flat);
+  }
 }

extensions-core/orc-extensions/src/test/java/org/apache/druid/data/input/orc/OrcReaderTest.java

@@ -29,6 +29,7 @@ import org.apache.druid.data.input.InputRow;
 import org.apache.druid.data.input.InputRowSchema;
 import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.data.input.impl.FileEntity;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
 import org.apache.druid.data.input.impl.TimestampSpec;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.parsers.CloseableIterator;
@@ -569,6 +570,44 @@ public class OrcReaderTest extends InitializedNullHandlingTest
     }
   }
 
+  @Test
+  public void testSimpleNullValues() throws IOException
+  {
+    final InputFormat inputFormat = new OrcInputFormat(
+        new JSONPathSpec(
+            true,
+            ImmutableList.of()
+        ),
+        null,
+        new Configuration()
+    );
+    final InputEntityReader reader = createReader(
+        new TimestampSpec("timestamp", "auto", null),
+        new DimensionsSpec(
+            ImmutableList.of(
+                new StringDimensionSchema("c1"),
+                new StringDimensionSchema("c2")
+            )
+        ),
+        inputFormat,
+        "example/test_simple.orc"
+    );
+    try (CloseableIterator<InputRow> iterator = reader.read()) {
+      Assert.assertTrue(iterator.hasNext());
+      InputRow row = iterator.next();
+
+      Assert.assertEquals(DateTimes.of("2022-01-01T00:00:00.000Z"), row.getTimestamp());
+      Assert.assertEquals("true", Iterables.getOnlyElement(row.getDimension("c1")));
+      Assert.assertEquals("str1", Iterables.getOnlyElement(row.getDimension("c2")));
+
+      row = iterator.next();
+      Assert.assertEquals(DateTimes.of("2022-01-02T00:00:00.000Z"), row.getTimestamp());
+      Assert.assertEquals(ImmutableList.of(), row.getDimension("c1"));
+      Assert.assertEquals(ImmutableList.of(), row.getDimension("c2"));
+      Assert.assertFalse(iterator.hasNext());
+    }
+  }
+
   private InputEntityReader createReader(
       TimestampSpec timestampSpec,
       DimensionsSpec dimensionsSpec,