Add InputSource and InputFormat interfaces

core/src/main/java/org/apache/druid/data/input/FiniteFirehoseFactory.java

@@ -33,6 +33,7 @@
  * @param <T> parser type
  * @param <S> input split type
  */
+@Deprecated
 public interface FiniteFirehoseFactory<T extends InputRowParser, S> extends FirehoseFactory<T>
 {
   /**

core/src/main/java/org/apache/druid/data/input/FirehoseFactory.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.data.input;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import org.apache.druid.data.input.impl.InputRowParser;
 import org.apache.druid.data.input.impl.prefetch.PrefetchableTextFilesFirehoseFactory;
@@ -85,6 +86,7 @@ default Firehose connectForSampler(T parser, @Nullable File temporaryDirectory)
     return connect(parser, temporaryDirectory);
   }
 
+  @JsonIgnore
   default boolean isSplittable()
   {
     return false;

core/src/main/java/org/apache/druid/data/input/FirehoseFactoryToInputSourceAdaptor.java

@@ -0,0 +1,101 @@
+/*
+ * 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.data.input;
+
+import com.google.common.base.Preconditions;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.FirehoseToInputSourceReaderAdaptor;
+import org.apache.druid.data.input.impl.InputFormat;
+import org.apache.druid.data.input.impl.InputRowParser;
+import org.apache.druid.data.input.impl.SplittableInputSource;
+import org.apache.druid.data.input.impl.TimestampSpec;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.IOException;
+import java.util.stream.Stream;
+
+public class FirehoseFactoryToInputSourceAdaptor implements SplittableInputSource
+{
+  private final FiniteFirehoseFactory firehoseFactory;
+  private final InputRowParser inputRowParser;
+
+  public FirehoseFactoryToInputSourceAdaptor(FiniteFirehoseFactory firehoseFactory, InputRowParser inputRowParser)
+  {
+    this.firehoseFactory = firehoseFactory;
+    this.inputRowParser = Preconditions.checkNotNull(inputRowParser, "inputRowParser");
+  }
+
+  public FiniteFirehoseFactory getFirehoseFactory()
+  {
+    return firehoseFactory;
+  }
+
+  @Override
+  public boolean isSplittable()
+  {
+    return firehoseFactory.isSplittable();
+  }
+
+  @Override
+  public Stream<InputSplit> createSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec)
+      throws IOException
+  {
+    if (firehoseFactory.isSplittable()) {
+      return firehoseFactory.getSplits(splitHintSpec);
+    } else {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  @Override
+  public int getNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException
+  {
+    if (firehoseFactory.isSplittable()) {
+      return firehoseFactory.getNumSplits(splitHintSpec);
+    } else {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  @Override
+  public SplittableInputSource withSplit(InputSplit split)
+  {
+    if (firehoseFactory.isSplittable()) {
+      return new FirehoseFactoryToInputSourceAdaptor(
+          firehoseFactory.withSplit(split),
+          inputRowParser
+      );
+    } else {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  @Override
+  public InputSourceReader reader(
+      TimestampSpec timestampSpec,
+      DimensionsSpec dimensionsSpec,
+      @Nullable InputFormat inputFormat, // inputFormat will be ignored
+      @Nullable File temporaryDirectory
+  )
+  {
+    return new FirehoseToInputSourceReaderAdaptor(firehoseFactory, inputRowParser, temporaryDirectory);
+  }
+}

core/src/main/java/org/apache/druid/data/input/InputSource.java

@@ -0,0 +1,72 @@
+/*
+ * 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.data.input;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.HttpInputSource;
+import org.apache.druid.data.input.impl.InputFormat;
+import org.apache.druid.data.input.impl.LocalInputSource;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.guice.annotations.ExtensionPoint;
+
+import javax.annotation.Nullable;
+import java.io.File;
+
+/**
+ * InputSource abstracts the storage system where input data is stored.
+ * It creates an {@link InputSourceReader} to read data from the given input source.
+ * The most common use case would be:
+ *
+ * <pre>{@code
+ *   InputSourceReader reader = inputSource.reader();
+ *   try (CloseableIterator<InputRow> iterator = reader.read()) {
+ *     while (iterator.hasNext()) {
+ *       InputRow row = iterator.next();
+ *       processRow(row);
+ *     }
+ *   }
+ * }</pre>
+ */
+@ExtensionPoint
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+    @Type(name = "local", value = LocalInputSource.class),
+    @Type(name = "http", value = HttpInputSource.class)
+})
+public interface InputSource
+{
+  /**
+   * Returns true if this inputSource can be processed in parallel using ParallelIndexSupervisorTask.
+   */
+  default boolean isSplittable()
+  {
+    return false;
+  }
+
+  InputSourceReader reader(
+      TimestampSpec timestampSpec,
+      DimensionsSpec dimensionsSpec,
+      InputFormat inputFormat,
+      @Nullable File temporaryDirectory
+  );
+}

core/src/main/java/org/apache/druid/data/input/InputSourceReader.java

@@ -0,0 +1,41 @@
+/*
+ * 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.data.input;
+
+import org.apache.druid.guice.annotations.ExtensionPoint;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+
+import java.io.IOException;
+
+/**
+ * InputSourceReader reads data from {@link InputSource} and returns a {@link CloseableIterator} of
+ * {@link InputRow}. See {@link InputSource} for an example usage.
+ *
+ * Implementations of this class can use {@link SplitSource} and {@link SplitReader}.
+ *
+ * See {@link org.apache.druid.data.input.impl.SplitIteratingReader} as an example.
+ */
+@ExtensionPoint
+public interface InputSourceReader
+{
+  CloseableIterator<InputRow> read() throws IOException;
+
+  CloseableIterator<InputRowPlusRaw> sample() throws IOException;
+}

core/src/main/java/org/apache/druid/data/input/SplitHintSpec.java

@@ -31,6 +31,8 @@
  *
  * @see FiniteFirehoseFactory#getSplits(SplitHintSpec)
  * @see FiniteFirehoseFactory#getNumSplits(SplitHintSpec)
+ * @see org.apache.druid.data.input.impl.SplittableInputSource#createSplits
+ * @see org.apache.druid.data.input.impl.SplittableInputSource#getNumSplits
  */
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes(value = {

core/src/main/java/org/apache/druid/data/input/SplitReader.java

@@ -0,0 +1,40 @@
+/*
+ * 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.data.input;
+
+import org.apache.druid.guice.annotations.ExtensionPoint;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+
+import java.io.File;
+import java.io.IOException;
+
+/**
+ * SplitReader knows how to parse data into {@link InputRow}.
+ * This class is <i>stateful</i> and a new SplitReader should be created per {@link InputSplit}.
+ *
+ * @see TextReader for text format readers
+ */
+@ExtensionPoint
+public interface SplitReader
+{
+  CloseableIterator<InputRow> read(SplitSource source, File temporaryDirectory) throws IOException;
+
+  CloseableIterator<InputRowPlusRaw> sample(SplitSource source, File temporaryDirectory) throws IOException;
+}