[Backport] Fix a bug for CSVParser/DelimitedParser when empty column exists in the header row

api/src/main/java/io/druid/data/input/impl/CSVParseSpec.java

@@ -21,7 +21,6 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import io.druid.java.util.common.parsers.CSVParser;
 import io.druid.java.util.common.parsers.Parser;
@@ -114,7 +113,7 @@ public void verify(List<String> usedCols)
   @Override
   public Parser<String, Object> makeParser()
   {
-    return new CSVParser(Optional.fromNullable(listDelimiter), columns, hasHeaderRow, skipHeaderRows);
+    return new CSVParser(listDelimiter, columns, hasHeaderRow, skipHeaderRows);
   }
 
   @Override

api/src/main/java/io/druid/data/input/impl/DelimitedParseSpec.java

@@ -21,7 +21,6 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import io.druid.java.util.common.parsers.DelimitedParser;
 import io.druid.java.util.common.parsers.Parser;
@@ -125,8 +124,8 @@ public void verify(List<String> usedCols)
   public Parser<String, Object> makeParser()
   {
     return new DelimitedParser(
-        Optional.fromNullable(delimiter),
-        Optional.fromNullable(listDelimiter),
+        delimiter,
+        listDelimiter,
         columns,
         hasHeaderRow,
         skipHeaderRows

extensions-core/lookups-cached-global/src/main/java/io/druid/query/lookup/namespace/UriExtractionNamespace.java

@@ -29,7 +29,6 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
@@ -295,7 +294,7 @@ public CSVFlatDataParser(
       );
 
       this.parser = new DelegateParser(
-          new CSVParser(Optional.absent(), columns, hasHeaderRow, skipHeaderRows),
+          new CSVParser(null, columns, hasHeaderRow, skipHeaderRows),
           this.keyColumn,
           this.valueColumn
       );
@@ -395,8 +394,8 @@ public TSVFlatDataParser(
           "Must specify more than one column to have a key value pair"
       );
       final DelimitedParser delegate = new DelimitedParser(
-          Optional.fromNullable(Strings.emptyToNull(delimiter)),
-          Optional.fromNullable(Strings.emptyToNull(listDelimiter)),
+          Strings.emptyToNull(delimiter),
+          Strings.emptyToNull(listDelimiter),
           hasHeaderRow,
           skipHeaderRows
       );

indexing-service/src/test/java/io/druid/indexing/common/task/IndexTaskTest.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.common.io.Files;
 import io.druid.data.input.impl.CSVParseSpec;
 import io.druid.data.input.impl.DimensionsSpec;
@@ -39,6 +40,7 @@
 import io.druid.indexing.common.actions.TaskActionClient;
 import io.druid.indexing.common.task.IndexTask.IndexIngestionSpec;
 import io.druid.indexing.overlord.SegmentPublishResult;
+import io.druid.java.util.common.StringUtils;
 import io.druid.java.util.common.granularity.Granularities;
 import io.druid.java.util.common.parsers.ParseException;
 import io.druid.query.aggregation.AggregatorFactory;
@@ -74,8 +76,10 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public class IndexTaskTest
 {
@@ -601,6 +605,139 @@ public void testReportParseException() throws Exception
     runTask(indexTask);
   }
 
+  @Test
+  public void testCsvWithHeaderOfEmptyColumns() throws Exception
+  {
+    final File tmpDir = temporaryFolder.newFolder();
+
+    File tmpFile = File.createTempFile("druid", "index", tmpDir);
+
+    try (BufferedWriter writer = Files.newWriter(tmpFile, StandardCharsets.UTF_8)) {
+      writer.write("time,,\n");
+      writer.write("2014-01-01T00:00:10Z,a,1\n");
+    }
+
+    tmpFile = File.createTempFile("druid", "index", tmpDir);
+
+    try (BufferedWriter writer = Files.newWriter(tmpFile, StandardCharsets.UTF_8)) {
+      writer.write("time,dim,\n");
+      writer.write("2014-01-01T00:00:10Z,a,1\n");
+    }
+
+    tmpFile = File.createTempFile("druid", "index", tmpDir);
+
+    try (BufferedWriter writer = Files.newWriter(tmpFile, StandardCharsets.UTF_8)) {
+      writer.write("time,,val\n");
+      writer.write("2014-01-01T00:00:10Z,a,1\n");
+    }
+
+    final IndexIngestionSpec parseExceptionIgnoreSpec = createIngestionSpec(
+        tmpDir,
+        new CSVParseSpec(
+            new TimestampSpec(
+                "time",
+                "auto",
+                null
+            ),
+            new DimensionsSpec(
+                null,
+                null,
+                null
+            ),
+            null,
+            null,
+            true,
+            0
+        ),
+        null,
+        2,
+        null,
+        false,
+        false,
+        true // report parse exception
+    );
+
+    IndexTask indexTask = new IndexTask(
+        null,
+        null,
+        parseExceptionIgnoreSpec,
+        null,
+        jsonMapper
+    );
+
+    final List<DataSegment> segments = runTask(indexTask);
+    // the order of result segments can be changed because hash shardSpec is used.
+    // the below loop is to make this test deterministic.
+    Assert.assertEquals(2, segments.size());
+    Assert.assertNotEquals(segments.get(0), segments.get(1));
+
+    for (int i = 0; i < 2; i++) {
+      final DataSegment segment = segments.get(i);
+      final Set<String> dimensions = new HashSet<>(segment.getDimensions());
+
+      Assert.assertTrue(
+          StringUtils.safeFormat("Actual dimensions: %s", dimensions),
+          dimensions.equals(Sets.newHashSet("dim", "column_3")) ||
+          dimensions.equals(Sets.newHashSet("column_2", "column_3"))
+      );
+
+      Assert.assertEquals(Arrays.asList("val"), segment.getMetrics());
+      Assert.assertEquals(new Interval("2014/P1D"), segment.getInterval());
+    }
+  }
+
+  @Test
+  public void testCsvWithHeaderOfEmptyTimestamp() throws Exception
+  {
+    expectedException.expect(ParseException.class);
+    expectedException.expectMessage("Unparseable timestamp found!");
+
+    final File tmpDir = temporaryFolder.newFolder();
+
+    final File tmpFile = File.createTempFile("druid", "index", tmpDir);
+
+    try (BufferedWriter writer = Files.newWriter(tmpFile, StandardCharsets.UTF_8)) {
+      writer.write(",,\n");
+      writer.write("2014-01-01T00:00:10Z,a,1\n");
+    }
+
+    final IndexIngestionSpec parseExceptionIgnoreSpec = createIngestionSpec(
+        tmpDir,
+        new CSVParseSpec(
+            new TimestampSpec(
+                "time",
+                "auto",
+                null
+            ),
+            new DimensionsSpec(
+                null,
+                Lists.<String>newArrayList(),
+                Lists.<SpatialDimensionSchema>newArrayList()
+            ),
+            null,
+            Arrays.asList("time", "", ""),
+            true,
+            0
+        ),
+        null,
+        2,
+        null,
+        false,
+        false,
+        true // report parse exception
+    );
+
+    IndexTask indexTask = new IndexTask(
+        null,
+        null,
+        parseExceptionIgnoreSpec,
+        null,
+        jsonMapper
+    );
+
+    runTask(indexTask);
+  }
+
   private final List<DataSegment> runTask(final IndexTask indexTask) throws Exception
   {
     final List<DataSegment> segments = Lists.newArrayList();

java-util/src/main/java/io/druid/java/util/common/parsers/AbstractFlatTextFormatParser.java

@@ -0,0 +1,165 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.java.util.common.parsers;
+
+import com.google.common.base.Function;
+import com.google.common.base.Splitter;
+import com.google.common.base.Strings;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import io.druid.java.util.common.collect.Utils;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public abstract class AbstractFlatTextFormatParser implements Parser<String, Object>
+{
+  public enum FlatTextFormat
+  {
+    CSV(","),
+    DELIMITED("\t");
+
+    private final String defaultDelimiter;
+
+    FlatTextFormat(String defaultDelimiter)
+    {
+      this.defaultDelimiter = defaultDelimiter;
+    }
+
+    public String getDefaultDelimiter()
+    {
+      return defaultDelimiter;
+    }
+  }
+
+  private final String listDelimiter;
+  private final Splitter listSplitter;
+  private final Function<String, Object> valueFunction;
+  private final boolean hasHeaderRow;
+  private final int maxSkipHeaderRows;
+
+  private List<String> fieldNames = null;
+  private boolean hasParsedHeader = false;
+  private int skippedHeaderRows;
+  private boolean supportSkipHeaderRows;
+
+  public AbstractFlatTextFormatParser(
+      @Nullable final String listDelimiter,
+      final boolean hasHeaderRow,
+      final int maxSkipHeaderRows
+  )
+  {
+    this.listDelimiter = listDelimiter != null ? listDelimiter : Parsers.DEFAULT_LIST_DELIMITER;
+    this.listSplitter = Splitter.on(this.listDelimiter);
+    this.valueFunction = ParserUtils.getMultiValueFunction(this.listDelimiter, this.listSplitter);
+
+    this.hasHeaderRow = hasHeaderRow;
+    this.maxSkipHeaderRows = maxSkipHeaderRows;
+  }
+
+  @Override
+  public void startFileFromBeginning()
+  {
+    if (hasHeaderRow) {
+      fieldNames = null;
+    }
+    hasParsedHeader = false;
+    skippedHeaderRows = 0;
+    supportSkipHeaderRows = true;
+  }
+
+  public String getListDelimiter()
+  {
+    return listDelimiter;
+  }
+
+  @Override
+  public List<String> getFieldNames()
+  {
+    return fieldNames;
+  }
+
+  @Override
+  public void setFieldNames(final Iterable<String> fieldNames)
+  {
+    if (fieldNames != null) {
+      final List<String> fieldsList = Lists.newArrayList(fieldNames);
+      this.fieldNames = new ArrayList<>(fieldsList.size());
+      for (int i = 0; i < fieldsList.size(); i++) {
+        if (Strings.isNullOrEmpty(fieldsList.get(i))) {
+          this.fieldNames.add(ParserUtils.getDefaultColumnName(i));
+        } else {
+          this.fieldNames.add(fieldsList.get(i));
+        }
+      }
+      ParserUtils.validateFields(this.fieldNames);
+    }
+  }
+
+  public void setFieldNames(final String header)
+  {
+    try {
+      setFieldNames(parseLine(header));
+    }
+    catch (Exception e) {
+      throw new ParseException(e, "Unable to parse header [%s]", header);
+    }
+  }
+
+  @Override
+  public Map<String, Object> parse(final String input)
+  {
+    if (!supportSkipHeaderRows && (hasHeaderRow || maxSkipHeaderRows > 0)) {
+      throw new UnsupportedOperationException("hasHeaderRow or maxSkipHeaderRows is not supported. "
+                                              + "Please check the indexTask supports these options.");
+    }
+
+    try {
+      List<String> values = parseLine(input);
+
+      if (skippedHeaderRows < maxSkipHeaderRows) {
+        skippedHeaderRows++;
+        return null;
+      }
+
+      if (hasHeaderRow && !hasParsedHeader) {
+        if (fieldNames == null) {
+          setFieldNames(values);
+        }
+        hasParsedHeader = true;
+        return null;
+      }
+
+      if (fieldNames == null) {
+        setFieldNames(ParserUtils.generateFieldNames(values.size()));
+      }
+
+      return Utils.zipMapPartial(fieldNames, Iterables.transform(values, valueFunction));
+    }
+    catch (Exception e) {
+      throw new ParseException(e, "Unable to parse row [%s]", input);
+    }
+  }
+
+  protected abstract List<String> parseLine(String input) throws IOException;
+}