add explodeSpec

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

@@ -57,6 +57,7 @@ public Parser getFlatParser()
         new TimestampSpec("ts", "iso", null),
         new DimensionsSpec(null, null, null),
         null,
+        null,
         null
     );
     return spec.makeParser();
@@ -71,6 +72,7 @@ public Parser getFieldDiscoveryParser()
         new TimestampSpec("ts", "iso", null),
         new DimensionsSpec(null, null, null),
         flattenSpec,
+        null,
         null
     );
 
@@ -114,6 +116,7 @@ public Parser getNestedParser()
         new TimestampSpec("ts", "iso", null),
         new DimensionsSpec(null, null, null),
         flattenSpec,
+        null,
         null
     );
 
@@ -157,6 +160,7 @@ public Parser getForcedPathParser()
         new TimestampSpec("ts", "iso", null),
         new DimensionsSpec(null, null, null),
         flattenSpec,
+        null,
         null
     );
 
@@ -198,6 +202,7 @@ public Parser getJqParser()
         new TimestampSpec("ts", "iso", null),
         new DimensionsSpec(null, null, null),
         flattenSpec,
+        null,
         null
     );
 

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

@@ -29,14 +29,16 @@
 import javax.annotation.Nullable;
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
 
 public class FileIteratingFirehose implements Firehose
 {
   private final Iterator<LineIterator> lineIterators;
   private final StringInputRowParser parser;
-
+  private Iterator<InputRow> parsedInputRows = new ArrayList<InputRow>().iterator();
+  private String raw = null;
   private LineIterator lineIterator = null;
 
   private final Closeable closer;
@@ -63,11 +65,12 @@ public FileIteratingFirehose(
   @Override
   public boolean hasMore() throws IOException
   {
+
     while ((lineIterator == null || !lineIterator.hasNext()) && lineIterators.hasNext()) {
       lineIterator = getNextLineIterator();
     }
 
-    return lineIterator != null && lineIterator.hasNext();
+    return (lineIterator != null && lineIterator.hasNext()) || parsedInputRows.hasNext();
   }
 
   @Nullable
@@ -77,8 +80,17 @@ public InputRow nextRow() throws IOException
     if (!hasMore()) {
       throw new NoSuchElementException();
     }
+    if (parsedInputRows.hasNext()) {
+      return parsedInputRows.next();
+    }
+    return getNextRow();
+  }
 
-    return parser.parse(lineIterator.next());
+  private InputRow getNextRow()
+  {
+    raw = lineIterator.next();
+    parsedInputRows = parser.parseBatch(raw).iterator();
+    return parsedInputRows.hasNext() ? parsedInputRows.next() : null;
   }
 
   @Override
@@ -87,10 +99,8 @@ public InputRowPlusRaw nextRowWithRaw() throws IOException
     if (!hasMore()) {
       throw new NoSuchElementException();
     }
-
-    String raw = lineIterator.next();
     try {
-      return InputRowPlusRaw.of(parser.parse(raw), StringUtils.toUtf8(raw));
+      return InputRowPlusRaw.of(nextRow(), StringUtils.toUtf8(raw));
     }
     catch (ParseException e) {
       return InputRowPlusRaw.of(StringUtils.toUtf8(raw), e);

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

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonParser.Feature;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.java.util.common.parsers.JSONExplodeSpec;
 import org.apache.druid.java.util.common.parsers.JSONPathParser;
 import org.apache.druid.java.util.common.parsers.JSONPathSpec;
 import org.apache.druid.java.util.common.parsers.Parser;
@@ -33,17 +34,20 @@
 import java.util.Objects;
 
 /**
+ *
  */
 public class JSONParseSpec extends NestedDataParseSpec<JSONPathSpec>
 {
   private final ObjectMapper objectMapper;
   private final Map<String, Boolean> featureSpec;
+  private final List<JSONExplodeSpec> explodeSpec;
 
   @JsonCreator
   public JSONParseSpec(
       @JsonProperty("timestampSpec") TimestampSpec timestampSpec,
       @JsonProperty("dimensionsSpec") DimensionsSpec dimensionsSpec,
       @JsonProperty("flattenSpec") JSONPathSpec flattenSpec,
+      @JsonProperty("explodeSpec") List<JSONExplodeSpec> explodeSpec,
       @JsonProperty("featureSpec") Map<String, Boolean> featureSpec
   )
   {
@@ -54,12 +58,19 @@ public JSONParseSpec(
       Feature feature = Feature.valueOf(entry.getKey());
       objectMapper.configure(feature, entry.getValue());
     }
+    this.explodeSpec = explodeSpec;
+  }
+
+  @JsonProperty
+  public List<JSONExplodeSpec> getExplodeSpec()
+  {
+    return explodeSpec;
   }
 
   @Deprecated
   public JSONParseSpec(TimestampSpec ts, DimensionsSpec dims)
   {
-    this(ts, dims, null, null);
+    this(ts, dims, null, null, null);
   }
 
   @Override
@@ -70,19 +81,19 @@ public void verify(List<String> usedCols)
   @Override
   public Parser<String, Object> makeParser()
   {
-    return new JSONPathParser(getFlattenSpec(), objectMapper);
+    return new JSONPathParser(getFlattenSpec(), getExplodeSpec(), objectMapper);
   }
 
   @Override
   public ParseSpec withTimestampSpec(TimestampSpec spec)
   {
-    return new JSONParseSpec(spec, getDimensionsSpec(), getFlattenSpec(), getFeatureSpec());
+    return new JSONParseSpec(spec, getDimensionsSpec(), getFlattenSpec(), getExplodeSpec(), getFeatureSpec());
   }
 
   @Override
   public ParseSpec withDimensionsSpec(DimensionsSpec spec)
   {
-    return new JSONParseSpec(getTimestampSpec(), spec, getFlattenSpec(), getFeatureSpec());
+    return new JSONParseSpec(getTimestampSpec(), spec, getFlattenSpec(), getExplodeSpec(), getFeatureSpec());
   }
 
   @JsonProperty
@@ -120,6 +131,7 @@ public String toString()
            "timestampSpec=" + getTimestampSpec() +
            ", dimensionsSpec=" + getDimensionsSpec() +
            ", flattenSpec=" + getFlattenSpec() +
+           ", explodeSpec=" + getExplodeSpec() +
            ", featureSpec=" + featureSpec +
            '}';
   }

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

@@ -28,6 +28,7 @@
 import org.apache.druid.java.util.common.parsers.Parser;
 
 import java.util.List;
+import java.util.Objects;
 
 @ExtensionPoint
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "format")
@@ -98,12 +99,10 @@ public boolean equals(Object o)
 
     ParseSpec parseSpec = (ParseSpec) o;
 
-    if (timestampSpec != null ? !timestampSpec.equals(parseSpec.timestampSpec) : parseSpec.timestampSpec != null) {
+    if (!Objects.equals(timestampSpec, parseSpec.timestampSpec)) {
       return false;
     }
-    return !(dimensionsSpec != null
-             ? !dimensionsSpec.equals(parseSpec.dimensionsSpec)
-             : parseSpec.dimensionsSpec != null);
+    return !(!Objects.equals(dimensionsSpec, parseSpec.dimensionsSpec));
 
   }
 

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

@@ -25,7 +25,6 @@
 import com.google.common.collect.Iterators;
 import org.apache.druid.data.input.ByteBufferInputRowParser;
 import org.apache.druid.data.input.InputRow;
-import org.apache.druid.java.util.common.collect.Utils;
 import org.apache.druid.java.util.common.parsers.ParseException;
 import org.apache.druid.java.util.common.parsers.Parser;
 
@@ -36,10 +35,12 @@
 import java.nio.charset.CoderResult;
 import java.nio.charset.CodingErrorAction;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
 /**
+ *
  */
 public class StringInputRowParser implements ByteBufferInputRowParser
 {
@@ -60,7 +61,6 @@ public StringInputRowParser(
   {
     this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec");
     this.mapParser = new MapInputRowParser(parseSpec);
-
     if (encoding != null) {
       this.charset = Charset.forName(encoding);
     } else {
@@ -77,7 +77,9 @@ public StringInputRowParser(ParseSpec parseSpec)
   @Override
   public List<InputRow> parseBatch(ByteBuffer input)
   {
-    return Utils.nullableListOf(parseMap(buildStringKeyMap(input)));
+    final List<InputRow> theList = new ArrayList<>();
+    buildStringKeyMapList(input).forEach(e -> theList.add(parseMap(e)));
+    return theList;
   }
 
   @JsonProperty
@@ -127,6 +129,34 @@ private Map<String, Object> buildStringKeyMap(ByteBuffer input)
     return theMap;
   }
 
+  private List<Map<String, Object>> buildStringKeyMapList(ByteBuffer input)
+  {
+    int payloadSize = input.remaining();
+    if (chars == null || chars.remaining() < payloadSize) {
+      chars = CharBuffer.allocate(payloadSize);
+    }
+
+    final CoderResult coderResult = charset.newDecoder()
+                                           .onMalformedInput(CodingErrorAction.REPLACE)
+                                           .onUnmappableCharacter(CodingErrorAction.REPLACE)
+                                           .decode(input, chars, true);
+
+    List<Map<String, Object>> theMapList;
+    if (coderResult.isUnderflow()) {
+      chars.flip();
+      try {
+        initializeParser();
+        theMapList = parser.parseToMapList(chars.toString());
+      }
+      finally {
+        chars.clear();
+      }
+    } else {
+      throw new ParseException("Failed with CoderResult[%s]", coderResult);
+    }
+    return theMapList;
+  }
+
   public void initializeParser()
   {
     if (parser == null) {
@@ -142,6 +172,15 @@ public void startFileFromBeginning()
     parser.startFileFromBeginning();
   }
 
+
+  public List<InputRow> parseBatch(@Nullable String input)
+  {
+    initializeParser();
+    List<InputRow> returnList = new ArrayList<>();
+    parser.parseToMapList(input).forEach(e -> returnList.add(parseMap(e)));
+    return returnList;
+  }
+
   @Nullable
   public InputRow parse(@Nullable String input)
   {

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

@@ -126,6 +126,12 @@ public void setFieldNames(final String header)
     }
   }
 
+  @Override
+  public List<Map<String, Object>> parseToMapList(final String input)
+  {
+    return Utils.nullableListOf(parseToMap(input));
+  }
+
   @Override
   public Map<String, Object> parseToMap(final String input)
   {

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

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.java.util.common.parsers;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+public class JSONExplodeSpec
+{
+  public static final JSONExplodeSpec DEFAULT =
+      new JSONExplodeSpec(null, null);
+
+  private final String path;
+  private final String type;
+
+  @JsonCreator
+  public JSONExplodeSpec(
+      @JsonProperty("path") String path,
+      @JsonProperty("type") String type
+  )
+  {
+    this.path = path;
+    this.type = type;
+  }
+
+  @JsonProperty
+  public String getPath()
+  {
+    return path;
+  }
+
+  @JsonProperty
+  public String getType()
+  {
+    return type;
+  }
+
+
+  @Override
+  public boolean equals(final Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final JSONExplodeSpec that = (JSONExplodeSpec) o;
+    return path.equals(that.path) &&
+           type.equals(that.type);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(path, type);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "JSONExplodeSpec{" +
+           "path=" + path +
+           ", type=" + type +
+           '}';
+  }
+}
+